diff --git a/src/main/agent-hooks/server-codex-transcript-turn-end.test.ts b/src/main/agent-hooks/server-codex-transcript-turn-end.test.ts new file mode 100644 index 000000000000..5abf42f4ee04 --- /dev/null +++ b/src/main/agent-hooks/server-codex-transcript-turn-end.test.ts @@ -0,0 +1,319 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { AgentHookServer } from './server' +import { makePaneKey } from '../../shared/stable-pane-id' + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' +// Longer than one transcript poll, so "still working" means the poll ran and declined. +const POLL_SETTLE_MS = 1_500 + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +function taskStarted(turnId: string): string { + return line({ type: 'event_msg', payload: { type: 'task_started', turn_id: turnId } }) +} + +// Shape Codex 0.156 writes when a turn ends; `error` is present only when the turn failed. +function taskComplete(turnId: string, failed: boolean): string { + return line({ + type: 'event_msg', + payload: { + type: 'task_complete', + turn_id: turnId, + last_agent_message: null, + ...(failed + ? { + error: { + message: + 'We’re currently experiencing high demand, which may cause temporary errors.', + codex_error_info: 'internal_server_error' + } + } + : {}) + } + }) +} + +describe('AgentHookServer settles a Codex turn from its rollout', () => { + const dirs: string[] = [] + const servers: AgentHookServer[] = [] + + afterEach(() => { + for (const server of servers) { + server.stop() + } + servers.length = 0 + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + async function startRig(initialRollout = ''): Promise<{ + rollout: string + dir: string + server: AgentHookServer + post: (payload: Record) => Promise + row: () => ReturnType[number] | undefined + }> { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-turn-end-')) + dirs.push(dir) + const rollout = join(dir, 'rollout-parent.jsonl') + writeFileSync(rollout, initialRollout) + const server = new AgentHookServer() + servers.push(server) + await server.start({ env: 'production' }) + const env = server.buildPtyEnv() + const post = (payload: Record): Promise => + fetch(`http://127.0.0.1:${env.ORCA_AGENT_HOOK_PORT}/hook/codex`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Orca-Agent-Hook-Token': env.ORCA_AGENT_HOOK_TOKEN + }, + body: JSON.stringify({ + paneKey: PANE_KEY, + tabId: 'tab-1', + worktreeId: 'wt-1', + payload: { session_id: 'root-session', transcript_path: rollout, ...payload } + }) + }) + return { rollout, dir, server, post, row: () => server.getStatusSnapshot()[0] } + } + + async function startFailingTurn( + rig: Awaited>, + turnId: string + ): Promise { + appendFileSync(rig.rollout, taskStarted(turnId)) + await rig.post({ hook_event_name: 'UserPromptSubmit', turn_id: turnId, prompt: 'run it' }) + await rig.post({ hook_event_name: 'PreToolUse', turn_id: turnId, tool_name: 'Bash' }) + await rig.post({ hook_event_name: 'PostToolUse', turn_id: turnId, tool_name: 'Bash' }) + expect(rig.row()).toMatchObject({ state: 'working', mainAgent: { state: 'working' } }) + } + + // STA-7949: Codex runs no hook at all when a turn fails, so nothing but the rollout ends it. + it('settles a turn that failed after a tool call, with no Stop hook', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ + state: 'done', + prompt: 'run it', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + }) + + it('settles a turn the rollout completed without an outcome when its Stop never arrived', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + + appendFileSync(rig.rollout, taskComplete('turn-1', false)) + + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ state: 'done', mainAgent: { state: 'done' } }) + }, + { timeout: 3_000, interval: 50 } + ) + expect(rig.row()?.mainAgent?.outcome).toBeUndefined() + }) + + it('never settles the open turn from an earlier turn’s end', async () => { + const rig = await startRig(taskStarted('turn-0') + taskComplete('turn-0', true)) + await startFailingTurn(rig, 'turn-1') + + await new Promise((resolve) => setTimeout(resolve, POLL_SETTLE_MS)) + expect(rig.row()).toMatchObject({ state: 'working', mainAgent: { state: 'working' } }) + }) + + it('keeps a new prompt working when it lands before the poll reads the old turn’s end', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + + appendFileSync(rig.rollout, taskComplete('turn-1', true) + taskStarted('turn-2')) + await rig.post({ hook_event_name: 'UserPromptSubmit', turn_id: 'turn-2', prompt: 'again' }) + + await new Promise((resolve) => setTimeout(resolve, POLL_SETTLE_MS)) + expect(rig.row()).toMatchObject({ + state: 'working', + prompt: 'again', + mainAgent: { state: 'working' } + }) + + appendFileSync(rig.rollout, taskComplete('turn-2', false)) + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ state: 'done', prompt: 'again' }) + }, + { timeout: 3_000, interval: 50 } + ) + }) + + it('treats a late Stop for the settled turn as a restatement, not a new transition', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.row()?.state).toBe('done') + }, + { timeout: 3_000, interval: 50 } + ) + const settled = rig.row() + + await rig.post({ hook_event_name: 'Stop', turn_id: 'turn-1' }) + + expect(rig.row()).toMatchObject({ + state: 'done', + stateStartedAt: settled?.stateStartedAt, + mainAgent: { + state: 'done', + outcome: 'failure', + stateStartedAt: settled?.mainAgent?.stateStartedAt + } + }) + }) + + it('leaves the turn working when its rollout cannot be read', async () => { + const rig = await startRig() + const missing = join(rig.dir, 'rollout-missing.jsonl') + await rig.post({ + hook_event_name: 'UserPromptSubmit', + turn_id: 'turn-1', + prompt: 'run it', + transcript_path: missing + }) + await rig.post({ + hook_event_name: 'PostToolUse', + turn_id: 'turn-1', + tool_name: 'Bash', + transcript_path: missing + }) + + await new Promise((resolve) => setTimeout(resolve, POLL_SETTLE_MS)) + expect(rig.row()).toMatchObject({ state: 'working', mainAgent: { state: 'working' } }) + }) + + it('ends the root turn but keeps the row working while a child still runs', async () => { + const rig = await startRig( + line({ + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: CHILD_ID, + agent_path: '/root/pr_review', + kind: 'started' + } + }) + ) + const childRollout = join(rig.dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync(childRollout, taskStarted('child-turn')) + await startFailingTurn(rig, 'turn-1') + expect(rig.row()?.subagents).toHaveLength(1) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ + state: 'working', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + + appendFileSync(childRollout, taskComplete('child-turn', false)) + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ state: 'done', mainAgent: { state: 'done' } }) + }, + { timeout: 3_000, interval: 50 } + ) + }) + + // A hook-reported child keeps posting its own hooks while the root turn fails, so the root fires none after. + it('settles the root turn when a child hook, not a root hook, is the latest event', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + const childRollout = join(rig.dir, 'rollout-child.jsonl') + writeFileSync(childRollout, '') + const child = { + agent_id: 'child-1', + agent_type: 'worker', + turn_id: 'child-turn', + transcript_path: childRollout + } + await rig.post({ hook_event_name: 'SubagentStart', ...child }) + await rig.post({ hook_event_name: 'PostToolUse', tool_name: 'Bash', ...child }) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ + state: 'working', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + + await rig.post({ hook_event_name: 'SubagentStop', ...child }) + expect(rig.row()).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }) + + it('settles the root turn when a finished child’s SubagentStop is the latest event', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + const child = { agent_id: 'child-1', agent_type: 'worker', turn_id: 'child-turn' } + await rig.post({ hook_event_name: 'SubagentStart', ...child }) + await rig.post({ hook_event_name: 'SubagentStop', ...child }) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + }) + + // Codex fires SessionStart(compact) mid-turn with no turn_id, then the turn can fail with no further hook. + it('settles a turn that failed after a mid-turn compaction', async () => { + const rig = await startRig() + await startFailingTurn(rig, 'turn-1') + await rig.post({ hook_event_name: 'SessionStart', source: 'compact', model: 'gpt-5.5' }) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + + await vi.waitFor( + () => { + expect(rig.row()).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + }) +}) diff --git a/src/relay/agent-hook-server-codex-transcript-turn-end.test.ts b/src/relay/agent-hook-server-codex-transcript-turn-end.test.ts new file mode 100644 index 000000000000..cfeb0f19ca87 --- /dev/null +++ b/src/relay/agent-hook-server-codex-transcript-turn-end.test.ts @@ -0,0 +1,221 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { RelayAgentHookServer } from './agent-hook-server' +import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay' +import { makePaneKey } from '../shared/stable-pane-id' +import { AgentHookServer } from '../main/agent-hooks/server' + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +function taskComplete(turnId: string, failed: boolean): string { + return line({ + type: 'event_msg', + payload: { + type: 'task_complete', + turn_id: turnId, + ...(failed ? { error: { message: 'upstream failure' } } : {}) + } + }) +} + +describe('RelayAgentHookServer settles a Codex turn from its rollout', () => { + const dirs: string[] = [] + const stops: (() => void)[] = [] + + afterEach(() => { + for (const stop of stops) { + stop() + } + stops.length = 0 + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + /** A real relay whose forwards feed a real desktop hook server, as over SSH. */ + async function startRelayToDesktop(initialRollout: string): Promise<{ + dir: string + rollout: string + post: (payload: Record) => Promise + desktopRow: () => ReturnType[number] | undefined + }> { + const dir = mkdtempSync(join(tmpdir(), 'relay-hook-codex-turn-end-')) + dirs.push(dir) + const rollout = join(dir, 'rollout-parent.jsonl') + writeFileSync(rollout, initialRollout) + const desktop = new AgentHookServer() + const relay = new RelayAgentHookServer({ + endpointDir: dir, + forward: (envelope) => desktop.ingestRemote(envelope, 'conn-test') + }) + stops.push( + () => relay.stop(), + () => desktop.stop() + ) + await relay.start() + const { port, token } = relay.getCoordinates() + const post = (payload: Record): Promise => + fetch(`http://127.0.0.1:${port}/hook/codex`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token }, + body: JSON.stringify({ + paneKey: PANE_KEY, + tabId: 'tab-1', + worktreeId: 'wt-1', + payload: { session_id: 'root-session', transcript_path: rollout, ...payload } + }) + }) + return { dir, rollout, post, desktopRow: () => desktop.getStatusSnapshot()[0] } + } + + // STA-7949: only the execution host can read the rollout, so the relay must publish the turn end itself. + it('forwards a failed turn as a Stop carrying the failure, then stops polling', async () => { + const dir = mkdtempSync(join(tmpdir(), 'relay-hook-codex-turn-end-')) + dirs.push(dir) + const rollout = join(dir, 'rollout-parent.jsonl') + writeFileSync( + rollout, + line({ type: 'event_msg', payload: { type: 'task_started', turn_id: 'turn-1' } }) + ) + const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>() + const server = new RelayAgentHookServer({ endpointDir: dir, forward }) + await server.start() + try { + const { port, token } = server.getCoordinates() + const post = (payload: Record): Promise => + fetch(`http://127.0.0.1:${port}/hook/codex`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': token }, + body: JSON.stringify({ + paneKey: PANE_KEY, + tabId: 'tab-1', + worktreeId: 'wt-1', + payload: { + session_id: 'root-session', + transcript_path: rollout, + turn_id: 'turn-1', + ...payload + } + }) + }) + await post({ hook_event_name: 'UserPromptSubmit', prompt: 'run it' }) + await post({ hook_event_name: 'PostToolUse', tool_name: 'Bash' }) + expect(forward.mock.calls.at(-1)?.[0]).toMatchObject({ + hookEventName: 'PostToolUse', + payload: { state: 'working' } + }) + const forwardsBeforeEnd = forward.mock.calls.length + + appendFileSync( + rollout, + line({ + type: 'event_msg', + payload: { + type: 'task_complete', + turn_id: 'turn-1', + error: { message: 'upstream failure', codex_error_info: 'internal_server_error' } + } + }) + ) + + await vi.waitFor( + () => { + expect(forward.mock.calls.at(-1)?.[0]).toMatchObject({ + hookEventName: 'Stop', + payload: { state: 'done', mainAgent: { state: 'done', outcome: 'failure' } } + }) + }, + { timeout: 3_000, interval: 50 } + ) + expect(forward).toHaveBeenCalledTimes(forwardsBeforeEnd + 1) + await new Promise((resolve) => setTimeout(resolve, 1_500)) + expect(forward).toHaveBeenCalledTimes(forwardsBeforeEnd + 1) + } finally { + server.stop() + } + }) + + // The relay publishes later child-only updates too; each must keep the desktop's root done. + it('keeps the desktop row settled when a child outlives the failed root turn', async () => { + const rig = await startRelayToDesktop( + line({ + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: CHILD_ID, + agent_path: '/root/worker', + kind: 'started' + } + }) + ) + const childRollout = join(rig.dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync( + childRollout, + line({ type: 'event_msg', payload: { type: 'task_started', turn_id: 'child-turn' } }) + ) + await rig.post({ hook_event_name: 'UserPromptSubmit', turn_id: 'turn-1', prompt: 'run it' }) + await rig.post({ hook_event_name: 'PostToolUse', turn_id: 'turn-1', tool_name: 'Bash' }) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.desktopRow()).toMatchObject({ + state: 'working', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + + appendFileSync(childRollout, taskComplete('child-turn', false)) + await vi.waitFor( + () => { + expect(rig.desktopRow()).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + }) + + it('settles the desktop row when the root turn ends under a hook-reported child', async () => { + const rig = await startRelayToDesktop('') + const child = { + agent_id: 'child-1', + agent_type: 'worker', + turn_id: 'child-turn', + transcript_path: join(rig.dir, 'rollout-child.jsonl') + } + await rig.post({ hook_event_name: 'UserPromptSubmit', turn_id: 'turn-1', prompt: 'run it' }) + await rig.post({ hook_event_name: 'SubagentStart', ...child }) + await rig.post({ hook_event_name: 'PostToolUse', tool_name: 'Bash', ...child }) + + appendFileSync(rig.rollout, taskComplete('turn-1', true)) + await vi.waitFor( + () => { + expect(rig.desktopRow()).toMatchObject({ + state: 'working', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }, + { timeout: 3_000, interval: 50 } + ) + + await rig.post({ hook_event_name: 'SubagentStop', ...child }) + expect(rig.desktopRow()).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }) +}) diff --git a/src/shared/agent-hook-listener-codex-main-agent.test.ts b/src/shared/agent-hook-listener-codex-main-agent.test.ts index e7448c0f866b..f3e28086d62e 100644 --- a/src/shared/agent-hook-listener-codex-main-agent.test.ts +++ b/src/shared/agent-hook-listener-codex-main-agent.test.ts @@ -1,4 +1,7 @@ -import { beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { createHookListenerState, type HookListenerState @@ -8,7 +11,8 @@ import { reconcileRemoteCodexState, seedCodexStateFromSnapshot } from './agent-hook-listener/providers/codex-state' -import { PANE_KEY } from './agent-hook-listener-test-harness' +import { shouldPollHookTranscript } from './agent-hook-listener/transcript-poll-policy' +import { normalizeAndAccept, PANE_KEY } from './agent-hook-listener-test-harness' describe('the Codex root record seeded from a durable row', () => { let state: HookListenerState @@ -65,6 +69,34 @@ describe('the Codex root record seeded from a durable row', () => { expect(reconciled.mainAgent).toMatchObject({ state: 'done', outcome: 'cancellation' }) }) + it('keeps the failure a relay read from the rollout on its Stop', () => { + reconcileRemoteCodexState( + state, + PANE_KEY, + 'PostToolUse', + undefined, + { state: 'working', prompt: 'ship', agentType: 'codex' }, + undefined + ) + const reconciled = reconcileRemoteCodexState( + state, + PANE_KEY, + 'Stop', + undefined, + { + state: 'done', + prompt: 'ship', + agentType: 'codex', + mainAgent: { state: 'done', outcome: 'failure', stateStartedAt: 7 } + }, + undefined + ) + expect(reconciled).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }) + }) + it('folds a relayed waiting child through the shared rule, keeping the root fact', () => { // The relay's aggregate says `working`; main re-derives the row from the roster instead. const reconciled = reconcileRemoteCodexState( @@ -83,3 +115,41 @@ describe('the Codex root record seeded from a durable row', () => { expect(reconciled).toMatchObject({ state: 'waiting', mainAgent: { state: 'working' } }) }) }) + +describe('the Codex rollout poll for an open root turn', () => { + let dir: string + + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'codex-turn-poll-')) + }) + + afterEach(() => { + rmSync(dir, { recursive: true, force: true }) + }) + + function pollsAfter(hook: Record): boolean { + const rollout = join(dir, 'rollout.jsonl') + writeFileSync(rollout, '') + const state = createHookListenerState() + const event = normalizeAndAccept(state, 'codex', { + hook_event_name: 'PostToolUse', + session_id: 'root-session', + transcript_path: rollout, + tool_name: 'Bash', + ...hook + }) + if (!event) { + throw new Error('expected a Codex status event') + } + return shouldPollHookTranscript(state, 'codex', event) + } + + it('runs while the turn is open and its id can match a rollout end', () => { + expect(pollsAfter({ turn_id: 'turn-1' })).toBe(true) + }) + + // Hooks from a Codex that sends no turn_id can never match a rollout end. + it('does not run when the hooks carry no turn id', () => { + expect(pollsAfter({})).toBe(false) + }) +}) diff --git a/src/shared/agent-hook-listener/main-agent-turn-state.ts b/src/shared/agent-hook-listener/main-agent-turn-state.ts index c077860f8467..af981f6116cf 100644 --- a/src/shared/agent-hook-listener/main-agent-turn-state.ts +++ b/src/shared/agent-hook-listener/main-agent-turn-state.ts @@ -32,4 +32,6 @@ export type CodexLeadTurnState = { /** When `state` first appeared; the root's own clock, published as `mainAgent.stateStartedAt`. */ stateStartedAt: number model?: string + /** Codex `turn_id` of the root turn this record describes; only the execution host sets it. */ + turnId?: string } diff --git a/src/shared/agent-hook-listener/providers/codex-events.ts b/src/shared/agent-hook-listener/providers/codex-events.ts index c7f615213704..16aa6f6255a1 100644 --- a/src/shared/agent-hook-listener/providers/codex-events.ts +++ b/src/shared/agent-hook-listener/providers/codex-events.ts @@ -11,7 +11,10 @@ import { finishCodexSubagent, upsertCodexSubagent } from '../../codex-subagent-roster' -import { reconcileCodexSubagentTranscript } from '../../codex-subagent-transcript' +import { + codexTranscriptTurnEnd, + reconcileCodexSubagentTranscript +} from '../../codex-subagent-transcript' import { codexTurnApprovalsAreAutoReviewed, reconcileCodexSubagentReviewer @@ -78,6 +81,31 @@ export function buildCodexChildDrivenStatusPayload( }) } +/** Settles the open root turn once its rollout records the end; a child hook can be the pane's + * last event when the root turn fails, and Codex runs no root hook for that failure. */ +function settleCodexRootTurnFromTranscript(state: HookListenerState, paneKey: string): void { + const lead = state.codexLeadStateByPaneKey.get(paneKey) + const transcriptState = state.codexSubagentTranscriptByPaneKey.get(paneKey) + const parentPath = transcriptState?.parent.filePath + if (!lead?.turnId || lead.state === 'done' || !transcriptState || !parentPath) { + return + } + reconcileCodexSubagentTranscript( + transcriptState, + getOrCreateCodexSubagentRoster(state, paneKey), + parentPath + ) + const turnEnd = codexTranscriptTurnEnd(transcriptState, lead.turnId) + if (turnEnd) { + setCodexMainAgentTurnState(state, paneKey, { + state: 'done', + ...(turnEnd === 'failed' ? { outcome: 'failure' as const } : {}), + model: lead.model, + turnId: lead.turnId + }) + } +} + export function normalizeCodexSubagentLifecycleEvent( state: HookListenerState, eventName: 'SubagentStart' | 'SubagentStop', @@ -103,6 +131,7 @@ export function normalizeCodexSubagentLifecycleEvent( } else { finishCodexSubagent(roster, agentId) } + settleCodexRootTurnFromTranscript(state, paneKey) return buildCodexChildDrivenStatusPayload(state, eventName, paneKey, hookPayload) } @@ -166,6 +195,11 @@ export function normalizeCodexEvent( } const agentId = readString(hookPayload, 'agent_id') + if (eventName === 'SessionStart' && hookPayload['source'] === 'compact') { + // Why: Codex fires it mid-turn after compacting, with no `turn_id` and no new process, so the + // turn's last real hook must stay the one the rollout poll re-reads. + return null + } const transcriptPath = readFirstString(hookPayload, ['transcript_path', 'transcriptPath']) if (eventName === 'SessionStart' && !agentId) { // Why: a pane can host a new Codex process after the old one exited without child Stop hooks. @@ -210,31 +244,38 @@ export function normalizeCodexEvent( }, Date.now() ) + settleCodexRootTurnFromTranscript(state, paneKey) return buildCodexChildDrivenStatusPayload(state, eventName, paneKey, hookPayload) } - if (eventName === 'Stop' && !hasCodexTranscriptSubagents(state, paneKey)) { + const turnId = readString(hookPayload, 'turn_id') + // Why: Codex runs no hook when a turn fails, so its rollout's record of this turn's end stands in for Stop. + const transcriptTurnEnd = + stateName !== 'done' && turnId + ? codexTranscriptTurnEnd(state.codexSubagentTranscriptByPaneKey.get(paneKey), turnId) + : undefined + const leadEventName = transcriptTurnEnd ? 'Stop' : eventName + if (leadEventName === 'Stop' && !hasCodexTranscriptSubagents(state, paneKey)) { // Why: Codex CLI 0.144 can omit child Stop hooks; later child activity safely recreates any agent still running. state.codexSubagentRosterByPaneKey.delete(paneKey) } // Why: resolved after the transcript reconcile above, so this turn's reviewer is read from the // rollout during the very PermissionRequest being classified, not from a prior event. - const ownedState = resolveCodexApprovalOwnedState( - state, - eventName, - paneKey, - transcriptPath, - stateName - ) + const ownedState = transcriptTurnEnd + ? 'done' + : resolveCodexApprovalOwnedState(state, eventName, paneKey, transcriptPath, stateName) const previousLead = state.codexLeadStateByPaneKey.get(paneKey) const record = setCodexMainAgentTurnState(state, paneKey, { state: ownedState, - ...codexOutcomeRestatedByStop(previousLead, ownedState), + ...(transcriptTurnEnd === 'failed' + ? { outcome: 'failure' as const } + : codexOutcomeRestatedByStop(previousLead, ownedState)), model: normalizeOptionalField(hookPayload['model'], AGENT_MODEL_MAX_LENGTH) ?? - (eventName === 'SessionStart' ? undefined : previousLead?.model) + (eventName === 'SessionStart' ? undefined : previousLead?.model), + turnId }) - return buildCodexStatusPayload(state, eventName, promptText, paneKey, hookPayload, { + return buildCodexStatusPayload(state, leadEventName, promptText, paneKey, hookPayload, { ...resolveCodexPaneStatus(state, paneKey, record), updateLead: true }) diff --git a/src/shared/agent-hook-listener/providers/codex-state.ts b/src/shared/agent-hook-listener/providers/codex-state.ts index e2f7e2f5ae3e..5ca61ba752e2 100644 --- a/src/shared/agent-hook-listener/providers/codex-state.ts +++ b/src/shared/agent-hook-listener/providers/codex-state.ts @@ -47,6 +47,19 @@ export function hasCodexTranscriptSubagents(state: HookListenerState, paneKey: s return hasTrackedCodexTranscriptSubagents(state.codexSubagentTranscriptByPaneKey.get(paneKey)) } +/** Whether the root turn is still open, with the id and rollout needed to find its end. */ +export function codexLeadTurnAwaitsTranscriptEnd( + state: HookListenerState, + paneKey: string +): boolean { + const lead = state.codexLeadStateByPaneKey.get(paneKey) + return ( + lead?.turnId !== undefined && + lead.state !== 'done' && + state.codexSubagentTranscriptByPaneKey.get(paneKey)?.parent.filePath !== undefined + ) +} + /** The only writer of the root record; the root's clock keeps continuity across same-state writes. */ export function setCodexMainAgentTurnState( state: HookListenerState, @@ -60,7 +73,8 @@ export function setCodexMainAgentTurnState( state: next.state, ...(continued.outcome ? { outcome: continued.outcome } : {}), stateStartedAt: continued.stateStartedAt, - model: next.model + model: next.model, + ...(next.turnId ? { turnId: next.turnId } : {}) } state.codexLeadStateByPaneKey.set(paneKey, record) return record @@ -201,6 +215,15 @@ export function reconcileRemoteCodexState( if (eventName === 'SubagentStop') { finishCodexSubagent(roster, agentId) } + // Why: the relay alone reads the rollout, and a child hook can carry the root turn's end. + const lead = state.codexLeadStateByPaneKey.get(paneKey) + if (payload.mainAgent?.state === 'done' && lead && lead.state !== 'done') { + setCodexMainAgentTurnState(state, paneKey, { + state: 'done', + ...(payload.mainAgent.outcome ? { outcome: payload.mainAgent.outcome } : {}), + model: lead.model + }) + } } else { const leadState = codexLeadStateForHookEvent(eventName, payload.state) if (eventName === 'SessionStart' || (eventName === 'Stop' && !payload.subagents)) { @@ -208,9 +231,13 @@ export function reconcileRemoteCodexState( } if (leadState) { const previousLead = state.codexLeadStateByPaneKey.get(paneKey) + // Why: the relay alone reads the rollout, so a turn it saw fail arrives as a Stop carrying the verdict. + const outcome = + codexOutcomeRestatedByStop(previousLead, leadState).outcome ?? + (leadState === 'done' ? payload.mainAgent?.outcome : undefined) setCodexMainAgentTurnState(state, paneKey, { state: leadState, - ...codexOutcomeRestatedByStop(previousLead, leadState), + ...(outcome ? { outcome } : {}), model: payload.model ?? previousLead?.model }) } diff --git a/src/shared/agent-hook-listener/transcript-poll-policy.ts b/src/shared/agent-hook-listener/transcript-poll-policy.ts index d27814cf928d..58b1bf014915 100644 --- a/src/shared/agent-hook-listener/transcript-poll-policy.ts +++ b/src/shared/agent-hook-listener/transcript-poll-policy.ts @@ -1,7 +1,10 @@ import type { AgentHookSource } from '../agent-hook-relay' import type { AgentHookEventPayload } from './listener-event' import type { HookListenerState } from './listener-state' -import { hasCodexTranscriptSubagents } from './providers/codex-state' +import { + codexLeadTurnAwaitsTranscriptEnd, + hasCodexTranscriptSubagents +} from './providers/codex-state' import { hasMuseSessionLog } from './providers/muse-events' /** Whether a pane's last hook body should be re-normalized on a timer to pick up transcript-only state. */ @@ -11,7 +14,10 @@ export function shouldPollHookTranscript( event: AgentHookEventPayload ): boolean { if (source === 'codex') { - return hasCodexTranscriptSubagents(state, event.paneKey) + return ( + hasCodexTranscriptSubagents(state, event.paneKey) || + codexLeadTurnAwaitsTranscriptEnd(state, event.paneKey) + ) } if (source === 'muse') { // Why: Muse's question tool fires no hook, so only its session log shows the wait and its answer. @@ -35,7 +41,16 @@ export function transcriptPollUpdate( ? { ...polled, hasExplicitPrompt: undefined, hookEventName: undefined } : undefined } - const subagentsChanged = + const rootEnded = polled.payload.mainAgent?.state === 'done' + const changed = + (rootEnded && original.payload.mainAgent?.state !== 'done') || JSON.stringify(polled.payload.subagents) !== JSON.stringify(original.payload.subagents) - return subagentsChanged ? polled : undefined + if (!changed) { + return undefined + } + // Why: a root hook re-read after its turn ended in the rollout restates that turn's Stop, so + // every host version keeps the root done, including on later child-only updates. + return rootEnded && !polled.toolAgentId + ? { ...polled, hasExplicitPrompt: undefined, hookEventName: 'Stop' } + : polled } diff --git a/src/shared/codex-subagent-transcript.ts b/src/shared/codex-subagent-transcript.ts index 5ce841ca2527..ed1b8e119a16 100644 --- a/src/shared/codex-subagent-transcript.ts +++ b/src/shared/codex-subagent-transcript.ts @@ -20,6 +20,8 @@ import { // Why: retire a child whose rollout stays unreadable this long, else a deleted/never-written file pins a phantom row forever. const CHILD_UNREADABLE_GRACE_MS = 60_000 +// Why: only the open turn is ever looked up; a small window covers a batch read after a restart. +const MAX_TRACKED_TURN_ENDS = 16 const SAFE_THREAD_ID = /^[A-Za-z0-9-]{1,64}$/ type TrackedTranscriptSubagent = JsonlCursor & { @@ -32,8 +34,13 @@ type TrackedTranscriptSubagent = JsonlCursor & { unresolvedSince?: number } +/** How the root agent's turn ended, as its rollout recorded it. */ +export type CodexTranscriptTurnEnd = 'completed' | 'failed' + export type CodexSubagentTranscriptState = { parent: JsonlCursor + /** Root turn ends read from the parent rollout, keyed by Codex `turn_id`. */ + turnEnds: Map subagents: Map /** Incremental reviewer cursors for child rollouts, which must not replace the parent cursor. */ reviewerCursorsByPath: Map @@ -149,6 +156,20 @@ function readChildModel(records: JsonRecord[]): string | undefined { return model } +function readTurnEnd( + recordValue: JsonRecord +): { turnId: string; end: CodexTranscriptTurnEnd } | undefined { + if (recordValue.type !== 'event_msg') { + return undefined + } + const payload = record(recordValue.payload) + const turnId = typeof payload?.turn_id === 'string' ? payload.turn_id.trim() : '' + if (payload?.type !== 'task_complete' || !turnId) { + return undefined + } + return { turnId, end: record(payload.error) ? 'failed' : 'completed' } +} + function normalizedTranscriptPath(transcriptPath: string | undefined): string | undefined { const normalizedPath = transcriptPath?.trim() return normalizedPath && isAbsolute(normalizedPath) && extname(normalizedPath) === '.jsonl' @@ -175,6 +196,7 @@ function childIsComplete(records: JsonRecord[]): boolean { export function createCodexSubagentTranscriptState(): CodexSubagentTranscriptState { return { parent: { offset: 0, carry: '' }, + turnEnds: new Map(), subagents: new Map(), reviewerCursorsByPath: new Map(), reviewersByPath: new Map() @@ -187,6 +209,13 @@ export function hasTrackedCodexTranscriptSubagents( return Boolean(state && state.subagents.size > 0) } +export function codexTranscriptTurnEnd( + state: CodexSubagentTranscriptState | undefined, + turnId: string +): CodexTranscriptTurnEnd | undefined { + return state?.turnEnds.get(turnId) +} + export function reconcileCodexSubagentTranscript( state: CodexSubagentTranscriptState, roster: CodexSubagentRoster, @@ -202,6 +231,7 @@ export function reconcileCodexSubagentTranscript( } state.parent = { filePath: normalizedPath, offset: 0, carry: '' } state.subagents.clear() + state.turnEnds.clear() state.reviewerCursorsByPath.clear() state.reviewersByPath.clear() // Why: a different rollout is a different session, so its predecessor's reviewer is void. @@ -214,6 +244,16 @@ export function reconcileCodexSubagentTranscript( ? undefined : (readApprovalsReviewer(parentRecords) ?? state.approvalsReviewer) for (const recordValue of parentRecords ?? []) { + const turnEnd = readTurnEnd(recordValue) + if (turnEnd) { + state.turnEnds.delete(turnEnd.turnId) + state.turnEnds.set(turnEnd.turnId, turnEnd.end) + const oldest = state.turnEnds.keys().next() + if (state.turnEnds.size > MAX_TRACKED_TURN_ENDS && !oldest.done) { + state.turnEnds.delete(oldest.value) + } + continue + } const activity = readActivity(recordValue) if (!activity) { continue