diff --git a/src/main/runtime/orca-runtime-on-pty-data.ts b/src/main/runtime/orca-runtime-on-pty-data.ts index c3dfd7b2cac3..ffd89a645457 100644 --- a/src/main/runtime/orca-runtime-on-pty-data.ts +++ b/src/main/runtime/orca-runtime-on-pty-data.ts @@ -3,7 +3,7 @@ import { OrcaRuntimeWithPreparePtyExecutionContext } from './orca-runtime-prepar import type { TerminalOutputSourceRange } from '../../shared/terminal-output-source-range' import { advertisedUrlWatcher } from '../ports/advertised-url-watcher' import { appendNormalizedToTailBuffer } from './terminal-tail-buffer' -import { normalizeTerminalChunk } from './terminal-ansi-normalization' +import { normalizeTerminalChunk, parseAnsiControlSequence } from './terminal-ansi-normalization' import { observeTerminalCommandPaint } from './terminal-command-paint' import { appendCompletedTerminalTranscript, @@ -16,6 +16,73 @@ import { } from './terminal-wait-tail-state' import { extractOscTitleScanTail } from '../../shared/osc-title-scan-tail' +const CODEX_RECONNECT_FAILURE_BANNER = [ + 'Automatic reconnect could not restore this session', + 'app-server session could not be restored', + 'Reconnect failed — check the endpoint, then relaunch' +] +export const CODEX_RECONNECT_SCAN_CONTEXT_CHARS = 512 + +export type CodexReconnectOutputScan = { + tail: string + dispatchId: string | null + pendingDispatchId: string | null + handledDispatchId: string | null +} + +export function createCodexReconnectOutputScan(): CodexReconnectOutputScan { + return { tail: '', dispatchId: null, pendingDispatchId: null, handledDispatchId: null } +} + +export const codexReconnectOutputScans = new WeakMap() + +export function stripTerminalControlSequences(value: string): string { + const visibleParts: string[] = [] + let textStart = 0 + for (let index = 0; index < value.length; index += 1) { + if (value.charCodeAt(index) !== 0x1b) { + continue + } + visibleParts.push(value.slice(textStart, index)) + const sequence = parseAnsiControlSequence(value, index) + if (sequence) { + index = sequence.endIndex + } + textStart = index + 1 + } + visibleParts.push(value.slice(textStart)) + return visibleParts.join('') +} + +export function findCodexReconnectFailureBanner( + output: string +): { start: number; end: number } | undefined { + const lines: { text: string; start: number; end: number }[] = [] + let lineStart = 0 + for (let index = 0; index <= output.length; index += 1) { + const character = output[index] + if (index !== output.length && character !== '\r' && character !== '\n') { + continue + } + lines.push({ text: output.slice(lineStart, index).trim(), start: lineStart, end: index }) + if (character === '\r' && output[index + 1] === '\n') { + index += 1 + } + lineStart = index + 1 + } + for (let index = 0; index <= lines.length - CODEX_RECONNECT_FAILURE_BANNER.length; index += 1) { + const start = lines[index] + if ( + CODEX_RECONNECT_FAILURE_BANNER.every( + (expected, offset) => lines[index + offset].text === expected + ) + ) { + return { start: start.start, end: lines[index + 2].end } + } + } + return undefined +} + export class OrcaRuntimeWithOnPtyData extends OrcaRuntimeWithPreparePtyExecutionContext { /** Arrival-order mode scan: the settled tracker plus any in-flight snapshot capture's. */ protected scanProviderModeTrackers(ptyId: string, data: string): void { @@ -115,6 +182,12 @@ export class OrcaRuntimeWithOnPtyData extends OrcaRuntimeWithPreparePtyExecution pty.tailLinesTotal += nextTail.newCompleteLines pty.preview = buildPreview(pty.tailBuffer, pty.tailPartialLine) this.scheduleWaitBlockedCheck(ptyId, normalized.text, at) + const leaf = this.getLeavesForPty(ptyId)[0] + const handle = this.handleByPtyId.get(ptyId) ?? this.findHandleForPtyRecord(ptyId) + const paneKey = leaf ? this.makeRuntimePaneKey(leaf) : (pty.paneKey ?? null) + if (handle) { + this.observeCodexReconnectFailureOutput(handle, paneKey, pty, normalized.text) + } } for (const leaf of this.getLeavesForPty(ptyId)) { diff --git a/src/main/runtime/orca-runtime-subscribe-to-terminal-resize.ts b/src/main/runtime/orca-runtime-subscribe-to-terminal-resize.ts index 1cef2bbf23ab..747bf84e3287 100644 --- a/src/main/runtime/orca-runtime-subscribe-to-terminal-resize.ts +++ b/src/main/runtime/orca-runtime-subscribe-to-terminal-resize.ts @@ -22,6 +22,13 @@ import { classifyWorkerTerminalProcessIncarnation, parseWorkerTerminalHostScope } from './orchestration/worker-terminal-process-liveness' +import { + CODEX_RECONNECT_SCAN_CONTEXT_CHARS, + codexReconnectOutputScans, + createCodexReconnectOutputScan, + findCodexReconnectFailureBanner, + stripTerminalControlSequences +} from './orca-runtime-on-pty-data' import { getRepoIdFromWorktreeId } from '../../shared/worktree/id' import { buildOrchestrationTaskDisplayMetadata } from '../../shared/orchestration-task-display' @@ -92,12 +99,122 @@ export class OrcaRuntimeWithSubscribeToTerminalResize extends OrcaRuntimeWithApp return } - // Why: create an escalation message so the coordinator is notified about - // the unexpected exit on its next check cycle, even if the circuit breaker - // hasn't tripped yet. + this.notifyWorkerDispatchFailure({ + dispatch, + handle, + failureLogLabel: 'worker exit', + subject: `Agent exited unexpectedly (${errorContext})`, + reason: errorContext, + bodyPrefix: `Worker ${handle} stopped while running task`, + payload: { + taskId: dispatch.task_id, + dispatchId: dispatch.id, + exitCode, + exitCause: cause, + handle + }, + settledStatus: settled?.status + }) + } + + protected observeCodexReconnectFailureOutput( + handle: string, + paneKey: string | null, + pty: object, + output: string + ): void { + const db = this._orchestrationDb + const scan = codexReconnectOutputScans.get(pty) ?? createCodexReconnectOutputScan() + const dispatch = db?.getActiveDispatchForTerminal(handle, paneKey ?? undefined) + const dispatchId = dispatch?.id ?? null + if (dispatchId !== scan.dispatchId) { + scan.dispatchId = dispatchId + scan.tail = '' + scan.pendingDispatchId = null + scan.handledDispatchId = null + } + const combinedOutput = scan.tail + stripTerminalControlSequences(output) + if ( + dispatch && + scan.pendingDispatchId !== dispatchId && + scan.handledDispatchId !== dispatchId + ) { + const match = findCodexReconnectFailureBanner(combinedOutput) + if (match && match.end > scan.tail.length) { + scan.pendingDispatchId = dispatchId + } + } + if (dispatch && scan.pendingDispatchId === dispatchId) { + const result = this.failCodexSessionUnrecoverableDispatch(dispatch, handle) + if (result === 'settled' || result === 'ignored') { + scan.handledDispatchId = dispatchId + scan.pendingDispatchId = null + } + } + scan.tail = dispatch ? combinedOutput.slice(-CODEX_RECONNECT_SCAN_CONTEXT_CHARS) : '' + codexReconnectOutputScans.set(pty, scan) + } + + protected failCodexSessionUnrecoverableDispatch( + dispatch: { id: string; run_id: string; task_id: string }, + handle: string + ): 'settled' | 'retry' | 'ignored' { + const db = this._orchestrationDb + if (!db) { + return 'ignored' + } + const reason = 'Codex could not restore its app-server session' + const settledStatus = db.failCodexSessionUnrecoverableDispatch(dispatch.id, reason) + if (settledStatus === 'retry') { + return 'retry' + } + if (settledStatus !== 'failed' && settledStatus !== 'circuit_broken') { + return 'ignored' + } + this.notifyWorkerDispatchFailure({ + dispatch, + handle, + failureLogLabel: 'worker failure', + taskDispositionNote: + ' The task was marked failed pending coordinator review and will not be retried automatically.', + subject: `Agent session failed (${reason})`, + reason, + bodyPrefix: `Worker ${handle} could not continue task`, + bodySuffix: ' The terminal and worktree were kept for diagnosis.', + payload: { + taskId: dispatch.task_id, + dispatchId: dispatch.id, + handle, + failureKind: 'codex_session_unrecoverable', + terminalPreserved: true, + worktreePreserved: true + }, + settledStatus + }) + return 'settled' + } + + private notifyWorkerDispatchFailure(params: { + dispatch: { id: string; run_id: string; task_id: string } + handle: string + failureLogLabel: 'worker exit' | 'worker failure' + subject: string + reason: string + bodyPrefix: string + bodySuffix?: string + taskDispositionNote?: string + payload: Record + settledStatus?: string + }): void { + const db = this._orchestrationDb + if (!db) { + return + } try { - const owningRun = this._orchestrationDb.getRun?.(dispatch.run_id) - const active = this._orchestrationDb.getActiveCoordinatorRun?.() + // Why: a lightweight Run owns its own mailbox; legacy dispatches still fall back to the + // active coordinator address read by `orchestration check`. + const owningRun = db.getRun?.(params.dispatch.run_id) + const active = db.getActiveCoordinatorRun?.() const recipient = owningRun && owningRun.legacy !== 1 ? { to: `run:${owningRun.id}`, runId: owningRun.id } @@ -107,7 +224,7 @@ export class OrcaRuntimeWithSubscribeToTerminalResize extends OrcaRuntimeWithApp if (!recipient) { return } - const task = this._orchestrationDb.getTask?.(dispatch.task_id, dispatch.run_id) + const task = db.getTask?.(params.dispatch.task_id, params.dispatch.run_id) // Why: prefer the explicit task title and keep the derived one single-line and bounded; // a raw multi-paragraph spec inlined here breaks the coordinator's escalation banner. const title = @@ -118,28 +235,29 @@ export class OrcaRuntimeWithSubscribeToTerminalResize extends OrcaRuntimeWithApp displayName: task.display_name }).taskTitle : '' - const named = title ? `"${title}" (${dispatch.task_id})` : dispatch.task_id - const escalation = this._orchestrationDb.insertMessage({ - from: handle, + const named = title ? `"${title}" (${params.dispatch.task_id})` : params.dispatch.task_id + const settlementNote = + params.taskDispositionNote ?? + (params.settledStatus === 'circuit_broken' + ? ' This task has now failed too many times, so it will not be retried automatically.' + : params.settledStatus === 'failed' + ? ' The task is ready to be dispatched again.' + : '') + const escalation = db.insertMessage({ + from: params.handle, to: recipient.to, - subject: `Agent exited unexpectedly (${errorContext})`, - body: `Worker ${handle} stopped while running task ${named}. ${errorContext}.${settled?.status === 'circuit_broken' ? ' This task has now failed too many times, so it will not be retried automatically.' : settled?.status === 'failed' ? ' The task is ready to be dispatched again.' : ''}`, + subject: params.subject, + body: `${params.bodyPrefix} ${named}. ${params.reason}.${settlementNote}${params.bodySuffix ?? ''}`, type: 'escalation', priority: 'high', - payload: JSON.stringify({ - taskId: dispatch.task_id, - dispatchId: dispatch.id, - exitCode, - exitCause: cause, - handle - }), - runId: dispatch.run_id + payload: JSON.stringify(params.payload), + runId: params.dispatch.run_id }) this.notifyMessageArrived(escalation.to_handle, escalation.type) } catch (error) { - console.warn('[orchestration] failed to escalate worker exit', { - dispatchId: dispatch.id, - runId: dispatch.run_id, + console.warn(`[orchestration] failed to escalate ${params.failureLogLabel}`, { + dispatchId: params.dispatch.id, + runId: params.dispatch.run_id, error }) } diff --git a/src/main/runtime/orchestration/codex-reconnect-failed-dispatch.test.ts b/src/main/runtime/orchestration/codex-reconnect-failed-dispatch.test.ts new file mode 100644 index 000000000000..6e7483ba4aa5 --- /dev/null +++ b/src/main/runtime/orchestration/codex-reconnect-failed-dispatch.test.ts @@ -0,0 +1,341 @@ +import { describe, expect, it, vi } from 'vitest' +import { OrcaRuntimeService } from '../orca-runtime' +import { LEGACY_RUN_ID, OrchestrationDb } from './db' +import { makePaneKey } from '../../../shared/stable-pane-id' +import { getDefaultWorkspaceSession } from '../../../shared/constants' +import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types' + +const WORKTREE_ID = 'repo-1::/tmp/codex-reconnect-failed' +const WORKER_LEAF_ID = '11111111-1111-4111-8111-111111111111' +const COORDINATOR_LEAF_ID = '22222222-2222-4222-8222-222222222222' +const WORKER_PTY_ID = 'pty-codex-worker' +const COORDINATOR_PTY_ID = 'pty-coordinator' +const WORKER_PANE_KEY = makePaneKey('tab-worker', WORKER_LEAF_ID) +const COORDINATOR_PANE_KEY = makePaneKey('tab-coordinator', COORDINATOR_LEAF_ID) + +const RECONNECT_FAILURE_BANNER = [ + 'Automatic reconnect could not restore this session', + 'app-server session could not be restored', + 'Reconnect failed — check the endpoint, then relaunch' +].join('\r\n') + +function makeStore() { + const session: WorkspaceSessionState = getDefaultWorkspaceSession() + return { + getWorkspaceSession: vi.fn(() => session), + setWorkspaceSession: vi.fn(), + getRepos: vi.fn(() => [ + { + id: 'repo-1', + path: '/tmp/codex-reconnect-failed', + displayName: 'codex-reconnect-failed', + badgeColor: '#000000', + addedAt: 0 + } + ]), + getAllWorktreeMeta: vi.fn(() => ({})), + getWorktreeMeta: vi.fn(() => undefined), + setWorktreeMeta: vi.fn(), + removeWorktreeMeta: vi.fn(), + getSettings: vi.fn(() => ({ workspaceDir: '/tmp/workspaces' })), + getProjects: vi.fn(() => []) + } +} + +function makeRuntime() { + const kill = vi.fn(() => true) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the runtime constructor only reaches the WorkspaceSession, repo, worktree metadata, settings, and project reads supplied by this fixture. + const runtime = new OrcaRuntimeService(makeStore() as never) + const ptyController = { + spawn: vi.fn(async () => ({ id: 'never' })), + write: vi.fn(() => true), + kill, + getForegroundProcess: async () => null, + listProcesses: vi.fn(async () => []) + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this fixture implements the spawn, write, kill, foreground-process, and process-list operations reached by the runtime path under test. + runtime.setPtyController(ptyController as never) + const workerHandle = runtime.preAllocateHandleForPty(WORKER_PTY_ID) + const coordinatorHandle = runtime.preAllocateHandleForPty(COORDINATOR_PTY_ID) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { + tabs: [ + { + tabId: 'tab-worker', + worktreeId: WORKTREE_ID, + title: 'Worker', + activeLeafId: WORKER_LEAF_ID, + layout: null + }, + { + tabId: 'tab-coordinator', + worktreeId: WORKTREE_ID, + title: 'Coordinator', + activeLeafId: COORDINATOR_LEAF_ID, + layout: null + } + ], + leaves: [ + { + tabId: 'tab-worker', + worktreeId: WORKTREE_ID, + leafId: WORKER_LEAF_ID, + paneRuntimeId: 1, + ptyId: WORKER_PTY_ID, + paneTitle: null + }, + { + tabId: 'tab-coordinator', + worktreeId: WORKTREE_ID, + leafId: COORDINATOR_LEAF_ID, + paneRuntimeId: 2, + ptyId: COORDINATOR_PTY_ID, + paneTitle: null + } + ] + }) + return { runtime, kill, workerHandle, coordinatorHandle } +} + +function attachOrchestrationDb(runtime: OrcaRuntimeService, db: OrchestrationDb): void { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: db is the real OrchestrationDb implementation; the assertion bridges the runtime's narrower test-facing storage port. + runtime.setOrchestrationDb(db as never) +} + +describe('Codex reconnect failure settles its worker dispatch', () => { + it('fails and escalates the explicit unrecoverable-session banner but keeps the PTY alive', async () => { + const { runtime, kill, workerHandle, coordinatorHandle } = makeRuntime() + const db = new OrchestrationDb(':memory:') + try { + const run = db.createRun({ + objective: 'settle unrecoverable Codex session', + coordinatorHandle, + coordinatorPaneKey: COORDINATOR_PANE_KEY + }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskSpec: 'run work in a Codex worker', + taskRunId: run.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: workerHandle, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'runtime:pty:codex-worker', + worktreeId: WORKTREE_ID, + setupState: 'not_applicable', + effects: [{ kind: 'terminal', action: 'created', id: workerHandle }], + terminalOwnership: 'created' + }) + db.markWorkerDispatchReady(started.dispatch.id) + attachOrchestrationDb(runtime, db) + + expect(() => + db.failDispatch(started.dispatch.id, 'no authoritative session failure') + ).toThrow(/active supervised worker/) + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: emulate a remaining synthetic PTY handle after its cached PTY ID index is absent. + const runtimeHandleIndexes = runtime as unknown as { + handleByPtyId: Map + handles: Map + } + const syntheticHandle = runtimeHandleIndexes.handles.get(workerHandle) + expect(syntheticHandle?.ptyId).toBe(WORKER_PTY_ID) + if (syntheticHandle) { + syntheticHandle.tabId = `pty:${WORKER_PTY_ID}` + } + runtimeHandleIndexes.handleByPtyId.delete(WORKER_PTY_ID) + + runtime.onPtyData( + WORKER_PTY_ID, + '\x1b[1mAutomatic reconnect could not restore this session\x1b[0m\r\napp-server session could not be restored\r\nReconnect failed — check the endpoint, then re', + 100 + ) + runtime.onPtyData(WORKER_PTY_ID, 'launch', 101) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(db.getDispatchContextById(started.dispatch.id)).toMatchObject({ + status: 'failed', + termination_reason: null + }) + expect(db.getTask(started.task.id)?.status).toBe('failed') + expect(db.getWorkerDispatch(started.dispatch.id)).toMatchObject({ + state: 'failed', + stage: 'session_unrecoverable' + }) + expect(db.getWorkerTerminalResourceByOwner(started.dispatch.id)).toMatchObject({ + release_state: 'not_requested', + ownership_state: 'owned' + }) + expect( + db.getUnreadRunMailbox(run.id, 100, ['escalation']).map(({ subject, body }) => ({ + subject, + body + })) + ).toEqual([ + { + subject: 'Agent session failed (Codex could not restore its app-server session)', + body: expect.stringContaining( + 'The task was marked failed pending coordinator review and will not be retried automatically.' + ) + } + ]) + expect(kill).not.toHaveBeenCalled() + + const nextTask = db.createTask({ runId: run.id, spec: 'run the next Codex worker task' }) + const next = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: nextTask.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: next.dispatch.id, + handle: workerHandle, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'runtime:pty:codex-worker', + worktreeId: WORKTREE_ID, + setupState: 'not_applicable', + effects: [], + terminalOwnership: 'external' + }) + db.markWorkerDispatchReady(next.dispatch.id) + + runtime.onPtyData(WORKER_PTY_ID, 'next task started\r\n', 102) + expect(db.getDispatchContextById(next.dispatch.id)?.status).toBe('dispatched') + expect(db.getWorkerDispatch(next.dispatch.id)?.state).toBe('ready') + + runtime.onPtyData(WORKER_PTY_ID, RECONNECT_FAILURE_BANNER, 103) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(db.getDispatchContextById(next.dispatch.id)?.status).toBe('failed') + expect(db.getTask(nextTask.id)?.status).toBe('failed') + expect( + db + .getUnreadRunMailbox(run.id, 100, ['escalation']) + .filter(({ subject }) => subject.includes('Agent session failed')) + ).toHaveLength(2) + } finally { + db.close() + } + }) + + it('ignores a reconnect banner that began before a later dispatch started', () => { + const { runtime, kill, workerHandle, coordinatorHandle } = makeRuntime() + const db = new OrchestrationDb(':memory:') + try { + const run = db.createRun({ + objective: 'start a Codex worker after old terminal output', + coordinatorHandle, + coordinatorPaneKey: COORDINATOR_PANE_KEY + }) + attachOrchestrationDb(runtime, db) + runtime.onPtyData( + WORKER_PTY_ID, + 'Automatic reconnect could not restore this session\r\napp-server session could not be restored\r\n', + 100 + ) + + const task = db.createTask({ runId: run.id, spec: 'keep the new dispatch active' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: workerHandle, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'runtime:pty:codex-worker', + worktreeId: WORKTREE_ID, + setupState: 'not_applicable', + effects: [], + terminalOwnership: 'external' + }) + db.markWorkerDispatchReady(started.dispatch.id) + + runtime.onPtyData(WORKER_PTY_ID, 'Reconnect failed — check the endpoint, then relaunch', 101) + + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + expect(db.getWorkerDispatch(started.dispatch.id)?.state).toBe('ready') + expect(kill).not.toHaveBeenCalled() + } finally { + db.close() + } + }) + + it('does not settle a worker from a partial reconnect banner', () => { + const { runtime, kill, workerHandle } = makeRuntime() + const db = new OrchestrationDb(':memory:') + try { + const task = db.createTask({ runId: LEGACY_RUN_ID, spec: 'keep a live worker active' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: workerHandle, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'runtime:pty:codex-worker', + worktreeId: WORKTREE_ID, + setupState: 'not_applicable', + effects: [], + terminalOwnership: 'created' + }) + db.markWorkerDispatchReady(started.dispatch.id) + attachOrchestrationDb(runtime, db) + + runtime.onPtyData( + WORKER_PTY_ID, + 'Automatic reconnect could not restore this session\r\napp-server session could not be restored', + 100 + ) + + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + expect(db.getWorkerDispatch(started.dispatch.id)?.state).toBe('ready') + expect(kill).not.toHaveBeenCalled() + } finally { + db.close() + } + }) + + it('does not apply the Codex banner rule to another agent', () => { + const { runtime, kill, workerHandle } = makeRuntime() + const db = new OrchestrationDb(':memory:') + try { + const task = db.createTask({ runId: LEGACY_RUN_ID, spec: 'keep a Claude worker active' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'claude' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: workerHandle, + paneKey: WORKER_PANE_KEY, + processIncarnation: 'runtime:pty:claude-worker', + worktreeId: WORKTREE_ID, + setupState: 'not_applicable', + effects: [], + terminalOwnership: 'created' + }) + db.markWorkerDispatchReady(started.dispatch.id) + attachOrchestrationDb(runtime, db) + + runtime.onPtyData(WORKER_PTY_ID, RECONNECT_FAILURE_BANNER, 100) + + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + expect(db.getWorkerDispatch(started.dispatch.id)?.state).toBe('ready') + expect(kill).not.toHaveBeenCalled() + } finally { + db.close() + } + }) +}) diff --git a/src/main/runtime/orchestration/db/dispatch-context/dispatch-completion.ts b/src/main/runtime/orchestration/db/dispatch-context/dispatch-completion.ts index be1cf99d2b45..1f03f18ec363 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/dispatch-completion.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/dispatch-completion.ts @@ -112,7 +112,11 @@ export function failDispatch( this: OrchestrationDb, ctxId: string, error: string, - options: { workerProcessExited?: boolean; terminationReason?: string } = {} + options: { + workerProcessExited?: boolean + workerSessionUnrecoverable?: boolean + terminationReason?: string + } = {} ): DispatchContextRow | undefined { // Why: reserve the WAL writer before lifecycle reads so a concurrent commit cannot cause SQLITE_BUSY_SNAPSHOT. const transaction = beginLifecycleWriteTransaction(this.db, FAIL_DISPATCH_SAVEPOINT) @@ -121,13 +125,18 @@ export function failDispatch( | DispatchContextRow | undefined const workerBefore = this.getWorkerDispatch(ctxId) + const settlesWorker = options.workerProcessExited || options.workerSessionUnrecoverable if (!before || !['pending', 'dispatched'].includes(before.status)) { + if (options.workerSessionUnrecoverable) { + commitLifecycleWriteTransaction(this.db, transaction) + return undefined + } const worker = workerBefore if ( before && worker && !['failed', 'succeeded', 'stopped', 'abandoned'].includes(worker.state) && - !options.workerProcessExited + !settlesWorker ) { throw new OrchestrationError( 'task_not_startable', @@ -138,8 +147,15 @@ export function failDispatch( commitLifecycleWriteTransaction(this.db, transaction) return before } + // Why: an explicit, unrecoverable agent-session verdict can settle the worker while its PTY + // remains live. It is valid only for a ready worker; a concurrent stop or other settlement + // keeps ownership and the Dispatch untouched. + if (options.workerSessionUnrecoverable && workerBefore?.state !== 'ready') { + commitLifecycleWriteTransaction(this.db, transaction) + return undefined + } if ( - !options.workerProcessExited && + !settlesWorker && workerBefore && !['failed', 'succeeded', 'stopped', 'abandoned'].includes(workerBefore.state) ) { @@ -184,10 +200,25 @@ export function failDispatch( updated_at: new Date().toISOString() } }) + } else if (worker && options.workerSessionUnrecoverable) { + transitionLifecycleWithDb(this.db, { + entity: 'worker', + id: ctxId, + from: 'ready', + to: 'failed', + projection: { + stage: 'session_unrecoverable', + last_error: error, + updated_at: new Date().toISOString() + } + }) } - // Why: back to 'ready' not 'pending' — 'pending' would strand it since promoteReadyTasks only runs when a dep completes. - const taskStatus: TaskStatus = ctx.status === 'circuit_broken' ? 'failed' : 'ready' + // Why: a terminal banner can be emitted by displayed worker output, so avoid automatically + // replaying side effects while the original PTY remains alive; a coordinator can review and retry. + // Worker process exits keep the existing retry behavior until the circuit breaker trips. + const taskStatus: TaskStatus = + ctx.status === 'circuit_broken' || options.workerSessionUnrecoverable ? 'failed' : 'ready' // Why: the status guard keeps a late failure from reopening a task that already completed or was retried elsewhere. const task = this.getTask(ctx.task_id) if ( @@ -218,6 +249,42 @@ export function failDispatch( } } +export function failCodexSessionUnrecoverableDispatch( + this: OrchestrationDb, + dispatchId: string, + error: string +): 'retry' | 'ignored' | 'failed' | 'circuit_broken' { + const worker = this.getWorkerDispatch(dispatchId) + if (!worker || !['starting', 'start_unknown', 'ready'].includes(worker.state)) { + return 'ignored' + } + if (worker.state !== 'ready') { + return 'retry' + } + if (worker.stage !== 'input_accepted') { + return 'ignored' + } + let startOptions: unknown + try { + startOptions = JSON.parse(worker.start_options) + } catch { + return 'ignored' + } + if ( + !startOptions || + typeof startOptions !== 'object' || + !('agent' in startOptions) || + startOptions.agent !== 'codex' + ) { + return 'ignored' + } + const settled = this.failDispatch(dispatchId, error, { workerSessionUnrecoverable: true }) + if (settled?.status === 'failed' || settled?.status === 'circuit_broken') { + return settled.status + } + return 'ignored' +} + export type DispatchCompletionMethods = { completeDispatch: typeof completeDispatch completeActiveDispatchesForTask: typeof completeActiveDispatchesForTask @@ -225,6 +292,7 @@ export type DispatchCompletionMethods = { recordHeartbeat: typeof recordHeartbeat getStaleDispatches: typeof getStaleDispatches failDispatch: typeof failDispatch + failCodexSessionUnrecoverableDispatch: typeof failCodexSessionUnrecoverableDispatch } export function attachDispatchCompletion(ctor: { prototype: object }): void { @@ -234,6 +302,7 @@ export function attachDispatchCompletion(ctor: { prototype: object }): void { failActiveDispatchForTask, recordHeartbeat, getStaleDispatches, - failDispatch + failDispatch, + failCodexSessionUnrecoverableDispatch }) }