diff --git a/src/main/agent-hooks/__fixtures__/codex-review-child-0159.jsonl b/src/main/agent-hooks/__fixtures__/codex-review-child-0159.jsonl new file mode 100644 index 00000000000..1b52af31e5e --- /dev/null +++ b/src/main/agent-hooks/__fixtures__/codex-review-child-0159.jsonl @@ -0,0 +1,2 @@ +{"type":"event_msg","payload":{"type":"task_started","turn_id":"child-turn","root_turn_id":"root-turn"}} +{"type":"event_msg","payload":{"type":"task_complete","turn_id":"child-turn"}} diff --git a/src/main/agent-hooks/server-codex-review-turn-end.test.ts b/src/main/agent-hooks/server-codex-review-turn-end.test.ts new file mode 100644 index 00000000000..9099490d648 --- /dev/null +++ b/src/main/agent-hooks/server-codex-review-turn-end.test.ts @@ -0,0 +1,163 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + appendFileSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync +} from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { RelayAgentHookServer } from '../../wsl-guest/agent-hook-server' +import { AgentHookServer } from './server' +import { buildBody, postHookEvent } from './server.test-fixtures' + +// Captured from a completed Codex CLI 0.159.0 /review; session and turn IDs were replaced. +const reviewRecords = readFileSync( + new URL('./__fixtures__/codex-review-child-0159.jsonl', import.meta.url), + 'utf8' +) + .trimEnd() + .split('\n') +const event = (payload: Record): string => + `${JSON.stringify({ type: 'event_msg', payload })}\n` +const settle = (ms = 1_200) => new Promise((resolve) => setTimeout(resolve, ms)) + +describe('Codex turn completion without Stop', () => { + let desktop: AgentHookServer + let relay: RelayAgentHookServer | undefined + let directory: string + let published: (string | undefined)[] + + async function startHost(relayed: boolean): Promise { + directory = mkdtempSync(join(tmpdir(), 'codex-turn-end-')) + desktop = new AgentHookServer() + published = [] + desktop.setListener((status) => published.push(status.hookEventName)) + if (!relayed) { + await desktop.start({ env: 'production' }) + return + } + relay = new RelayAgentHookServer({ + endpointDir: directory, + forward: (envelope) => desktop.ingestRemote(envelope, 'ssh-connection') + }) + await relay.start() + } + + async function hook(payload: Record): Promise { + const body = buildBody({ session_id: 'root-session', ...payload }) + const coordinates = relay?.getCoordinates() + const response = coordinates + ? await fetch(`http://127.0.0.1:${coordinates.port}/hook/codex`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Orca-Agent-Hook-Token': coordinates.token + }, + body: JSON.stringify(body) + }) + : await postHookEvent(desktop, body, '/hook/codex') + expect(response.status).toBe(204) + } + + afterEach(() => { + desktop.stop() + relay?.stop() + relay = undefined + rmSync(directory, { recursive: true, force: true }) + }) + + it('settles a /review from its child rollout and leaves later turns working', async () => { + await startHost(false) + const childPath = join(directory, 'review-child.jsonl') + writeFileSync(childPath, `${reviewRecords[0]}\n`) + // Real review hooks carry the root session ID and child turn/path, with no agent_id or Stop. + const review = { transcript_path: childPath, turn_id: 'child-turn' } + await hook({ ...review, hook_event_name: 'UserPromptSubmit', prompt: 'Review changes' }) + for (const hookEventName of ['PreToolUse', 'PostToolUse', 'PreToolUse', 'PostToolUse']) { + await hook({ ...review, hook_event_name: hookEventName, tool_name: 'Bash' }) + } + expect(desktop.getStatusSnapshot()[0]).toMatchObject({ + state: 'working', + providerSession: { id: 'root-session', transcriptPath: childPath } + }) + + appendFileSync(childPath, `${reviewRecords[1]}\n`) + await vi.waitFor( + () => + expect(desktop.getStatusSnapshot()[0]).toMatchObject({ + state: 'done', + mainAgent: { state: 'done' } + }), + { timeout: 3_000 } + ) + // Desktops that predate host-owned turn verdicts still recognize a labelled Stop. + expect(published.at(-1)).toBe('Stop') + + // Neither a compaction queued before the next prompt nor a newer turn replays the review's end. + await hook({ hook_event_name: 'SessionStart', source: 'compact', transcript_path: childPath }) + await settle() + expect(published.at(-1)).toBe('SessionStart') + appendFileSync(childPath, event({ type: 'task_started', turn_id: 'next-turn' })) + await hook({ + ...review, + turn_id: 'next-turn', + hook_event_name: 'UserPromptSubmit', + prompt: 'Next' + }) + await settle() + expect(desktop.getStatusSnapshot()[0]).toMatchObject({ mainAgent: { state: 'working' } }) + // A delayed hook from the review's ended turn does not end the newer turn. + await hook({ ...review, hook_event_name: 'PostToolUse', tool_name: 'Bash' }) + expect(desktop.getStatusSnapshot()[0]).toMatchObject({ mainAgent: { state: 'working' } }) + + // A real Stop that lands before its rollout end, then a compaction, never rearms that turn. + await hook({ ...review, turn_id: 'next-turn', hook_event_name: 'Stop' }) + await hook({ hook_event_name: 'SessionStart', source: 'compact', transcript_path: childPath }) + appendFileSync(childPath, event({ type: 'task_complete', turn_id: 'next-turn' })) + await settle() + expect(published.slice(-2)).toEqual(['Stop', 'SessionStart']) + }) + + it.each([ + ['locally', 'in an earlier poll', false, true], + ['locally', 'in the same poll', false, false], + ['through a relay', 'in an earlier poll', true, true] + ])( + 'settles a failed turn %s after a child finished %s', + async (_, __, relayed, separatePolls) => { + await startHost(relayed) + const day = join(directory, '2026', '10', '16') + mkdirSync(day, { recursive: true }) + const rootPath = join(day, 'rollout-2026-10-16T00-00-00-root.jsonl') + const childId = '0199aaaa-bbbb-cccc-dddd-eeeeffff0001' + const childPath = join(day, `rollout-2026-10-16T00-00-01-${childId}.jsonl`) + const root = { transcript_path: rootPath, turn_id: 'root-turn' } + writeFileSync(rootPath, event({ type: 'task_started', turn_id: 'root-turn' })) + await hook({ ...root, hook_event_name: 'UserPromptSubmit', prompt: 'Do the task' }) + const started = { agent_thread_id: childId, kind: 'started', occurred_at_ms: Date.now() } + appendFileSync(rootPath, event({ type: 'sub_agent_activity', ...started })) + writeFileSync(childPath, event({ type: 'task_started', turn_id: 'child-turn' })) + await hook({ ...root, hook_event_name: 'PostToolUse', tool_name: 'spawn_agent' }) + + appendFileSync(childPath, event({ type: 'task_complete', turn_id: 'child-turn' })) + if (separatePolls) { + await settle() + } + const failure = { error: { message: 'usage limit' } } + appendFileSync(rootPath, event({ type: 'task_complete', turn_id: 'root-turn', ...failure })) + await vi.waitFor( + () => + expect(desktop.getStatusSnapshot()[0]).toMatchObject({ + state: 'done', + mainAgent: { state: 'done', outcome: 'failure' } + }), + { timeout: 3_000 } + ) + // Desktops derive a labelled Stop's outcome as success, so a failure goes out unlabelled. + expect(published.at(-1)).toBeUndefined() + } + ) +}) diff --git a/src/main/agent-hooks/server/server-status-retries.ts b/src/main/agent-hooks/server/server-status-retries.ts index 3007d862395..19b4718d01d 100644 --- a/src/main/agent-hooks/server/server-status-retries.ts +++ b/src/main/agent-hooks/server/server-status-retries.ts @@ -81,7 +81,7 @@ export abstract class AgentHookServerStatusRetries extends AgentHookServerStatus } const normalized = source === 'codex' - ? pollCodexTranscriptStatus(this.state, original) + ? pollCodexTranscriptStatus(this.state, original, body, this.env) : normalizeHookPayload(this.state, source, body, this.env) if (!normalized) { return diff --git a/src/shared/agent-hook-listener.ts b/src/shared/agent-hook-listener.ts index 6c6f9465ef5..3658235a23b 100644 --- a/src/shared/agent-hook-listener.ts +++ b/src/shared/agent-hook-listener.ts @@ -304,13 +304,13 @@ export function normalizeHookPayload( : undefined : hasExplicitUserPrompt( source, - eventName, + dispatched.eventName, extractedPrompt, dispatched.resolvedPromptText, dispatched.hasTranscriptPromptEvidence ), promptInteractionKey: dispatched.promptInteractionKey, - hookEventName: typeof eventName === 'string' ? eventName : undefined, + hookEventName: typeof dispatched.eventName === 'string' ? dispatched.eventName : undefined, providerPromptId: source === 'grok' ? (grokActiveTurn?.promptId ?? providerPromptId) : providerPromptId, grokPromptBoundary: grokActiveTurn ? true : undefined, 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 af7fdf8cc61..5a899e0af3e 100644 --- a/src/shared/agent-hook-listener/main-agent-turn-state.ts +++ b/src/shared/agent-hook-listener/main-agent-turn-state.ts @@ -43,4 +43,7 @@ export type CodexLeadTurnState = { /** When `state` first appeared; the root's own clock, published as `mainAgent.stateStartedAt`. */ stateStartedAt: number model?: string + /** Execution-host-only identity of the root session and turn whose rollout end can be replayed as Stop. */ + sessionId?: string + turnId?: string } diff --git a/src/shared/agent-hook-listener/provider-dispatch.ts b/src/shared/agent-hook-listener/provider-dispatch.ts index 52585c1dd05..142858ca3f8 100644 --- a/src/shared/agent-hook-listener/provider-dispatch.ts +++ b/src/shared/agent-hook-listener/provider-dispatch.ts @@ -13,7 +13,7 @@ import { readJcodeTurnPrompt } from './providers/jcode-turn-prompt' import { normalizeAntigravityEvent } from './providers/antigravity-events' import { normalizeAmpEvent } from './providers/amp-events' import { normalizeClaudeEvent } from './providers/claude-events' -import { normalizeCodexEvent } from './providers/codex-events' +import { codexHookEventForEndedTurn, normalizeCodexEvent } from './providers/codex-events' import { normalizeGeminiEvent } from './providers/gemini-events' import { normalizeOpenCodeFamilyEvent } from './providers/opencode-family-events' import { normalizeCursorEvent } from './providers/cursor-events' @@ -36,6 +36,8 @@ export type ProviderDispatchResult = { resolvedPromptText: string promptInteractionKey?: string hasTranscriptPromptEvidence: boolean + /** The event the payload was normalized as, which a provider may restate (Codex ended turns). */ + eventName: unknown } /** Exhaustive provider routing with provider-specific transcript locality and attribution. */ @@ -50,8 +52,11 @@ export function normalizeProviderEvent(input: { extractedPrompt: ExtractedPromptText previousOpenCodeMainAgent?: AgentMainAgentStatus }): ProviderDispatchResult { - const { state, source, eventName, promptText, paneKey, hookPayload, envelope, extractedPrompt } = - input + const { state, source, promptText, paneKey, hookPayload, envelope, extractedPrompt } = input + const eventName = + source === 'codex' + ? codexHookEventForEndedTurn(state, paneKey, input.eventName, hookPayload) + : input.eventName let resolvedPromptText = promptText let promptInteractionKey: string | undefined let hasTranscriptPromptEvidence = false @@ -213,5 +218,11 @@ export function normalizeProviderEvent(input: { payload = normalizeKiroEvent(state, eventName, promptText, paneKey, hookPayload) } - return { payload, resolvedPromptText, promptInteractionKey, hasTranscriptPromptEvidence } + return { + payload, + resolvedPromptText, + promptInteractionKey, + hasTranscriptPromptEvidence, + eventName + } } diff --git a/src/shared/agent-hook-listener/providers/codex-events.ts b/src/shared/agent-hook-listener/providers/codex-events.ts index 6781ce0d425..aa2f0f08e52 100644 --- a/src/shared/agent-hook-listener/providers/codex-events.ts +++ b/src/shared/agent-hook-listener/providers/codex-events.ts @@ -14,7 +14,10 @@ import { finishCodexSubagent, upsertCodexSubagent } from '../../codex-subagent-roster' -import { reconcileCodexSubagentTranscript } from '../../codex-subagent-transcript' +import { + codexTranscriptTurnEnd, + reconcileCodexSubagentTranscript +} from '../../codex-subagent-transcript' import { codexTurnApprovalsAreAutoReviewed, reconcileCodexSubagentReviewer @@ -239,7 +242,8 @@ export function normalizeCodexEvent( if (eventName === '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) + // Cleared, not deleted, so a reconcile cannot re-seed retired children from the previous row. + getOrCreateCodexSubagentRoster(state, paneKey).clear() } // 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. @@ -251,15 +255,60 @@ export function normalizeCodexEvent( stateName ) const previousLead = state.codexLeadStateByPaneKey.get(paneKey) + const transcript = state.codexSubagentTranscriptByPaneKey.get(paneKey) + const turnId = readString(hookPayload, 'turn_id') + // Why: Codex fires no Stop for a failed turn, so a Stop for one is the rollout's end replayed. + const failedTurn = + eventName === 'Stop' && + previousLead?.state !== 'done' && + turnId !== undefined && + codexTranscriptTurnEnd(transcript, turnId) === 'failed' + // Why: a compaction inside a running turn keeps that turn; one queued before the next turn follows an ended one. + const startsTurn = + eventName === 'UserPromptSubmit' || + (eventName === 'SessionStart' && + (readString(hookPayload, 'source') !== 'compact' || + previousLead?.state === 'done' || + !previousLead?.turnId || + codexTranscriptTurnEnd(transcript, previousLead.turnId) !== undefined)) const record = setCodexMainAgentTurnState(state, paneKey, { state: ownedState, - ...codexLeadOutcomeForEvent(eventName, previousLead, ownedState), + ...(failedTurn + ? { outcome: 'failure' as const } + : codexLeadOutcomeForEvent(eventName, previousLead, ownedState)), model: normalizeOptionalField(hookPayload['model'], AGENT_MODEL_MAX_LENGTH) ?? - (eventName === 'SessionStart' ? undefined : previousLead?.model) + (eventName === 'SessionStart' ? undefined : previousLead?.model), + sessionId: readString(hookPayload, 'session_id'), + turnId: turnId ?? (startsTurn ? null : undefined) }) return buildCodexStatusPayload(state, eventName, promptText, paneKey, hookPayload, { ...resolveCodexPaneStatus(state, paneKey, record), updateLead: true }) } + +/** A root hook for a turn whose rollout already recorded its end restates that end as Stop, + * since Codex skips Stop for internal subagents such as /review. A delayed hook for an ended + * turn the lead has moved past is dropped, so it can neither end nor rearm the current turn. */ +export function codexHookEventForEndedTurn( + state: HookListenerState, + paneKey: string, + eventName: unknown, + hookPayload: Record +): unknown { + const turnId = readString(hookPayload, 'turn_id') + if ( + !turnId || + eventName === 'Stop' || + eventName === 'Interrupt' || + readString(hookPayload, 'agent_id') + ) { + return eventName + } + if (!codexTranscriptTurnEnd(state.codexSubagentTranscriptByPaneKey.get(paneKey), turnId)) { + return eventName + } + const leadTurnId = state.codexLeadStateByPaneKey.get(paneKey)?.turnId + return leadTurnId && leadTurnId !== turnId ? undefined : 'Stop' +} diff --git a/src/shared/agent-hook-listener/providers/codex-state.ts b/src/shared/agent-hook-listener/providers/codex-state.ts index c13fdbf95ed..a22df449fe9 100644 --- a/src/shared/agent-hook-listener/providers/codex-state.ts +++ b/src/shared/agent-hook-listener/providers/codex-state.ts @@ -51,16 +51,24 @@ export function hasCodexTranscriptSubagents(state: HookListenerState, paneKey: s export function setCodexMainAgentTurnState( state: HookListenerState, paneKey: string, - next: Omit & { stateStartedAt?: number }, + next: Omit & { + stateStartedAt?: number + /** Omitted keeps the current turn; null starts a turn Codex sent no id for. */ + turnId?: string | null + }, now = Date.now() ): CodexLeadTurnState { const previous = state.codexLeadStateByPaneKey.get(paneKey) const continued = continueMainAgentStatus(previous, next, now) + const turnId = next.turnId === null ? undefined : (next.turnId ?? previous?.turnId) + const sessionId = next.sessionId ?? previous?.sessionId const record: CodexLeadTurnState = { state: next.state, ...(continued.outcome ? { outcome: continued.outcome } : {}), stateStartedAt: continued.stateStartedAt, - model: next.model + model: next.model, + ...(sessionId ? { sessionId } : {}), + ...(turnId ? { turnId } : {}) } state.codexLeadStateByPaneKey.set(paneKey, record) return record @@ -221,6 +229,10 @@ export function reconcileRemoteCodexState( } if (!eventName && !agentId && payload.mainAgent && payload.mainAgent.state !== 'blocked') { + if (payload.mainAgent.state === 'done' && !payload.subagents) { + // Why: an unlabelled host turn end (failed or cancelled) reporting no children has none left. + roster.clear() + } setCodexMainAgentTurnState(state, paneKey, { ...payload.mainAgent, state: payload.mainAgent.state, diff --git a/src/shared/agent-hook-listener/providers/codex-transcript-poll.ts b/src/shared/agent-hook-listener/providers/codex-transcript-poll.ts index 73a1b740d90..2ee6d495873 100644 --- a/src/shared/agent-hook-listener/providers/codex-transcript-poll.ts +++ b/src/shared/agent-hook-listener/providers/codex-transcript-poll.ts @@ -1,14 +1,22 @@ -import { reconcileCodexSubagentTranscript } from '../../codex-subagent-transcript' +import { record } from '../../codex-rollout-jsonl-cursor' +import { + codexTranscriptTurnEnd, + reconcileCodexSubagentTranscript +} from '../../codex-subagent-transcript' +import { normalizeHookPayload } from '../../agent-hook-listener' import type { AgentHookEventPayload } from '../listener-event' import type { HookListenerState } from '../listener-state' import { buildCodexChildDrivenStatusPayload } from './codex-events' import { getOrCreateCodexSubagentRoster, markCodexLeadTurnInterrupted } from './codex-state' -/** Polling reads new host-owned records without replaying the hook that started the turn. */ -export function pollCodexTranscriptStatus( +/** Polling reads new host-owned records without replaying the hook that started the turn; a turn + * end the rollout records is replayed as the root Stop that Codex skips for internal subagents. */ +export function pollCodexTranscriptStatus( state: HookListenerState, - original: T -): T | undefined { + original: AgentHookEventPayload, + body: unknown, + env: string +): AgentHookEventPayload | undefined { const transcript = state.codexSubagentTranscriptByPaneKey.get(original.paneKey) if (!transcript?.parent.filePath) { return undefined @@ -18,9 +26,29 @@ export function pollCodexTranscriptStatus( getOrCreateCodexSubagentRoster(state, original.paneKey), transcript.parent.filePath ) - const interrupted = - transcript.rootTurn.interrupted && - state.codexLeadStateByPaneKey.get(original.paneKey)?.state !== 'done' + const lead = state.codexLeadStateByPaneKey.get(original.paneKey) + const interrupted = transcript.rootTurn.interrupted && lead?.state !== 'done' + const envelope = record(body) + if ( + !interrupted && + envelope && + lead?.turnId && + lead.state !== 'done' && + codexTranscriptTurnEnd(transcript, lead.turnId) + ) { + const stop = { + ...envelope, + hook_event_name: undefined, + hookEventName: undefined, + payload: { + hook_event_name: 'Stop', + ...(lead.sessionId ? { session_id: lead.sessionId } : {}), + turn_id: lead.turnId, + transcript_path: transcript.parent.filePath + } + } + return normalizeHookPayload(state, 'codex', stop, env) ?? undefined + } if (!changed && !interrupted) { return original } @@ -28,5 +56,6 @@ export function pollCodexTranscriptStatus( markCodexLeadTurnInterrupted(state, original.paneKey) } const payload = buildCodexChildDrivenStatusPayload(state, undefined, original.paneKey, {}) - return payload ? { ...original, payload } : undefined + // Why: only a replayed Stop may carry an event name; this row restates the last hook's. + return payload ? { ...original, hookEventName: undefined, payload } : undefined } diff --git a/src/shared/agent-hook-listener/transcript-poll-policy.ts b/src/shared/agent-hook-listener/transcript-poll-policy.ts index 31cf93dac26..4397fd6524e 100644 --- a/src/shared/agent-hook-listener/transcript-poll-policy.ts +++ b/src/shared/agent-hook-listener/transcript-poll-policy.ts @@ -58,7 +58,18 @@ export function transcriptPollUpdate( const leadChanged = polled.payload.mainAgent?.state !== original.payload.mainAgent?.state || polled.payload.mainAgent?.outcome !== original.payload.mainAgent?.outcome - return subagentsChanged || leadChanged - ? { ...polled, hasExplicitPrompt: undefined, hookEventName: undefined } - : undefined + if (!subagentsChanged && !leadChanged) { + return undefined + } + // Why: a replayed root Stop keeps its name so every desktop sees a turn end; a failed one goes + // unlabelled like a cancellation, because desktops derive a labelled Stop's outcome as success. + const replayedStop = + polled.hookEventName === 'Stop' && + original.payload.mainAgent?.state !== 'done' && + polled.payload.mainAgent?.outcome !== 'failure' + return { + ...polled, + hasExplicitPrompt: undefined, + hookEventName: replayedStop ? 'Stop' : undefined + } } diff --git a/src/shared/codex-subagent-transcript.ts b/src/shared/codex-subagent-transcript.ts index 9f3f2ec81dc..62273cbba0a 100644 --- a/src/shared/codex-subagent-transcript.ts +++ b/src/shared/codex-subagent-transcript.ts @@ -24,6 +24,7 @@ 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 +const MAX_TRACKED_TURN_ENDS = 16 const SAFE_THREAD_ID = /^[A-Za-z0-9-]{1,64}$/ type TrackedTranscriptSubagent = JsonlCursor & { @@ -39,6 +40,7 @@ type TrackedTranscriptSubagent = JsonlCursor & { export type CodexSubagentTranscriptState = { parent: JsonlCursor rootTurn: CodexTranscriptTurn + turnEnds: Map subagents: Map /** Incremental reviewer cursors for child rollouts, which must not replace the parent cursor. */ reviewerCursorsByPath: Map @@ -161,6 +163,20 @@ function normalizedTranscriptPath(transcriptPath: string | undefined): string | : undefined } +function readTurnEnd( + recordValue: JsonRecord +): { turnId: string; end: 'completed' | 'failed' } | 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' && payload?.type !== 'turn_complete') || !turnId) { + return undefined + } + return { turnId, end: payload.error == null ? 'completed' : 'failed' } +} + function childIsComplete(records: JsonRecord[]): boolean { let complete = false for (const recordValue of records) { @@ -181,12 +197,20 @@ export function createCodexSubagentTranscriptState(): CodexSubagentTranscriptSta return { parent: { offset: 0, carry: '' }, rootTurn: { interrupted: false }, + turnEnds: new Map(), subagents: new Map(), reviewerCursorsByPath: new Map(), reviewersByPath: new Map() } } +export function codexTranscriptTurnEnd( + state: CodexSubagentTranscriptState | undefined, + turnId: string +): 'completed' | 'failed' | undefined { + return state?.turnEnds.get(turnId) +} + export function hasTrackedCodexTranscriptSubagents( state: CodexSubagentTranscriptState | undefined ): boolean { @@ -210,6 +234,7 @@ export function reconcileCodexSubagentTranscript( changed = true state.parent = { filePath: normalizedPath, offset: 0, carry: '' } state.rootTurn = { interrupted: false } + state.turnEnds.clear() state.subagents.clear() state.reviewerCursorsByPath.clear() state.reviewersByPath.clear() @@ -227,6 +252,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 diff --git a/src/wsl-guest/agent-hook-result-retry-scheduler.ts b/src/wsl-guest/agent-hook-result-retry-scheduler.ts index 0ba0b6f270f..242ac1e37d8 100644 --- a/src/wsl-guest/agent-hook-result-retry-scheduler.ts +++ b/src/wsl-guest/agent-hook-result-retry-scheduler.ts @@ -41,7 +41,7 @@ export type AgentHookResultRetryHost = { source: AgentHookSource, env?: string, version?: string - ) => void + ) => AgentHookEventPayload | undefined } export class AgentHookResultRetryScheduler { @@ -136,17 +136,17 @@ export class AgentHookResultRetryScheduler { } const event = source === 'codex' - ? pollCodexTranscriptStatus(this.host.state, original) + ? pollCodexTranscriptStatus(this.host.state, original, body, this.host.env) : normalizeHookPayload(this.host.state, source, body, this.host.env) if (!event) { return } const update = transcriptPollUpdate(source, original, event) - const next = update ?? original - if (update) { - this.host.applyEvent(update, source, env, version) + // Why: the next tick's identity check compares against the stored row, not the update it was copied from. + const next = update ? this.host.applyEvent(update, source, env, version) : original + if (next) { + this.scheduleTranscriptPoll(source, body, next, env, version) } - this.scheduleTranscriptPoll(source, body, next, env, version) } scheduleAssistantMessageRetry( diff --git a/src/wsl-guest/agent-hook-server.ts b/src/wsl-guest/agent-hook-server.ts index 301dde16671..c4116608c55 100644 --- a/src/wsl-guest/agent-hook-server.ts +++ b/src/wsl-guest/agent-hook-server.ts @@ -101,9 +101,8 @@ export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus { state: this.state, env: this.env, isListening: () => this.server !== null, - applyEvent: (event, source, env, version) => { + applyEvent: (event, source, env, version) => this.applyEvent(event, source, env, version, { checkPresence: false }) - } }) }