fix(agent-hooks): harden Codex restart reconciliation

This commit is contained in:
Brennan Benson
2026-08-27 15:18:57 -07:00
parent f8ad411e2c
commit 4f3e1f43e7
12 changed files with 452 additions and 105 deletions
@@ -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<void> {
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<string, Record<string, unknown>>
}
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)
})
})
@@ -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({
+45 -14
View File
@@ -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<string, ReturnType<typeof setTimeout>>()
private codexRestartReconcileTimers = new Map<string, ReturnType<typeof setTimeout>>()
private codexPaneGenerations = new Map<string, number>()
private codexPaneGenerationCounter = 0
private codexRestartReconcileNextRunAt = 0
private promptSentDedupeByPaneKey = new Map<string, AgentPromptSentDedupeEntry>()
private activeHookTurnCompletedAtByPaneKey = new Map<string, number>()
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()
}
@@ -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<string, ReturnType<typeof setTimeout>>
metadata: ReadonlyMap<string, RelayHookStatusMeta>
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 }
}
@@ -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<typeof vi.fn>
restart: ReturnType<typeof vi.fn>
} {
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()
})
})
+7 -2
View File
@@ -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)
}
}
@@ -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()
}
+10 -8
View File
@@ -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<string, ReturnType<typeof setTimeout>>()
private scheduleCodexRestartReconciliation: (paneKey: string) => void
private codexReconcileGate = { nextRunAt: 0 }
private codexReconciliationSchedulers: ReturnType<typeof createRelayCodexReconciliationSchedulers>
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<void> {
@@ -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)
})
+59 -27
View File
@@ -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<string, { source: 'codex' }>()
+20 -51
View File
@@ -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<string, RelayHookStatusMeta>
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()
@@ -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)) {
+6 -1
View File
@@ -35,6 +35,7 @@ export type CodexSubagentTranscriptState = {
parent: JsonlCursor
subagents: Map<string, TrackedTranscriptSubagent>
parentTerminalObserved?: boolean
parentReadable?: boolean
}
type JsonRecord = Record<string, unknown>
@@ -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') {