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

cli/src/codex/runCodex.ts

@@ -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 = {
@@ -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);

cli/src/modules/pingPeer/pingPeer.test.ts

@@ -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}`)
@@ -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}`)

cli/src/modules/pingPeer/pingPeer.ts

@@ -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'
@@ -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,

cli/src/utils/MessageQueue2.test.ts

@@ -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 }
+        ]);
+    });
 });

cli/src/utils/MessageQueue2.ts

@@ -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;
 }
@@ -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>>();
@@ -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
@@ -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,
@@ -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
@@ -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,
@@ -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) {
@@ -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,
@@ -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
@@ -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,
@@ -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
@@ -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,
@@ -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) {

hub/src/sync/messageService.test.ts

@@ -1249,6 +1249,59 @@ describe('MessageService.sendMessage deliveryMode', () => {
         })
     })
 
+    it('persists steer for arrival-steerable flavors and downgrades it for cursor and 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 cursorSession = store.sessions.getOrCreateSession(
+            'delivery-mode-cursor',
+            { path: '/tmp/delivery-mode-cursor', host: 'localhost', flavor: 'cursor' },
+            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'
+        })
+        await service.sendMessage(cursorSession.id, {
+            text: 'cursor steers only via the manual button',
+            localId: 'cursor-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' }
+        })
+        expect(store.messages.getUninvokedLocalMessages(cursorSession.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(
@@ -1359,7 +1412,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)

hub/src/sync/messageService.ts

@@ -124,7 +124,14 @@ function getNormalizedDeliveryMode(
         return 'queue'
     }
 
-    return isObject(metadata) && metadata.flavor === 'pi' ? 'steer' : 'queue'
+    // Persist steer provenance only for flavors whose CLI consumes the hint on
+    // arrival (codex turn/steer, pi native steer). Cursor ACP steers only via
+    // the manual button today — its inbound path drops the hint, so keep its
+    // rows queued rather than advertising an inject that never happens.
+    return isObject(metadata)
+        && ((metadata.flavor === 'pi' || metadata.flavor === 'codex'))
+        ? 'steer'
+        : 'queue'
 }
 
 /**

shared/src/sessionCitation.ts

@@ -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 codex and pi 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.'