Skip to content

feat(cli): auto-steer ping_peer messages into an active turn #1708

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KevinFun wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat(cli): auto-steer ping_peer messages into an active turn #1708

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b9f084b
feat(cli): auto-steer ping_peer messages into an active turn
KevinFun Aug 28, 2026
73630bb
fix(hub): persist steer provenance for arrival-wired flavors only
KevinFun Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
278 changes: 151 additions & 127 deletions cli/src/codex/codexRemoteLauncher.ts
View file
Open in desktop

Large diffs are not rendered by default.

12 changes: 10 additions & 2 deletions cli/src/codex/runCodex.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,10 @@ export async function runCodex(opts: {
messageQueue.pushIsolateAndClear(isolatedCommandText, enhancedMode, localId);
return;
}
messageQueue.push(text, enhancedMode, localId);
// Peer nudges arrive tagged deliveryMode 'steer' (ping_peer):
// the remote launcher's arrival hook injects them into the
// active turn when one is in flight.
messageQueue.push(text, enhancedMode, localId, message.meta?.deliveryMode === 'steer');
} catch (error) {
logger.debug('[Codex] Failed to handle user message', error);
const enhancedMode: EnhancedMode = {
Expand All @@ -295,7 +298,12 @@ export async function runCodex(opts: {
serviceTier: currentServiceTier,
personality: currentPersonality
};
messageQueue.push(formatMessageWithAttachments(message.content.text, message.content.attachments), enhancedMode, localId);
messageQueue.push(
formatMessageWithAttachments(message.content.text, message.content.attachments),
enhancedMode,
localId,
message.meta?.deliveryMode === 'steer'
);
}
}).catch((error) => {
logger.debug('[Codex] User message handler chain failed', error);
Expand Down
9 changes: 7 additions & 2 deletions cli/src/modules/pingPeer/pingPeer.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,8 @@ describe('pingPeer', () => {
return { status: 200, data: { token: 'jwt' } }
}
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
expect(body).toEqual({ text: 'hello peer' })
expect(body).toMatchObject({ text: 'hello peer', deliveryMode: 'steer' })
expect(typeof (body as { localId?: unknown }).localId).toBe('string')
return { status: 200, data: { ok: true } }
}
throw new Error(`unexpected POST ${url}`)
Expand Down Expand Up @@ -632,7 +633,11 @@ describe('listSessions query params', () => {
return { status: 200, data: { token: 'jwt' } }
}
if (url.endsWith(`/api/sessions/${sessionId}/messages`)) {
expect(body).toMatchObject({ text: 'hi' })
// Steer-tagged + queue-shaped: lets the peer CLI inject the
// message into an active turn and the UI render the waiting row.
expect(body).toMatchObject({ text: 'hi', deliveryMode: 'steer' })
expect(typeof (body as { localId?: unknown }).localId).toBe('string')
expect((body as { localId?: string }).localId?.startsWith('ping-peer-')).toBe(true)
return { status: 200, data: { ok: true } }
}
throw new Error(`unexpected POST ${url}`)
Expand Down
12 changes: 11 additions & 1 deletion cli/src/modules/pingPeer/pingPeer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
*/

import axios, { type AxiosInstance } from 'axios'
import { randomUUID } from 'node:crypto'
import { extractAssistantPlainText, isObject } from '@hapi/protocol'
import { normalizeSessionIdPrefix } from '@hapi/protocol/sessionCitation'
import { configuration } from '@/configuration'
Expand Down Expand Up @@ -349,7 +350,16 @@ async function sendMessage(
): Promise<void> {
const response = await http.post(
`${apiUrl}/api/sessions/${encodeURIComponent(sessionId)}/messages`,
{ text: message },
{
text: message,
// A localId makes the row queue-shaped (invokedAt null) so the
// receiving UI can render it in the waiting bar, and deliveryMode
// 'steer' asks the peer CLI to inject it into an active turn
// instead of waiting for turn end (flavors that cannot steer
// store an ordinary queue row; hub + CLI both downgrade safely).
localId: `ping-peer-${randomUUID()}`,
deliveryMode: 'steer'
},
{
headers: authHeaders(jwt),
timeout: 30_000,
Expand Down
19 changes: 19 additions & 0 deletions cli/src/utils/MessageQueue2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -692,4 +692,23 @@ describe('MessageQueue2', () => {
expect(batch3?.message).toBe('after-isolated');
expect(batch3?.mode.type).toBe('B');
});

it('should pass the queued item with steerHint to the onMessage handler', () => {
const queue = new MessageQueue2<string>(mode => mode);
const seen: Array<{ message: string; steerHint?: boolean }> = [];

queue.setOnMessage((_message, _mode, item) => {
seen.push({ message: item.message, steerHint: item.steerHint });
});

queue.push('plain', 'local', 'lid-1');
queue.push('nudge', 'local', 'lid-2', true);
queue.pushIsolateAndClear('command', 'local', 'lid-3');

expect(seen).toEqual([
{ message: 'plain', steerHint: undefined },
{ message: 'nudge', steerHint: true },
{ message: 'command', steerHint: undefined }
]);
});
});
33 changes: 18 additions & 15 deletions cli/src/utils/MessageQueue2.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ export interface QueueItem<T> {
modeHash: string;
localId?: string;
isolate?: boolean; // If true, this message must be processed alone
/** Request mid-turn steer into the active turn instead of waiting for turn end. */
steerHint?: boolean;
/** Stable FIFO key used when an async reservation is restored later. */
enqueueOrder?: number;
}
Expand All @@ -29,7 +31,7 @@ export class MessageQueue2<T> {
public queue: QueueItem<T>[] = []; // Made public for testing
private waiter: ((hasMessages: boolean) => void) | null = null;
private closed = false;
private onMessageHandler: ((message: string, mode: T) => void) | null = null;
private onMessageHandler: ((message: string, mode: T, item: QueueItem<T>) => void) | null = null;
onBatchConsumed: ((localIds: string[]) => void) | null = null;
modeHasher: (mode: T) => string;
private readonly reservations = new Map<string, QueueReservation<T>>();
Expand All @@ -49,35 +51,36 @@ export class MessageQueue2<T> {
/**
* Set a handler that will be called when a message arrives
*/
setOnMessage(handler: ((message: string, mode: T) => void) | null): void {
setOnMessage(handler: ((message: string, mode: T, item: QueueItem<T>) => void) | null): void {
this.onMessageHandler = handler;
}

/**
* Push a message to the queue with a mode.
*/
push(message: string, mode: T, localId?: string): void {
push(message: string, mode: T, localId?: string, steerHint?: boolean): void {
if (this.closed) {
throw new Error('Cannot push to closed queue');
}

const modeHash = this.modeHasher(mode);
logger.debug(`[MessageQueue2] push() called with mode hash: ${modeHash}`);

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
localId,
isolate: false,
...(steerHint === true ? { steerHint: true } : {}),
enqueueOrder: this.nextEnqueueOrder++
};
Object.defineProperty(item, 'enqueueOrder', { value: item.enqueueOrder, enumerable: false, writable: true });
this.queue.push(item);

// Trigger message handler if set
if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

// Notify waiter if any
Expand All @@ -103,7 +106,7 @@ export class MessageQueue2<T> {
const modeHash = this.modeHasher(mode);
logger.debug(`[MessageQueue2] pushImmediate() called with mode hash: ${modeHash}`);

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
Expand All @@ -116,7 +119,7 @@ export class MessageQueue2<T> {

// Trigger message handler if set
if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

// Notify waiter if any
Expand Down Expand Up @@ -145,7 +148,7 @@ export class MessageQueue2<T> {
const modeHash = this.modeHasher(mode);
logger.debug(`[MessageQueue2] pushIsolated() called with mode hash: ${modeHash} - preserving ${this.queue.length} pending messages`);

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
Expand All @@ -157,7 +160,7 @@ export class MessageQueue2<T> {
this.queue.push(item);

if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

if (this.waiter) {
Expand Down Expand Up @@ -189,7 +192,7 @@ export class MessageQueue2<T> {
// rejection must not restore a prompt the clear command discarded.
this.cancelReservations();

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
Expand All @@ -202,7 +205,7 @@ export class MessageQueue2<T> {

// Trigger message handler if set
if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

// Notify waiter if any
Expand All @@ -227,7 +230,7 @@ export class MessageQueue2<T> {
const modeHash = this.modeHasher(mode);
logger.debug(`[MessageQueue2] unshift() called with mode hash: ${modeHash}`);

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
Expand All @@ -240,7 +243,7 @@ export class MessageQueue2<T> {

// Trigger message handler if set
if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

// Notify waiter if any
Expand Down Expand Up @@ -269,7 +272,7 @@ export class MessageQueue2<T> {
const modeHash = this.modeHasher(mode);
logger.debug(`[MessageQueue2] unshiftIsolated() called with mode hash: ${modeHash}`);

const item = {
const item: QueueItem<T> = {
message,
mode,
modeHash,
Expand All @@ -281,7 +284,7 @@ export class MessageQueue2<T> {
this.queue.unshift(item);

if (this.onMessageHandler) {
this.onMessageHandler(message, mode);
this.onMessageHandler(message, mode, item);
}

if (this.waiter) {
Expand Down
40 changes: 39 additions & 1 deletion hub/src/sync/messageService.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1249,6 +1249,44 @@ describe('MessageService.sendMessage deliveryMode', () => {
})
})

it('persists steer for steering-capable flavors and downgrades it for claude', async () => {
const store = makeStore()
const codexSession = store.sessions.getOrCreateSession(
'delivery-mode-codex',
{ path: '/tmp/delivery-mode-codex', host: 'localhost', flavor: 'codex' },
null,
'default'
)
const claudeSession = store.sessions.getOrCreateSession(
'delivery-mode-claude',
{ path: '/tmp/delivery-mode-claude', host: 'localhost', flavor: 'claude' },
null,
'default'
)
const { io } = makeTrackingIo()
const service = new MessageService(store, io, makePublisher() as any)

await service.sendMessage(codexSession.id, {
text: 'mid-turn nudge',
localId: 'codex-steer',
deliveryMode: 'steer'
})
await service.sendMessage(claudeSession.id, {
text: 'claude cannot steer',
localId: 'claude-steer',
deliveryMode: 'steer'
})

expect(store.messages.getUninvokedLocalMessages(codexSession.id)[0]?.content).toMatchObject({
role: 'user',
meta: { sentFrom: 'webapp', deliveryMode: 'steer' }
})
expect(store.messages.getUninvokedLocalMessages(claudeSession.id)[0]?.content).toMatchObject({
role: 'user',
meta: { sentFrom: 'webapp', deliveryMode: 'queue' }
})
})

it('delivers a duplicate-localId retry as queue even when the stored row retains steer', async () => {
const store = makeStore()
const session = store.sessions.getOrCreateSession(
Expand Down Expand Up @@ -1359,7 +1397,7 @@ describe('MessageService.sendMessage deliveryMode', () => {
])
})

it('downgrades forged steer intent for non-Pi sessions and defaults omitted intent to queue', async () => {
it('downgrades forged steer intent for non-steerable sessions and defaults omitted intent to queue', async () => {
const store = makeStore()
const session = makeSession(store, 'delivery-mode-non-pi')
const service = new MessageService(store, makeTrackingIo().io, makePublisher() as any)
Expand Down
9 changes: 7 additions & 2 deletions hub/src/sync/messageService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import {
isRedundantGoalStatusEventContent,
unwrapRoleWrappedRecordEnvelope
} from '@hapi/protocol/messages'
import { isObject } from '@hapi/protocol'
import { isObject, isSteeringSupportedForSession } from '@hapi/protocol'
import type { MessageDeliveryMode, MessagesResponse, QueuedStateResponse } from '@hapi/protocol/apiTypes'
import type { Server } from 'socket.io'
import { randomUUID } from 'node:crypto'
Expand Down Expand Up @@ -124,7 +124,12 @@ function getNormalizedDeliveryMode(
return 'queue'
}

return isObject(metadata) && metadata.flavor === 'pi' ? 'steer' : 'queue'
// Steer provenance is only meaningful for flavors whose CLI can act on it
// (codex turn/steer, pi native steer, cursor ACP soft-send). Everything
// else (claude, unknown flavors) stores an ordinary queue row.
return isObject(metadata) && isSteeringSupportedForSession(metadata as Parameters<typeof isSteeringSupportedForSession>[0])

@github-actions github-actions Bot Aug 28, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[MAJOR] Cursor ACP is included here, but its inbound path never auto-steers this intent

isSteeringSupportedForSession() returns true for Cursor ACP, so this preserves deliveryMode: "steer". However, cli/src/cursor/runCursor.ts:100-108 ignores message.meta.deliveryMode, enqueueCursorUserMessage() calls messageQueue.push(...) without a steer hint, and the Cursor launcher only steers when the explicit SteerQueuedMessage RPC is invoked. A ping_peer message sent during an active Cursor turn therefore waits for turn completion despite the new tool description promising immediate injection.

Suggested fix (until Cursor gets equivalent arrival-hook plumbing):

const flavor = isObject(metadata) ? metadata.flavor : null
return flavor === "pi" || flavor === "codex" ? "steer" : "queue"

Also remove Cursor from PING_PEER_TOOL_DESCRIPTION; alternatively, pass the delivery mode through Cursor’s queue and reuse its existing soft-steer handler from an arrival hook.

? 'steer'
: 'queue'
}

/**
Expand Down
1 change: 1 addition & 0 deletions shared/src/sessionCitation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ export const INSPECT_PEER_TOOL_DESCRIPTION =
/** MCP `ping_peer` tool description (same citation forms as inspect_peer). */
export const PING_PEER_TOOL_DESCRIPTION =
'Send a message to another HAPI session (peer handoff / nudge). Resolves by session id prefix, resumes if inactive, then POSTs on the same hub/namespace. ' +
'Delivery is steer-tagged: on steerable flavors (codex / pi / cursor) the message is injected into the peer\'s running turn when one is active, otherwise it becomes the peer\'s next prompt. ' +
'When the user cites a peer via [title](/sessions/<id>), Copy-reference prose See session "…" (/sessions/<id>) for context, or a bare /sessions/<id>, ' +
'extract <id> and pass it as sessionIdPrefix. /sessions/<id> is a hub path - do NOT search the local filesystem for it. ' +
'Prefer this (or `hapi ping-peer`) over reinventing JWT+curl. Targets another session - not the current chat.'
Expand Down
Loading