From 4f3e1f43e7835c85c22fe7da7c099cf702a0c859 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:18:57 -0700 Subject: [PATCH] fix(agent-hooks): harden Codex restart reconciliation --- .../server-codex-restart-diagnostic.test.ts | 135 ++++++++++++++++++ ...ver-last-status-hydrate-validation.test.ts | 34 +++++ src/main/agent-hooks/server.ts | 59 ++++++-- src/relay/agent-hook-codex-reconciliation.ts | 75 ++++++++++ .../agent-hook-event-application.test.ts | 50 +++++++ src/relay/agent-hook-event-application.ts | 9 +- ...k-server-codex-subagent-transcript.test.ts | 12 +- src/relay/agent-hook-server.ts | 18 +-- src/relay/agent-hook-status-cache.test.ts | 86 +++++++---- src/relay/agent-hook-status-cache.ts | 71 +++------ .../codex-subagent-transcript-seeding.ts | 1 + src/shared/codex-subagent-transcript.ts | 7 +- 12 files changed, 452 insertions(+), 105 deletions(-) create mode 100644 src/main/agent-hooks/server-codex-restart-diagnostic.test.ts create mode 100644 src/relay/agent-hook-codex-reconciliation.ts create mode 100644 src/relay/agent-hook-event-application.test.ts diff --git a/src/main/agent-hooks/server-codex-restart-diagnostic.test.ts b/src/main/agent-hooks/server-codex-restart-diagnostic.test.ts new file mode 100644 index 00000000000..36397172ae0 --- /dev/null +++ b/src/main/agent-hooks/server-codex-restart-diagnostic.test.ts @@ -0,0 +1,135 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { mkdtempSync, readFileSync, 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') + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +async function postCodexStatus(server: AgentHookServer, transcriptPath: string): Promise { + const env = server.buildPtyEnv() + await 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: { + hook_event_name: 'PostToolUse', + session_id: 'root-session', + transcript_path: transcriptPath, + tool_name: 'collaborationspawn_agent' + } + }) + }) +} + +describe('AgentHookServer Codex restart diagnostics', () => { + const dirs: string[] = [] + + afterEach(() => { + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + it('marks a vanished parent transcript unreadable without a child roster', async () => { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-missing-parent-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + writeFileSync(parentPath, line({ type: 'event_msg', payload: { type: 'task_started' } })) + const first = new AgentHookServer() + await first.start({ env: 'production', userDataPath: dir }) + try { + await postCodexStatus(first, parentPath) + } finally { + first.stop() + } + rmSync(parentPath) + + const second = new AgentHookServer() + await second.start({ env: 'production', userDataPath: dir }) + try { + await vi.waitFor( + () => + expect(second.getStatusSnapshot()[0]?.reconcileDiagnostic).toEqual({ + kind: 'unverifiable', + reason: 'transcript-unreadable', + observedAt: expect.any(Number) + }), + { timeout: 7_000, interval: 100 } + ) + } finally { + second.stop() + } + }) + + it('clears a persisted transcript diagnostic after the parent becomes readable', async () => { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-parent-recovery-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const cachePath = join(dir, 'agent-hooks', 'last-status.json') + const seed = new AgentHookServer() + await seed.start({ env: 'production', userDataPath: dir }) + try { + await postCodexStatus(seed, parentPath) + } finally { + seed.stop() + } + const persisted = JSON.parse(readFileSync(cachePath, 'utf8')) as { + entries: Record> + } + persisted.entries[PANE_KEY] = { + ...persisted.entries[PANE_KEY], + reconcileDiagnostic: { + kind: 'unverifiable', + reason: 'transcript-unreadable', + observedAt: 123 + } + } + writeFileSync(cachePath, JSON.stringify(persisted)) + writeFileSync(parentPath, line({ type: 'event_msg', payload: { type: 'task_started' } })) + + const recovered = new AgentHookServer() + await recovered.start({ env: 'production', userDataPath: dir }) + try { + await vi.waitFor( + () => expect(recovered.getStatusSnapshot()[0]?.reconcileDiagnostic).toBeNull(), + { timeout: 2_500, interval: 50 } + ) + } finally { + recovered.stop() + } + }) + + it('releases generation fences when unique Codex panes are cleared', () => { + const server = new AgentHookServer() + for (let index = 0; index < 512; index += 1) { + const paneKey = makePaneKey(`tab-${index}`, '11111111-1111-4111-8111-111111111111') + server.ingestRemote( + { + paneKey, + tabId: `tab-${index}`, + hookEventName: 'SessionStart', + source: 'codex', + payload: { state: 'working', prompt: 'new turn', agentType: 'codex' } + }, + 'connection-1' + ) + server.clearPaneState(paneKey) + } + + expect(server._getCodexPaneGenerationCountForTests()).toBe(0) + }) +}) diff --git a/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts b/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts index 3ad27b64a38..f060e1fd590 100644 --- a/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts +++ b/src/main/agent-hooks/server-last-status-hydrate-validation.test.ts @@ -386,6 +386,40 @@ describe('Last-status persistence', () => { } }) + it('drops a hydrated remote entry with an empty connection identity', async () => { + mkdirSync(join(userDataPath, 'agent-hooks'), { recursive: true }) + writeFileSync( + lastStatusPath(), + JSON.stringify({ + version: 2, + entries: { + [PANE]: { + paneKey: PANE, + tabId: 'tab-1', + worktreeId: 'wt-1', + connectionId: ' ', + receivedAt: recentTs(), + stateStartedAt: recentTs(-1000), + providerSession: { + key: 'session_id', + id: 'remote-codex-session', + transcriptPath: '/remote/codex/rollout.jsonl' + }, + payload: { state: 'working', prompt: 'remote cached', agentType: 'codex' } + } + } + }), + 'utf8' + ) + const server = new AgentHookServer() + await server.start({ env: 'production', userDataPath }) + try { + expect(server.getStatusSnapshot()).toEqual([]) + } finally { + server.stop() + } + }) + it('clearPaneState evicts the entry from the on-disk file', async () => { const server = new AgentHookServer() await server.start({ diff --git a/src/main/agent-hooks/server.ts b/src/main/agent-hooks/server.ts index 0293ee72f25..c6c6456a431 100644 --- a/src/main/agent-hooks/server.ts +++ b/src/main/agent-hooks/server.ts @@ -227,6 +227,7 @@ const ASSISTANT_MESSAGE_RETRY_MS = 50 const CODEX_SUBAGENT_POLL_MS = 1_000 const CODEX_RESTART_RECONCILE_ATTEMPTS = 5 const CODEX_RESTART_RECONCILE_MS = 1_000 +const CODEX_RESTART_RECONCILE_YIELD_MS = 25 const INTERRUPTED_DONE_LATE_WORKING_SUPPRESSION_MS = 15_000 // Why: starts at 2 — pre-merge v1 lacked receivedAt/stateStartedAt (never shipped); a mismatched version hydrates empty (treated as corrupt). @@ -351,13 +352,16 @@ function sanitizeHydratedEntry( ) { return null } - // Why: connectionId is null (local) or string (relay); any other shape is rejected to keep the typed surface honest. + // Why: only null is local; an empty relay id must not grant local transcript access. const connectionIdRaw = record.connectionId let connectionId: string | null if (connectionIdRaw === null || connectionIdRaw === undefined) { connectionId = null } else if (typeof connectionIdRaw === 'string') { - connectionId = connectionIdRaw + connectionId = connectionIdRaw.trim() + if (!connectionId) { + return null + } } else { return null } @@ -755,6 +759,8 @@ export class AgentHookServer { private codexSubagentPollTimers = new Map>() private codexRestartReconcileTimers = new Map>() private codexPaneGenerations = new Map() + private codexPaneGenerationCounter = 0 + private codexRestartReconcileNextRunAt = 0 private promptSentDedupeByPaneKey = new Map() private activeHookTurnCompletedAtByPaneKey = new Map() private promptSentHashSalt = randomBytes(16).toString('hex') @@ -1677,11 +1683,16 @@ export class AgentHookServer { } private bumpCodexPaneGeneration(paneKey: string): number { - const generation = (this.codexPaneGenerations.get(paneKey) ?? 0) + 1 + const generation = ++this.codexPaneGenerationCounter this.codexPaneGenerations.set(paneKey, generation) return generation } + private retireCodexPaneGeneration(paneKey: string): void { + this.bumpCodexPaneGeneration(paneKey) + this.codexPaneGenerations.delete(paneKey) + } + private scheduleCodexRestartReconciliation(paneKey: string): void { const entry = this.state.lastStatusByPaneKey.get(paneKey) as | EnrichedAgentHookEventPayload @@ -1692,7 +1703,7 @@ export class AgentHookServer { } // Transcript paths for relay/SSH/WSL panes belong to the execution host; reconciliation is // performed there and only its result crosses the wire. - if (entry.connectionId) { + if (entry.connectionId !== null) { return } if (!transcriptPath?.trim()) { @@ -1713,11 +1724,21 @@ export class AgentHookServer { return } this.clearCodexRestartReconcile(paneKey) - const generation = this.codexPaneGenerations.get(paneKey) ?? 0 + const generation = + this.codexPaneGenerations.get(paneKey) ?? this.bumpCodexPaneGeneration(paneKey) let original = entry let attempt = 0 const reconcile = (): void => { this.codexRestartReconcileTimers.delete(paneKey) + const gateDelay = this.codexRestartReconcileNextRunAt - Date.now() + if (gateDelay > 0) { + const timer = setTimeout(reconcile, gateDelay) + this.codexRestartReconcileTimers.set(paneKey, timer) + if (typeof timer.unref === 'function') { + timer.unref() + } + return + } const current = this.state.lastStatusByPaneKey.get(paneKey) as | EnrichedAgentHookEventPayload | undefined @@ -1727,8 +1748,12 @@ export class AgentHookServer { const transcript = getOrCreateCodexSubagentTranscriptState(this.state, paneKey) const roster = getOrCreateCodexSubagentRoster(this.state, paneKey) reconcileCodexSubagentTranscript(transcript, roster, transcriptPath) + this.codexRestartReconcileNextRunAt = Date.now() + CODEX_RESTART_RECONCILE_YIELD_MS const subagents = codexRosterToSnapshots(roster) const terminal = transcript.parentTerminalObserved === true + const transcriptUnreadable = + transcript.parentReadable === false || + [...transcript.subagents.values()].some((child) => child.unresolvedSince) const nextPayload = { ...current.payload, ...(terminal @@ -1744,7 +1769,8 @@ export class AgentHookServer { let latest = current if ( JSON.stringify(nextPayload) !== JSON.stringify(current.payload) || - (terminal && current.restoredUnconfirmed) + (terminal && current.restoredUnconfirmed) || + (current.reconcileDiagnostic?.reason === 'transcript-unreadable' && !transcriptUnreadable) ) { const next: EnrichedAgentHookEventPayload = { ...current, @@ -1761,9 +1787,8 @@ export class AgentHookServer { } attempt += 1 if (!terminal && attempt >= CODEX_RESTART_RECONCILE_ATTEMPTS) { - const unreadable = [...transcript.subagents.values()].some((child) => child.unresolvedSince) if ( - unreadable && + transcriptUnreadable && (latest.reconcileDiagnostic === undefined || latest.reconcileDiagnostic === null) ) { const diagnostic: EnrichedAgentHookEventPayload = { @@ -2097,7 +2122,7 @@ export class AgentHookServer { this.clearAssistantMessageRetry(previousOwnerPaneKey) this.clearCodexSubagentPoll(previousOwnerPaneKey) this.clearCodexRestartReconcile(previousOwnerPaneKey) - this.bumpCodexPaneGeneration(previousOwnerPaneKey) + this.retireCodexPaneGeneration(previousOwnerPaneKey) this.scheduleCodexRestartReconciliation(toPaneKey) // Why: the live process keeps posting the physical source key after detach; persist a chain-safe mapping to the current owner. this.legacyPaneKeyAliases.set(physicalPaneKey, { @@ -2138,7 +2163,7 @@ export class AgentHookServer { this.clearAssistantMessageRetry(key) this.clearCodexSubagentPoll(key) this.clearCodexRestartReconcile(key) - this.bumpCodexPaneGeneration(key) + this.retireCodexPaneGeneration(key) clearPaneCacheState(this.state, key) this.activeHookTurnCompletedAtByPaneKey.delete(key) this.runtimeObservedStatusPaneKeys.delete(key) @@ -2875,6 +2900,8 @@ export class AgentHookServer { } this.codexRestartReconcileTimers.clear() this.codexPaneGenerations.clear() + this.codexPaneGenerationCounter = 0 + this.codexRestartReconcileNextRunAt = 0 // Why: don't unlink the endpoint file — a stale file matches fail-open and avoids a TOCTOU race with a concurrent Orca. this.endpointDir = null this.endpointFilePathCache = null @@ -3004,7 +3031,7 @@ export class AgentHookServer { // Why: a replacement remote process may reuse the pane; don't merge it with the lost connection's children. this.clearCodexSubagentPoll(paneKey) this.clearCodexRestartReconcile(paneKey) - this.bumpCodexPaneGeneration(paneKey) + this.retireCodexPaneGeneration(paneKey) this.state.codexSubagentRosterByPaneKey.delete(paneKey) this.state.codexSubagentTranscriptByPaneKey.delete(paneKey) this.state.codexLeadStateByPaneKey.delete(paneKey) @@ -3055,7 +3082,7 @@ export class AgentHookServer { this.clearAssistantMessageRetry(resolvedPaneKey) this.clearCodexSubagentPoll(resolvedPaneKey) this.clearCodexRestartReconcile(resolvedPaneKey) - this.bumpCodexPaneGeneration(resolvedPaneKey) + this.retireCodexPaneGeneration(resolvedPaneKey) this.runtimeObservedStatusPaneKeys.delete(resolvedPaneKey) this.currentAuthorityObservations.delete(resolvedPaneKey) if (existing.payload.state === 'done') { @@ -3130,7 +3157,7 @@ export class AgentHookServer { this.clearAssistantMessageRetry(paneKey) this.clearCodexSubagentPoll(paneKey) this.clearCodexRestartReconcile(paneKey) - this.bumpCodexPaneGeneration(paneKey) + this.retireCodexPaneGeneration(paneKey) clearPaneCacheState(this.state, paneKey) this.activeHookTurnCompletedAtByPaneKey.delete(paneKey) this.runtimeObservedStatusPaneKeys.delete(paneKey) @@ -3155,7 +3182,7 @@ export class AgentHookServer { this.clearAssistantMessageRetry(resolvedPaneKey) this.clearCodexSubagentPoll(resolvedPaneKey) this.clearCodexRestartReconcile(resolvedPaneKey) - this.bumpCodexPaneGeneration(resolvedPaneKey) + this.retireCodexPaneGeneration(resolvedPaneKey) clearPaneCacheState(this.state, resolvedPaneKey) this.activeHookTurnCompletedAtByPaneKey.delete(resolvedPaneKey) this.currentAuthorityObservations.delete(resolvedPaneKey) @@ -3662,6 +3689,10 @@ export class AgentHookServer { return this.state } + _getCodexPaneGenerationCountForTests(): number { + return this.codexPaneGenerations.size + } + _resetPromptSentDedupeForTests(): void { this.promptSentDedupeByPaneKey.clear() } diff --git a/src/relay/agent-hook-codex-reconciliation.ts b/src/relay/agent-hook-codex-reconciliation.ts new file mode 100644 index 00000000000..e97804ae64a --- /dev/null +++ b/src/relay/agent-hook-codex-reconciliation.ts @@ -0,0 +1,75 @@ +import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import type { HookListenerState } from '../shared/agent-hook-listener/listener-state' +import { + getOrCreateCodexSubagentRoster, + getOrCreateCodexSubagentTranscriptState, + seedCodexStateFromSnapshot +} from '../shared/agent-hook-listener/providers/codex-state' +import { codexRosterToSnapshots } from '../shared/codex-subagent-roster' +import { reconcileCodexSubagentTranscript } from '../shared/codex-subagent-transcript' +import { seedCodexSubagentTranscriptFromSnapshot } from '../shared/codex-subagent-transcript-seeding' +import { createRelayCodexReconciler, type RelayHookStatusMeta } from './agent-hook-status-cache' +import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay' + +export function createRelayCodexReconciliationSchedulers(options: { + state: HookListenerState + isListening: () => boolean + timers: Map> + metadata: ReadonlyMap + forward: (envelope: AgentHookRelayEnvelope) => void + persist: () => void + gate: { nextRunAt: number } +}): { live: (paneKey: string) => void; restart: (paneKey: string) => void } { + return { + live: createRelayCodexReconciler({ + ...options, + reconcile: (event) => reconcileRelayCodexEvent(options.state, event) + }), + restart: createRelayCodexReconciler({ + ...options, + isReplay: true, + reconcile: (event) => + reconcileRelayCodexEvent(options.state, event, { reconcileParentState: true }) + }) + } +} + +export function reconcileRelayCodexEvent( + state: HookListenerState, + event: AgentHookEventPayload, + options: { reconcileParentState?: boolean } = {} +): AgentHookEventPayload { + const transcriptPath = event.providerSession?.transcriptPath + if (!transcriptPath || event.payload.agentType !== 'codex') { + return event + } + seedCodexStateFromSnapshot(state, event.paneKey, event.payload) + const transcript = getOrCreateCodexSubagentTranscriptState(state, event.paneKey) + if (event.payload.subagents?.length) { + seedCodexSubagentTranscriptFromSnapshot(transcript, event.payload.subagents, transcriptPath) + } + const roster = getOrCreateCodexSubagentRoster(state, event.paneKey) + reconcileCodexSubagentTranscript(transcript, roster, transcriptPath) + const subagents = codexRosterToSnapshots(roster) + const payload = { + ...event.payload, + ...(subagents ? { subagents } : { subagents: undefined }), + ...(options.reconcileParentState + ? transcript.parentTerminalObserved === true + ? { state: 'done' as const } + : transcript.parentTerminalObserved === false + ? { + state: event.payload.state === 'waiting' ? ('waiting' as const) : ('working' as const) + } + : {} + : {}) + } + const transcriptUnreadable = + transcript.parentReadable === false || + [...transcript.subagents.values()].some((child) => child.unresolvedSince) + return transcriptUnreadable + ? { ...event, payload } + : event.reconcileDiagnostic?.reason === 'transcript-unreadable' + ? { ...event, payload, reconcileDiagnostic: null } + : { ...event, payload } +} diff --git a/src/relay/agent-hook-event-application.test.ts b/src/relay/agent-hook-event-application.test.ts new file mode 100644 index 00000000000..0c68bb1277d --- /dev/null +++ b/src/relay/agent-hook-event-application.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it, vi } from 'vitest' + +import { createHookListenerState } from '../shared/agent-hook-listener/listener-state' +import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import { applyRelayEvent } from './agent-hook-event-application' + +const PANE_KEY = 'tab-1:11111111-1111-4111-8111-111111111111' + +function apply(isReplay: boolean): { + live: ReturnType + restart: ReturnType +} { + const scheduleCodexReconciliation = vi.fn() + const scheduleCodexRestartReconciliation = vi.fn() + const event: AgentHookEventPayload = { + paneKey: PANE_KEY, + source: 'codex', + connectionId: null, + hookEventName: 'UserPromptSubmit', + payload: { state: 'working', prompt: 'new turn', agentType: 'codex' } + } + applyRelayEvent({ + state: createHookListenerState(), + event, + source: 'codex', + isReplay, + metadata: new Map(), + persist: vi.fn(), + clearPaneState: vi.fn(), + forward: vi.fn(), + scheduleCodexReconciliation, + scheduleCodexRestartReconciliation, + clearAssistantMessageRetry: vi.fn() + }) + return { live: scheduleCodexReconciliation, restart: scheduleCodexRestartReconciliation } +} + +describe('relay hook event application', () => { + it('uses live reconciliation for live events', () => { + const scheduled = apply(false) + expect(scheduled.live).toHaveBeenCalledWith(PANE_KEY) + expect(scheduled.restart).not.toHaveBeenCalled() + }) + + it('preserves replay provenance for spool replay reconciliation', () => { + const scheduled = apply(true) + expect(scheduled.restart).toHaveBeenCalledWith(PANE_KEY) + expect(scheduled.live).not.toHaveBeenCalled() + }) +}) diff --git a/src/relay/agent-hook-event-application.ts b/src/relay/agent-hook-event-application.ts index 0d94d76a2d9..20a668cb379 100644 --- a/src/relay/agent-hook-event-application.ts +++ b/src/relay/agent-hook-event-application.ts @@ -1,4 +1,5 @@ -import { applyRelayHookEvent, reconcileRelayCodexEvent } from './agent-hook-status-cache' +import { applyRelayHookEvent } from './agent-hook-status-cache' +import { reconcileRelayCodexEvent } from './agent-hook-codex-reconciliation' import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' import type { AgentHookSource, AgentHookRelayEnvelope } from '../shared/agent-hook-relay' import type { HookListenerState } from '../shared/agent-hook-listener/listener-state' @@ -15,6 +16,7 @@ export function applyRelayEvent(options: { persist: () => void clearPaneState: (paneKey: string) => void forward: (envelope: AgentHookRelayEnvelope) => void + scheduleCodexReconciliation: (paneKey: string) => void scheduleCodexRestartReconciliation: (paneKey: string) => void clearAssistantMessageRetry: (paneKey: string) => void }): void { @@ -45,6 +47,9 @@ export function applyRelayEvent(options: { options.forward(options.isReplay ? { ...envelope, isReplay: true } : envelope) }) if (reconciled.payload.agentType === 'codex') { - options.scheduleCodexRestartReconciliation(reconciled.paneKey) + const schedule = options.isReplay + ? options.scheduleCodexRestartReconciliation + : options.scheduleCodexReconciliation + schedule(reconciled.paneKey) } } diff --git a/src/relay/agent-hook-server-codex-subagent-transcript.test.ts b/src/relay/agent-hook-server-codex-subagent-transcript.test.ts index 699301ec616..f78cb2c782a 100644 --- a/src/relay/agent-hook-server-codex-subagent-transcript.test.ts +++ b/src/relay/agent-hook-server-codex-subagent-transcript.test.ts @@ -130,8 +130,16 @@ describe('RelayAgentHookServer Codex subagent transcript polling', () => { await second.start() try { expect(second.replayCachedPayloadsForPanes()).toBe(1) - expect(secondForward.mock.calls.at(-1)?.[0].payload.subagents).toBeUndefined() - expect(secondForward.mock.calls.at(-1)?.[0].payload.state).toBe('working') + expect(secondForward.mock.calls.at(-1)?.[0].payload.subagents).toHaveLength(1) + await vi.waitFor( + () => { + expect(secondForward.mock.calls.at(-1)?.[0].payload.subagents).toBeUndefined() + expect(secondForward.mock.calls.at(-1)?.[0].payload.state).toBe('working') + expect(secondForward.mock.calls.at(-1)?.[0].isReplay).toBe(true) + expect(secondForward).toHaveBeenCalledTimes(2) + }, + { timeout: 2_000, interval: 50 } + ) } finally { second.stop() } diff --git a/src/relay/agent-hook-server.ts b/src/relay/agent-hook-server.ts index 814b9f4eeec..1840c94a21e 100644 --- a/src/relay/agent-hook-server.ts +++ b/src/relay/agent-hook-server.ts @@ -37,12 +37,11 @@ import { buildRelayHookEnvelope, hookBodyEnv, hookBodyVersion } from './agent-ho import { AgentHookResultRetryScheduler } from './agent-hook-result-retry-scheduler' import { listenRelayHttpServer } from './relay-http-listener' import { - createRelayCodexReconciler, hydrateRelayHookStatusCache, persistRelayHookStatusCache, - reconcileRelayCodexEvent, type RelayHookStatusMeta } from './agent-hook-status-cache' +import { createRelayCodexReconciliationSchedulers } from './agent-hook-codex-reconciliation' import { RelayHookStatusCacheWriter } from './agent-hook-status-cache-writer' import { handleRelayHookRequest } from './agent-hook-request' import { applyRelayEvent } from './agent-hook-event-application' @@ -80,7 +79,8 @@ export class RelayAgentHookServer { private retryScheduler: AgentHookResultRetryScheduler private cacheFilePath: string private codexRestartReconcileTimers = new Map>() - private scheduleCodexRestartReconciliation: (paneKey: string) => void + private codexReconcileGate = { nextRunAt: 0 } + private codexReconciliationSchedulers: ReturnType private statusCacheWriter: RelayHookStatusCacheWriter constructor(options: RelayHookServerOptions) { this.env = options.env ?? REMOTE_AGENT_HOOK_ENV @@ -106,14 +106,14 @@ export class RelayAgentHookServer { this.applyEvent(event, source, env, version) } }) - this.scheduleCodexRestartReconciliation = createRelayCodexReconciler({ + this.codexReconciliationSchedulers = createRelayCodexReconciliationSchedulers({ state: this.state, isListening: () => this.server !== null, timers: this.codexRestartReconcileTimers, - reconcile: (event) => reconcileRelayCodexEvent(this.state, event), metadata: this.lastEnvelopeMetaByPaneKey, forward: this.forward, - persist: () => this.statusCacheWriter.schedule() + persist: () => this.statusCacheWriter.schedule(), + gate: this.codexReconcileGate }) } async start(options: RelayHookServerStartOptions = {}): Promise { @@ -137,7 +137,7 @@ export class RelayAgentHookServer { const hydratedMetadata = hydrateRelayHookStatusCache( this.cacheFilePath, this.state, - (paneKey) => this.scheduleCodexRestartReconciliation(paneKey) + (paneKey) => this.codexReconciliationSchedulers.restart(paneKey) ) this.lastEnvelopeMetaByPaneKey.clear() for (const [paneKey, metadata] of hydratedMetadata) { @@ -208,6 +208,7 @@ export class RelayAgentHookServer { clearTimeout(timer) } this.codexRestartReconcileTimers.clear() + this.codexReconcileGate.nextRunAt = 0 clearAllListenerCaches(this.state) this.lastEnvelopeMetaByPaneKey.clear() } @@ -298,7 +299,8 @@ export class RelayAgentHookServer { persist: () => this.statusCacheWriter.schedule(), clearPaneState: (paneKey) => this.clearPaneState(paneKey), forward: this.forward, - scheduleCodexRestartReconciliation: this.scheduleCodexRestartReconciliation, + scheduleCodexReconciliation: this.codexReconciliationSchedulers.live, + scheduleCodexRestartReconciliation: this.codexReconciliationSchedulers.restart, clearAssistantMessageRetry: (paneKey) => this.retryScheduler.clearAssistantMessageRetry(paneKey) }) diff --git a/src/relay/agent-hook-status-cache.test.ts b/src/relay/agent-hook-status-cache.test.ts index bef52e4aeba..406083c3c8f 100644 --- a/src/relay/agent-hook-status-cache.test.ts +++ b/src/relay/agent-hook-status-cache.test.ts @@ -5,7 +5,8 @@ import { join } from 'node:path' import { createHookListenerState } from '../shared/agent-hook-listener/listener-state' import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' -import { applyRelayHookEvent, reconcileRelayCodexEvent } from './agent-hook-status-cache' +import { applyRelayHookEvent } from './agent-hook-status-cache' +import { reconcileRelayCodexEvent } from './agent-hook-codex-reconciliation' import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay' const PANE_KEY = 'tab-1:11111111-1111-4111-8111-111111111111' @@ -35,22 +36,26 @@ describe('relay agent-hook status cache', () => { ) const state = createHookListenerState() - const reconciled = reconcileRelayCodexEvent(state, { - ...event('PostToolUse'), - providerSession: { key: 'session_id', id: 'session-1', transcriptPath }, - payload: { - state: 'done', - prompt: 'prompt', - agentType: 'codex', - subagents: [ - { - id: '019fa65f-3144-7151-9c02-cff7a28f316f', - state: 'working', - startedAt: 1234 - } - ] - } - }) + const reconciled = reconcileRelayCodexEvent( + state, + { + ...event('PostToolUse'), + providerSession: { key: 'session_id', id: 'session-1', transcriptPath }, + payload: { + state: 'done', + prompt: 'prompt', + agentType: 'codex', + subagents: [ + { + id: '019fa65f-3144-7151-9c02-cff7a28f316f', + state: 'working', + startedAt: 1234 + } + ] + } + }, + { reconcileParentState: true } + ) expect(reconciled.payload.state).toBe('working') } finally { @@ -84,16 +89,20 @@ describe('relay agent-hook status cache', () => { ) const state = createHookListenerState() - const reconciled = reconcileRelayCodexEvent(state, { - ...event('PostToolUse'), - providerSession: { key: 'session_id', id: 'session-1', transcriptPath }, - payload: { - state: 'waiting', - prompt: 'permission', - agentType: 'codex', - subagents: [{ id: childId, state: 'waiting', startedAt: 1234 }] - } - }) + const reconciled = reconcileRelayCodexEvent( + state, + { + ...event('PostToolUse'), + providerSession: { key: 'session_id', id: 'session-1', transcriptPath }, + payload: { + state: 'waiting', + prompt: 'permission', + agentType: 'codex', + subagents: [{ id: childId, state: 'waiting', startedAt: 1234 }] + } + }, + { reconcileParentState: true } + ) expect(reconciled.payload.state).toBe('waiting') expect(reconciled.payload.subagents?.[0]?.state).toBe('waiting') @@ -102,6 +111,29 @@ describe('relay agent-hook status cache', () => { } }) + it('does not rewrite a live turn from a historical parent terminal record', () => { + const dir = mkdtempSync(join(tmpdir(), 'relay-hook-status-cache-live-turn-')) + try { + const transcriptPath = join(dir, 'rollout-parent.jsonl') + writeFileSync( + transcriptPath, + `${JSON.stringify({ type: 'event_msg', payload: { type: 'task_complete' } })}\n` + ) + const state = createHookListenerState() + const live = { + ...event('UserPromptSubmit'), + providerSession: { key: 'session_id' as const, id: 'session-1', transcriptPath } + } + + expect(reconcileRelayCodexEvent(state, live).payload.state).toBe('working') + expect( + reconcileRelayCodexEvent(state, live, { reconcileParentState: true }).payload.state + ).toBe('done') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + it('clears an inherited Codex reconciliation diagnostic at SessionStart', () => { const state = createHookListenerState() const metadata = new Map() diff --git a/src/relay/agent-hook-status-cache.ts b/src/relay/agent-hook-status-cache.ts index b29865a6912..cf6066cc1e9 100644 --- a/src/relay/agent-hook-status-cache.ts +++ b/src/relay/agent-hook-status-cache.ts @@ -8,15 +8,8 @@ import { type AgentHookRelayEnvelope, type AgentHookSource } from '../shared/agent-hook-relay' -import { codexRosterToSnapshots } from '../shared/codex-subagent-roster' -import { reconcileCodexSubagentTranscript } from '../shared/codex-subagent-transcript' import { buildRelayHookEnvelope } from './agent-hook-envelope-build' import { parsePaneKey } from '../shared/stable-pane-id' -import { - getOrCreateCodexSubagentTranscriptState, - seedCodexStateFromSnapshot -} from '../shared/agent-hook-listener/providers/codex-state' -import { seedCodexSubagentTranscriptFromSnapshot } from '../shared/codex-subagent-transcript-seeding' import { normalizeAgentProviderSession } from '../shared/agent-session-resume' import { normalizeAgentReconcileDiagnostic } from '../shared/agent-reconcile-diagnostic' @@ -35,6 +28,7 @@ const CACHE_VERSION = 1 const MAX_CACHED_PANES = 256 const MAX_RECONCILE_ATTEMPTS = 5 const RECONCILE_INTERVAL_MS = 1_000 +const RECONCILE_YIELD_MS = 25 function sanitizeHydratedEntry( rawEntry: unknown @@ -137,43 +131,6 @@ export function applyRelayHookEvent(options: { options.forward(buildRelayHookEnvelope(cachedEvent, options.source, options.env, options.version)) } -export function reconcileRelayCodexEvent( - state: HookListenerState, - event: AgentHookEventPayload -): AgentHookEventPayload { - const transcriptPath = event.providerSession?.transcriptPath - if (!transcriptPath || event.payload.agentType !== 'codex') { - return event - } - seedCodexStateFromSnapshot(state, event.paneKey, event.payload) - const transcript = getOrCreateCodexSubagentTranscriptState(state, event.paneKey) - if (event.payload.subagents?.length) { - seedCodexSubagentTranscriptFromSnapshot(transcript, event.payload.subagents, transcriptPath) - } - const roster = state.codexSubagentRosterByPaneKey.get(event.paneKey) - if (!transcript || !roster) { - return event - } - reconcileCodexSubagentTranscript(transcript, roster, transcriptPath) - const subagents = codexRosterToSnapshots(roster) - const payload = { - ...event.payload, - ...(subagents ? { subagents } : { subagents: undefined }), - ...(transcript.parentTerminalObserved === true - ? { state: 'done' as const } - : transcript.parentTerminalObserved === false - ? { - state: event.payload.state === 'waiting' ? ('waiting' as const) : ('working' as const) - } - : {}) - } - return [...transcript.subagents.values()].some((child) => child.unresolvedSince) - ? { ...event, payload } - : event.reconcileDiagnostic !== undefined - ? { ...event, payload, reconcileDiagnostic: null } - : { ...event, payload } -} - export function hydrateRelayHookStatusCache( filePath: string, state: HookListenerState, @@ -201,10 +158,7 @@ export function hydrateRelayHookStatusCache( if (!entry) { continue } - const event = - entry.event.payload.agentType === 'codex' - ? reconcileRelayCodexEvent(state, entry.event) - : entry.event + const event = entry.event state.lastStatusByPaneKey.set(event.paneKey, event) metadata.set(event.paneKey, entry.meta) if (event.payload.agentType === 'codex') { @@ -248,7 +202,10 @@ export function scheduleRelayCodexReconciliation(options: { metadata: ReadonlyMap forward: (envelope: AgentHookRelayEnvelope) => void persist: () => void + gate: { nextRunAt: number } + isReplay?: boolean attempt?: number + delayMs?: number }): void { const prior = options.timers.get(options.paneKey) if (prior) { @@ -256,6 +213,11 @@ export function scheduleRelayCodexReconciliation(options: { } const timer = setTimeout(() => { options.timers.delete(options.paneKey) + const gateDelay = options.gate.nextRunAt - Date.now() + if (gateDelay > 0) { + scheduleRelayCodexReconciliation({ ...options, delayMs: gateDelay }) + return + } if (!options.isListening()) { return } @@ -264,11 +226,13 @@ export function scheduleRelayCodexReconciliation(options: { return } const next = options.reconcile(current) + options.gate.nextRunAt = Date.now() + RECONCILE_YIELD_MS const transcript = options.state.codexSubagentTranscriptByPaneKey.get(options.paneKey) const attempt = options.attempt ?? 0 const unreadable = attempt + 1 >= MAX_RECONCILE_ATTEMPTS && (!current.providerSession?.transcriptPath || + transcript?.parentReadable === false || Boolean( transcript && [...transcript.subagents.values()].some((child) => child.unresolvedSince) )) @@ -287,13 +251,18 @@ export function scheduleRelayCodexReconciliation(options: { options.persist() const meta = options.metadata.get(options.paneKey) if (meta) { - options.forward(buildRelayHookEnvelope(updated, meta.source, meta.env, meta.version)) + const envelope = buildRelayHookEnvelope(updated, meta.source, meta.env, meta.version) + options.forward(options.isReplay ? { ...envelope, isReplay: true } : envelope) } } if (!unreadable && attempt + 1 < MAX_RECONCILE_ATTEMPTS) { - scheduleRelayCodexReconciliation({ ...options, attempt: attempt + 1 }) + scheduleRelayCodexReconciliation({ + ...options, + attempt: attempt + 1, + delayMs: RECONCILE_INTERVAL_MS + }) } - }, RECONCILE_INTERVAL_MS) + }, options.delayMs ?? RECONCILE_INTERVAL_MS) options.timers.set(options.paneKey, timer) if (typeof timer.unref === 'function') { timer.unref() diff --git a/src/shared/codex-subagent-transcript-seeding.ts b/src/shared/codex-subagent-transcript-seeding.ts index d3ad165bde3..ae7c9e9847b 100644 --- a/src/shared/codex-subagent-transcript-seeding.ts +++ b/src/shared/codex-subagent-transcript-seeding.ts @@ -21,6 +21,7 @@ export function seedCodexSubagentTranscriptFromSnapshot( state.parent = { filePath: normalizedPath, offset: 0, carry: '' } state.subagents.clear() state.parentTerminalObserved = undefined + state.parentReadable = undefined } for (const snapshot of snapshots) { if (!SAFE_THREAD_ID.test(snapshot.id) || state.subagents.has(snapshot.id)) { diff --git a/src/shared/codex-subagent-transcript.ts b/src/shared/codex-subagent-transcript.ts index 7e0c5b3f15c..2fc8be6deaa 100644 --- a/src/shared/codex-subagent-transcript.ts +++ b/src/shared/codex-subagent-transcript.ts @@ -35,6 +35,7 @@ export type CodexSubagentTranscriptState = { parent: JsonlCursor subagents: Map parentTerminalObserved?: boolean + parentReadable?: boolean } type JsonRecord = Record @@ -187,6 +188,7 @@ export function reconcileCodexSubagentTranscript( ): void { const normalizedPath = transcriptPath?.trim() if (!normalizedPath || !isAbsolute(normalizedPath) || extname(normalizedPath) !== '.jsonl') { + state.parentReadable = false return } if (state.parent.filePath !== normalizedPath) { @@ -196,8 +198,11 @@ export function reconcileCodexSubagentTranscript( state.parent = { filePath: normalizedPath, offset: 0, carry: '' } state.subagents.clear() state.parentTerminalObserved = undefined + state.parentReadable = undefined } - for (const recordValue of readJsonlCursor(state.parent) ?? []) { + const parentRecords = readJsonlCursor(state.parent) + state.parentReadable = parentRecords !== undefined + for (const recordValue of parentRecords ?? []) { if (recordValue.type === 'event_msg') { const eventType = record(recordValue.payload)?.type if (eventType === 'task_started' || eventType === 'task_complete') {