diff --git a/cli/README.md b/cli/README.md index 318ab3c7..a9729204 100644 --- a/cli/README.md +++ b/cli/README.md @@ -55,7 +55,7 @@ Use explicit `queue` for non-urgent updates that can wait. Use `steer` for important, time-sensitive findings the peer needs to know now, not after its current turn: a correction that invalidates its assumptions, a blocker, or a result that changes its next action. State why it matters now and what the peer -should adjust. Steer requires insertion into an active Pi, Codex, or Cursor ACP +should adjust. Steer requires insertion into an active Claude (remote), Pi, Codex, or Cursor ACP turn; it does not abort that turn or guarantee an immediate response. MCP `ping_peer` accepts the same optional `deliveryMode: "auto" | "queue" | "steer"` field. Auto is resolved by the peer sender; ordinary web/native chat delivery is unchanged. diff --git a/cli/src/claude/claudeRemote.steer.test.ts b/cli/src/claude/claudeRemote.steer.test.ts new file mode 100644 index 00000000..d1d8ddf0 --- /dev/null +++ b/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 = []; + let wake: (() => void) | null = null; + let options: Record | null = null; + const emit = (message: SDKMessage | Error | null) => { + outbox.push(message); + wake?.(); + wake = null; + }; + const query = ({ prompt, options: queryOptions }: { prompt: AsyncIterable; options: Record }) => { + 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((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 { + 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 = []; + 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 = []; + 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; + }); +}); diff --git a/cli/src/claude/claudeRemote.test.ts b/cli/src/claude/claudeRemote.test.ts index cc123df0..49c92f3b 100644 --- a/cli/src/claude/claudeRemote.test.ts +++ b/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, diff --git a/cli/src/claude/claudeRemote.ts b/cli/src/claude/claudeRemote.ts index 8bca8d82..7e23f59e 100644 --- a/cli/src/claude/claudeRemote.ts +++ b/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 } | 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(); + // 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(); + 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 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((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; + 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 | 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]); diff --git a/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts b/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts index 1fddb56e..ee1d6945 100644 --- a/cli/src/claude/claudeRemoteLauncher.launchFailure.test.ts +++ b/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'); } diff --git a/cli/src/claude/claudeRemoteLauncher.steer.test.ts b/cli/src/claude/claudeRemoteLauncher.steer.test.ts new file mode 100644 index 00000000..9954c6dd --- /dev/null +++ b/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((resolve) => harness.deliveries.push(resolve)) } + } + }) + await new Promise((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 { 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 {} + destroy(): void {} + } +})) + +import { claudeRemoteLauncher } from './claudeRemoteLauncher' +import { Session } from './session' + +type Handler = (payload?: unknown) => unknown + +function createClientStub() { + const rpcHandlers = new Map() + let agentState: Record = {} + 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((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() + } + }) +}) diff --git a/cli/src/claude/claudeRemoteLauncher.ts b/cli/src/claude/claudeRemoteLauncher.ts index 8fa1bd30..a3bcc649 100644 --- a/cli/src/claude/claudeRemoteLauncher.ts +++ b/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 | 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 => { + 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 { this.clearAbortHandlers(this.session.client.rpcHandlerManager); + this.steerTarget = null; if (this.handleSessionFound) { this.session.removeSessionFoundCallback(this.handleSessionFound); diff --git a/cli/src/claude/sdk/query.ts b/cli/src/claude/sdk/query.ts index 6f69a95a..e8d08431 100644 --- a/cli/src/claude/sdk/query.ts +++ b/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 diff --git a/cli/src/claude/sdk/types.ts b/cli/src/claude/sdk/types.ts index 3fb03fa5..0a7828e0 100644 --- a/cli/src/claude/sdk/types.ts +++ b/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 } diff --git a/cli/src/modules/pingPeer/pingPeer.test.ts b/cli/src/modules/pingPeer/pingPeer.test.ts index 2f1f89c8..23a26770 100644 --- a/cli/src/modules/pingPeer/pingPeer.test.ts +++ b/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 } } diff --git a/cli/src/modules/pingPeer/pingPeer.ts b/cli/src/modules/pingPeer/pingPeer.ts index e6a2d813..ec470f6c 100644 --- a/cli/src/modules/pingPeer/pingPeer.ts +++ b/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' diff --git a/docs/api/client-contract/pagination.md b/docs/api/client-contract/pagination.md index 024ce7d5..21f462e0 100644 --- a/docs/api/client-contract/pagination.md +++ b/docs/api/client-contract/pagination.md @@ -191,7 +191,7 @@ Response `{"ok": true}`. Sending to an inactive session returns `409 {"error":"S Other subscribers learn the same outcome via `message-cancelled` / `messages-consumed` SSE events. -**Steer a queued message into the current turn**: `POST /api/sessions/:id/messages/:messageId/steer` → `SteerQueuedMessageResponseSchema`. Unlike the send-time `deliveryMode` option above, this endpoint supports Pi, Codex, and Cursor ACP sessions (`isSteeringSupportedForSession` in `shared/src/modes.ts`). It rejects all scheduled messages, and rejects terminal-controlled sessions unless they advertise `concurrentClients`. +**Steer a queued message into the current turn**: `POST /api/sessions/:id/messages/:messageId/steer` → `SteerQueuedMessageResponseSchema`. Unlike the send-time `deliveryMode` option above, this endpoint supports Claude (remote mode), Pi, Codex, and Cursor ACP sessions (`isSteeringSupportedForSession` in `shared/src/modes.ts`). It rejects all scheduled messages, and rejects terminal-controlled sessions unless they advertise `concurrentClients`. | Response | Client action | |---|---| diff --git a/hub/src/sync/steerQueuedMessage.test.ts b/hub/src/sync/steerQueuedMessage.test.ts index 1704a959..a9ce0925 100644 --- a/hub/src/sync/steerQueuedMessage.test.ts +++ b/hub/src/sync/steerQueuedMessage.test.ts @@ -52,8 +52,8 @@ describe('SyncEngine.steerQueuedMessage', () => { const { store, engine } = createEngine() try { const session = engine.getOrCreateSession( - 'steer-claude', - { path: '/tmp/project', host: 'localhost', flavor: 'claude' }, + 'steer-opencode', + { path: '/tmp/project', host: 'localhost', flavor: 'opencode' }, { requests: {}, completedRequests: {} }, 'default' ) @@ -63,7 +63,7 @@ describe('SyncEngine.steerQueuedMessage', () => { expect(result).toEqual({ status: 'failed', - error: 'Steering is only supported for Pi, Codex, and Cursor ACP sessions', + error: 'Steering is only supported for Claude, Pi, Codex, and Cursor ACP sessions', localId: null }) } finally { diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index f30021bb..ba869a04 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -1228,7 +1228,7 @@ export class SyncEngine { /** * Ask the CLI to deliver one waiting-queue message into the active turn - * (native steer). Supported for Pi, Codex, and Cursor ACP sessions; the + * (native steer). Supported for Claude, Pi, Codex, and Cursor ACP sessions; the * CLI's `steer-queued-message` handler is registered per flavor. Legacy * stream-json Cursor sessions and other flavors are rejected by the * capability gate. @@ -1242,7 +1242,7 @@ export class SyncEngine { return { status: 'failed', error: 'Session not found', localId: null } } if (!isSteeringSupportedForSession(session.metadata)) { - return { status: 'failed', error: 'Steering is only supported for Pi, Codex, and Cursor ACP sessions', localId: null } + return { status: 'failed', error: 'Steering is only supported for Claude, Pi, Codex, and Cursor ACP sessions', localId: null } } if (session.agentState?.controlledByUser === true && !session.metadata?.capabilities?.concurrentClients) { return { status: 'failed', error: 'Steering is only available for remote sessions', localId: null } diff --git a/shared/src/modes.test.ts b/shared/src/modes.test.ts index bd11f668..8381ff7c 100644 --- a/shared/src/modes.test.ts +++ b/shared/src/modes.test.ts @@ -149,7 +149,7 @@ describe('isSteeringSupportedForFlavor', () => { expect(isSteeringSupportedForFlavor('codex')).toBe(true) expect(isSteeringSupportedForFlavor('cursor')).toBe(true) expect(isSteeringSupportedForFlavor('pi')).toBe(true) - expect(isSteeringSupportedForFlavor('claude')).toBe(false) + expect(isSteeringSupportedForFlavor('claude')).toBe(true) expect(isSteeringSupportedForFlavor('opencode')).toBe(false) expect(isSteeringSupportedForFlavor(undefined)).toBe(false) expect(isSteeringSupportedForFlavor(null)).toBe(false) @@ -160,6 +160,7 @@ describe('isSteeringSupportedForSession', () => { it('supports codex and pi sessions', () => { expect(isSteeringSupportedForSession({ flavor: 'codex' })).toBe(true) expect(isSteeringSupportedForSession({ flavor: 'pi' })).toBe(true) + expect(isSteeringSupportedForSession({ flavor: 'claude' })).toBe(true) }) it('supports Cursor ACP sessions', () => { @@ -184,7 +185,7 @@ describe('isSteeringSupportedForSession', () => { }) it('rejects non-steerable flavors', () => { - expect(isSteeringSupportedForSession({ flavor: 'claude' })).toBe(false) + expect(isSteeringSupportedForSession({ flavor: 'opencode' })).toBe(false) expect(isSteeringSupportedForSession(null)).toBe(false) }) }) diff --git a/shared/src/modes.ts b/shared/src/modes.ts index 8f5c0515..eba6964c 100644 --- a/shared/src/modes.ts +++ b/shared/src/modes.ts @@ -194,10 +194,12 @@ export function getCodexCollaborationModeOptions(): CodexCollaborationModeOption * - Cursor ACP: concurrent `session/prompt` soft-send (no cancel). Legacy * stream-json Cursor sessions are NOT steerable — gate with * {@link isSteeringSupportedForSession}. + * - Claude: stream-json stdin write during the turn (TUI Ctrl+X Ctrl+S); + * Claude folds it in at the next tool boundary. Remote mode only. * - * Claude / others: not supported (no reachable soft-steer path) — UI hides Steer. + * Others: not supported — UI hides Steer. */ -export const STEERING_SUPPORTED_FLAVORS = ['codex', 'cursor', 'pi'] as const +export const STEERING_SUPPORTED_FLAVORS = ['claude', 'codex', 'cursor', 'pi'] as const export function isSteeringSupportedForFlavor(flavor?: string | null): boolean { return (STEERING_SUPPORTED_FLAVORS as readonly string[]).includes(flavor ?? '') @@ -216,7 +218,7 @@ export function isSteeringSupportedForSession(metadata?: { cursorSessionId?: string | null cursorSessionProtocol?: 'acp' | 'stream-json' | null } | null): boolean { - if (metadata?.flavor === 'codex' || metadata?.flavor === 'pi') { + if (metadata?.flavor === 'claude' || metadata?.flavor === 'codex' || metadata?.flavor === 'pi') { return true } if (metadata?.flavor !== 'cursor') { diff --git a/shared/src/sessionCitation.ts b/shared/src/sessionCitation.ts index 2662403f..8af95c79 100644 --- a/shared/src/sessionCitation.ts +++ b/shared/src/sessionCitation.ts @@ -32,7 +32,7 @@ export const PEER_DELIVERY_MODE_DESCRIPTION = 'steer: use for important, time-sensitive findings the peer needs to know now, not after finishing its current turn ' + '(e.g. a correction that invalidates its assumptions, a blocker, or a result that changes its next action). ' + 'State why the finding matters now and what the peer should adjust. ' + - 'Steer requires insertion into an active Pi, Codex, or Cursor ACP turn; unsupported, idle, or inactive targets are rejected before sending. ' + + 'Steer requires insertion into an active Claude, Pi, Codex, or Cursor ACP turn; unsupported, idle, or inactive targets are rejected before sending. ' + 'Steering does not abort the current turn or guarantee an immediate response. ' + 'If send or steer is unconfirmed after submission, inspect the target before retrying; do not resend automatically.'