From 68d5b9206e9bc79a4522d6466debe2b9fa356b8f Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:44:40 -0700 Subject: [PATCH] fix(agent-prompt): stop reporting delivered prompts as stalled (#16095) (#16590) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(agent-prompt): stop reporting delivered prompts as stalled (#16095) Enter is written before verification runs, so `agent_prompt_stalled` can only ever mean "turn start not observed" — never "prompt not delivered". Three of the verifier's blind spots made that misreading routine, and the coordinator then treated it as non-delivery and pasted the whole preamble a second time into a worker already running it. - Accept a hook-reported `working` recorded after the baseline. Hook rows reach the runtime through getAgentStatusSnapshot with no window involved, unlike the synthetic-title route that feeds workingSequence (suppressed for codex, absent for kimi, and gated on window visibility for everyone else). - Accept pane output after Enter when the agent was already working: a `->working` edge is unreachable there, so the old predicate could never be satisfied by a follow-up prompt. An idle agent still owes a real turn start, so a swallowed Enter stays detectable. - Give codex/kimi panes a longer effect window; their only turn-start proof is an out-of-process hook round-trip, not a TUI repaint. - Coordinator dispatch no longer fails (and re-dispatches) a task whose prompt stalled; the dispatch stays active with its capability intact so the worker's own report settles it. * fix(orchestration): let a worker's own report correct an unobserved prompt (#16095) Follow-up to f9f973c on two review findings. Anchor the hook signal on a turn, not a refresh: `receivedAt`/`updatedAt` bump on every same-state hook ping, so an in-progress turn could have passed for a new one and silently accepted every prompt to a working agent. `stateStartedAt` is the documented per-turn identity (pinned across same-state pings), so the verifier now reads that. Close the worker-start path: a `dispatch_input` stall settled the dispatch as failed *and* revoked the capability, so a worker that ran the preamble to completion had its result rejected. Revocation is now skipped for that cause, and a worker report can re-settle a dispatch whose `last_failure` is `agent_prompt_stalled` — retaining the capability alone was not enough, because settlement also gates on dispatch/task status. * fix(orchestration): let a failed worker report correct a stalled-prompt record (#16095) The duplicate short-circuit ran before the unobserved-prompt branch, and for outcome 'failed' both expected statuses are exactly the state failWorkerStart leaves behind. A worker that reported a real failure was answered duplicate:true, so its cause and result body were dropped and the record kept 'agent_prompt_stalled'. Evaluate settledByUnobservedPrompt first so one failure report can re-settle that dispatch; a repeat report is still a duplicate. Also derive the previous dispatch/worker states once instead of two parallel ternaries, reuse getPtyAgent in createAgentPromptRenderGate, cite the real 30s relay request budget the hook window is sized against, and make the coordinator test settle an actual worker report rather than calling completeDispatch under a name that promised otherwise. * test(orchestration): carry the new dispatch-depth fields into this PR's fixtures main added required creator/maxDepth on createStartingWorkerDispatch and nestedWorkerMaxDepth on dispatchTaskToWorker while this branch was open. The two fixtures added here predate them, so the merge typechecked clean on each side and failed once combined. Mechanical; no behaviour asserted here changes. --- .../agent-prompt-submission-runtime.test.ts | 118 +++++++++++++-- ...ent-prompt-submission-verification.test.ts | 110 ++++++++++++++ .../agent-prompt-submission-verification.ts | 76 +++++++++- src/main/runtime/orca-runtime.ts | 32 ++++- ...dinator-dispatch-unobserved-prompt.test.ts | 134 ++++++++++++++++++ .../coordinator-task-dispatch.ts | 15 +- .../worker-report-settlement.ts | 43 ++++-- .../worker-dispatch-outcome.ts | 11 +- ...start-unobserved-prompt-settlement.test.ts | 102 +++++++++++++ ...orker-start-outcome-classification.test.ts | 32 +++++ ...ation-worker-start-prompt-contract.test.ts | 4 +- .../orchestration-worker-start-receipt.ts | 7 +- 12 files changed, 648 insertions(+), 36 deletions(-) create mode 100644 src/main/runtime/orchestration/coordinator-dispatch-unobserved-prompt.test.ts create mode 100644 src/main/runtime/orchestration/worker-start-unobserved-prompt-settlement.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-worker-start-outcome-classification.test.ts diff --git a/src/main/runtime/agent-prompt-submission-runtime.test.ts b/src/main/runtime/agent-prompt-submission-runtime.test.ts index 69eb0a01079..3b8fe67bea7 100644 --- a/src/main/runtime/agent-prompt-submission-runtime.test.ts +++ b/src/main/runtime/agent-prompt-submission-runtime.test.ts @@ -403,15 +403,7 @@ describe('agent prompt submission runtime', () => { it('does not treat an unchanged newer working status as submission evidence', async () => { vi.useFakeTimers() vi.setSystemTime(1_000) - const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => { - if (data === '\r') { - runtime.onPtyData( - 'pty-prompt', - '\x1b]9999;{"state":"working","agentType":"aider"}\x07', - Date.now() - ) - } - }) + const { runtime, handle, writes } = await createPromptRuntime(() => undefined) runtime.onPtyData('pty-prompt', '\x1b]0;Codex waiting for permission\x07', Date.now()) vi.setSystemTime(2_000) runtime.onPtyData( @@ -428,6 +420,114 @@ describe('agent prompt submission runtime', () => { expect(writes.filter((data) => data === '\r')).toHaveLength(1) }) + // Why (#16095): a still-working agent can never produce a `→working` edge, so the old predicate + // was unsatisfiable for every follow-up prompt; pane output after Enter is the evidence left. + it('accepts pane output after Enter while the agent is already working', async () => { + vi.useFakeTimers() + vi.setSystemTime(1_000) + const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => { + if (data === '\r') { + runtime.onPtyData('pty-prompt', 'queued for the current turn', Date.now()) + } + }) + runtime.onPtyData( + 'pty-prompt', + '\x1b]9999;{"state":"working","agentType":"aider"}\x07', + Date.now() + ) + + const submission = runtime.sendTerminalAgentPrompt(handle, 'review this') + await vi.runAllTimersAsync() + + await expect(submission).resolves.toMatchObject({ accepted: true }) + expect(writes.filter((data) => data === '\r')).toHaveLength(1) + }) + + // Why: hook rows reach the runtime through this provider, which has no window and no OSC title — + // the same path a headless `orca serve` host and a minimized desktop window take. + async function createHookOnlyPromptRuntime(hook: { + state: 'done' | 'working' + stateStartedAt: number + }): Promise<{ runtime: OrcaRuntimeService; handle: string; writes: string[] }> { + let handle = '' + const writes: string[] = [] + const runtime = new OrcaRuntimeService(makeStore() as never, undefined, { + getAgentStatusSnapshot: () => [ + { + paneKey: 'prompt-pane', + terminalHandle: handle, + state: hook.state, + prompt: '', + agentType: 'kimi', + connectionId: null, + // Why: every hook ping refreshes receivedAt, including same-state tool pings. + receivedAt: Date.now(), + stateStartedAt: hook.stateStartedAt + } + ] + }) + runtime.setPtyController({ + spawn: vi.fn().mockResolvedValue({ id: 'pty-prompt' }), + write: (_ptyId, data) => { + writes.push(data) + return true + }, + kill: () => true, + getForegroundProcess: async () => null + }) + handle = ( + await runtime.createTerminal(`path:${AGENT_PROMPT_TEST_WORKTREE_PATH}`, { + launchAgent: 'kimi' + }) + ).handle + return { runtime, handle, writes } + } + + it('accepts a hook working status with no window and no title coverage', async () => { + vi.useFakeTimers() + vi.setSystemTime(1_000) + const hook = { state: 'done' as 'done' | 'working', stateStartedAt: 1_000 } + const { runtime, handle, writes } = await createHookOnlyPromptRuntime(hook) + runtime.setPtyController({ + spawn: vi.fn().mockResolvedValue({ id: 'pty-prompt' }), + write: (_ptyId, data) => { + writes.push(data) + if (data === '\r') { + vi.setSystemTime(3_000) + hook.state = 'working' + hook.stateStartedAt = 3_000 + } + return true + }, + kill: () => true, + getForegroundProcess: async () => null + }) + + const submission = runtime.sendTerminalAgentPrompt(handle, 'review this') + await vi.runAllTimersAsync() + + await expect(submission).resolves.toMatchObject({ accepted: true }) + expect(writes.filter((data) => data === '\r')).toHaveLength(1) + }) + + // Why: same-state pings keep refreshing receivedAt on a turn that started before the prompt; + // only the pinned stateStartedAt separates that from a turn this prompt started. + it('does not accept a hook row refreshed without a new working turn', async () => { + vi.useFakeTimers() + vi.setSystemTime(1_000) + const { runtime, handle, writes } = await createHookOnlyPromptRuntime({ + state: 'working', + stateStartedAt: 1_000 + }) + + const submission = runtime.sendTerminalAgentPrompt(handle, 'review this') + const rejected = expect(submission).rejects.toThrow('agent_prompt_stalled') + await vi.runAllTimersAsync() + + await rejected + expect(writes.filter((data) => data === '\r')).toHaveLength(1) + }) + it('does not write Enter after the PTY generation changes during settlement', async () => { vi.useFakeTimers() const { runtime, handle, writes } = await createPromptRuntime(() => undefined) diff --git a/src/main/runtime/agent-prompt-submission-verification.test.ts b/src/main/runtime/agent-prompt-submission-verification.test.ts index a0f3e1a5213..4a802df8d33 100644 --- a/src/main/runtime/agent-prompt-submission-verification.test.ts +++ b/src/main/runtime/agent-prompt-submission-verification.test.ts @@ -1,7 +1,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { AGENT_PROMPT_EFFECT_TIMEOUT_MS, + AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS, type AgentPromptActivity, + isAgentPromptStalledError, + resolveAgentPromptEffectTimeoutMs, verifyAgentPromptSubmission } from './agent-prompt-submission-verification' @@ -10,6 +13,8 @@ function activity(overrides: Partial = {}): AgentPromptActi generation: 1, permissionSequence: 2, workingSequence: 4, + explicitWorkingStartedAt: null, + outputSequence: 7, status: 'idle', ...overrides } @@ -127,6 +132,111 @@ describe('agent prompt submission verification', () => { await rejected }) + it('accepts a hook working status recorded after the baseline', async () => { + vi.useFakeTimers() + let current = activity() + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current + }) + + // No workingSequence edge: the window-gated synthetic title never ran (hidden window/headless). + current = activity({ explicitWorkingStartedAt: 2_000, status: 'working' }) + await vi.advanceTimersByTimeAsync(50) + + await expect(verification).resolves.toBeUndefined() + }) + + it('does not accept a hook working status that predates the baseline', async () => { + vi.useFakeTimers() + const current = activity({ explicitWorkingStartedAt: 2_000, status: 'working' }) + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current + }) + const rejected = expect(verification).rejects.toThrow('agent_prompt_stalled') + + await vi.advanceTimersByTimeAsync(AGENT_PROMPT_EFFECT_TIMEOUT_MS) + + await rejected + }) + + // Why: same-state hook pings refresh the row without starting a turn, so only the pinned + // stateStartedAt may satisfy the check — a refreshed row must stay unproven. + it('does not accept a refreshed hook row whose working turn did not restart', async () => { + vi.useFakeTimers() + let current = activity({ explicitWorkingStartedAt: 2_000 }) + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current + }) + const rejected = expect(verification).rejects.toThrow('agent_prompt_stalled') + + current = activity({ explicitWorkingStartedAt: 2_000, outputSequence: 40 }) + await vi.advanceTimersByTimeAsync(AGENT_PROMPT_EFFECT_TIMEOUT_MS) + + await rejected + }) + + it('accepts pane output after Enter when the agent was already working', async () => { + vi.useFakeTimers() + let current = activity({ status: 'working' }) + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current + }) + + current = activity({ status: 'working', outputSequence: 8 }) + await vi.advanceTimersByTimeAsync(50) + + await expect(verification).resolves.toBeUndefined() + }) + + it('does not accept pane output when the agent was idle at submit', async () => { + vi.useFakeTimers() + let current = activity() + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current + }) + const rejected = expect(verification).rejects.toThrow('agent_prompt_stalled') + + current = activity({ outputSequence: 9 }) + await vi.advanceTimersByTimeAsync(AGENT_PROMPT_EFFECT_TIMEOUT_MS) + + await rejected + }) + + it('holds the longer hook window open past the default timeout', async () => { + vi.useFakeTimers() + let current = activity() + const verification = verifyAgentPromptSubmission({ + baseline: current, + readActivity: () => current, + timeoutMs: AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS + }) + + await vi.advanceTimersByTimeAsync(AGENT_PROMPT_EFFECT_TIMEOUT_MS + 1_000) + current = activity({ explicitWorkingStartedAt: 9_000, status: 'working' }) + await vi.advanceTimersByTimeAsync(50) + + await expect(verification).resolves.toBeUndefined() + }) + + it('gives hook-observed agents the longer effect window', () => { + expect(resolveAgentPromptEffectTimeoutMs('codex')).toBe(AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS) + expect(resolveAgentPromptEffectTimeoutMs('kimi')).toBe(AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS) + expect(resolveAgentPromptEffectTimeoutMs('claude')).toBe(AGENT_PROMPT_EFFECT_TIMEOUT_MS) + expect(resolveAgentPromptEffectTimeoutMs(null)).toBe(AGENT_PROMPT_EFFECT_TIMEOUT_MS) + }) + + it('recognizes a stalled verdict from a message or a relayed error code', () => { + expect(isAgentPromptStalledError(new Error('agent_prompt_stalled'))).toBe(true) + expect(isAgentPromptStalledError({ code: 'agent_prompt_stalled' })).toBe(true) + expect(isAgentPromptStalledError(new Error('terminal_not_writable'))).toBe(false) + expect(isAgentPromptStalledError(null)).toBe(false) + }) + it('rejects a replaced terminal generation', async () => { const baseline = activity() diff --git a/src/main/runtime/agent-prompt-submission-verification.ts b/src/main/runtime/agent-prompt-submission-verification.ts index 41b5f5b1732..e8373bbaf1f 100644 --- a/src/main/runtime/agent-prompt-submission-verification.ts +++ b/src/main/runtime/agent-prompt-submission-verification.ts @@ -1,31 +1,68 @@ +import type { TuiAgent } from '../../shared/tui-agent' + export const AGENT_PROMPT_EFFECT_TIMEOUT_MS = 5_000 +// Why: these panes prove a turn start only through the out-of-process hook — kimi has no synthetic +// title profile and codex suppresses the hook-driven working frame (synthesizeWorkingTitle: false), +// so the first proof lags Enter by agent startup, not by one TUI repaint. Capped so the worst case +// (8s render gate + this wait + chunked paste) still fits RELAY_TO_CLIENT_REQUEST_TIMEOUT_MS +// (30s, src/relay/dispatcher.ts), the budget a paired client's submission runs under. +export const AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS = 15_000 const AGENT_PROMPT_EFFECT_POLL_MS = 50 +const HOOK_OBSERVED_TURN_START_AGENTS = new Set(['codex', 'kimi']) + +/** The prompt bytes are written before verification, so this only ever means "not observed". */ +export const AGENT_PROMPT_STALLED_ERROR = 'agent_prompt_stalled' + export type AgentPromptActivity = Readonly<{ generation: number permissionSequence: number workingSequence: number + /** When the hook's current `working` turn began; reaches the runtime with no window and no + * title coverage. Pinned across same-state pings, so a refresh alone cannot move it. */ + explicitWorkingStartedAt: number | null + /** PTY bytes seen on this pane; delivery evidence when a turn-start edge cannot be observed. */ + outputSequence: number status: 'working' | 'permission' | 'idle' | null }> type AgentPromptVerificationOptions = { baseline: AgentPromptActivity readActivity: () => AgentPromptActivity + timeoutMs?: number signal?: AbortSignal } +export function resolveAgentPromptEffectTimeoutMs(agent: TuiAgent | null | undefined): number { + return agent && HOOK_OBSERVED_TURN_START_AGENTS.has(agent) + ? AGENT_PROMPT_HOOK_EFFECT_TIMEOUT_MS + : AGENT_PROMPT_EFFECT_TIMEOUT_MS +} + +export function isAgentPromptStalledError(error: unknown): boolean { + if (error instanceof Error && error.message === AGENT_PROMPT_STALLED_ERROR) { + return true + } + // Why: a relayed submission surfaces the same verdict as an RPC error code, not a message. + return ( + typeof error === 'object' && + error !== null && + (error as { code?: unknown }).code === AGENT_PROMPT_STALLED_ERROR + ) +} + export async function verifyAgentPromptSubmission( options: AgentPromptVerificationOptions ): Promise { throwIfAgentPromptAborted(options.signal) assertPromptNotBlocked(options.baseline, options.baseline) - const deadline = Date.now() + AGENT_PROMPT_EFFECT_TIMEOUT_MS + const deadline = Date.now() + (options.timeoutMs ?? AGENT_PROMPT_EFFECT_TIMEOUT_MS) while (Date.now() < deadline) { const current = options.readActivity() assertSamePromptGeneration(options.baseline, current) assertPromptNotBlocked(options.baseline, current) - if (agentPromptLifecycleChanged(options.baseline, current)) { + if (agentPromptEffectObserved(options.baseline, current)) { return } await waitForAgentPromptPoll(options.signal) @@ -34,17 +71,44 @@ export async function verifyAgentPromptSubmission( const current = options.readActivity() assertSamePromptGeneration(options.baseline, current) assertPromptNotBlocked(options.baseline, current) - if (agentPromptLifecycleChanged(options.baseline, current)) { + if (agentPromptEffectObserved(options.baseline, current)) { return } - throw new Error('agent_prompt_stalled') + throw new Error(AGENT_PROMPT_STALLED_ERROR) } -function agentPromptLifecycleChanged( +function agentPromptEffectObserved( baseline: AgentPromptActivity, current: AgentPromptActivity ): boolean { - return current.workingSequence > baseline.workingSequence + return ( + current.workingSequence > baseline.workingSequence || + observedHookWorkingAfterBaseline(baseline, current) || + observedDeliveryEvidence(baseline, current) + ) +} + +// Why: hook status reaches the runtime directly, so it survives a hidden window and headless serve — +// the synthetic-title route that feeds workingSequence does not (#16095). Only a turn that started +// after the baseline counts, so a same-state ping on the turn already running is not evidence. +function observedHookWorkingAfterBaseline( + baseline: AgentPromptActivity, + current: AgentPromptActivity +): boolean { + return ( + current.explicitWorkingStartedAt !== null && + current.explicitWorkingStartedAt > (baseline.explicitWorkingStartedAt ?? 0) + ) +} + +// Why: a `→working` edge is unreachable for an agent that is already working, so the honest proof +// that the prompt landed is the pane emitting bytes after Enter. An idle agent still owes a real +// turn start, which keeps a swallowed Enter detectable. +function observedDeliveryEvidence( + baseline: AgentPromptActivity, + current: AgentPromptActivity +): boolean { + return baseline.status === 'working' && current.outputSequence > baseline.outputSequence } function assertSamePromptGeneration( diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 71c71b7291d..54c6b711173 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -106,6 +106,7 @@ import { } from '../../shared/agent-prompt-injection' import { type AgentPromptActivity, + resolveAgentPromptEffectTimeoutMs, verifyAgentPromptSubmission } from './agent-prompt-submission-verification' import { @@ -19780,16 +19781,20 @@ export class OrcaRuntimeService { private getFreshExplicitAgentStatusForHandle(handle: string): { status: NonNullable updatedAt: number + /** When this state was entered. Pinned across same-state pings, so it identifies the turn. */ + stateStartedAt: number } | null { const paneKey = this.getPaneKeyForTerminalHandle(handle) const now = Date.now() let bestStatus: NonNullable | null = null let bestUpdatedAt = -1 + let bestStateStartedAt = -1 const consider = ( state: AgentStatusEntry['state'] | undefined, updatedAt: number | null | undefined, - restoredUnconfirmed = false + restoredUnconfirmed = false, + stateStartedAt?: number | null ): void => { if (!state || restoredUnconfirmed) { return @@ -19803,22 +19808,25 @@ export class OrcaRuntimeService { if (updatedAt > bestUpdatedAt || (updatedAt === bestUpdatedAt && status === 'permission')) { bestStatus = status bestUpdatedAt = updatedAt + bestStateStartedAt = typeof stateStartedAt === 'number' ? stateStartedAt : updatedAt } } if (paneKey) { const retained = this.latestAgentStatusByPaneKey.get(paneKey) - consider(retained?.payload.state, retained?.updatedAt) + consider(retained?.payload.state, retained?.updatedAt, false, retained?.stateStartedAt) } for (const entry of this.getAgentStatusSnapshotFn?.() ?? []) { if (entry.terminalHandle !== handle && (!paneKey || entry.paneKey !== paneKey)) { continue } - consider(entry.state, entry.receivedAt, entry.restoredUnconfirmed) + consider(entry.state, entry.receivedAt, entry.restoredUnconfirmed, entry.stateStartedAt) } - return bestStatus ? { status: bestStatus, updatedAt: bestUpdatedAt } : null + return bestStatus + ? { status: bestStatus, updatedAt: bestUpdatedAt, stateStartedAt: bestStateStartedAt } + : null } private async writeTerminalAction( @@ -20025,6 +20033,7 @@ export class OrcaRuntimeService { await verifyAgentPromptSubmission({ baseline, readActivity: () => this.getAgentPromptActivity(handle, ptyId), + timeoutMs: resolveAgentPromptEffectTimeoutMs(this.getPtyAgent(ptyId)), signal: options.signal }) return 1 @@ -20081,10 +20090,21 @@ export class OrcaRuntimeService { generation: this.getPtyLifecycleGeneration(ptyId), permissionSequence: this.agentPromptPermissionSequenceByPtyId.get(ptyId) ?? 0, workingSequence: lifecycle?.workingSequence ?? 0, + // Why: hook status is the only turn-start signal agents without title coverage have, and it + // reaches here without the window-gated synthetic title frame (#16095). Anchored on + // stateStartedAt, not updatedAt — same-state tool/prompt pings refresh updatedAt and would + // otherwise pass off an in-progress turn as a new one. + explicitWorkingStartedAt: explicit?.status === 'working' ? explicit.stateStartedAt : null, + outputSequence: this.getPtyOutputSequence(ptyId), status } } + private getPtyAgent(ptyId: string): TuiAgent | null { + const pty = this.ptysById.get(ptyId) + return pty?.launchAgent ?? pty?.foregroundAgent ?? null + } + private assertAgentPromptPermissionSafe( baseline: AgentPromptActivity, current: AgentPromptActivity @@ -20114,9 +20134,7 @@ export class OrcaRuntimeService { wait: () => Promise dispose: () => void } | null { - const pty = this.ptysById.get(ptyId) - const agent = pty?.launchAgent ?? pty?.foregroundAgent - if (!isTerminalSendSettlementAgent(agent)) { + if (!isTerminalSendSettlementAgent(this.getPtyAgent(ptyId))) { return null } let armed = false diff --git a/src/main/runtime/orchestration/coordinator-dispatch-unobserved-prompt.test.ts b/src/main/runtime/orchestration/coordinator-dispatch-unobserved-prompt.test.ts new file mode 100644 index 00000000000..5f99e283de3 --- /dev/null +++ b/src/main/runtime/orchestration/coordinator-dispatch-unobserved-prompt.test.ts @@ -0,0 +1,134 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' +import type { CoordinatorRuntime } from './coordinator-runtime-contract' +import { dispatchTaskToWorker } from './coordinator-task-dispatch' + +const WORKER_PANE_KEY = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' +let db: OrchestrationDb + +function createRuntime(promptError: Error | null): CoordinatorRuntime & { prompts: string[] } { + const prompts: string[] = [] + return { + prompts, + async sendTerminalAgentPrompt(_handle: string, prompt: string) { + prompts.push(prompt) + if (promptError) { + throw promptError + } + return { accepted: true } + }, + async listTerminals() { + return { terminals: [] } + }, + async createTerminal() { + return { handle: 'term_a', worktreeId: 'wt1' } + }, + async waitForTerminal(handle: string) { + return { handle, condition: 'exit' } + }, + async probeWorktreeDrift() { + return null + }, + getTerminalPaneKey() { + return WORKER_PANE_KEY + }, + getOrchestrationDispatchAuthority() { + return { + paneKey: WORKER_PANE_KEY, + processIncarnation: 'incarnation-1', + launchTokenHash: null + } + } + } +} + +async function dispatch( + runtime: CoordinatorRuntime, + taskId: string, + logs: string[] +): Promise { + return dispatchTaskToWorker({ + db, + runtime, + task: db.getTask(taskId)!, + targetHandle: 'term_a', + nestedWorkerMaxDepth: Number.MAX_SAFE_INTEGER, + baseDrift: null, + coordinatorHandle: 'coord', + worktree: undefined, + onLog: (message) => logs.push(message), + onCircuitBroken: () => undefined + }) +} + +describe('coordinator dispatch with an unobserved prompt', () => { + afterEach(() => db?.close()) + + it('never re-pastes a preamble whose turn start was not observed', async () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'do the work' }) + const runtime = createRuntime(new Error('agent_prompt_stalled')) + const logs: string[] = [] + + const result = await dispatch(runtime, task.id, logs) + + expect(result).toBe('dispatched-unobserved') + expect(runtime.prompts).toHaveLength(1) + // The task stays dispatched, so the next coordinator tick cannot pick it up again. + expect(db.getTask(task.id)?.status).toBe('dispatched') + const ctx = db.getDispatchContext(task.id) + expect(ctx).toMatchObject({ + status: 'dispatched', + failure_count: 0, + capability_revoked_at: null + }) + expect(db.listTasks({ status: 'ready' })).toEqual([]) + expect(logs.join('\n')).toContain('turn start was not observed') + }) + + it('lets a late worker report settle a dispatch whose prompt was unobserved', async () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'do the work' }) + await dispatch(createRuntime(new Error('agent_prompt_stalled')), task.id, []) + const dispatchId = db.getDispatchContext(task.id)!.id + const minted = db.mintDispatchCapability({ + dispatchId, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'incarnation-1' + }) + + expect( + db.verifyDispatchCapability({ + dispatchId, + capability: minted, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'incarnation-1' + }) + ).toEqual({ valid: true }) + expect( + db.settleWorkerReport({ + taskId: task.id, + dispatchId, + outcome: 'succeeded', + result: 'done the work' + }) + ).toEqual({ action: 'settled', outcome: 'succeeded', duplicate: false }) + expect(db.getTask(task.id)).toMatchObject({ status: 'completed', result: 'done the work' }) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed') + }) + + it('still fails the dispatch when the prompt was never delivered', async () => { + db = new OrchestrationDb(':memory:') + const task = db.createTask({ spec: 'do the work' }) + const runtime = createRuntime(new Error('terminal_not_writable')) + + await expect(dispatch(runtime, task.id, [])).rejects.toThrow('terminal_not_writable') + + expect(db.getTask(task.id)?.status).toBe('ready') + expect(db.getDispatchContext(task.id)).toMatchObject({ + status: 'failed', + failure_count: 1, + last_failure: 'terminal_not_writable' + }) + }) +}) diff --git a/src/main/runtime/orchestration/coordinator-task-dispatch.ts b/src/main/runtime/orchestration/coordinator-task-dispatch.ts index 8dae105726a..685552cc2e9 100644 --- a/src/main/runtime/orchestration/coordinator-task-dispatch.ts +++ b/src/main/runtime/orchestration/coordinator-task-dispatch.ts @@ -7,8 +7,10 @@ import { DISPATCH_STALE_THRESHOLD, parseAllowStaleBaseFromSpec } from './coordinator-stale-base-flag' +import { isAgentPromptStalledError } from '../agent-prompt-submission-verification' -export type TaskDispatchResult = 'dispatched' | 'stale-base-refused' +/** `dispatched-unobserved`: the preamble landed but the worker's turn start was never observed. */ +export type TaskDispatchResult = 'dispatched' | 'dispatched-unobserved' | 'stale-base-refused' // Why: 10 min = documented heartbeat cadence (5 min) × 2, so one missed heartbeat is the earliest a dispatch can look stale. const HUNG_THRESHOLD_MS = 10 * 60 * 1000 @@ -136,6 +138,17 @@ export async function dispatchTaskToWorker(params: { try { await runtime.sendTerminalAgentPrompt(targetHandle, preamble + gateContext) } catch (err) { + // Why (#16095): Enter is written before submission is verified, so a stall is only ever an + // unobserved turn start — never proof the preamble is missing. Failing here would reset the + // task to 'ready' and paste the whole preamble a second time into a worker already running it, + // and would revoke the capability its worker_done needs. + if (isAgentPromptStalledError(err)) { + onLog( + `Dispatched task ${task.id} to ${targetHandle}; turn start was not observed. ` + + `The preamble is already in the pane, so the dispatch stays active instead of being resent.` + ) + return 'dispatched-unobserved' + } const updated = db.failDispatch(dispatch.id, err instanceof Error ? err.message : String(err)) if (updated?.status === 'circuit_broken') { params.onCircuitBroken(task.id) diff --git a/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts b/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts index 734c25b1264..9d6b643c070 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/worker-report-settlement.ts @@ -1,5 +1,6 @@ import type { WorkerReportOutcome, WorkerReportSettlement } from '../../types' import type { OrchestrationDb } from '../orchestration-db' +import { AGENT_PROMPT_STALLED_ERROR } from '../../../agent-prompt-submission-verification' import { settleActiveDispatchesForTask } from './dispatch-completion' import { getActiveDispatchForTask } from './task-dispatch-reconciliation' @@ -54,10 +55,26 @@ export function settleWorkerReportInTransaction( const expectedDispatchStatus = params.outcome === 'succeeded' ? 'completed' : 'failed' const expectedTaskStatus = params.outcome === 'succeeded' ? 'completed' : 'failed' - if (dispatch.status === expectedDispatchStatus && task.status === expectedTaskStatus) { + // Why (#16095): worker-start records a stalled prompt as failed, but the preamble was written + // before verification ran — the worker may have been executing it the whole time. Its own report + // is first-hand evidence and must be able to correct that record instead of being thrown away. + // Checked before the duplicate short-circuit: a `failed` report lands on the very statuses that + // short-circuit reads as already settled, dropping the worker's real cause and result body. + const settledByUnobservedPrompt = + dispatch.status === 'failed' && + dispatch.last_failure === AGENT_PROMPT_STALLED_ERROR && + task.status === 'failed' + if ( + !settledByUnobservedPrompt && + dispatch.status === expectedDispatchStatus && + task.status === expectedTaskStatus + ) { return { action: 'settled', outcome: params.outcome, duplicate: true } } - if (dispatch.status !== 'dispatched' || task.status !== 'dispatched') { + const previous = settledByUnobservedPrompt + ? { status: 'failed', workerState: 'failed' } + : { status: 'dispatched', workerState: 'ready' } + if (dispatch.status !== previous.status || task.status !== previous.status) { return { action: 'rejected', code: 'inactive_dispatch', @@ -105,16 +122,22 @@ export function settleWorkerReportInTransaction( SET status = ?, completed_at = datetime('now'), last_failure = CASE WHEN ? = 'failed' THEN ? ELSE last_failure END, capability_revoked_at = COALESCE(capability_revoked_at, datetime('now')) - WHERE id = ? AND status = 'dispatched'` + WHERE id = ? AND status = ?` + ) + .run( + expectedDispatchStatus, + expectedDispatchStatus, + params.result, + params.dispatchId, + previous.status ) - .run(expectedDispatchStatus, expectedDispatchStatus, params.result, params.dispatchId) const taskUpdate = this.db .prepare( `UPDATE tasks SET status = ?, result = ?, completed_at = datetime('now') - WHERE id = ? AND status = 'dispatched'` + WHERE id = ? AND status = ?` ) - .run(expectedTaskStatus, params.result, params.taskId) + .run(expectedTaskStatus, params.result, params.taskId, previous.status) if (dispatchUpdate.changes !== 1 || taskUpdate.changes !== 1) { this.db.exec('ROLLBACK TO settle_worker_report') this.db.exec('RELEASE settle_worker_report') @@ -128,9 +151,13 @@ export function settleWorkerReportInTransaction( .prepare( `UPDATE worker_dispatches SET state = ?, stage = 'settled', updated_at = datetime('now') - WHERE dispatch_id = ? AND state = 'ready'` + WHERE dispatch_id = ? AND state = ?` + ) + .run( + params.outcome === 'succeeded' ? 'succeeded' : 'failed', + params.dispatchId, + previous.workerState ) - .run(params.outcome === 'succeeded' ? 'succeeded' : 'failed', params.dispatchId) settleActiveDispatchesForTask( this, params.taskId, diff --git a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-outcome.ts b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-outcome.ts index d9a8b23a39b..3ff796c097d 100644 --- a/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-outcome.ts +++ b/src/main/runtime/orchestration/db/worker-dispatch/worker-dispatch-outcome.ts @@ -37,7 +37,11 @@ export function failWorkerStart( this: OrchestrationDb, dispatchId: string, stage: string, - reason: string + reason: string, + // Why (#16095): revocation exists to stop a worker acting on a dispatch that never landed. A + // prompt whose turn start went unobserved provably landed, so its worker keeps the authority its + // own report needs. + options: { retainCapability?: boolean } = {} ): WorkerDispatchRow { this.db.exec('BEGIN IMMEDIATE') try { @@ -50,10 +54,11 @@ export function failWorkerStart( .prepare( `UPDATE dispatch_contexts SET status = 'failed', last_failure = ?, completed_at = datetime('now'), - capability_revoked_at = COALESCE(capability_revoked_at, datetime('now')) + capability_revoked_at = CASE WHEN ? = 1 THEN capability_revoked_at + ELSE COALESCE(capability_revoked_at, datetime('now')) END WHERE id = ?` ) - .run(reason, dispatchId) + .run(reason, options.retainCapability ? 1 : 0, dispatchId) this.db .prepare( `UPDATE worker_dispatches diff --git a/src/main/runtime/orchestration/worker-start-unobserved-prompt-settlement.test.ts b/src/main/runtime/orchestration/worker-start-unobserved-prompt-settlement.test.ts new file mode 100644 index 00000000000..35cfb94b74c --- /dev/null +++ b/src/main/runtime/orchestration/worker-start-unobserved-prompt-settlement.test.ts @@ -0,0 +1,102 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { OrchestrationDb } from './db' + +const WORKER_PANE_KEY = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' +const INCARNATION = 'runtime_test:term_worker:1' +let db: OrchestrationDb + +function startWorker(spec: string): { taskId: string; dispatchId: string; capability: string } { + const task = db.createTask({ spec }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: {} + }) + const capability = db.mintDispatchCapability({ + dispatchId: started.dispatch.id, + paneKey: WORKER_PANE_KEY, + processIncarnation: INCARNATION + }) + return { taskId: task.id, dispatchId: started.dispatch.id, capability } +} + +function verify(dispatchId: string, capability: string): { valid: boolean; reason?: string } { + return db.verifyDispatchCapability({ + dispatchId, + capability, + paneKey: WORKER_PANE_KEY, + processIncarnation: INCARNATION + }) +} + +describe('worker start settled by an unobserved prompt', () => { + afterEach(() => db?.close()) + + it('keeps the capability and lets the worker report correct the record', () => { + db = new OrchestrationDb(':memory:') + const { taskId, dispatchId, capability } = startWorker('run to completion') + + db.failWorkerStart(dispatchId, 'dispatch_input', 'agent_prompt_stalled', { + retainCapability: true + }) + + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + status: 'failed', + last_failure: 'agent_prompt_stalled', + capability_revoked_at: null + }) + expect(verify(dispatchId, capability)).toEqual({ valid: true }) + + expect( + db.settleWorkerReport({ + taskId, + dispatchId, + outcome: 'succeeded', + result: 'done the work' + }) + ).toEqual({ action: 'settled', outcome: 'succeeded', duplicate: false }) + expect(db.getTask(taskId)).toMatchObject({ status: 'completed', result: 'done the work' }) + expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed') + expect(db.getWorkerDispatch(dispatchId)).toMatchObject({ state: 'succeeded', stage: 'settled' }) + }) + + it('revokes and stays settled when the start failed for any other cause', () => { + db = new OrchestrationDb(':memory:') + const { taskId, dispatchId, capability } = startWorker('never became ready') + + db.failWorkerStart(dispatchId, 'agent_readiness', 'Agent did not become ready (idle).') + + expect(db.getDispatchContextById(dispatchId)?.capability_revoked_at).toEqual(expect.any(String)) + expect(verify(dispatchId, capability).valid).toBe(false) + expect( + db.settleWorkerReport({ taskId, dispatchId, outcome: 'succeeded', result: 'done' }) + ).toMatchObject({ action: 'rejected', code: 'inactive_dispatch' }) + expect(db.getTask(taskId)?.status).toBe('failed') + }) + + it('lets a failure report replace the unobserved-prompt cause with the real one', () => { + db = new OrchestrationDb(':memory:') + const { taskId, dispatchId } = startWorker('reports its own failure') + + db.failWorkerStart(dispatchId, 'dispatch_input', 'agent_prompt_stalled', { + retainCapability: true + }) + + expect( + db.settleWorkerReport({ taskId, dispatchId, outcome: 'failed', result: 'build broke on X' }) + ).toEqual({ action: 'settled', outcome: 'failed', duplicate: false }) + expect(db.getTask(taskId)).toMatchObject({ status: 'failed', result: 'build broke on X' }) + expect(db.getDispatchContextById(dispatchId)).toMatchObject({ + status: 'failed', + last_failure: 'build broke on X' + }) + expect(db.getWorkerDispatch(dispatchId)).toMatchObject({ state: 'failed', stage: 'settled' }) + + // The stalled cause is gone, so a repeat report has nothing left to correct. + expect( + db.settleWorkerReport({ taskId, dispatchId, outcome: 'failed', result: 'again' }) + ).toEqual({ action: 'settled', outcome: 'failed', duplicate: true }) + expect(db.getTask(taskId)?.result).toBe('build broke on X') + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration-worker-start-outcome-classification.test.ts b/src/main/runtime/rpc/methods/orchestration-worker-start-outcome-classification.test.ts new file mode 100644 index 00000000000..112df925f3e --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-worker-start-outcome-classification.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest' +import { isUnknownWorkerStartOutcome } from './orchestration-worker-topology' + +describe('worker start outcome classification', () => { + it('treats an explicit operation_unknown code as unknown at any stage', () => { + const error = Object.assign(new Error('relay dropped'), { code: 'operation_unknown' }) + + expect(isUnknownWorkerStartOutcome(error, 'dispatch_input')).toBe(true) + expect(isUnknownWorkerStartOutcome(error, 'worktree_create')).toBe(true) + }) + + it('treats a lost connection during worktree create as unknown', () => { + expect(isUnknownWorkerStartOutcome(new Error('connection reset'), 'worktree_create')).toBe(true) + expect(isUnknownWorkerStartOutcome(new Error('request timed out'), 'worktree_create')).toBe( + true + ) + }) + + it('keeps a definite failure definite', () => { + expect(isUnknownWorkerStartOutcome(new Error('connection reset'), 'dispatch_input')).toBe(false) + expect(isUnknownWorkerStartOutcome(new Error('worktree exists'), 'worktree_create')).toBe(false) + }) + + // Why: a stalled prompt still reports a definite failure to the caller — the correction path is + // the worker's own report, which keeps its capability and can re-settle the dispatch (see + // worker-start-unobserved-prompt-settlement.test.ts), not an outcome_unknown receipt. + it('does not class a stalled dispatch prompt as unknown', () => { + expect(isUnknownWorkerStartOutcome(new Error('agent_prompt_stalled'), 'dispatch_input')).toBe( + false + ) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration-worker-start-prompt-contract.test.ts b/src/main/runtime/rpc/methods/orchestration-worker-start-prompt-contract.test.ts index 89455f9cc0a..826eb57ea55 100644 --- a/src/main/runtime/rpc/methods/orchestration-worker-start-prompt-contract.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-worker-start-prompt-contract.test.ts @@ -254,10 +254,12 @@ describe('orchestration worker-start prompt contract', () => { expect(harness.writes.filter((data) => data === '\r')).toHaveLength(1) const persisted = reopenPromptContractDb(harness) expect(persisted.getTask(harness.taskId)?.status).toBe('failed') + // Why (#16095): the receipt still reports the failure, but Enter was written before it was + // verified — so the capability survives and the worker's own report can correct the record. expect(persisted.getDispatchContextById(dispatchId)).toMatchObject({ status: 'failed', last_failure: 'agent_prompt_stalled', - capability_revoked_at: expect.any(String) + capability_revoked_at: null }) expect(persisted.getWorkerDispatch(dispatchId)).toMatchObject({ state: 'failed', diff --git a/src/main/runtime/rpc/methods/orchestration-worker-start-receipt.ts b/src/main/runtime/rpc/methods/orchestration-worker-start-receipt.ts index e58f9fff01e..cde2ea9a22f 100644 --- a/src/main/runtime/rpc/methods/orchestration-worker-start-receipt.ts +++ b/src/main/runtime/rpc/methods/orchestration-worker-start-receipt.ts @@ -1,4 +1,5 @@ import type { OrchestrationDb } from '../../orchestration/db' +import { isAgentPromptStalledError } from '../../agent-prompt-submission-verification' import { isUnknownWorkerStartOutcome, type WorkerSetupReceipt @@ -19,7 +20,11 @@ export function failWorkerStartWithReceipt(args: { const unknown = isUnknownWorkerStartOutcome(args.error, args.failedStage) const worker = unknown ? args.db.markWorkerStartUnknown(args.dispatchId, args.failedStage, reason) - : args.db.failWorkerStart(args.dispatchId, args.failedStage, reason) + : args.db.failWorkerStart(args.dispatchId, args.failedStage, reason, { + // Why (#16095): the preamble is written before submission is verified, so a stalled + // verdict never means the worker lacks its task — keep the authority its report needs. + retainCapability: isAgentPromptStalledError(args.error) + }) return { runId: args.runId, taskId: args.taskId,