From 48e31b3fc0ae70689d27dc5bd2c10b14e75afc7c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 27 Jul 2026 23:50:25 -0700 Subject: [PATCH] fix(agent-status): show Codex v2 subagents (#11059) * fix(agent-status): track Codex rollout subagents * fix(agent-status): resolve cross-day Codex child rollouts and unblock CI gate Codex files each rollout under its own local start date, so a session that runs past midnight spawns children into a sibling day directory. Scanning only the parent's directory left 13% of real subagent spawns (48/371 across local rollouts) permanently unresolved, which pinned a phantom "working" row and re-ran readdirSync every poll tick forever. Resolve the child's own day directory from occurred_at_ms, and time-box a child whose rollout stays unreadable so a deleted or never-written file can't leak a working row. Also make the hook HTTP handler return void: the changed-code quality gate keys findings by span overlap, so this PR's added line inside the pre-existing async createServer callback resurfaced no-misused-promises as a new finding. Tests cover cross-day resolution, grace-period retirement, and that the poll re-arms across successive roster changes (the prior tests passed even when the poll died after its first change). * fix(agent-status): keep the Codex subagent poll alive across nested hooks A nested non-codex CLI inherits its parent's ORCA_PANE_KEY, so its hook POST reached scheduleCodexSubagentPoll and tore the timer down before the source guard, silently ending polling while a rollout child was still live. --- .../server-codex-subagent-transcript.test.ts | 207 ++++++++++++ src/main/agent-hooks/server.ts | 62 +++- ...k-server-codex-subagent-transcript.test.ts | 85 +++++ src/relay/agent-hook-server.ts | 56 ++++ src/shared/agent-hook-listener.ts | 42 ++- .../codex-subagent-rollout-lifecycle.test.ts | 92 ++++++ src/shared/codex-subagent-roster.test.ts | 2 + src/shared/codex-subagent-roster.ts | 14 +- src/shared/codex-subagent-transcript.test.ts | 158 +++++++++ src/shared/codex-subagent-transcript.ts | 299 ++++++++++++++++++ 10 files changed, 1013 insertions(+), 4 deletions(-) create mode 100644 src/main/agent-hooks/server-codex-subagent-transcript.test.ts create mode 100644 src/relay/agent-hook-server-codex-subagent-transcript.test.ts create mode 100644 src/shared/codex-subagent-rollout-lifecycle.test.ts create mode 100644 src/shared/codex-subagent-transcript.test.ts create mode 100644 src/shared/codex-subagent-transcript.ts diff --git a/src/main/agent-hooks/server-codex-subagent-transcript.test.ts b/src/main/agent-hooks/server-codex-subagent-transcript.test.ts new file mode 100644 index 00000000000..dd771d8b2be --- /dev/null +++ b/src/main/agent-hooks/server-codex-subagent-transcript.test.ts @@ -0,0 +1,207 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { AgentHookServer } from './server' +import { makePaneKey } from '../../shared/stable-pane-id' + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' +const SECOND_CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f3170' + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +function spawnLine(threadId: string, agentPath: string): string { + return line({ + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: threadId, + agent_path: agentPath, + kind: 'started' + } + }) +} + +describe('AgentHookServer Codex subagent transcript polling', () => { + const dirs: string[] = [] + + afterEach(() => { + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + it('publishes rollout-only children and removes them after their task completes', async () => { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-subagent-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync( + parentPath, + line({ + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: CHILD_ID, + agent_path: '/root/pr_review', + kind: 'started' + } + }) + ) + writeFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_started' } })) + const server = new AgentHookServer() + await server.start({ env: 'production' }) + try { + const env = server.buildPtyEnv() + const response = 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: parentPath, + tool_name: 'collaborationspawn_agent' + } + }) + }) + + expect(response.status).toBe(204) + expect(server.getStatusSnapshot()[0]?.subagents).toEqual([ + expect.objectContaining({ id: CHILD_ID, description: '/root/pr_review' }) + ]) + + appendFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_complete' } })) + await vi.waitFor( + () => { + expect(server.getStatusSnapshot()[0]?.subagents).toBeUndefined() + }, + { timeout: 2_000, interval: 50 } + ) + } finally { + server.stop() + } + }) + + // Why: the poll re-arms off the object it just stored; if that identity ever drifts it stops after the first change. + it('keeps polling across successive roster changes', async () => { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-subagent-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + const secondChildPath = join(dir, `rollout-child-${SECOND_CHILD_ID}.jsonl`) + const started = line({ type: 'event_msg', payload: { type: 'task_started' } }) + writeFileSync(parentPath, spawnLine(CHILD_ID, '/root/pr_review')) + writeFileSync(childPath, started) + writeFileSync(secondChildPath, started) + const server = new AgentHookServer() + await server.start({ env: 'production' }) + try { + 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: parentPath, + tool_name: 'collaborationspawn_agent' + } + }) + }) + expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(1) + + appendFileSync(parentPath, spawnLine(SECOND_CHILD_ID, '/root/perf_audit')) + await vi.waitFor( + () => { + expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(2) + }, + { timeout: 3_000, interval: 50 } + ) + + const complete = line({ type: 'event_msg', payload: { type: 'task_complete' } }) + appendFileSync(childPath, complete) + appendFileSync(secondChildPath, complete) + await vi.waitFor( + () => { + expect(server.getStatusSnapshot()[0]?.subagents).toBeUndefined() + }, + { timeout: 3_000, interval: 50 } + ) + } finally { + server.stop() + } + }) + + // Why: a nested non-codex CLI inherits the pane's ORCA_PANE_KEY, so its hook must not tear down the codex poll. + it('keeps polling when a nested non-codex hook lands on the same pane', async () => { + const dir = mkdtempSync(join(tmpdir(), 'agent-hook-codex-subagent-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync(parentPath, spawnLine(CHILD_ID, '/root/pr_review')) + writeFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_started' } })) + const server = new AgentHookServer() + await server.start({ env: 'production' }) + try { + const env = server.buildPtyEnv() + const post = (path: string, payload: unknown): Promise => + fetch(`http://127.0.0.1:${env.ORCA_AGENT_HOOK_PORT}${path}`, { + 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 }) + }) + + await post('/hook/codex', { + hook_event_name: 'PostToolUse', + session_id: 'root-session', + transcript_path: parentPath, + tool_name: 'collaborationspawn_agent' + }) + expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(1) + + const nested = await post('/hook/copilot', { + hook_event_name: 'Stop', + session_id: 'nested-session', + transcript_path: join(dir, 'nested-copilot-transcript.jsonl') + }) + expect(nested.status).toBe(204) + // The nested completion is suppressed, so the pane is still the same live codex turn. + expect(server.getStatusSnapshot()[0]?.agentType).toBe('codex') + expect(server.getStatusSnapshot()[0]?.subagents).toHaveLength(1) + + appendFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_complete' } })) + await vi.waitFor( + () => { + expect(server.getStatusSnapshot()[0]?.subagents).toBeUndefined() + }, + { timeout: 3_000, interval: 50 } + ) + } finally { + server.stop() + } + }) +}) diff --git a/src/main/agent-hooks/server.ts b/src/main/agent-hooks/server.ts index a9395cf44de..8887a285235 100644 --- a/src/main/agent-hooks/server.ts +++ b/src/main/agent-hooks/server.ts @@ -15,6 +15,7 @@ import { clearClaudeAnsweredQuestionWait, createHookListenerState, getEndpointFileName, + hasCodexTranscriptSubagents, hasPendingAgentResultText, HOOK_REQUEST_SLOWLORIS_MS, markClaudeLeadTurnInterrupted, @@ -106,6 +107,7 @@ type PaneKeyAliasEntry = { const LAST_STATUS_FILE_NAME = 'last-status.json' const ASSISTANT_MESSAGE_RETRY_ATTEMPTS = 5 const ASSISTANT_MESSAGE_RETRY_MS = 50 +const CODEX_SUBAGENT_POLL_MS = 1_000 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). @@ -484,6 +486,7 @@ export class AgentHookServer { // Why: trailing-edge debounce timer, per-instance so test servers in one process don't share state. private statusPersistTimer: ReturnType | null = null private assistantMessageRetryTimers = new Map>() + private codexSubagentPollTimers = new Map>() private promptSentDedupeByPaneKey = new Map() private promptSentHashSalt = randomBytes(16).toString('hex') private closedAgentStatusTabIds = new Set() @@ -1040,6 +1043,49 @@ export class AgentHookServer { this.assistantMessageRetryTimers.delete(paneKey) } + private clearCodexSubagentPoll(paneKey: string): void { + const timer = this.codexSubagentPollTimers.get(paneKey) + if (!timer) { + return + } + clearTimeout(timer) + this.codexSubagentPollTimers.delete(paneKey) + } + + private scheduleCodexSubagentPoll( + source: AgentHookSource, + body: unknown, + original: EnrichedAgentHookEventPayload + ): void { + // Why: a nested non-codex CLI inherits ORCA_PANE_KEY, so clearing here would silently end a live codex poll. + if (source !== 'codex') { + return + } + this.clearCodexSubagentPoll(original.paneKey) + if (!hasCodexTranscriptSubagents(this.state, original.paneKey)) { + return + } + const timer = setTimeout(() => { + this.codexSubagentPollTimers.delete(original.paneKey) + const current = this.state.lastStatusByPaneKey.get(original.paneKey) + if (!this.server || current !== original) { + return + } + const normalized = normalizeHookPayload(this.state, source, body, this.env) + if (!normalized) { + return + } + const subagentsChanged = + JSON.stringify(normalized.payload.subagents) !== JSON.stringify(original.payload.subagents) + const next = subagentsChanged ? this.applyNormalizedStatus(normalized) : original + this.scheduleCodexSubagentPoll(source, body, next) + }, CODEX_SUBAGENT_POLL_MS) + this.codexSubagentPollTimers.set(original.paneKey, timer) + if (typeof timer.unref === 'function') { + timer.unref() + } + } + private scheduleAssistantMessageRetry( source: AgentHookSource, body: unknown, @@ -1264,6 +1310,7 @@ export class AgentHookServer { this.promptSentDedupeByPaneKey.set(toPaneKey, promptDedupe) } this.clearAssistantMessageRetry(previousOwnerPaneKey) + this.clearCodexSubagentPoll(previousOwnerPaneKey) // 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, { stablePaneKey: toPaneKey, @@ -1296,6 +1343,7 @@ export class AgentHookServer { for (const key of paneKeys) { this.markPaneClosedForAgentStatus(key) this.clearAssistantMessageRetry(key) + this.clearCodexSubagentPoll(key) clearPaneCacheState(this.state, key) this.runtimeObservedStatusPaneKeys.delete(key) this.promptSentDedupeByPaneKey.delete(key) @@ -1584,7 +1632,7 @@ export class AgentHookServer { if (this.lastStatusFilePath) { this.hydrateLastStatusFromDisk() } - this.server = createServer(async (req: IncomingMessage, res: ServerResponse) => { + const handleRequest = async (req: IncomingMessage, res: ServerResponse): Promise => { if (req.method !== 'POST') { res.writeHead(404) res.end() @@ -1627,6 +1675,7 @@ export class AgentHookServer { if (normalized && !this.shouldSuppressClosedTabStatus(normalized.paneKey)) { const enriched = this.applyNormalizedStatus(normalized) this.scheduleAssistantMessageRetry(source, aliasedBody, enriched) + this.scheduleCodexSubagentPoll(source, aliasedBody, enriched) } res.writeHead(204) @@ -1636,6 +1685,10 @@ export class AgentHookServer { res.writeHead(204) res.end() } + } + // Why: node ignores a returned promise, so the handler must settle it itself; handleRequest never rejects. + this.server = createServer((req, res) => { + void handleRequest(req, res) }) await new Promise((resolve, reject) => { @@ -1675,6 +1728,10 @@ export class AgentHookServer { clearTimeout(timer) } this.assistantMessageRetryTimers.clear() + for (const timer of this.codexSubagentPollTimers.values()) { + clearTimeout(timer) + } + this.codexSubagentPollTimers.clear() // 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 @@ -1751,6 +1808,7 @@ export class AgentHookServer { } this.state.lastStatusByPaneKey.delete(resolvedPaneKey) this.clearAssistantMessageRetry(resolvedPaneKey) + this.clearCodexSubagentPoll(resolvedPaneKey) this.runtimeObservedStatusPaneKeys.delete(resolvedPaneKey) if (existing.payload.state === 'done') { this.promptSentDedupeByPaneKey.delete(resolvedPaneKey) @@ -1816,6 +1874,7 @@ export class AgentHookServer { statusChanged = true } this.clearAssistantMessageRetry(paneKey) + this.clearCodexSubagentPoll(paneKey) clearPaneCacheState(this.state, paneKey) this.runtimeObservedStatusPaneKeys.delete(paneKey) this.promptSentDedupeByPaneKey.delete(paneKey) @@ -1834,6 +1893,7 @@ export class AgentHookServer { // Why: only persist when a status entry was actually evicted; dropping prompt/tool caches doesn't change the file. const hadStatus = this.state.lastStatusByPaneKey.has(resolvedPaneKey) this.clearAssistantMessageRetry(resolvedPaneKey) + this.clearCodexSubagentPoll(resolvedPaneKey) clearPaneCacheState(this.state, resolvedPaneKey) this.promptSentDedupeByPaneKey.delete(resolvedPaneKey) let clearedAlias = false diff --git a/src/relay/agent-hook-server-codex-subagent-transcript.test.ts b/src/relay/agent-hook-server-codex-subagent-transcript.test.ts new file mode 100644 index 00000000000..e9d20d51db2 --- /dev/null +++ b/src/relay/agent-hook-server-codex-subagent-transcript.test.ts @@ -0,0 +1,85 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { RelayAgentHookServer } from './agent-hook-server' +import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay' +import { makePaneKey } from '../shared/stable-pane-id' + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' + +function line(record: unknown): string { + return `${JSON.stringify(record)}\n` +} + +describe('RelayAgentHookServer Codex subagent transcript polling', () => { + const dirs: string[] = [] + + afterEach(() => { + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + it('forwards completion discovered from a child rollout', async () => { + const dir = mkdtempSync(join(tmpdir(), 'relay-hook-codex-subagent-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync( + parentPath, + line({ + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: CHILD_ID, + agent_path: '/root/pr_review', + kind: 'started' + } + }) + ) + writeFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_started' } })) + const forward = vi.fn<(envelope: AgentHookRelayEnvelope) => void>() + const server = new RelayAgentHookServer({ endpointDir: dir, forward }) + await server.start() + try { + const { port, token } = server.getCoordinates() + const response = await fetch(`http://127.0.0.1:${port}/hook/codex`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Orca-Agent-Hook-Token': token + }, + body: JSON.stringify({ + paneKey: PANE_KEY, + tabId: 'tab-1', + worktreeId: 'wt-1', + payload: { + hook_event_name: 'PostToolUse', + session_id: 'root-session', + transcript_path: parentPath, + tool_name: 'collaborationspawn_agent' + } + }) + }) + + expect(response.status).toBe(204) + expect(forward.mock.calls[0]?.[0].payload.subagents).toHaveLength(1) + + appendFileSync(childPath, line({ type: 'event_msg', payload: { type: 'task_complete' } })) + await vi.waitFor( + () => { + expect(forward.mock.calls.at(-1)?.[0].payload.subagents).toBeUndefined() + expect(forward).toHaveBeenCalledTimes(2) + }, + { timeout: 2_000, interval: 50 } + ) + } finally { + server.stop() + } + }) +}) diff --git a/src/relay/agent-hook-server.ts b/src/relay/agent-hook-server.ts index 40775b5bd6a..fb938f55d2b 100644 --- a/src/relay/agent-hook-server.ts +++ b/src/relay/agent-hook-server.ts @@ -14,6 +14,7 @@ import { clearPaneCacheState, createHookListenerState, getEndpointFileName, + hasCodexTranscriptSubagents, hasPendingAgentResultText, HOOK_REQUEST_SLOWLORIS_MS, normalizeHookPayload, @@ -37,6 +38,7 @@ const RELAY_HOOKS_DIR_NAME = '.orca-relay' const RELAY_HOOKS_SUBDIR = 'agent-hooks' const ASSISTANT_MESSAGE_RETRY_ATTEMPTS = 5 const ASSISTANT_MESSAGE_RETRY_MS = 50 +const CODEX_SUBAGENT_POLL_MS = 1_000 // Why: cap env/version at 64 chars so a misbehaving agent CLI can't grow the meta cache unboundedly; canonical values are short. const MAX_HOOK_META_LEN = 64 @@ -101,6 +103,7 @@ export class RelayAgentHookServer { { source: AgentHookSource; env?: string; version?: string } > = new Map() private assistantMessageRetryTimers = new Map>() + private codexSubagentPollTimers = new Map>() private forward: RelayHookForward private fixedToken: string | undefined private preferredPort: number @@ -193,6 +196,10 @@ export class RelayAgentHookServer { clearTimeout(timer) } this.assistantMessageRetryTimers.clear() + for (const timer of this.codexSubagentPollTimers.values()) { + clearTimeout(timer) + } + this.codexSubagentPollTimers.clear() clearAllListenerCaches(this.state) this.lastEnvelopeMetaByPaneKey.clear() } @@ -216,6 +223,7 @@ export class RelayAgentHookServer { /** Drop a paneKey's cached entries on PTY exit so a terminated pane can't resurface as a ghost event on reconnect. */ clearPaneState(paneKey: string): void { this.clearAssistantMessageRetry(paneKey) + this.clearCodexSubagentPoll(paneKey) clearPaneCacheState(this.state, paneKey) this.lastEnvelopeMetaByPaneKey.delete(paneKey) } @@ -274,6 +282,7 @@ export class RelayAgentHookServer { const version = this.bodyVersion(body) this.applyEvent(event, source, env, version) this.scheduleAssistantMessageRetry(source, body, event, env, version) + this.scheduleCodexSubagentPoll(source, body, event, env, version) } res.writeHead(204) res.end() @@ -350,6 +359,53 @@ export class RelayAgentHookServer { this.assistantMessageRetryTimers.delete(paneKey) } + private clearCodexSubagentPoll(paneKey: string): void { + const timer = this.codexSubagentPollTimers.get(paneKey) + if (!timer) { + return + } + clearTimeout(timer) + this.codexSubagentPollTimers.delete(paneKey) + } + + private scheduleCodexSubagentPoll( + source: AgentHookSource, + body: unknown, + original: AgentHookEventPayload, + env?: string, + version?: string + ): void { + // Why: a nested non-codex CLI inherits ORCA_PANE_KEY, so clearing here would silently end a live codex poll. + if (source !== 'codex') { + return + } + this.clearCodexSubagentPoll(original.paneKey) + if (!hasCodexTranscriptSubagents(this.state, original.paneKey)) { + return + } + const timer = setTimeout(() => { + this.codexSubagentPollTimers.delete(original.paneKey) + if (!this.server || this.state.lastStatusByPaneKey.get(original.paneKey) !== original) { + return + } + const event = normalizeHookPayload(this.state, source, body, this.env) + if (!event) { + return + } + const subagentsChanged = + JSON.stringify(event.payload.subagents) !== JSON.stringify(original.payload.subagents) + const next = subagentsChanged ? event : original + if (subagentsChanged) { + this.applyEvent(event, source, env, version) + } + this.scheduleCodexSubagentPoll(source, body, next, env, version) + }, CODEX_SUBAGENT_POLL_MS) + this.codexSubagentPollTimers.set(original.paneKey, timer) + if (typeof timer.unref === 'function') { + timer.unref() + } + } + private scheduleAssistantMessageRetry( source: AgentHookSource, body: unknown, diff --git a/src/shared/agent-hook-listener.ts b/src/shared/agent-hook-listener.ts index d26268d120e..ee3536bb932 100644 --- a/src/shared/agent-hook-listener.ts +++ b/src/shared/agent-hook-listener.ts @@ -45,6 +45,12 @@ import { upsertCodexSubagent, type CodexSubagentRoster } from './codex-subagent-roster' +import { + createCodexSubagentTranscriptState, + hasTrackedCodexTranscriptSubagents, + reconcileCodexSubagentTranscript, + type CodexSubagentTranscriptState +} from './codex-subagent-transcript' import { ORCA_HOOK_PROTOCOL_VERSION } from './agent-hook-types' import { REMOTE_AGENT_HOOK_ENV, type AgentHookSource } from './agent-hook-relay' import { @@ -111,6 +117,8 @@ export type HookListenerState = { claudeLeadStateByPaneKey: Map /** Live thread-spawn children per Codex pane. */ codexSubagentRosterByPaneKey: Map + /** Incremental parent/child rollout cursors for Codex collaboration v2. */ + codexSubagentTranscriptByPaneKey: Map /** Root Codex state/model, kept separate from child hook traffic. */ codexLeadStateByPaneKey: Map } @@ -141,6 +149,7 @@ export function createHookListenerState(): HookListenerState { claudeSubagentRosterByPaneKey: new Map(), claudeLeadStateByPaneKey: new Map(), codexSubagentRosterByPaneKey: new Map(), + codexSubagentTranscriptByPaneKey: new Map(), codexLeadStateByPaneKey: new Map() } } @@ -154,6 +163,7 @@ export function clearPaneCacheState(state: HookListenerState, paneKey: string): state.claudeSubagentRosterByPaneKey.delete(paneKey) state.claudeLeadStateByPaneKey.delete(paneKey) state.codexSubagentRosterByPaneKey.delete(paneKey) + state.codexSubagentTranscriptByPaneKey.delete(paneKey) state.codexLeadStateByPaneKey.delete(paneKey) } @@ -197,6 +207,7 @@ export function movePaneCacheState( movePaneScopedMapEntries(state.claudeSubagentRosterByPaneKey, fromPaneKey, toPaneKey) movePaneScopedMapEntries(state.claudeLeadStateByPaneKey, fromPaneKey, toPaneKey) movePaneScopedMapEntries(state.codexSubagentRosterByPaneKey, fromPaneKey, toPaneKey) + movePaneScopedMapEntries(state.codexSubagentTranscriptByPaneKey, fromPaneKey, toPaneKey) movePaneScopedMapEntries(state.codexLeadStateByPaneKey, fromPaneKey, toPaneKey) } @@ -238,6 +249,7 @@ export function clearAllListenerCaches(state: HookListenerState): void { state.claudeSubagentRosterByPaneKey.clear() state.claudeLeadStateByPaneKey.clear() state.codexSubagentRosterByPaneKey.clear() + state.codexSubagentTranscriptByPaneKey.clear() state.codexLeadStateByPaneKey.clear() } @@ -3098,6 +3110,22 @@ function getOrCreateCodexSubagentRoster( return roster } +function getOrCreateCodexSubagentTranscriptState( + state: HookListenerState, + paneKey: string +): CodexSubagentTranscriptState { + let transcriptState = state.codexSubagentTranscriptByPaneKey.get(paneKey) + if (!transcriptState) { + transcriptState = createCodexSubagentTranscriptState() + state.codexSubagentTranscriptByPaneKey.set(paneKey, transcriptState) + } + return transcriptState +} + +export function hasCodexTranscriptSubagents(state: HookListenerState, paneKey: string): boolean { + return hasTrackedCodexTranscriptSubagents(state.codexSubagentTranscriptByPaneKey.get(paneKey)) +} + export function seedCodexStateFromSnapshot( state: HookListenerState, paneKey: string, @@ -3178,7 +3206,7 @@ export function reconcileRemoteCodexState( } } else { const leadState = codexLeadStateForHookEvent(eventName) - if (eventName === 'SessionStart' || eventName === 'Stop') { + if (eventName === 'SessionStart' || (eventName === 'Stop' && !payload.subagents)) { roster.clear() } if (leadState) { @@ -3325,7 +3353,17 @@ function normalizeCodexEvent( if (eventName === 'SessionStart') { // Why: a pane can host a new Codex process after the old one exited without child Stop hooks. state.codexSubagentRosterByPaneKey.delete(paneKey) - } else if (eventName === 'Stop') { + state.codexSubagentTranscriptByPaneKey.delete(paneKey) + } + const transcriptPath = readFirstString(hookPayload, ['transcript_path', 'transcriptPath']) + if (transcriptPath) { + reconcileCodexSubagentTranscript( + getOrCreateCodexSubagentTranscriptState(state, paneKey), + getOrCreateCodexSubagentRoster(state, paneKey), + transcriptPath + ) + } + 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) } diff --git a/src/shared/codex-subagent-rollout-lifecycle.test.ts b/src/shared/codex-subagent-rollout-lifecycle.test.ts new file mode 100644 index 00000000000..9d476b85504 --- /dev/null +++ b/src/shared/codex-subagent-rollout-lifecycle.test.ts @@ -0,0 +1,92 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { createHookListenerState, normalizeHookPayload } from './agent-hook-listener' +import { makePaneKey } from './stable-pane-id' + +const PANE_KEY = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111') +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' + +function jsonl(records: unknown[]): string { + return `${records.map((record) => JSON.stringify(record)).join('\n')}\n` +} + +describe('Codex rollout subagent lifecycle', () => { + const dirs: string[] = [] + + afterEach(() => { + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + it('surfaces a collaboration child without lifecycle hooks and retires it on task_complete', () => { + const dir = mkdtempSync(join(tmpdir(), 'codex-rollout-lifecycle-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync( + parentPath, + jsonl([ + { + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: 1234, + agent_thread_id: CHILD_ID, + agent_path: '/root/pr_review', + kind: 'started' + } + } + ]) + ) + writeFileSync(childPath, jsonl([{ type: 'event_msg', payload: { type: 'task_started' } }])) + const state = createHookListenerState() + const event = (hookEventName: string): ReturnType => + normalizeHookPayload( + state, + 'codex', + { + paneKey: PANE_KEY, + payload: { + hook_event_name: hookEventName, + session_id: 'root-session', + transcript_path: parentPath, + tool_name: 'collaborationspawn_agent' + } + }, + 'production' + ) + + const spawned = event('PostToolUse') + expect(spawned?.payload).toMatchObject({ + state: 'working', + subagents: [ + { + id: CHILD_ID, + description: '/root/pr_review', + state: 'working', + startedAt: 1234 + } + ] + }) + + const leadStopped = event('Stop') + expect(leadStopped?.payload.state).toBe('working') + expect(leadStopped?.payload.subagents).toHaveLength(1) + + writeFileSync( + childPath, + jsonl([ + { type: 'event_msg', payload: { type: 'task_started' } }, + { type: 'event_msg', payload: { type: 'task_complete' } } + ]) + ) + const completed = event('Stop') + expect(completed?.payload.state).toBe('done') + expect(completed?.payload.subagents).toBeUndefined() + }) +}) diff --git a/src/shared/codex-subagent-roster.test.ts b/src/shared/codex-subagent-roster.test.ts index ebb947dedf6..fde2c8cf849 100644 --- a/src/shared/codex-subagent-roster.test.ts +++ b/src/shared/codex-subagent-roster.test.ts @@ -20,6 +20,7 @@ describe('Codex subagent roster', () => { ' child-1 ', { agentType: `reviewer\n${'x'.repeat(AGENT_TYPE_MAX_LENGTH * 2)}`, + description: 'Review the sidebar lifecycle', model: `gpt-model-${'x'.repeat(AGENT_MODEL_MAX_LENGTH * 2)}`, state: 'working' }, @@ -30,6 +31,7 @@ describe('Codex subagent roster', () => { expect([...roster.keys()]).toEqual(['child-1']) expect(snapshot?.agentType).toHaveLength(AGENT_TYPE_MAX_LENGTH) expect(snapshot?.agentType).not.toContain('\n') + expect(snapshot?.description).toBe('Review the sidebar lifecycle') expect(snapshot?.model).toHaveLength(AGENT_MODEL_MAX_LENGTH) finishCodexSubagent(roster, ' child-1 ') diff --git a/src/shared/codex-subagent-roster.ts b/src/shared/codex-subagent-roster.ts index 5c4d3e6d7a0..72cc216b76e 100644 --- a/src/shared/codex-subagent-roster.ts +++ b/src/shared/codex-subagent-roster.ts @@ -1,6 +1,7 @@ import { AGENT_MODEL_MAX_LENGTH, AGENT_STATUS_MAX_SUBAGENTS, + AGENT_STATUS_TOOL_INPUT_MAX_LENGTH, AGENT_TYPE_MAX_LENGTH, type AgentSubagentSnapshot } from './agent-status-types' @@ -12,6 +13,7 @@ export type CodexSubagentRoster = Map type TrackedCodexSubagent = { agentType?: string + description?: string model?: string state: 'working' | 'waiting' startedAt: number @@ -22,6 +24,7 @@ export function upsertCodexSubagent( id: string, fields: { agentType?: string + description?: string model?: string state: 'working' | 'waiting' }, @@ -32,10 +35,12 @@ export function upsertCodexSubagent( return } const agentType = normalizeOptionalField(fields.agentType, AGENT_TYPE_MAX_LENGTH) + const description = normalizeOptionalField(fields.description, AGENT_STATUS_TOOL_INPUT_MAX_LENGTH) const model = normalizeOptionalField(fields.model, AGENT_MODEL_MAX_LENGTH) const existing = roster.get(normalizedId) if (existing) { existing.agentType = agentType ?? existing.agentType + existing.description = description ?? existing.description existing.model = model ?? existing.model existing.state = fields.state return @@ -45,6 +50,7 @@ export function upsertCodexSubagent( } roster.set(normalizedId, { agentType, + description, model, state: fields.state, startedAt: now @@ -66,7 +72,12 @@ export function seedCodexSubagentRoster( upsertCodexSubagent( roster, snapshot.id, - { agentType: snapshot.agentType, model: snapshot.model, state: snapshot.state }, + { + agentType: snapshot.agentType, + description: snapshot.description, + model: snapshot.model, + state: snapshot.state + }, snapshot.startedAt ) } @@ -81,6 +92,7 @@ export function codexRosterToSnapshots( const snapshots = Array.from(roster, ([id, tracked]) => ({ id, agentType: tracked.agentType, + description: tracked.description, model: tracked.model, state: tracked.state, startedAt: tracked.startedAt diff --git a/src/shared/codex-subagent-transcript.test.ts b/src/shared/codex-subagent-transcript.test.ts new file mode 100644 index 00000000000..d2f729912b5 --- /dev/null +++ b/src/shared/codex-subagent-transcript.test.ts @@ -0,0 +1,158 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + createCodexSubagentTranscriptState, + hasTrackedCodexTranscriptSubagents, + reconcileCodexSubagentTranscript +} from './codex-subagent-transcript' +import { codexRosterToSnapshots, type CodexSubagentRoster } from './codex-subagent-roster' + +const CHILD_ID = '019fa65f-3144-7151-9c02-cff7a28f316f' + +function jsonl(records: unknown[]): string { + return `${records.map((record) => JSON.stringify(record)).join('\n')}\n` +} + +function activity(kind: string, occurredAtMs = 1234): unknown { + return { + type: 'event_msg', + payload: { + type: 'sub_agent_activity', + occurred_at_ms: occurredAtMs, + agent_thread_id: CHILD_ID, + agent_path: '/root/sidebar_repro', + kind + } + } +} + +/** `/YYYY/MM/DD` for a timestamp, matching how Codex buckets rollouts by local start date. */ +function dayDirectory(root: string, atMs: number): string { + const at = new Date(atMs) + const pad = (value: number): string => String(value).padStart(2, '0') + return join(root, String(at.getFullYear()), pad(at.getMonth() + 1), pad(at.getDate())) +} + +describe('Codex subagent transcript reconciliation', () => { + const dirs: string[] = [] + + afterEach(() => { + for (const dir of dirs) { + rmSync(dir, { recursive: true, force: true }) + } + dirs.length = 0 + }) + + it('adds a child from the parent rollout and removes it after task completion', () => { + const dir = mkdtempSync(join(tmpdir(), 'codex-subagent-transcript-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + const childPath = join(dir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync(parentPath, jsonl([activity('started')])) + writeFileSync(childPath, jsonl([{ type: 'event_msg', payload: { type: 'task_started' } }])) + const state = createCodexSubagentTranscriptState() + const roster: CodexSubagentRoster = new Map() + + reconcileCodexSubagentTranscript(state, roster, parentPath) + + expect(hasTrackedCodexTranscriptSubagents(state)).toBe(true) + expect(codexRosterToSnapshots(roster)).toEqual([ + { + id: CHILD_ID, + description: '/root/sidebar_repro', + state: 'working', + startedAt: 1234, + agentType: undefined, + model: undefined + } + ]) + + writeFileSync( + childPath, + jsonl([ + { type: 'event_msg', payload: { type: 'task_started' } }, + { type: 'event_msg', payload: { type: 'task_complete' } } + ]) + ) + reconcileCodexSubagentTranscript(state, roster, parentPath) + + expect(hasTrackedCodexTranscriptSubagents(state)).toBe(false) + expect(codexRosterToSnapshots(roster)).toBeUndefined() + }) + + it('resolves a child rollout filed under a later session day than the parent', () => { + const root = mkdtempSync(join(tmpdir(), 'codex-subagent-transcript-')) + dirs.push(root) + const childStartedAt = Date.now() + const parentDir = dayDirectory(root, childStartedAt - 24 * 60 * 60 * 1000) + const childDir = dayDirectory(root, childStartedAt) + mkdirSync(parentDir, { recursive: true }) + mkdirSync(childDir, { recursive: true }) + const parentPath = join(parentDir, 'rollout-parent.jsonl') + const childPath = join(childDir, `rollout-child-${CHILD_ID}.jsonl`) + writeFileSync(parentPath, jsonl([activity('started', childStartedAt)])) + writeFileSync(childPath, jsonl([{ type: 'event_msg', payload: { type: 'task_started' } }])) + const state = createCodexSubagentTranscriptState() + const roster: CodexSubagentRoster = new Map() + + reconcileCodexSubagentTranscript(state, roster, parentPath) + writeFileSync( + childPath, + jsonl([ + { type: 'event_msg', payload: { type: 'task_started' } }, + { type: 'event_msg', payload: { type: 'task_complete' } } + ]) + ) + reconcileCodexSubagentTranscript(state, roster, parentPath) + + // Why: only a cross-day lookup can observe the completion; the parent-directory scan never finds this file. + expect(roster.size).toBe(0) + expect(hasTrackedCodexTranscriptSubagents(state)).toBe(false) + }) + + it('retires a child whose rollout never becomes readable', () => { + vi.useFakeTimers() + try { + const dir = mkdtempSync(join(tmpdir(), 'codex-subagent-transcript-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + writeFileSync(parentPath, jsonl([activity('started')])) + const state = createCodexSubagentTranscriptState() + const roster: CodexSubagentRoster = new Map() + + reconcileCodexSubagentTranscript(state, roster, parentPath) + expect(roster.size).toBe(1) + + // Why: within the grace window a slow-to-appear rollout must not drop a live child. + vi.advanceTimersByTime(30_000) + reconcileCodexSubagentTranscript(state, roster, parentPath) + expect(roster.size).toBe(1) + + vi.advanceTimersByTime(31_000) + reconcileCodexSubagentTranscript(state, roster, parentPath) + expect(roster.size).toBe(0) + expect(hasTrackedCodexTranscriptSubagents(state)).toBe(false) + } finally { + vi.useRealTimers() + } + }) + + it('removes a child when Codex reports it interrupted', () => { + const dir = mkdtempSync(join(tmpdir(), 'codex-subagent-transcript-')) + dirs.push(dir) + const parentPath = join(dir, 'rollout-parent.jsonl') + writeFileSync(parentPath, jsonl([activity('started')])) + const state = createCodexSubagentTranscriptState() + const roster: CodexSubagentRoster = new Map() + reconcileCodexSubagentTranscript(state, roster, parentPath) + + writeFileSync(parentPath, jsonl([activity('started'), activity('interrupted')])) + reconcileCodexSubagentTranscript(state, roster, parentPath) + + expect(hasTrackedCodexTranscriptSubagents(state)).toBe(false) + expect(roster.size).toBe(0) + }) +}) diff --git a/src/shared/codex-subagent-transcript.ts b/src/shared/codex-subagent-transcript.ts new file mode 100644 index 00000000000..3f9b30898c1 --- /dev/null +++ b/src/shared/codex-subagent-transcript.ts @@ -0,0 +1,299 @@ +import { closeSync, openSync, readSync, readdirSync, statSync, type Stats } from 'node:fs' +import { basename, dirname, extname, isAbsolute, join } from 'node:path' + +import { + finishCodexSubagent, + upsertCodexSubagent, + type CodexSubagentRoster +} from './codex-subagent-roster' + +const TRANSCRIPT_READ_MAX_BYTES = 1024 * 1024 +const TRANSCRIPT_LINE_MAX_BYTES = 256 * 1024 +const TRANSCRIPT_DIRECTORY_MAX_ENTRIES = 4096 +// 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 SAFE_THREAD_ID = /^[A-Za-z0-9-]{1,64}$/ + +type JsonlCursor = { + filePath?: string + offset: number + carry: string +} + +type TrackedTranscriptSubagent = JsonlCursor & { + description?: string + startedAt: number + unresolvedSince?: number +} + +export type CodexSubagentTranscriptState = { + parent: JsonlCursor + subagents: Map +} + +type JsonRecord = Record + +function record(value: unknown): JsonRecord | undefined { + return typeof value === 'object' && value !== null ? (value as JsonRecord) : undefined +} + +/** Returns undefined when the file is unreadable, distinguishing a vanished rollout from one with no new lines. */ +function readJsonlCursor(cursor: JsonlCursor): JsonRecord[] | undefined { + if (!cursor.filePath) { + return undefined + } + let stats: Stats + try { + stats = statSync(cursor.filePath) + } catch { + return undefined + } + if (!stats.isFile()) { + return undefined + } + if (stats.size < cursor.offset) { + cursor.offset = 0 + cursor.carry = '' + } + if (stats.size === cursor.offset) { + return [] + } + const bytesToRead = Math.min(stats.size - cursor.offset, TRANSCRIPT_READ_MAX_BYTES) + const start = stats.size - cursor.offset > bytesToRead ? stats.size - bytesToRead : cursor.offset + const buffer = Buffer.allocUnsafe(bytesToRead) + let bytesRead = 0 + let fd: number | undefined + try { + fd = openSync(cursor.filePath, 'r') + bytesRead = readSync(fd, buffer, 0, bytesToRead, start) + } catch { + return undefined + } finally { + if (fd !== undefined) { + closeSync(fd) + } + } + const skippedPrefix = start !== cursor.offset + const content = `${skippedPrefix ? '' : cursor.carry}${buffer.toString('utf8', 0, bytesRead)}` + const lines = content.split('\n') + cursor.offset = start + bytesRead + cursor.carry = lines.pop() ?? '' + if (skippedPrefix) { + lines.shift() + } + const records: JsonRecord[] = [] + for (const line of lines) { + if (Buffer.byteLength(line, 'utf8') > TRANSCRIPT_LINE_MAX_BYTES) { + continue + } + try { + const parsed = record(JSON.parse(line) as unknown) + if (parsed) { + records.push(parsed) + } + } catch { + // A malformed rollout line must not block later lifecycle events. + } + } + return records +} + +function readTranscriptDirectory(directory: string): string[] { + let entries: string[] + try { + entries = readdirSync(directory) + } catch { + return [] + } + if (entries.length > TRANSCRIPT_DIRECTORY_MAX_ENTRIES) { + entries = entries.slice(-TRANSCRIPT_DIRECTORY_MAX_ENTRIES) + } + return entries +} + +// Why: Codex files each rollout under its OWN local start date, so a session running past midnight spawns children into a sibling day directory. +function childDayDirectory(parentPath: string, startedAt: number): string | undefined { + const dayDir = dirname(parentPath) + const monthDir = dirname(dayDir) + const yearDir = dirname(monthDir) + if ( + !/^\d{2}$/.test(basename(dayDir)) || + !/^\d{2}$/.test(basename(monthDir)) || + !/^\d{4}$/.test(basename(yearDir)) || + !Number.isFinite(startedAt) + ) { + return undefined + } + const startedOn = new Date(startedAt) + if (Number.isNaN(startedOn.getTime())) { + return undefined + } + const pad = (value: number): string => String(value).padStart(2, '0') + return join( + dirname(yearDir), + String(startedOn.getFullYear()).padStart(4, '0'), + pad(startedOn.getMonth() + 1), + pad(startedOn.getDate()) + ) +} + +function resolveChildTranscript( + parentPath: string, + threadId: string, + startedAt: number, + entriesByDirectory: Map +): string | undefined { + if (!SAFE_THREAD_ID.test(threadId)) { + return undefined + } + const suffix = `-${threadId}.jsonl` + const parentDir = dirname(parentPath) + const childDir = childDayDirectory(parentPath, startedAt) + const directories = childDir && childDir !== parentDir ? [parentDir, childDir] : [parentDir] + for (const directory of directories) { + let entries = entriesByDirectory.get(directory) + if (!entries) { + entries = readTranscriptDirectory(directory) + entriesByDirectory.set(directory, entries) + } + const fileName = entries.find((entry) => entry.endsWith(suffix)) + if (fileName) { + return join(directory, fileName) + } + } + return undefined +} + +function readActivity(recordValue: JsonRecord): + | { + id: string + description?: string + kind: 'started' | 'interacted' | 'interrupted' + startedAt: number + } + | undefined { + if (recordValue.type !== 'event_msg') { + return undefined + } + const payload = record(recordValue.payload) + if (payload?.type !== 'sub_agent_activity') { + return undefined + } + const id = typeof payload.agent_thread_id === 'string' ? payload.agent_thread_id.trim() : '' + const rawKind = typeof payload.kind === 'string' ? payload.kind.toLowerCase() : '' + if ( + !SAFE_THREAD_ID.test(id) || + (rawKind !== 'started' && rawKind !== 'interacted' && rawKind !== 'interrupted') + ) { + return undefined + } + return { + id, + description: + typeof payload.agent_path === 'string' ? payload.agent_path.trim() || undefined : undefined, + kind: rawKind, + startedAt: + typeof payload.occurred_at_ms === 'number' && Number.isFinite(payload.occurred_at_ms) + ? payload.occurred_at_ms + : Date.now() + } +} + +function childIsComplete(records: JsonRecord[]): boolean { + let complete = false + for (const recordValue of records) { + if (recordValue.type !== 'event_msg') { + continue + } + const payload = record(recordValue.payload) + if (payload?.type === 'task_started') { + complete = false + } else if (payload?.type === 'task_complete') { + complete = true + } + } + return complete +} + +export function createCodexSubagentTranscriptState(): CodexSubagentTranscriptState { + return { + parent: { offset: 0, carry: '' }, + subagents: new Map() + } +} + +export function hasTrackedCodexTranscriptSubagents( + state: CodexSubagentTranscriptState | undefined +): boolean { + return Boolean(state && state.subagents.size > 0) +} + +export function reconcileCodexSubagentTranscript( + state: CodexSubagentTranscriptState, + roster: CodexSubagentRoster, + transcriptPath: string | undefined +): void { + const normalizedPath = transcriptPath?.trim() + if (!normalizedPath || !isAbsolute(normalizedPath) || extname(normalizedPath) !== '.jsonl') { + return + } + if (state.parent.filePath !== normalizedPath) { + for (const id of state.subagents.keys()) { + finishCodexSubagent(roster, id) + } + state.parent = { filePath: normalizedPath, offset: 0, carry: '' } + state.subagents.clear() + } + for (const recordValue of readJsonlCursor(state.parent) ?? []) { + const activity = readActivity(recordValue) + if (!activity) { + continue + } + if (activity.kind === 'interrupted') { + finishCodexSubagent(roster, activity.id) + state.subagents.delete(activity.id) + continue + } + const tracked = state.subagents.get(activity.id) ?? { + offset: 0, + carry: '', + startedAt: activity.startedAt + } + tracked.description = activity.description ?? tracked.description + state.subagents.set(activity.id, tracked) + upsertCodexSubagent( + roster, + activity.id, + { description: tracked.description, state: 'working' }, + tracked.startedAt + ) + } + const entriesByDirectory = new Map() + const now = Date.now() + for (const [id, tracked] of state.subagents) { + if (!tracked.filePath) { + tracked.filePath = resolveChildTranscript( + normalizedPath, + id, + tracked.startedAt, + entriesByDirectory + ) + } + const records = readJsonlCursor(tracked) + if (!records) { + // Why: a rollout that never appears (or is deleted) has no completion event, so time-box it instead of leaking a working row. + tracked.filePath = undefined + tracked.unresolvedSince ??= now + if (now - tracked.unresolvedSince <= CHILD_UNREADABLE_GRACE_MS) { + continue + } + } else { + tracked.unresolvedSince = undefined + if (!childIsComplete(records)) { + continue + } + } + finishCodexSubagent(roster, id) + state.subagents.delete(id) + } +}