Add Codex steer for queued messages

cli/src/codex/appServerTypes.ts

@@ -176,6 +176,17 @@ export interface TurnStartResponse {
     [key: string]: unknown;
 }
 
+export interface TurnSteerParams {
+    threadId: string;
+    input: UserInput[];
+    expectedTurnId: string;
+}
+
+export interface TurnSteerResponse {
+    turnId: string;
+    [key: string]: unknown;
+}
+
 export interface TurnInterruptParams {
     threadId: string;
     turnId: string;

cli/src/codex/codexAppServerClient.ts

@@ -13,6 +13,8 @@ import type {
     ThreadResumeResponse,
     TurnStartParams,
     TurnStartResponse,
+    TurnSteerParams,
+    TurnSteerResponse,
     TurnInterruptParams,
     TurnInterruptResponse,
     ThreadCompactStartParams,
@@ -200,6 +202,13 @@ export class CodexAppServerClient {
         return response as TurnStartResponse;
     }
 
+    async steerTurn(params: TurnSteerParams): Promise<TurnSteerResponse> {
+        const response = await this.sendRequest('turn/steer', params, {
+            timeoutMs: 30_000
+        });
+        return response as TurnSteerResponse;
+    }
+
     async interruptTurn(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
         const response = await this.sendRequest('turn/interrupt', params, {
             timeoutMs: 30_000

cli/src/codex/codexRemoteLauncher.ts

@@ -2549,6 +2549,39 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
             onSwitch: () => this.handleSwitchRequest()
         });
 
+        session.client.rpcHandlerManager.registerHandler('steer-queued-message', async (payload: unknown) => {
+            const record = asRecord(payload);
+            const localId = asString(record?.localId);
+            if (!localId) {
+                return { status: 'failed', error: 'Invalid localId' };
+            }
+
+            if (!turnInFlight || !this.currentThreadId || !this.currentTurnId) {
+                return { status: 'not-active' };
+            }
+
+            const taken = session.queue.takeByLocalIdWithIndex(localId);
+            if (!taken) {
+                return { status: 'not-found' };
+            }
+
+            try {
+                await appServerClient.steerTurn({
+                    threadId: this.currentThreadId,
+                    expectedTurnId: this.currentTurnId,
+                    input: [{ type: 'text', text: taken.item.message }]
+                });
+                session.client.emitMessagesConsumed([localId]);
+                return { status: 'steered', localId };
+            } catch (error) {
+                session.queue.restoreAt(taken.index, taken.item);
+                return {
+                    status: 'failed',
+                    error: error instanceof Error ? error.message : String(error)
+                };
+            }
+        });
+
         function logActiveHandles(tag: string) {
             if (!process.env.DEBUG) return;
             const anyProc: any = process as any;

cli/src/utils/MessageQueue2.test.ts

@@ -573,6 +573,45 @@ describe('MessageQueue2', () => {
         });
     });
 
+    describe('takeByLocalId', () => {
+        it('should remove and return the message with matching localId', () => {
+            const queue = new MessageQueue2<string>(mode => mode);
+            queue.push('msg1', 'local', 'id-abc');
+            queue.push('msg2', 'remote', 'id-def');
+
+            const taken = queue.takeByLocalId('id-def');
+            expect(taken?.message).toBe('msg2');
+            expect(taken?.mode).toBe('remote');
+            expect(taken?.localId).toBe('id-def');
+            expect(queue.size()).toBe(1);
+            expect(queue.queue[0].localId).toBe('id-abc');
+        });
+
+        it('should return null when localId is not found', () => {
+            const queue = new MessageQueue2<string>(mode => mode);
+            queue.push('msg1', 'local', 'id-abc');
+
+            const taken = queue.takeByLocalId('missing');
+            expect(taken).toBeNull();
+            expect(queue.size()).toBe(1);
+        });
+
+        it('should restore a taken item at its original position', () => {
+            const queue = new MessageQueue2<string>(mode => mode);
+            queue.push('msg1', 'local', 'id-abc');
+            queue.push('msg2', 'local', 'id-def');
+            queue.push('msg3', 'local', 'id-ghi');
+
+            const taken = queue.takeByLocalIdWithIndex('id-def');
+            expect(taken?.item.message).toBe('msg2');
+            expect(taken?.index).toBe(1);
+
+            queue.restoreAt(taken!.index, taken!.item);
+
+            expect(queue.queue.map(item => item.localId)).toEqual(['id-abc', 'id-def', 'id-ghi']);
+        });
+    });
+
     it('should differentiate between pushImmediate and pushIsolateAndClear behavior', async () => {
         const queue = new MessageQueue2<{ type: string }>((mode) => mode.type);
 
@@ -603,4 +642,18 @@ describe('MessageQueue2', () => {
         expect(batch3?.message).toBe('after-isolated');
         expect(batch3?.mode.type).toBe('B');
     });
-});
+
+    it('should preserve isolation when unshifting a message with isolate=true', async () => {
+        const queue = new MessageQueue2<string>(mode => mode);
+
+        queue.unshift('isolated', 'local', 'id-isolated', true);
+        queue.push('after', 'local', 'id-after');
+
+        const batch1 = await queue.waitForMessagesAndGetAsString();
+        expect(batch1?.message).toBe('isolated');
+        expect(batch1?.isolate).toBe(true);
+
+        const batch2 = await queue.waitForMessagesAndGetAsString();
+        expect(batch2?.message).toBe('after');
+    });
+});

