feat(claude): mid-turn Steer for remote Claude sessions (#2259)

* feat(claude): mid-turn Steer for remote Claude sessions

Queued messages can now be steered into a running Claude turn, the same as
Ctrl+X Ctrl+S in the Claude TUI. The row is written to the live stream-json
stdin and Claude folds it in at its next tool boundary.

- Spawn Claude with --replay-user-messages and tag every stdin message with
  a uuid; the echo acks pickup and is never relayed to chat
- A result that arrives while a steer is still unacked does not end the
  turn: Claude runs a follow-up turn for it, so ready waits for that result
- Launcher registers steer-queued-message: reserve the row, reject control
  commands and mode mismatches, ack the hub after the write, mark consumed
  (steered) on the echo; abort before pickup holds the row indeterminate,
  an unprompted exit requeues it
- Open the shared steer gate and peer/hub wording to Claude (remote only)

* fix(claude): retire completed input before a steer follow-up turn

- Generalize onCompactResultAccepted to onResultAccepted and fire it when a
  result arrives with a steer still unacked, so a stream failure during the
  follow-up turn recovers only the steer instead of replaying the finished
  prompt
- Close the steer target and settle pending steers when query() setup
  throws, so a retry backoff cannot accept a steer nobody will read

cli/src/claude/claudeRemote.steer.test.ts

@@ -0,0 +1,261 @@
+import { describe, it, expect, vi, afterEach } from 'vitest';
+import * as claudeSdk from '@/claude/sdk';
+import type { SDKMessage, SDKUserMessage } from '@/claude/sdk/types';
+import type { ClaudeSteerTarget } from './claudeRemote';
+
+vi.mock('@/claude/utils/claudeCheckSession', () => ({
+    claudeCheckSession: () => true
+}));
+
+vi.mock('@/modules/watcher/awaitFileExist', () => ({
+    awaitFileExist: async () => true
+}));
+
+vi.mock('@/claude/sdk/utils', () => ({
+    getDefaultClaudeCodePath: () => '/usr/bin/claude'
+}));
+
+/**
+ * Fake Claude process: records stdin messages and emits whatever the test
+ * scripts, so replay/result ordering can be driven explicitly.
+ */
+function createFakeClaude() {
+    const stdin: SDKUserMessage[] = [];
+    const outbox: Array<SDKMessage | Error | null> = [];
+    let wake: (() => void) | null = null;
+    let options: Record<string, unknown> | null = null;
+    const emit = (message: SDKMessage | Error | null) => {
+        outbox.push(message);
+        wake?.();
+        wake = null;
+    };
+    const query = ({ prompt, options: queryOptions }: { prompt: AsyncIterable<SDKUserMessage>; options: Record<string, unknown> }) => {
+        options = queryOptions;
+        void (async () => {
+            for await (const message of prompt) {
+                stdin.push(message);
+            }
+        })();
+        return {
+            async *[Symbol.asyncIterator]() {
+                while (true) {
+                    while (outbox.length === 0) {
+                        await new Promise<void>((resolve) => { wake = resolve; });
+                    }
+                    const next = outbox.shift()!;
+                    if (next === null) return;
+                    if (next instanceof Error) throw next;
+                    yield next;
+                }
+            },
+            setPermissionMode: async () => {},
+            getContextUsage: async () => ({})
+        };
+    };
+    return {
+        stdin,
+        emit,
+        query,
+        options: () => options,
+        replay: (message: SDKUserMessage) => emit({
+            type: 'user',
+            uuid: message.uuid,
+            isReplay: true,
+            message: message.message
+        } as SDKMessage),
+        result: () => emit({ type: 'result', subtype: 'success' } as SDKMessage),
+        close: () => emit(null),
+        fail: (error: Error) => emit(error)
+    };
+}
+
+async function waitFor(condition: () => boolean, timeoutMs = 1000): Promise<void> {
+    const startedAt = Date.now();
+    while (!condition()) {
+        if (Date.now() - startedAt > timeoutMs) {
+            throw new Error('Timed out waiting for condition');
+        }
+        await new Promise((resolve) => setTimeout(resolve, 5));
+    }
+}
+
+async function startRemote(firstMessage = 'Run the tests') {
+    const fake = createFakeClaude();
+    const querySpy = vi.spyOn(claudeSdk, 'query').mockImplementation(fake.query as unknown as typeof claudeSdk.query);
+    const { claudeRemote } = await import('./claudeRemote');
+    const targets: Array<ClaudeSteerTarget | null> = [];
+    const relayed: SDKMessage[] = [];
+    const thinking: boolean[] = [];
+    const onReady = vi.fn();
+    const onResultAccepted = vi.fn();
+    let delivered = false;
+    const run = claudeRemote({
+        sessionId: null,
+        path: process.cwd(),
+        allowedTools: [],
+        hookSettingsPath: '/tmp/hook.json',
+        canCallTool: async () => ({ behavior: 'allow', updatedInput: {} }),
+        // One prompt, then park until the stream closes.
+        nextMessage: async () => {
+            if (!delivered) {
+                delivered = true;
+                return { message: firstMessage, mode: { permissionMode: 'default' } };
+            }
+            return null;
+        },
+        onReady,
+        onResultAccepted,
+        isAborted: () => false,
+        onSessionFound: () => {},
+        onThinkingChange: (value) => thinking.push(value),
+        onMessage: (message) => relayed.push(message),
+        onSteerTargetChange: (target) => targets.push(target)
+    });
+    await waitFor(() => fake.stdin.length === 1);
+    return {
+        fake,
+        run,
+        targets,
+        relayed,
+        thinking,
+        onReady,
+        onResultAccepted,
+        currentTarget: () => targets.at(-1) ?? null,
+        restore: () => querySpy.mockRestore()
+    };
+}
+
+describe('claudeRemote steering', () => {
+    let restore: (() => void) | null = null;
+    afterEach(() => {
+        restore?.();
+        restore = null;
+    });
+
+    it('injects a steered message into the running turn and acks it by its echo', async () => {
+        const remote = await startRemote();
+        restore = remote.restore;
+        const { fake } = remote;
+        expect(fake.options()?.replayUserMessages).toBe(true);
+        fake.replay(fake.stdin[0]!);
+
+        const target = remote.currentTarget();
+        expect(target).not.toBeNull();
+        const delivery = target!.steer('Also run lint');
+        expect(delivery).not.toBeNull();
+        await waitFor(() => fake.stdin.length === 2);
+        expect(fake.stdin[1]!.message.content).toBe('Also run lint');
+        expect(fake.stdin[1]!.uuid).toEqual(expect.any(String));
+        expect(fake.stdin[1]!.uuid).not.toBe(fake.stdin[0]!.uuid);
+
+        fake.replay(fake.stdin[1]!);
+        await expect(delivery!.delivered).resolves.toBe(true);
+
+        fake.result();
+        await waitFor(() => remote.onReady.mock.calls.length === 1);
+        expect(remote.currentTarget()).toBeNull();
+        expect(target!.steer('too late')).toBeNull();
+        // Echoes are acks only, never relayed as chat messages.
+        expect(remote.relayed.filter((message) => message.type === 'user')).toEqual([]);
+
+        fake.close();
+        await remote.run;
+    });
+
+    it('keeps the turn open when the result arrives before Claude picks up a steer', async () => {
+        const remote = await startRemote();
+        restore = remote.restore;
+        const { fake } = remote;
+        fake.replay(fake.stdin[0]!);
+        const delivery = remote.currentTarget()!.steer('Then summarize')!;
+        await waitFor(() => fake.stdin.length === 2);
+
+        // Text-only turn: no tool boundary, so Claude finishes first and runs
+        // a follow-up turn for the steered message.
+        fake.result();
+        await new Promise((resolve) => setTimeout(resolve, 20));
+        expect(remote.onReady).not.toHaveBeenCalled();
+        expect(remote.thinking.at(-1)).toBe(true);
+        expect(remote.currentTarget()).not.toBeNull();
+
+        fake.replay(fake.stdin[1]!);
+        await expect(delivery.delivered).resolves.toBe(true);
+        fake.result();
+        await waitFor(() => remote.onReady.mock.calls.length === 1);
+        expect(remote.thinking.at(-1)).toBe(false);
+        expect(remote.currentTarget()).toBeNull();
+
+        fake.close();
+        await remote.run;
+    });
+
+    it('retires the completed prompt when the stream fails during a steer follow-up turn', async () => {
+        const remote = await startRemote();
+        restore = remote.restore;
+        const { fake } = remote;
+        fake.replay(fake.stdin[0]!);
+        const delivery = remote.currentTarget()!.steer('Then summarize')!;
+        await waitFor(() => fake.stdin.length === 2);
+
+        fake.result();
+        await waitFor(() => remote.onResultAccepted.mock.calls.length === 1);
+        expect(remote.onReady).not.toHaveBeenCalled();
+
+        fake.fail(new Error('Claude exited'));
+        await expect(remote.run).rejects.toThrow('Claude exited');
+        await expect(delivery.delivered).resolves.toBe(false);
+        expect(remote.currentTarget()).toBeNull();
+    });
+
+    it('closes the steer target when query setup throws', async () => {
+        const querySpy = vi.spyOn(claudeSdk, 'query').mockImplementation(() => {
+            throw new Error('Fallback model cannot be the same as the main model');
+        });
+        restore = () => querySpy.mockRestore();
+        const { claudeRemote } = await import('./claudeRemote');
+        const targets: Array<ClaudeSteerTarget | null> = [];
+        let delivered = false;
+        await expect(claudeRemote({
+            sessionId: null,
+            path: process.cwd(),
+            allowedTools: [],
+            hookSettingsPath: '/tmp/hook.json',
+            canCallTool: async () => ({ behavior: 'allow', updatedInput: {} }),
+            nextMessage: async () => {
+                if (delivered) return null;
+                delivered = true;
+                return { message: 'Run the tests', mode: { permissionMode: 'default' } };
+            },
+            onReady: () => {},
+            isAborted: () => false,
+            onSessionFound: () => {},
+            onMessage: () => {},
+            onSteerTargetChange: (target) => targets.push(target)
+        })).rejects.toThrow('Fallback model');
+
+        expect(targets).toHaveLength(2);
+        expect(targets[1]).toBeNull();
+        expect(targets[0]!.steer('stranded')).toBeNull();
+    });
+
+    it('reports an unacknowledged steer as undelivered when the stream closes', async () => {
+        const remote = await startRemote();
+        restore = remote.restore;
+        const { fake } = remote;
+        const delivery = remote.currentTarget()!.steer('Never seen')!;
+        await waitFor(() => fake.stdin.length === 2);
+
+        fake.close();
+        await remote.run;
+        await expect(delivery.delivered).resolves.toBe(false);
+        expect(remote.currentTarget()).toBeNull();
+    });
+
+    it('does not offer a steer target for a slash command turn', async () => {
+        const remote = await startRemote('/compact');
+        restore = remote.restore;
+        expect(remote.targets).toEqual([]);
+        remote.fake.close();
+        await remote.run;
+    });
+});

cli/src/claude/claudeRemote.test.ts

@@ -1208,7 +1208,7 @@ describe('claudeRemote compact summary promotion', () => {
                 readyCount += 1;
                 readyEvents.push(completionEvent);
             },
-            onCompactResultAccepted: () => {
+            onResultAccepted: () => {
                 acceptedCount += 1;
             },
             isAborted: () => false,

cli/src/claude/claudeRemote.ts

@@ -2,6 +2,7 @@ import { EnhancedMode, PermissionMode } from "./loop";
 import { query, type QueryOptions as Options, type SDKMessage, type SDKSystemMessage, AbortError, SDKUserMessage } from '@/claude/sdk'
 import { claudeCheckSession } from "./utils/claudeCheckSession";
 import { join } from 'node:path';
+import { randomUUID } from 'node:crypto';
 import { statSync } from 'node:fs';
 import { parseSpecialCommand } from "@/parsers/specialCommands";
 import { logger } from "@/lib";
@@ -28,6 +29,20 @@ interface CompactCompletion {
     contextTokens?: number;
 }
 
+/**
+ * Injects plain user messages into the running turn. Claude picks a message up
+ * at its next tool boundary, or opens a follow-up turn for it when the current
+ * one ends first.
+ */
+export interface ClaudeSteerTarget {
+    /**
+     * Null when the turn can no longer accept a message. `delivered` resolves
+     * true once Claude echoes the message (it is processing it), false when the
+     * stream closes first.
+     */
+    steer(message: string): { delivered: Promise<boolean> } | null;
+}
+
 interface ActiveCompact {
     baseline: number | null;
     failure: string | null;
@@ -73,9 +88,15 @@ export async function claudeRemote(opts: {
     onMessage: (message: SDKMessage) => void,
     onContextUsage?: (usage: { model?: string, maxTokens?: number }) => void,
     onFirstResult?: (initialMessage: string) => void,
-    onCompactResultAccepted?: () => void,
+    /**
+     * The delivered input finished at the SDK boundary although onReady is
+     * still deferred (compact summary lookup, or a steer follow-up turn).
+     */
+    onResultAccepted?: () => void,
     onCompletionEvent?: (message: string) => void,
-    onSessionReset?: () => void
+    onSessionReset?: () => void,
+    /** Open while a plain (non-command) turn runs; null otherwise. */
+    onSteerTargetChange?: (target: ClaudeSteerTarget | null) => void
 }) {
     const debugPrefix = '[claudeRemote][async-debug]';
 
@@ -141,6 +162,49 @@ export async function claudeRemote(opts: {
     let awaitingForkInit = forkSession;
 
     const messages = new PushableAsyncIterable<SDKUserMessage>();
+    // Every stdin message carries a uuid that `--replay-user-messages` echoes
+    // back. The echoes are acks only: relaying them would duplicate prompts.
+    const pushedUuids = new Set<string>();
+    const pushUserMessage = (content: string, uuid: string = randomUUID()) => {
+        pushedUuids.add(uuid);
+        messages.push({ type: 'user', uuid, message: { role: 'user', content } });
+    };
+    // Steered messages Claude has not echoed yet. A result while any is
+    // pending is not the end of the turn: Claude runs a follow-up turn for it.
+    const pendingSteers = new Map<string, (delivered: boolean) => void>();
+    let steerTarget: ClaudeSteerTarget | null = null;
+    let responseClosed = false;
+    let inputEnded = false;
+    const setSteerable = (open: boolean) => {
+        if (open === (steerTarget !== null)) return;
+        if (!open) {
+            steerTarget = null;
+            opts.onSteerTargetChange?.(null);
+            return;
+        }
+        const target: ClaudeSteerTarget = {
+            steer: (message) => {
+                if (steerTarget !== target || responseClosed || inputEnded) return null;
+                const uuid = randomUUID();
+                const delivered = new Promise<boolean>((resolve) => pendingSteers.set(uuid, resolve));
+                pushUserMessage(message, uuid);
+                logger.debug(`${debugPrefix} steered message pushed uuid=${uuid} length=${message.length}`);
+                return { delivered };
+            }
+        };
+        steerTarget = target;
+        opts.onSteerTargetChange?.(target);
+    };
+    // Once no reader drains stdin, close the target and settle every
+    // unacknowledged steer as undelivered.
+    const closeSteering = () => {
+        responseClosed = true;
+        setSteerable(false);
+        for (const resolveSteer of pendingSteers.values()) {
+            resolveSteer(false);
+        }
+        pendingSteers.clear();
+    };
 
     // Success-only: a failed compaction keeps its failure line, an unknown
     // session id or any transcript read problem falls back to the plain
@@ -249,13 +313,8 @@ export async function claudeRemote(opts: {
         }
 
         mode = next.mode;
-        messages.push({
-            type: 'user',
-            message: {
-                role: 'user',
-                content: next.message,
-            },
-        });
+        pushUserMessage(next.message);
+        setSteerable(specialCommand.type === null);
         return next;
     };
 
@@ -287,6 +346,8 @@ export async function claudeRemote(opts: {
         pathToClaudeCodeExecutable: getDefaultClaudeCodePath(),
         settingsPath: opts.hookSettingsPath,
         additionalDirectories: [getHapiBlobsDir()],
+        // Steering acks a mid-turn message by its echo.
+        replayUserMessages: true,
     }
 
     if (!awaitingForkInit) {
@@ -324,10 +385,17 @@ export async function claudeRemote(opts: {
     };
 
     // Start the loop
-    const response = query({
-        prompt: messages,
-        options: sdkOptions,
-    });
+    let response: ReturnType<typeof query>;
+    try {
+        response = query({
+            prompt: messages,
+            options: sdkOptions,
+        });
+    } catch (e) {
+        // The initial prompt already opened the steer target.
+        closeSteering();
+        throw e;
+    }
 
     // plan/auto are enforced by Claude itself, so a mid-turn switch into or out
     // of them must reach the running process; otherwise it waits for a restart.
@@ -363,12 +431,10 @@ export async function claudeRemote(opts: {
     });
 
     let nextMessageFetchInFlight = false;
-    let inputEnded = false;
     let nextMessageFetchSeq = 0;
     let streamMessageSeq = 0;
     let resultSeq = 0;
     const compactCompletionAbort = new AbortController();
-    let responseClosed = false;
     let compactCompletion: Promise<CompactCompletion> | null = null;
     const abortCompactCompletion = () => compactCompletionAbort.abort();
     opts.signal?.addEventListener('abort', abortCompactCompletion, { once: true });
@@ -408,7 +474,8 @@ export async function claudeRemote(opts: {
                 }
                 mode = next.mode;
                 specialCommand = nextSpecialCommand;
-                messages.push({ type: 'user', message: { role: 'user', content: next.message } });
+                pushUserMessage(next.message);
+                setSteerable(nextSpecialCommand.type === null);
                 logger.debug(
                     `${debugPrefix} nextMessage resolved fetchId=${fetchId} elapsedMs=${Date.now() - startedAt} ` +
                     `messageLength=${next.message.length} permissionMode=${next.mode.permissionMode}`
@@ -521,6 +588,19 @@ export async function claudeRemote(opts: {
             );
             logger.debugLargeJson(`[claudeRemote] Message ${message.type}`, message);
 
+            if (message.type === 'user' && (message as SDKUserMessage).isReplay === true) {
+                const uuid = (message as SDKUserMessage).uuid;
+                if (uuid && pushedUuids.delete(uuid)) {
+                    const resolveSteer = pendingSteers.get(uuid);
+                    if (resolveSteer) {
+                        pendingSteers.delete(uuid);
+                        logger.debug(`${debugPrefix} steered message picked up uuid=${uuid}`);
+                        resolveSteer(true);
+                    }
+                    continue;
+                }
+            }
+
             const compactMetadata =
                 message.type === 'system' && message.subtype === 'compact_boundary'
                     ? (message as any).compact_metadata
@@ -631,16 +711,27 @@ export async function claudeRemote(opts: {
             // Handle result messages
             if (message.type === 'result') {
                 resultSeq += 1;
+                if (resultSeq === 1 && specialCommand.type === null && initial) {
+                    opts.onFirstResult?.(initial.message);
+                }
+
+                if (pendingSteers.size > 0) {
+                    // The turn ended before Claude reached a tool boundary for
+                    // a steered message; it starts a follow-up turn for it.
+                    logger.debug(`${debugPrefix} result #${resultSeq} received with ${pendingSteers.size} pending steer(s); turn continues`);
+                    // The delivered prompt is complete; a later stream failure
+                    // must recover only the outstanding steer, not replay it.
+                    opts.onResultAccepted?.();
+                    continue;
+                }
+
                 updateThinking(false);
+                setSteerable(false);
                 logger.debug(
                     `${debugPrefix} result #${resultSeq} received; scheduling next user message ` +
                     `(nextInFlight=${nextMessageFetchInFlight}, inputEnded=${inputEnded})`
                 );
 
-                if (resultSeq === 1 && specialCommand.type === null && initial) {
-                    opts.onFirstResult?.(initial.message);
-                }
-
                 if (compactState.active) {
                     if (!compactState.active.sawCompactSignal) continue;
                     const compact = compactState.active;
@@ -656,7 +747,7 @@ export async function claudeRemote(opts: {
                     // bar with it, since the next real usage only arrives
                     // with the next model response.
                     compactState.active = null;
-                    opts.onCompactResultAccepted?.();
+                    opts.onResultAccepted?.();
 
                     compactCompletion = (async () => {
                         let compactSummary = await lookupCompactSummary(
@@ -723,7 +814,7 @@ export async function claudeRemote(opts: {
             throw e;
         }
     } finally {
-        responseClosed = true;
+        closeSteering();
         if (compactCompletion) {
             compactCompletionAbort.abort();
             await Promise.allSettled([compactCompletion]);

cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts

@@ -155,7 +155,7 @@ describe('claudeRemoteLauncher launch-failure recovery', () => {
             if (callCount === 1) {
                 const msg = await opts.nextMessage();
                 expect(msg?.message).toBe('/compact');
-                opts.onCompactResultAccepted();
+                opts.onResultAccepted();
                 throw new Error('stream failed after compact result');
             }
 

cli/src/claude/claudeRemoteLauncher.steer.test.ts

@@ -0,0 +1,230 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { RPC_METHODS } from '@hapi/protocol/rpcMethods'
+import { MessageQueue2 } from '@/utils/MessageQueue2'
+import type { EnhancedMode } from './loop'
+
+// Drives the launcher's steer-queued-message RPC against a scripted
+// claudeRemote(): one plain turn that exposes a steer target until the test
+// ends it, mirroring how the real stream opens/closes the target.
+const harness = vi.hoisted(() => ({
+    steered: [] as string[],
+    deliveries: [] as Array<(delivered: boolean) => void>,
+    endTurn: null as (() => void) | null,
+    signal: null as AbortSignal | null
+}))
+
+vi.mock('./claudeRemote', () => ({
+    claudeRemote: async (opts: any) => {
+        const initial = await opts.nextMessage()
+        if (!initial) return
+        harness.signal = opts.signal
+        let open = true
+        opts.onSteerTargetChange({
+            steer: (message: string) => {
+                if (!open) return null
+                harness.steered.push(message)
+                return { delivered: new Promise<boolean>((resolve) => harness.deliveries.push(resolve)) }
+            }
+        })
+        await new Promise<void>((resolve) => {
+            harness.endTurn = resolve
+            opts.signal.addEventListener('abort', () => resolve(), { once: true })
+        })
+        open = false
+        opts.onSteerTargetChange(null)
+        for (const resolve of harness.deliveries.splice(0)) resolve(false)
+    }
+}))
+
+vi.mock('./utils/permissionHandler', () => ({
+    PermissionHandler: class {
+        setOnPermissionRequest(): void {}
+        getResponses(): Map<string, unknown> { return new Map() }
+        onMessage(): void {}
+        handleToolCall = async () => ({ behavior: 'allow', updatedInput: {} })
+        reset(): void {}
+        isAborted(): boolean { return false }
+        handleModeChange(): void {}
+    }
+}))
+
+vi.mock('./utils/sdkToLogConverter', () => ({
+    SDKToLogConverter: class {
+        currentContextWindow(): undefined { return undefined }
+        updateSessionId(): void {}
+        resetParentChain(): void {}
+        convert(): null { return null }
+        convertSidechainUserMessage(): null { return null }
+        updateSelectedModel(): void {}
+        generateInterruptedToolResult(): null { return null }
+    }
+}))
+
+vi.mock('./utils/OutgoingMessageQueue', () => ({
+    OutgoingMessageQueue: class {
+        releaseToolCall(): void {}
+        enqueue(): void {}
+        async flush(): Promise<void> {}
+        destroy(): void {}
+    }
+}))
+
+import { claudeRemoteLauncher } from './claudeRemoteLauncher'
+import { Session } from './session'
+
+type Handler = (payload?: unknown) => unknown
+
+function createClientStub() {
+    const rpcHandlers = new Map<string, Handler>()
+    let agentState: Record<string, unknown> = {}
+    return {
+        rpcHandlerManager: {
+            registerHandler: (method: string, handler: Handler) => {
+                rpcHandlers.set(method, handler)
+            }
+        },
+        rpcHandlers,
+        agentState: () => agentState,
+        keepAlive: () => {},
+        updateMetadata: (mutator: (metadata: any) => any) => { mutator({}) },
+        updateAgentState: (mutator: (state: any) => any) => { agentState = mutator(agentState) },
+        emitMessagesConsumed: vi.fn(),
+        emitSteerIndeterminate: vi.fn(),
+        setSteerDeliveryState: vi.fn(async () => true),
+        sendClaudeSessionMessage: () => {},
+        sendSessionEvent: (_event: unknown) => {},
+        sendAgentMessage: vi.fn(),
+        notePendingHubPromptEcho: vi.fn(),
+        discardPendingHubPromptEcho: vi.fn(),
+        discardPendingHubPromptEchoText: () => {}
+    }
+}
+
+const DEFAULT_MODE: EnhancedMode = { permissionMode: 'default' }
+
+async function startTurn() {
+    const client = createClientStub()
+    const queue = new MessageQueue2<EnhancedMode>((mode) => JSON.stringify(mode))
+    const session = new Session({
+        api: {} as any,
+        client: client as any,
+        path: '/tmp/test',
+        logPath: '/tmp/test.log',
+        sessionId: null,
+        claudeEnvVars: {},
+        claudeArgs: [],
+        mcpServers: {},
+        messageQueue: queue,
+        onModeChange: () => {},
+        allowedTools: [],
+        mode: 'remote',
+        startedBy: 'runner',
+        startingMode: 'remote',
+        hookSettingsPath: '/tmp/hook.json',
+        permissionMode: 'default'
+    })
+    queue.push('Run the tests', DEFAULT_MODE, 'first')
+    const launched = claudeRemoteLauncher(session as any)
+    await vi.waitFor(() => expect(client.agentState().steeringActive).toBe(true))
+    const steer = (localId: string) => client.rpcHandlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId })
+    const stop = async () => {
+        client.rpcHandlers.get(RPC_METHODS.Switch)?.()
+        await launched
+        session.stopKeepAlive()
+    }
+    return { client, queue, steer, stop }
+}
+
+describe('claudeRemoteLauncher steer', () => {
+    afterEach(() => {
+        harness.steered = []
+        harness.deliveries = []
+        harness.endTurn = null
+        harness.signal = null
+    })
+
+    it('writes the queued row into the running turn and acks it once Claude picks it up', async () => {
+        const { client, queue, steer, stop } = await startTurn()
+        try {
+            queue.push('Also run lint', DEFAULT_MODE, 'steer-1')
+
+            await expect(steer('steer-1')).resolves.toEqual({ steered: true })
+            expect(harness.steered).toEqual(['Also run lint'])
+            expect(client.setSteerDeliveryState).toHaveBeenCalledWith(['steer-1'], 'dispatching')
+            expect(client.notePendingHubPromptEcho).toHaveBeenCalledWith('Also run lint', ['steer-1'])
+            expect(queue.size()).toBe(0)
+            expect(client.emitMessagesConsumed).not.toHaveBeenCalledWith(['steer-1'], { steered: true })
+
+            harness.deliveries[0]!(true)
+            await vi.waitFor(() => expect(client.emitMessagesConsumed).toHaveBeenCalledWith(['steer-1'], { steered: true }))
+            expect(queue.cancelByLocalId('steer-1')).toBe('consumed')
+        } finally {
+            await stop()
+        }
+    })
+
+    it('rejects rows that cannot join the active turn and keeps them queued', async () => {
+        const { queue, steer, stop } = await startTurn()
+        try {
+            queue.push('Plan it instead', { permissionMode: 'plan' }, 'other-mode')
+            queue.pushIsolated('/compact', DEFAULT_MODE, 'command')
+
+            await expect(steer('other-mode')).resolves.toEqual({
+                steered: false,
+                error: 'Queued message mode differs from the active turn'
+            })
+            await expect(steer('command')).resolves.toEqual({
+                steered: false,
+                error: 'Control commands cannot be steered'
+            })
+            await expect(steer('missing')).resolves.toEqual({ steered: false, error: 'Message not in queue' })
+            expect(harness.steered).toEqual([])
+            expect(queue.pendingLocalIds()).toEqual(['other-mode', 'command'])
+        } finally {
+            await stop()
+        }
+    })
+
+    it('reports no steerable turn once the turn has ended', async () => {
+        const { client, queue, steer, stop } = await startTurn()
+        try {
+            harness.endTurn!()
+            await vi.waitFor(() => expect(client.agentState().steeringActive).toBe(false))
+            queue.push('Late', DEFAULT_MODE, 'late')
+            await expect(steer('late')).resolves.toEqual({ steered: false, error: 'No active steerable turn' })
+        } finally {
+            await stop()
+        }
+    })
+
+    it('requeues a steer Claude never picked up when the process exits on its own', async () => {
+        const { client, queue, steer, stop } = await startTurn()
+        try {
+            queue.push('Also run lint', DEFAULT_MODE, 'steer-1')
+            await expect(steer('steer-1')).resolves.toEqual({ steered: true })
+
+            harness.deliveries[0]!(false)
+            await vi.waitFor(() => expect(client.setSteerDeliveryState).toHaveBeenCalledWith(['steer-1'], 'queued'))
+            expect(queue.pendingLocalIds()).toContain('steer-1')
+            expect(client.discardPendingHubPromptEcho).toHaveBeenCalledWith('steer-1')
+            expect(client.emitSteerIndeterminate).not.toHaveBeenCalled()
+        } finally {
+            await stop()
+        }
+    })
+
+    it('holds an unacknowledged steer for explicit resolution after an abort', async () => {
+        const { client, queue, steer, stop } = await startTurn()
+        try {
+            queue.push('Also run lint', DEFAULT_MODE, 'steer-1')
+            await expect(steer('steer-1')).resolves.toEqual({ steered: true })
+
+            await client.rpcHandlers.get(RPC_METHODS.Abort)!()
+            await vi.waitFor(() => expect(client.emitSteerIndeterminate).toHaveBeenCalledWith(['steer-1']))
+            expect(queue.pendingLocalIds()).not.toContain('steer-1')
+            expect(client.emitMessagesConsumed).not.toHaveBeenCalledWith(['steer-1'], { steered: true })
+        } finally {
+            await stop()
+        }
+    })
+})

cli/src/claude/claudeRemoteLauncher.ts

@@ -1,7 +1,7 @@
 import React from "react";
 import { Session } from "./session";
 import { RemoteModeDisplay } from "@/ui/ink/RemoteModeDisplay";
-import { claudeRemote, type CompactSummaryPayload } from "./claudeRemote";
+import { claudeRemote, type ClaudeSteerTarget, type CompactSummaryPayload } from "./claudeRemote";
 import { convertAgentMessage } from "@/agent/messageConverter";
 import { PermissionHandler } from "./utils/permissionHandler";
 import { Future } from "@/utils/future";
@@ -13,6 +13,8 @@ import { PLAN_FAKE_REJECT } from "./sdk/prompts";
 import { EnhancedMode } from "./loop";
 import { OutgoingMessageQueue } from "./utils/OutgoingMessageQueue";
 import type { ClaudePermissionMode } from "@hapi/protocol/types";
+import { RPC_METHODS } from "@hapi/protocol/rpcMethods";
+import { parseSpecialCommand } from "@/parsers/specialCommands";
 import { ClaudeToolResultImages } from './utils/toolResultImages';
 import { applySessionTitleFallback, readNativeSessionTitle, stripAttachmentReferences } from './utils/sessionTitleFallback';
 import {
@@ -53,6 +55,8 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
     private abortFuture: Future<void> | null = null;
     private permissionHandler: PermissionHandler | null = null;
     private handleSessionFound: ((sessionId: string) => void) | null = null;
+    /** The running plain turn that can take a steered queue row, if any. */
+    private steerTarget: { target: ClaudeSteerTarget; modeHash: string | null; signal: AbortSignal } | null = null;
 
     constructor(session: Session) {
         super(process.env.DEBUG ? session.logPath : undefined);
@@ -147,6 +151,106 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
             messageQueue.releaseToolCall(toolCallId);
         });
 
+        // Per-message steer from the waiting queue (web "Steer" button), the
+        // same as Ctrl+X Ctrl+S in the Claude TUI: the row is written to the
+        // running process and Claude folds it in at its next tool boundary.
+        session.client.rpcHandlerManager.registerHandler(
+            RPC_METHODS.SteerQueuedMessage,
+            async (payload: unknown) => {
+                const localId = typeof (payload as { localId?: unknown } | null)?.localId === 'string'
+                    ? (payload as { localId: string }).localId
+                    : '';
+                if (!localId) {
+                    return { steered: false, error: 'Missing localId' };
+                }
+                const active = this.steerTarget;
+                if (!active || active.signal.aborted) {
+                    return { steered: false, error: 'No active steerable turn' };
+                }
+                const taken = session.queue.takeByLocalId(localId);
+                if (!taken) {
+                    return { steered: false, error: 'Message not in queue' };
+                }
+                if (taken.item.isolate || parseSpecialCommand(taken.item.message).type !== null) {
+                    session.queue.restoreReservation(taken);
+                    return { steered: false, error: 'Control commands cannot be steered' };
+                }
+                if (taken.item.modeHash !== active.modeHash) {
+                    session.queue.restoreReservation(taken);
+                    return { steered: false, error: 'Queued message mode differs from the active turn' };
+                }
+                if (!session.queue.beginReservationDispatch(taken)) {
+                    return { steered: false, error: 'Steer cancelled' };
+                }
+                const dispatchStatePersisted = await session.client.setSteerDeliveryState([localId], 'dispatching');
+                if (!dispatchStatePersisted) {
+                    session.queue.markReservationIndeterminate(taken);
+                    session.client.emitSteerIndeterminate([localId]);
+                    return { steered: false, error: 'Steer state is indeterminate' };
+                }
+                const restoreQueuedReservation = async (): Promise<boolean> => {
+                    if (!taken.originIndeterminate) {
+                        const persisted = await session.client.setSteerDeliveryState([localId], 'queued');
+                        if (!persisted) {
+                            session.queue.markReservationIndeterminate(taken);
+                            session.client.emitSteerIndeterminate([localId]);
+                            return false;
+                        }
+                    }
+                    if (taken.state !== 'dispatching' || !session.queue.restoreReservation(taken)) {
+                        session.client.emitSteerIndeterminate([localId]);
+                        return false;
+                    }
+                    return true;
+                };
+                if (taken.state !== 'dispatching') {
+                    session.client.emitSteerIndeterminate([localId]);
+                    return { steered: false, error: 'Steer cancelled' };
+                }
+                if (this.steerTarget !== active || active.signal.aborted) {
+                    await restoreQueuedReservation();
+                    return { steered: false, error: 'Active turn changed' };
+                }
+
+                const deliveredText = session.expandSkillReference(taken.item.message);
+                session.client.notePendingHubPromptEcho(deliveredText, [localId]);
+                const delivery = active.target.steer(deliveredText);
+                if (!delivery) {
+                    session.client.discardPendingHubPromptEcho(localId);
+                    await restoreQueuedReservation();
+                    return { steered: false, error: 'Active turn changed' };
+                }
+                void delivery.delivered.then(async (delivered) => {
+                    if (delivered) {
+                        // Agent output preceding the pickup is already in the
+                        // outgoing queue; send it first so the steered row is
+                        // stamped after it, not above it.
+                        await messageQueue.flush();
+                        session.queue.commitReservation(taken);
+                        messageBuffer.addMessage(taken.item.message, 'user');
+                        session.client.emitMessagesConsumed([localId], { steered: true });
+                        return;
+                    }
+                    session.client.discardPendingHubPromptEcho(localId);
+                    if (active.signal.aborted) {
+                        // Abort killed the process with the row unacknowledged:
+                        // hold it for explicit Retry/Cancel instead of replaying
+                        // it into a turn the user just stopped.
+                        if (session.queue.markReservationIndeterminate(taken)) {
+                            session.client.emitSteerIndeterminate([localId]);
+                        }
+                        return;
+                    }
+                    // Claude exited before it ever picked the row up, so it was
+                    // never part of the conversation: queue it again.
+                    await restoreQueuedReservation();
+                }).catch((error) => {
+                    logger.debug('[remote]: steered message settlement failed', error);
+                });
+                return { steered: true };
+            }
+        );
+
         const sdkToLogConverter = new SDKToLogConverter({
             sessionId: session.sessionId || 'unknown',
             cwd: session.path,
@@ -509,10 +613,11 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                     nativeTitle ?? stripAttachmentReferences(initialMessage)
                                 ));
                         },
-                        onCompactResultAccepted: () => {
-                            // The command has completed at the SDK boundary even
-                            // if transcript summary lookup is still pending. A
-                            // later stream failure must not replay /compact.
+                        onResultAccepted: () => {
+                            // The input has completed at the SDK boundary even
+                            // if ready is deferred (compact summary lookup, or
+                            // a steer follow-up turn). A later stream failure
+                            // must not replay it.
                             reachedReadyThisAttempt = true;
                             inFlightMessage = null;
                         },
@@ -587,6 +692,12 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
                                 logger.debug('[claudeRemoteLauncher][async-debug] ready event suppressed (pending input exists)');
                             }
                         },
+                        onSteerTargetChange: (target) => {
+                            this.steerTarget = target
+                                ? { target, modeHash, signal: controller.signal }
+                                : null;
+                            session.client.updateAgentState((state) => ({ ...state, steeringActive: target !== null }));
+                        },
                         signal: controller.signal,
                     });
 
@@ -729,6 +840,7 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase {
 
     protected async cleanup(): Promise<void> {
         this.clearAbortHandlers(this.session.client.rpcHandlerManager);
+        this.steerTarget = null;
 
         if (this.handleSessionFound) {
             this.session.removeSessionFoundCallback(this.handleSessionFound);

cli/src/claude/sdk/query.ts

@@ -348,6 +348,7 @@ export function query(config: {
             fallbackModel,
             settingsPath,
             strictMcpConfig,
+            replayUserMessages,
             canCallTool,
             promptFailureCleanupTimeoutMs = DEFAULT_PROMPT_FAILURE_CLEANUP_TIMEOUT_MS
         } = {}
@@ -396,6 +397,7 @@ export function query(config: {
         args.push('--print', stripNewlinesForWindowsShellArg(prompt.trim()))
     } else {
         args.push('--input-format', 'stream-json')
+        if (replayUserMessages) args.push('--replay-user-messages')
     }
 
     // Determine how to spawn Claude Code

cli/src/claude/sdk/types.ts

@@ -24,6 +24,10 @@ export interface SDKUserMessage extends SDKMessage {
      */
     isSynthetic?: boolean
     isMeta?: boolean
+    /** Stdin messages may set it; `--replay-user-messages` echoes it back. */
+    uuid?: string
+    /** Set on the `--replay-user-messages` echo of a stdin message. */
+    isReplay?: boolean
     message: {
         role: 'user'
         content: string | Array<{
@@ -211,6 +215,11 @@ export interface QueryOptions {
     fallbackModel?: string
     settingsPath?: string
     strictMcpConfig?: boolean
+    /**
+     * Echo every stdin user message back (`isReplay: true`, same uuid) when
+     * Claude starts processing it. Requires a streaming prompt.
+     */
+    replayUserMessages?: boolean
     canCallTool?: CanCallToolCallback
     promptFailureCleanupTimeoutMs?: number
 }

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

@@ -562,7 +562,7 @@ describe('pingPeer delivery modes', () => {
     it.each([
         { ...active, active: false },
         { ...active, thinking: false, agentState: { steeringActive: false } },
-        { ...active, metadata: { flavor: 'claude' } },
+        { ...active, metadata: { flavor: 'opencode' } },
         { ...active, metadata: { flavor: 'cursor', cursorSessionProtocol: 'stream-json' as const } },
         { ...active, metadata: { flavor: 'cursor', cursorSessionId: 'legacy' } },
         { ...active, agentState: { controlledByUser: true } }

cli/src/modules/pingPeer/pingPeer.ts

@@ -552,7 +552,7 @@ function getSteeringUnavailableReason(session: PingPeerSessionSummary): string |
         return 'Steering requires an active turn; target is inactive, no message sent. Use queue to resume it'
     }
     if (!isSteeringSupportedForSession(session.metadata)) {
-        return 'Steering is only supported for Pi, Codex, and Cursor ACP sessions; no message sent'
+        return 'Steering is only supported for Claude, Pi, Codex, and Cursor ACP sessions; no message sent'
     }
     if (session.agentState?.controlledByUser && !session.metadata?.capabilities?.concurrentClients) {
         return 'Steering is only available for remote sessions; no message sent'