cli/src/utils/MessageQueue2.ts

@@ -187,7 +187,7 @@ export class MessageQueue2<T> {
     /**
      * Push a message to the beginning of the queue with a mode.
      */
-    unshift(message: string, mode: T, localId?: string): void {
+    unshift(message: string, mode: T, localId?: string, isolate = false): void {
         if (this.closed) {
             throw new Error('Cannot unshift to closed queue');
         }
@@ -200,7 +200,7 @@ export class MessageQueue2<T> {
             mode,
             modeHash,
             localId,
-            isolate: false
+            isolate
         });
 
         // Trigger message handler if set
@@ -233,6 +233,28 @@ export class MessageQueue2<T> {
         return true;
     }
 
+    /**
+     * Remove and return the first queued message that matches the given localId.
+     * Used when a caller needs to consume a queued item through a path other
+     * than the normal batch drain, such as steering it into an active turn.
+     */
+    takeByLocalId(localId: string): QueueItem<T> | null {
+        return this.takeByLocalIdWithIndex(localId)?.item ?? null;
+    }
+
+    takeByLocalIdWithIndex(localId: string): { item: QueueItem<T>; index: number } | null {
+        if (!localId) return null;
+        const index = this.queue.findIndex(item => item.localId === localId);
+        if (index === -1) return null;
+        const [item] = this.queue.splice(index, 1);
+        return item ? { item, index } : null;
+    }
+
+    restoreAt(index: number, item: QueueItem<T>): void {
+        const boundedIndex = Math.max(0, Math.min(index, this.queue.length));
+        this.queue.splice(boundedIndex, 0, item);
+    }
+
     /**
      * Reset the queue - clears all messages and resets to empty state
      */

hub/src/sync/rpcGateway.ts

@@ -79,6 +79,13 @@ export class RpcGateway {
         await this.sessionRpc(sessionId, RPC_METHODS.Abort, { reason: 'User aborted via Telegram Bot' })
     }
 
+    async steerQueuedMessage(
+        sessionId: string,
+        localId: string
+    ): Promise<unknown> {
+        return await this.sessionRpc(sessionId, 'steer-queued-message', { localId })
+    }
+
     async switchSession(sessionId: string, to: 'remote' | 'local'): Promise<void> {
         await this.sessionRpc(sessionId, RPC_METHODS.Switch, { to })
     }

hub/src/sync/syncEngine.ts

@@ -10,6 +10,7 @@
 import { isKnownFlavor, type LocalResumeTarget, type ResumableSession } from '@hapi/protocol'
 import type { SlashCommandsResponse } from '@hapi/protocol/apiTypes'
 import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types'
+import type { SteerQueuedMessageResponse } from '@hapi/protocol/schemas'
 import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages'
 import type { Server } from 'socket.io'
 import type { Store, CancelQueuedMessageResult } from '../store'
@@ -395,6 +396,62 @@ export class SyncEngine {
         this.messageService.sweepImmediateQueuedOnSessionEnd(sessionId, invokedAt)
     }
 
+    async steerQueuedMessage(
+        sessionId: string,
+        messageId: string
+    ): Promise<SteerQueuedMessageResponse> {
+        const lookup = this.store.messages.lookupQueuedMessage(sessionId, messageId)
+        if (lookup.status === 'absent') {
+            return { status: 'absent' }
+        }
+        if (lookup.status === 'invoked') {
+            return lookup
+        }
+        if (!lookup.localId) {
+            return { status: 'failed', error: 'Queued message has no localId' }
+        }
+
+        let result: unknown
+        try {
+            result = await this.rpcGateway.steerQueuedMessage(sessionId, lookup.localId)
+        } catch (error) {
+            return {
+                status: 'failed',
+                error: error instanceof Error ? error.message : String(error)
+            }
+        }
+
+        if (!result || typeof result !== 'object') {
+            return { status: 'failed', error: 'Unexpected steer response' }
+        }
+
+        const status = (result as { status?: unknown }).status
+        if (status === 'steered') {
+            const invokedAt = Date.now()
+            this.store.messages.markMessagesInvoked(sessionId, [lookup.localId], invokedAt)
+            this.recordSessionActivity(sessionId, invokedAt)
+            this.eventPublisher.emit({
+                type: 'messages-consumed',
+                sessionId,
+                localIds: [lookup.localId],
+                invokedAt
+            })
+            return { status: 'steered', localId: lookup.localId, invokedAt }
+        }
+        if (status === 'not-active' || status === 'not-found') {
+            return { status }
+        }
+        if (status === 'failed') {
+            const error = (result as { error?: unknown }).error
+            return {
+                status: 'failed',
+                error: typeof error === 'string' ? error : 'Steer failed'
+            }
+        }
+
+        return { status: 'failed', error: 'Unexpected steer response' }
+    }
+
     async approvePermission(
         sessionId: string,
         requestId: string,

hub/src/web/routes/messages.ts

@@ -48,6 +48,23 @@ export function createMessagesRoutes(getSyncEngine: () => SyncEngine | null): Ho
         return c.json(result)
     })
 
+    app.post('/sessions/:id/messages/:messageId/steer', async (c) => {
+        const engine = requireSyncEngine(c, getSyncEngine)
+        if (engine instanceof Response) {
+            return engine
+        }
+
+        const sessionResult = requireSessionFromParam(c, engine, { requireActive: true })
+        if (sessionResult instanceof Response) {
+            return sessionResult
+        }
+        const sessionId = sessionResult.sessionId
+        const messageId = c.req.param('messageId')
+
+        const result = await engine.steerQueuedMessage(sessionId, messageId)
+        return c.json(result)
+    })
+
     app.post('/sessions/:id/messages', async (c) => {
         const engine = requireSyncEngine(c, getSyncEngine)
         if (engine instanceof Response) {

shared/src/schemas.ts

@@ -374,3 +374,14 @@ export const CancelMessageResponseSchema = z.discriminatedUnion('status', [
 ])
 
 export type CancelMessageResponse = z.infer<typeof CancelMessageResponseSchema>
+
+export const SteerQueuedMessageResponseSchema = z.discriminatedUnion('status', [
+    z.object({ status: z.literal('steered'), localId: z.string(), invokedAt: z.number() }),
+    z.object({ status: z.literal('invoked'), message: DecryptedMessageSchema }),
+    z.object({ status: z.literal('absent') }),
+    z.object({ status: z.literal('not-active') }),
+    z.object({ status: z.literal('not-found') }),
+    z.object({ status: z.literal('failed'), error: z.string() }),
+])
+
+export type SteerQueuedMessageResponse = z.infer<typeof SteerQueuedMessageResponseSchema>

web/src/api/client.ts

@@ -36,7 +36,7 @@ import type {
     UploadFileResponse
 } from '@hapi/protocol/apiTypes'
 import type { AgentFlavor } from '@hapi/protocol'
-import type { CancelMessageResponse } from '@hapi/protocol/schemas'
+import type { CancelMessageResponse, SteerQueuedMessageResponse } from '@hapi/protocol/schemas'
 
 type ApiClientOptions = {
     baseUrl?: string
@@ -394,6 +394,14 @@ export class ApiClient {
         return response as CancelMessageResponse
     }
 
+    async steerQueuedMessage(sessionId: string, messageId: string): Promise<SteerQueuedMessageResponse> {
+        const response = await this.request(
+            `/api/sessions/${encodeURIComponent(sessionId)}/messages/${encodeURIComponent(messageId)}/steer`,
+            { method: 'POST', body: JSON.stringify({}) }
+        )
+        return response as SteerQueuedMessageResponse
+    }
+
     async abortSession(sessionId: string): Promise<void> {
         await this.request(`/api/sessions/${encodeURIComponent(sessionId)}/abort`, {
             method: 'POST',