From 9ccf8c04cb1f6397e156a5c470599c136482e0b2 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 14 Sep 2026 15:55:54 +0200 Subject: [PATCH] feat: Add expermental ACP session compaction updates Implements the agent side of the proposed RFD, gated by the capability --- docs/session-compaction.md | 23 + src/CodexAcpClient.ts | 8 +- src/CodexAcpServer.ts | 11 +- src/CodexAppServerClient.ts | 87 ++- src/CodexCommands.ts | 8 +- src/CodexEventHandler.ts | 65 +- src/CodexSessionCompactions.ts | 92 +++ .../compact-command-lifecycle.test.ts | 175 +++++ .../session-compaction-completed-only.json | 13 + .../data/session-compaction-deduplicated.json | 26 + .../data/session-compaction-error.json | 29 + .../data/session-compaction-failed.json | 27 + .../data/session-compaction-interrupted.json | 26 + .../data/session-compaction-legacy.json | 54 ++ .../data/session-compaction-lifecycle.json | 42 ++ .../data/session-compaction-multiple.json | 52 ++ ...n-native-child-collaboration-terminal.json | 54 ++ ...on-compaction-native-child-identities.json | 80 +++ ...n-compaction-native-child-interrupted.json | 54 ++ ...ssion-compaction-native-child-timeout.json | 41 ++ .../data/session-compaction-next-prompt.json | 54 ++ .../data/session-compaction-replay.json | 47 ++ .../data/session-compaction-retried.json | 28 + .../session-compaction-thread-completed.json | 13 + .../data/session-compaction-wire.json | 18 + .../CodexACPAgent/session-compaction.test.ts | 617 ++++++++++++++++++ src/__tests__/acp-test-utils.ts | 2 + src/subagents/CodexSubagentEventRouter.ts | 32 +- 28 files changed, 1741 insertions(+), 37 deletions(-) create mode 100644 docs/session-compaction.md create mode 100644 src/CodexSessionCompactions.ts create mode 100644 src/__tests__/CodexACPAgent/compact-command-lifecycle.test.ts create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-completed-only.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-deduplicated.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-error.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-failed.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-interrupted.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-legacy.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-lifecycle.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-multiple.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-native-child-collaboration-terminal.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-native-child-identities.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-native-child-interrupted.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-native-child-timeout.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-next-prompt.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-replay.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-retried.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-thread-completed.json create mode 100644 src/__tests__/CodexACPAgent/data/session-compaction-wire.json create mode 100644 src/__tests__/CodexACPAgent/session-compaction.test.ts diff --git a/docs/session-compaction.md b/docs/session-compaction.md new file mode 100644 index 00000000..ef03dc02 --- /dev/null +++ b/docs/session-compaction.md @@ -0,0 +1,23 @@ +# Session compaction + +The adapter implements the [ACP session compaction RFD](https://agentclientprotocol.com/rfds/session-compaction) for ACP v1. Clients opt in during initialization: + +```json +{ + "clientCapabilities": { + "session": { + "compaction": {} + } + } +} +``` + +For these clients, both automatic compaction and `/compact` produce `session/update` notifications with `sessionUpdate: "compaction_update"`. A Codex `contextCompaction` item starts an `in_progress` entity; its completion updates the same `compactionId` to `completed`. A failure or interruption closes an unfinished entity as `failed` or `cancelled`. Successful compactions stay completed if the surrounding turn subsequently fails or is interrupted. + +The adapter uses Codex's item ID as the compaction ID. It suppresses duplicate completion signals, including the older `thread/compacted` notification. If only that older notification is available, it emits a completed entity with an ID derived from the turn. That legacy signal cannot distinguish multiple compactions within one turn. + +Loading a session replays each persisted compaction as one completed update in its history position. Current Codex paginated history preserves the live item ID. Older legacy histories can reconstruct item IDs, and Codex does not persist failed or interrupted compaction items, so those entries are not available for replay. + +Codex's app-server compaction items expose lifecycle identity without a user-displayable summary. The adapter therefore omits `summary` and does not emit `compaction_summary_chunk`. It does not extract internal replacement history or encrypted compaction data. Context utilization continues to arrive separately through `usage_update`. + +When the client omits `session.compaction` or sets it to `null`, the adapter preserves its existing synthetic tool-call and text-message fallback. diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 7040057f..b0a978a6 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -662,8 +662,12 @@ export class CodexAcpClient { }, onTurnStarted); } - async runCompact(sessionId: string): Promise { - await this.codexClient.runCompact({threadId: sessionId}); + async runCompact( + sessionId: string, + onTurnStarted?: (turnId: string) => void, + ): Promise { + const completed = await this.codexClient.runCompact({threadId: sessionId}, onTurnStarted); + return completed.method === "turn/completed" ? completed.params : undefined; } async getGoal(sessionId: string): Promise { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 4ff0e1b4..341ebd57 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -144,6 +144,7 @@ import { } from "./AirExtension"; import {ASYNC_TASK_STOP_METHOD} from "./async-tasks/AsyncTaskExtension"; import {CodexBackgroundTerminalTasks} from "./async-tasks/CodexBackgroundTerminalTasks"; +import {clientSupportsCompaction, CodexSessionCompactions, createCompactionUpdate} from "./CodexSessionCompactions"; import { type AgentFileChangeReport, type AgentFileChangeReportRequest, @@ -184,6 +185,7 @@ export interface SessionState { titleGen?: TitleGenerator; subagents: CodexSubagentEventRouter; asyncTasks: CodexBackgroundTerminalTasks; + compactions: CodexSessionCompactions; } export type SessionFailureCategory = @@ -691,6 +693,7 @@ export class CodexAcpServer { new ACPSessionConnection(this.connection, sessionId), ), asyncTasks: this.createAsyncTasks(sessionId), + compactions: new CodexSessionCompactions(), }; sessionState.titleGen = new TitleGenerator( this.codexAcpClient.appServerClient, @@ -1948,6 +1951,7 @@ export class CodexAcpServer { new ACPSessionConnection(this.connection, sessionId), ), asyncTasks: this.createAsyncTasks(sessionId), + compactions: new CodexSessionCompactions(), }; sessionState.titleGen = new TitleGenerator( this.codexAcpClient.appServerClient, @@ -2276,7 +2280,9 @@ export class CodexAcpServer { case "exitedReviewMode": return [this.createReviewModeUpdate(item, false)]; case "contextCompaction": - return [createCompletedContextCompactionUpdate(item)]; + return [clientSupportsCompaction(this.clientCapabilities) + ? createCompactionUpdate(item.id, "completed") + : createCompletedContextCompactionUpdate(item)]; case "plan": return item.text.length > 0 ? [this.createPlanHistoryUpdate(item)] : []; } @@ -2792,6 +2798,7 @@ export class CodexAcpServer { this.sessionFailureEpoch, sessionState.subagents, (accountUpdated) => this.handleAccountUpdated(accountUpdated), + clientSupportsCompaction(this.clientCapabilities), ); eventHandler = promptEventHandler; const permissionLifecycle = this.permissionLifecycleContext(sessionState); @@ -3190,7 +3197,7 @@ export class CodexAcpServer { : "failed", ); } catch (error) { - logger.error("Failed to publish terminal subagent state during prompt cleanup", error); + logger.error("Failed to publish terminal compaction or subagent state during prompt cleanup", error); } if (agentFileChangeReportRequest !== null) { await this.publishAgentFileChangeReport( diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index daa7e875..03571bea 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -161,7 +161,7 @@ export class CodexAppServerClient { private readonly pendingTurnCompletionResolvers = new Map void>>(); private readonly pendingCompactionCompletionResolvers = new Map void>>(); private readonly turnCompletionCaptures = new Map void>>(); - private readonly turnRoutingCaptures = new Map void>>(); + private readonly turnRoutingCaptures = new Map void>>(); private readonly threadStatusCaptures = new Map void>>(); private readonly threadGoalUpdateCaptures = new Map void>>(); private readonly threadGoalClearedCaptures = new Map void>>(); @@ -204,7 +204,7 @@ export class CodexAppServerClient { if (this.handleStaleTurnNotification(serverNotification, routing)) { return; } - this.recordTurnRouting(routing); + this.recordTurnRouting(routing, serverNotification); if (this.handleStaleTurnNotification(serverNotification, routing)) { return; } @@ -518,10 +518,48 @@ export class CodexAppServerClient { }; } - async runCompact(params: ThreadCompactStartParams): Promise { - const compactionCompleted = this.awaitCompactionCompleted(params.threadId); - await this.threadCompactStart(params); - return await compactionCompleted; + async runCompact( + params: ThreadCompactStartParams, + onTurnStarted?: (turnId: string) => void, + ): Promise> { + type Result = CompactionCompletedNotification | Extract; + let compactTurnId: string | null = null; + let resolveCompleted: (event: Result) => void = () => {}; + let rejectCompleted: (error: Error) => void = () => {}; + const completed = new Promise((resolve, reject) => { + resolveCompleted = resolve; + rejectCompleted = reject; + }); + // The request acknowledgement can still be pending when the connection closes. + void completed.catch(() => {}); + const closed = this.connection.onClose?.(() => rejectCompleted(new Error("Codex connection closed during compaction."))); + const completeCompaction = (event: CompactionCompletedNotification) => { + if (compactTurnId !== null && event.params.turnId !== compactTurnId) return; + resolveCompleted(event); + }; + const releaseCompactionCapture = this.captureCompactionCompletions(params.threadId, completeCompaction); + const releaseTurnCapture = this.captureTurnCompletions(params.threadId, (event) => { + if (compactTurnId === null || event.turn.id !== compactTurnId) return; + if (event.turn.status !== "inProgress") { + resolveCompleted({method: "turn/completed", params: event}); + } + }); + const releaseRoutingCapture = this.captureTurnRoutings(params.threadId, (turnId, notification) => { + if (compactTurnId !== null) return; + if (notification.method !== "turn/started" + && !(notification.method === "item/started" && notification.params.item.type === "contextCompaction")) return; + compactTurnId = turnId; + onTurnStarted?.(turnId); + }); + try { + await this.threadCompactStart(params); + return await completed; + } finally { + releaseTurnCapture(); + releaseRoutingCapture(); + releaseCompactionCapture(); + closed?.dispose(); + } } async turnInterrupt(params: TurnInterruptParams): Promise { @@ -736,9 +774,10 @@ export class CodexAppServerClient { async awaitCompactionCompleted(threadId: string): Promise { return await new Promise((resolve) => { - const resolvers = this.pendingCompactionCompletionResolvers.get(threadId) ?? new Set(); - resolvers.add(resolve); - this.pendingCompactionCompletionResolvers.set(threadId, resolvers); + const releaseCapture = this.captureCompactionCompletions(threadId, (event) => { + releaseCapture(); + resolve(event); + }); }); } @@ -828,12 +867,29 @@ export class CodexAppServerClient { if (!resolvers) { return; } - this.pendingCompactionCompletionResolvers.delete(threadId); for (const resolve of resolvers) { resolve(event); } } + private captureCompactionCompletions( + threadId: string, + capture: (event: CompactionCompletedNotification) => void, + ): () => void { + const captures = this.pendingCompactionCompletionResolvers.get(threadId) ?? new Set(); + captures.add(capture); + this.pendingCompactionCompletionResolvers.set(threadId, captures); + let released = false; + return () => { + if (released) return; + released = true; + captures.delete(capture); + if (captures.size === 0) { + this.pendingCompactionCompletionResolvers.delete(threadId); + } + }; + } + private recordThreadStatusChanged(event: ThreadStatusChangedNotification): void { const captures = this.threadStatusCaptures.get(event.threadId); if (!captures) { @@ -864,7 +920,10 @@ export class CodexAppServerClient { } } - private recordTurnRouting(routing: { threadId: string | null, turnId: string | null }): void { + private recordTurnRouting( + routing: { threadId: string | null, turnId: string | null }, + notification: ServerNotification, + ): void { if (routing.threadId === null || routing.turnId === null) { return; } @@ -873,7 +932,7 @@ export class CodexAppServerClient { return; } for (const capture of captures) { - capture(routing.turnId); + capture(routing.turnId, notification); } } @@ -938,8 +997,8 @@ export class CodexAppServerClient { }; } - private captureTurnRoutings(threadId: string, capture: (turnId: string) => void): () => void { - const captures = this.turnRoutingCaptures.get(threadId) ?? new Set<(turnId: string) => void>(); + private captureTurnRoutings(threadId: string, capture: (turnId: string, notification: ServerNotification) => void): () => void { + const captures = this.turnRoutingCaptures.get(threadId) ?? new Set<(turnId: string, notification: ServerNotification) => void>(); captures.add(capture); this.turnRoutingCaptures.set(threadId, captures); let released = false; diff --git a/src/CodexCommands.ts b/src/CodexCommands.ts index 321ac5da..c99e338a 100644 --- a/src/CodexCommands.ts +++ b/src/CodexCommands.ts @@ -226,8 +226,12 @@ export class CodexCommands { return { handled: options.setConfigOption !== undefined }; } case "compact": { - await this.runWithProcessCheck(() => this.codexAcpClient.runCompact(sessionId)); - return { handled: true }; + options.onTurnStartPending?.(); + const turnCompleted = await this.runWithProcessCheck(() => this.codexAcpClient.runCompact( + sessionId, + (turnId) => options.onTurnStarted?.(turnId, sessionId), + )); + return { handled: true, ...(turnCompleted === undefined ? {} : {turnCompleted}) }; } case "goal": { return await this.runGoalCommand(sessionState, command.rest, options); diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index ae5e008b..c5662b9c 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -243,6 +243,7 @@ export class CodexEventHandler { new ACPSessionConnection(connection, sessionState.sessionId), ), onAccountUpdated?: (notification: AccountUpdatedNotification) => void, + private readonly supportsCompaction = false, ) { this.onAccountUpdated = onAccountUpdated; this.sessionState = sessionState; @@ -294,6 +295,7 @@ export class CodexEventHandler { await this.handleNotification(notification); return; } + await this.finishCompactionsForNotification(notification); if (notification.params.willRetry) { await this.session.update(this.createSessionFailureUpdate(this.recordRetryWarning(notification.params, false))); return; @@ -375,8 +377,13 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { await this.flushPendingErrors(); + await this.finishCompactionsForNotification(notification); const closingChildren = this.subagents.closingChildSessions(notification); for (const child of closingChildren) { + await this.finishOutstandingCompactions( + child.state === "cancelled" ? "cancelled" : "failed", + child.sessionId, + ); await this.sessionState.asyncTasks.reconcile(child.threadId, child.sessionId); } const handledBySubagents = await this.subagents.handle(notification); @@ -411,13 +418,49 @@ export class CodexEventHandler { } async waitForNativeSubagents(signal: AbortSignal): Promise { - await this.subagents.wait(signal); + if (await this.subagents.wait(signal) === "timed_out") { + await this.finishOutstandingNativeSubagents("failed"); + } } async finishOutstandingNativeSubagents(state: SubagentState): Promise { + await this.finishOutstandingCompactions(state === "cancelled" ? "cancelled" : "failed"); await this.subagents.finishOutstanding(state); } + async finishOutstandingCompactions(status: "failed" | "cancelled", sessionId?: string): Promise { + if (!this.supportsCompaction) return; + for (const {sessionId: targetSessionId, update} of this.sessionState.compactions.finishOutstanding(status, sessionId)) { + await this.session.update(update, targetSessionId); + } + } + + private async finishCompactionsForNotification(notification: ServerNotification): Promise { + if (!this.supportsCompaction) return; + let updates: UpdateSessionEvent[]; + const sessionId = this.subagents.notificationSessionId(notification); + if (notification.method === "turn/completed") { + const turn = notification.params.turn; + if (turn.status === "inProgress") return; + updates = this.sessionState.compactions.finishTurn( + sessionId, + turn.id, + turn.status === "interrupted" ? "cancelled" : "failed", + turn.error?.message ?? "Codex ended the turn before compaction completed.", + ); + } else if (notification.method === "error" && !notification.params.willRetry) { + updates = this.sessionState.compactions.finishTurn( + sessionId, + notification.params.turnId, + "failed", + notification.params.error.message, + ); + } else { + return; + } + for (const update of updates) await this.session.update(update, sessionId); + } + async flushPendingPlanUpdates(): Promise { this.cancelPlanUpdateTimer(); do { @@ -538,7 +581,11 @@ export class CodexEventHandler { case "item/autoApprovalReview/completed": return this.handleGuardianApprovalReviewCompleted(notification.params); case "thread/compacted": - return this.createContextCompactedEvent(); + return this.supportsCompaction + ? this.sessionState.compactions.completeLegacy( + this.subagents.notificationSessionId(notification), notification.params.turnId, + ) + : this.createContextCompactedEvent(); case "item/reasoning/summaryTextDelta": this.completeRetryIncidentOnTurnProgress(); return this.createReasoningDeltaEvent(notification.params); @@ -748,7 +795,12 @@ export class CodexEventHandler { this.rememberAgentMessagePhase(event.item); return null; case "contextCompaction": - return createContextCompactionStartUpdate(event.item); + return this.supportsCompaction + ? this.sessionState.compactions.start( + this.subagents.notificationSessionId({method: "item/started", params: event}), + event.turnId, event.item.id, + ) + : createContextCompactionStartUpdate(event.item); case "subAgentActivity": return this.subagents.legacyActivityStarted(event.item); case "sleep": @@ -817,7 +869,12 @@ export class CodexEventHandler { case "exitedReviewMode": return this.createExitedReviewModeEvent(event.item); case "contextCompaction": - return createContextCompactionCompleteUpdate(event.item); + return this.supportsCompaction + ? this.sessionState.compactions.complete( + this.subagents.notificationSessionId({method: "item/completed", params: event}), + event.turnId, event.item.id, + ) + : createContextCompactionCompleteUpdate(event.item); //ignored types case "subAgentActivity": return this.subagents.legacyActivityCompleted(event.item); diff --git a/src/CodexSessionCompactions.ts b/src/CodexSessionCompactions.ts new file mode 100644 index 00000000..14c4ed58 --- /dev/null +++ b/src/CodexSessionCompactions.ts @@ -0,0 +1,92 @@ +import type {ClientCapabilities, CompactionUpdate} from "@agentclientprotocol/sdk"; + +type Update = CompactionUpdate & {sessionUpdate: "compaction_update"}; +type TerminalStatus = "completed" | "failed" | "cancelled"; +type Compaction = { + sessionId: string; + turnId: string; + id: string; + terminal: boolean; + legacyOnly: boolean; +}; + +export function clientSupportsCompaction(capabilities: ClientCapabilities | null): boolean { + return capabilities?.session?.compaction != null; +} + +export function createCompactionUpdate(compactionId: string, status: "in_progress" | TerminalStatus, error?: string): Update { + return { + sessionUpdate: "compaction_update", + compactionId, + status, + ...(status === "failed" && error !== undefined ? {error} : {}), + }; +} + +/** Session-owned identity survives prompt handler replacement and late duplicate events. */ +export class CodexSessionCompactions { + private readonly items = new Map(); + private readonly latestByTurn = new Map(); + + start(sessionId: string, turnId: string, itemId: string): Update | null { + const key = JSON.stringify([sessionId, itemId]); + if (this.items.has(key)) return null; + const compaction = this.remember(sessionId, turnId, itemId); + return createCompactionUpdate(compaction.id, "in_progress"); + } + + complete(sessionId: string, turnId: string, itemId: string): Update | null { + const latest = this.latestByTurn.get(JSON.stringify([sessionId, turnId])); + if (!this.items.has(JSON.stringify([sessionId, itemId])) && latest?.legacyOnly) { + // Attaching mid-compaction can expose both completion surfaces with + // no start. Preserve the ID already published by the legacy signal. + this.items.delete(JSON.stringify([sessionId, latest.id])); + this.items.set(JSON.stringify([sessionId, itemId]), latest); + latest.legacyOnly = false; + } + const compaction = this.items.get(JSON.stringify([sessionId, itemId])) + ?? this.remember(sessionId, turnId, itemId); + return this.finish(compaction, "completed"); + } + + completeLegacy(sessionId: string, turnId: string): Update | null { + // Older Codex versions expose only a turn-addressed completion. When both + // surfaces arrive, they describe the same latest compaction in that turn. + const compaction = this.latestByTurn.get(JSON.stringify([sessionId, turnId])) + ?? this.remember(sessionId, turnId, `compaction:${turnId}`, true); + return this.finish(compaction, "completed"); + } + + finishTurn(sessionId: string, turnId: string, status: "failed" | "cancelled", error?: string): Update[] { + const updates: Update[] = []; + for (const compaction of this.items.values()) { + if (compaction.sessionId !== sessionId || compaction.turnId !== turnId) continue; + const update = this.finish(compaction, status, error); + if (update) updates.push(update); + } + return updates; + } + + finishOutstanding(status: "failed" | "cancelled", sessionId?: string): {sessionId: string; update: Update}[] { + const updates: {sessionId: string; update: Update}[] = []; + for (const compaction of this.items.values()) { + if (sessionId !== undefined && compaction.sessionId !== sessionId) continue; + const update = this.finish(compaction, status); + if (update) updates.push({sessionId: compaction.sessionId, update}); + } + return updates; + } + + private remember(sessionId: string, turnId: string, id: string, legacyOnly = false): Compaction { + const compaction = {sessionId, turnId, id, terminal: false, legacyOnly}; + this.items.set(JSON.stringify([sessionId, id]), compaction); + this.latestByTurn.set(JSON.stringify([sessionId, turnId]), compaction); + return compaction; + } + + private finish(compaction: Compaction, status: TerminalStatus, error?: string): Update | null { + if (compaction.terminal) return null; + compaction.terminal = true; + return createCompactionUpdate(compaction.id, status, error); + } +} diff --git a/src/__tests__/CodexACPAgent/compact-command-lifecycle.test.ts b/src/__tests__/CodexACPAgent/compact-command-lifecycle.test.ts new file mode 100644 index 00000000..ee598dd7 --- /dev/null +++ b/src/__tests__/CodexACPAgent/compact-command-lifecycle.test.ts @@ -0,0 +1,175 @@ +import {describe, expect, it, vi} from "vitest"; +import type {ServerNotification} from "../../app-server"; +import type {Turn} from "../../app-server/v2"; +import {createCodexMockTestFixture, createTestSessionState} from "../acp-test-utils"; + +const sessionId = "compact-command-session"; +const turnId = "compact-command-turn"; + +describe("compact command lifecycle", () => { + it.each(["interrupted", "failed"] as const)("captures an early %s turn before compact acknowledgement", async (status) => { + const fixture = createCodexMockTestFixture(); + const appServer = fixture.getCodexAppServerClient(); + const completed = turnCompleted(status); + vi.spyOn(appServer, "threadCompactStart").mockImplementation(async () => { + fixture.sendServerNotification({method: "turn/started", params: {threadId: sessionId, turn: turn("inProgress")}}); + fixture.sendServerNotification(completed); + return {}; + }); + const onTurnStarted = vi.fn(); + + await expect(appServer.runCompact({threadId: sessionId}, onTurnStarted)).resolves.toEqual(completed); + expect(onTurnStarted).toHaveBeenCalledExactlyOnceWith(turnId); + }); + + it("waits for the compact turn when another turn finishes", async () => { + const fixture = createCodexMockTestFixture(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadCompactStart").mockResolvedValue({}); + let settled = false; + const pending = appServer.runCompact({threadId: sessionId}).then(result => { + settled = true; + return result; + }); + // The preceding compact prompt may resolve on item/completed before this + // previous turn/completed reaches the next compact request. + fixture.sendServerNotification({ + method: "turn/completed", + params: {threadId: sessionId, turn: {...turn("completed"), id: "previous-turn"}}, + }); + await Promise.resolve(); + expect(settled).toBe(false); + fixture.sendServerNotification({method: "turn/started", params: {threadId: sessionId, turn: turn("inProgress")}}); + fixture.sendServerNotification({ + method: "item/completed", + params: { + threadId: sessionId, turnId: "other-turn", completedAtMs: 1, + item: {type: "contextCompaction", id: "other-compaction"}, + }, + }); + fixture.sendServerNotification({ + method: "turn/completed", + params: {threadId: sessionId, turn: {...turn("interrupted"), id: "other-turn"}}, + }); + await Promise.resolve(); + expect(settled).toBe(false); + + const completed: ServerNotification = { + method: "item/completed", + params: { + threadId: sessionId, turnId, completedAtMs: 2, + item: {type: "contextCompaction", id: "compaction-item"}, + }, + }; + fixture.sendServerNotification(completed); + await expect(pending).resolves.toEqual(completed); + }); + + it("releases compact observation after the request is rejected", async () => { + const fixture = createCodexMockTestFixture(); + const appServer = fixture.getCodexAppServerClient(); + const start = vi.spyOn(appServer, "threadCompactStart").mockRejectedValueOnce(new Error("Cannot compact")); + const onTurnStarted = vi.fn(); + await expect(appServer.runCompact({threadId: sessionId}, onTurnStarted)).rejects.toThrow("Cannot compact"); + fixture.sendServerNotification({method: "turn/started", params: {threadId: sessionId, turn: turn("inProgress")}}); + fixture.sendServerNotification(turnCompleted("interrupted")); + expect(onTurnStarted).not.toHaveBeenCalled(); + + start.mockResolvedValue({}); + const retry = appServer.runCompact({threadId: sessionId}); + const completed: ServerNotification = {method: "thread/compacted", params: {threadId: sessionId, turnId}}; + fixture.sendServerNotification(completed); + await expect(retry).resolves.toEqual(completed); + }); + + it("rejects a compact wait when the Codex connection closes", async () => { + const fixture = createCodexMockTestFixture(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "threadCompactStart").mockResolvedValue({}); + let close = () => {}; + const dispose = vi.fn(); + appServer.connection.onClose = vi.fn(listener => { + close = () => listener(undefined); + return {dispose}; + }); + const pending = appServer.runCompact({threadId: sessionId}); + await Promise.resolve(); + close(); + + await expect(pending).rejects.toThrow("Codex connection closed during compaction."); + expect(dispose).toHaveBeenCalledOnce(); + }); + + it("cancels the manual compact turn and finishes the ACP prompt", async () => { + const fixture = await commandFixture(); + const appServer = fixture.getCodexAppServerClient(); + const start = vi.spyOn(appServer, "threadCompactStart").mockResolvedValue({}); + const interrupt = vi.spyOn(appServer, "turnInterrupt").mockImplementation(async () => { + fixture.sendServerNotification(turnCompleted("interrupted")); + return {}; + }); + const prompt = fixture.getCodexAcpAgent().prompt({sessionId, prompt: [{type: "text", text: "/compact"}]}); + await vi.waitFor(() => expect(start).toHaveBeenCalled()); + startCompaction(fixture); + await fixture.getCodexAcpClient().waitForSessionNotifications(sessionId); + + await fixture.getCodexAcpAgent().cancel({sessionId}); + expect(interrupt).toHaveBeenCalledWith({threadId: sessionId, turnId}); + await expect(prompt).resolves.toMatchObject({stopReason: "cancelled"}); + expect(fixture.getAcpConnectionDump([])).toContain('"status": "cancelled"'); + }); + + it("returns the failed compact turn through ACP failure handling", async () => { + const fixture = await commandFixture(); + const start = vi.spyOn(fixture.getCodexAppServerClient(), "threadCompactStart").mockResolvedValue({}); + const prompt = fixture.getCodexAcpAgent().prompt({sessionId, prompt: [{type: "text", text: "/compact"}]}); + await vi.waitFor(() => expect(start).toHaveBeenCalled()); + startCompaction(fixture); + fixture.sendServerNotification(turnCompleted("failed")); + + await expect(prompt).resolves.toMatchObject({ + _meta: {jetbrains: {air: {sessionFailure: {severity: "error"}}}}, + }); + expect(fixture.getAcpConnectionDump([])).toContain('"status": "failed"'); + }); +}); + +async function commandFixture() { + const fixture = createCodexMockTestFixture(); + const session = createTestSessionState({sessionId}); + vi.spyOn(fixture.getCodexAcpAgent(), "getSessionState") + .mockReturnValue(session); + // @ts-expect-error - registering local session state for the ACP cancel path + fixture.getCodexAcpAgent().sessions.set(sessionId, session); + await fixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: { + session: {compaction: {}}, + _meta: {jetbrains: {air: {version: 1, capabilities: ["sessionFailure"]}}}, + }, + }); + fixture.clearAcpConnectionDump(); + return fixture; +} + +function startCompaction(fixture: ReturnType) { + fixture.sendServerNotification({method: "turn/started", params: {threadId: sessionId, turn: turn("inProgress")}}); + fixture.sendServerNotification({ + method: "item/started", + params: {threadId: sessionId, turnId, startedAtMs: 0, item: {type: "contextCompaction", id: "compaction-item"}}, + }); +} + +function turnCompleted(status: "failed" | "interrupted"): Extract { + return {method: "turn/completed", params: {threadId: sessionId, turn: turn(status)}}; +} + +function turn(status: Turn["status"]): Turn { + return { + id: turnId, items: [], itemsView: "full", status, + error: status === "failed" ? { + message: "Compaction service failed.", codexErrorInfo: "serverOverloaded", additionalDetails: null, misalignment: null, + } : null, + startedAt: null, completedAt: null, durationMs: null, + }; +} diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-completed-only.json b/src/__tests__/CodexACPAgent/data/session-compaction-completed-only.json new file mode 100644 index 00000000..faf7e432 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-completed-only.json @@ -0,0 +1,13 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-deduplicated.json b/src/__tests__/CodexACPAgent/data/session-compaction-deduplicated.json new file mode 100644 index 00000000..308d3c2f --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-deduplicated.json @@ -0,0 +1,26 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-error.json b/src/__tests__/CodexACPAgent/data/session-compaction-error.json new file mode 100644 index 00000000..300827ba --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-error.json @@ -0,0 +1,29 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "failed", + "error": "The compaction request failed." + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-failed.json b/src/__tests__/CodexACPAgent/data/session-compaction-failed.json new file mode 100644 index 00000000..b779f845 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-failed.json @@ -0,0 +1,27 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "failed", + "error": "The compaction request failed." + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-interrupted.json b/src/__tests__/CodexACPAgent/data/session-compaction-interrupted.json new file mode 100644 index 00000000..ff038d02 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-interrupted.json @@ -0,0 +1,26 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "cancelled" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-legacy.json b/src/__tests__/CodexACPAgent/data/session-compaction-legacy.json new file mode 100644 index 00000000..1e208197 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-legacy.json @@ -0,0 +1,54 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "tool_call", + "toolCallId": "compaction-item", + "kind": "think", + "title": "Compact conversation", + "status": "in_progress", + "_meta": { + "contextCompaction": { + "version": 1 + } + } + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "tool_call_update", + "toolCallId": "compaction-item", + "title": "Compact conversation", + "status": "completed", + "_meta": { + "contextCompaction": { + "version": 1 + } + } + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { + "type": "text", + "text": "*Context compacted to fit the model's context window.*\n\n" + } + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-lifecycle.json b/src/__tests__/CodexACPAgent/data/session-compaction-lifecycle.json new file mode 100644 index 00000000..d012c65a --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-lifecycle.json @@ -0,0 +1,42 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "next-message", + "content": { + "type": "text", + "text": "Continuing the task." + } + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-multiple.json b/src/__tests__/CodexACPAgent/data/session-compaction-multiple.json new file mode 100644 index 00000000..755040b3 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-multiple.json @@ -0,0 +1,52 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "second-compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "second-compaction-item", + "status": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-native-child-collaboration-terminal.json b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-collaboration-terminal.json new file mode 100644 index 00000000..7fd10bbe --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-collaboration-terminal.json @@ -0,0 +1,54 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "cancelled" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "failed-compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "failed" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_state_update", + "subagentSessionId": "compaction-child", + "state": "cancelled" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_state_update", + "subagentSessionId": "failed-compaction-child", + "state": "failed" + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-native-child-identities.json b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-identities.json new file mode 100644 index 00000000..66c636ff --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-identities.json @@ -0,0 +1,80 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_spawned", + "subagentSessionId": "compaction-child", + "name": "Compaction child", + "task": "Continue the delegated task.", + "capabilities": {} + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_state_update", + "subagentSessionId": "compaction-child", + "state": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-native-child-interrupted.json b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-interrupted.json new file mode 100644 index 00000000..a24fdde5 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-interrupted.json @@ -0,0 +1,54 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_spawned", + "subagentSessionId": "compaction-child", + "name": "Compaction child", + "task": "Continue the delegated task.", + "capabilities": {} + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "cancelled" + } + } + ] +} +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_state_update", + "subagentSessionId": "compaction-child", + "state": "cancelled" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-native-child-timeout.json b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-timeout.json new file mode 100644 index 00000000..6d7c2e45 --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-native-child-timeout.json @@ -0,0 +1,41 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-child", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "failed" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "subagent_state_update", + "subagentSessionId": "compaction-child", + "state": "failed" + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-next-prompt.json b/src/__tests__/CodexACPAgent/data/session-compaction-next-prompt.json new file mode 100644 index 00000000..5f7246eb --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-next-prompt.json @@ -0,0 +1,54 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "second-compaction-item", + "status": "in_progress" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "second-compaction-item", + "status": "completed" + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-replay.json b/src/__tests__/CodexACPAgent/data/session-compaction-replay.json new file mode 100644 index 00000000..a8dd59ad --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-replay.json @@ -0,0 +1,47 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "before-message", + "content": { + "type": "text", + "text": "Before compaction." + } + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "messageId": "after-message", + "content": { + "type": "text", + "text": "After compaction." + } + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-retried.json b/src/__tests__/CodexACPAgent/data/session-compaction-retried.json new file mode 100644 index 00000000..8c3b33ca --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-retried.json @@ -0,0 +1,28 @@ +[ + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "in_progress" + } + } + ] + }, + { + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction-item", + "status": "completed" + } + } + ] + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-thread-completed.json b/src/__tests__/CodexACPAgent/data/session-compaction-thread-completed.json new file mode 100644 index 00000000..5128b9ad --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-thread-completed.json @@ -0,0 +1,13 @@ +{ + "method": "sessionUpdate", + "args": [ + { + "sessionId": "compaction-session", + "update": { + "sessionUpdate": "compaction_update", + "compactionId": "compaction:compaction-turn", + "status": "completed" + } + } + ] +} \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/data/session-compaction-wire.json b/src/__tests__/CodexACPAgent/data/session-compaction-wire.json new file mode 100644 index 00000000..3acc9eee --- /dev/null +++ b/src/__tests__/CodexACPAgent/data/session-compaction-wire.json @@ -0,0 +1,18 @@ +[ + { + "sessionId": "compaction-session", + "update": { + "compactionId": "compaction-item", + "status": "in_progress", + "sessionUpdate": "compaction_update" + } + }, + { + "sessionId": "compaction-session", + "update": { + "compactionId": "compaction-item", + "status": "completed", + "sessionUpdate": "compaction_update" + } + } +] \ No newline at end of file diff --git a/src/__tests__/CodexACPAgent/session-compaction.test.ts b/src/__tests__/CodexACPAgent/session-compaction.test.ts new file mode 100644 index 00000000..93dc1b44 --- /dev/null +++ b/src/__tests__/CodexACPAgent/session-compaction.test.ts @@ -0,0 +1,617 @@ +import * as acp from "@agentclientprotocol/sdk"; +import {describe, expect, it, vi} from "vitest"; +import {CodexAcpClient} from "../../CodexAcpClient"; +import {CodexAcpServer} from "../../CodexAcpServer"; +import {CodexAppServerClient} from "../../CodexAppServerClient"; +import {ACPSessionConnection} from "../../ACPSessionConnection"; +import {CodexSubagentEventRouter} from "../../subagents/CodexSubagentEventRouter"; +import type {ServerNotification} from "../../app-server"; +import type {Thread, ThreadItem, Turn} from "../../app-server/v2"; +import { + createCodexMockTestFixture, + createTestModel, + createTestSessionState, + type CodexMockTestFixture, +} from "../acp-test-utils"; +import {createMockConnections} from "./test-utils"; + +const sessionId = "compaction-session"; +const turnId = "compaction-turn"; +const compactionId = "compaction-item"; +const childSessionId = "compaction-child"; +const childTurnId = "compaction-child-turn"; +const compactionCapabilities: acp.ClientCapabilities = { + session: {compaction: {}}, +}; + +describe("session compaction", () => { + it("keeps one timeline entity through item progress and completion", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [ + compactionStarted(), + { + method: "item/agentMessage/delta", + params: {threadId: sessionId, turnId, itemId: "next-message", delta: "Continuing the task."}, + }, + compactionCompleted(), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-lifecycle.json", + ); + }); + + it.each(["before", "after"])("deduplicates thread/compacted %s item completion", async (order) => { + const fixture = await createFixture(); + const compacted: ServerNotification = { + method: "thread/compacted", + params: {threadId: sessionId, turnId}, + }; + await sendNotifications(fixture, [ + compactionStarted(), + ...(order === "before" + ? [compacted, compactionCompleted()] + : [compactionCompleted(), compacted]), + compactionCompleted(), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-deduplicated.json", + ); + }); + + it("materializes an item completion without inventing an earlier start", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [compactionCompleted()]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-completed-only.json", + ); + }); + + it("materializes a legacy completion signal when no lifecycle item is available", async () => { + const fixture = await createFixture(); + const compacted: ServerNotification = { + method: "thread/compacted", + params: {threadId: sessionId, turnId}, + }; + await sendNotifications(fixture, [compacted, compacted]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-thread-completed.json", + ); + }); + + it("keeps the original entity when a native completion follows the legacy signal", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [ + {method: "thread/compacted", params: {threadId: sessionId, turnId}}, + compactionCompleted(), + compactionCompleted(), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-thread-completed.json", + ); + }); + + it("keeps separate compactions in the same turn distinct", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [ + compactionStarted(), + compactionCompleted(), + compactionStarted("second-compaction-item"), + compactionCompleted("second-compaction-item"), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-multiple.json", + ); + }); + + it("preserves compaction identity when a later prompt replaces its event handler", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [compactionStarted(), compactionCompleted()]); + const firstCompaction = fixture.getAcpConnectionEvents([]); + await fixture.getCodexAcpAgent().prompt({sessionId, prompt: [{type: "text", text: "Continue again."}]}); + fixture.clearAcpConnectionDump(); + await sendNotifications(fixture, [ + compactionStarted(), + compactionCompleted(), + compactionStarted("second-compaction-item"), + compactionCompleted("second-compaction-item"), + ]); + + await expect(JSON.stringify([...firstCompaction, ...fixture.getAcpConnectionEvents([])], null, 2)) + .toMatchFileSnapshot("data/session-compaction-next-prompt.json"); + }); + + it("delivers the lifecycle and finishes a manual /compact request", async () => { + const fixture = await createFixture(); + const compactStart = vi.spyOn(fixture.getCodexAppServerClient(), "threadCompactStart"); + const prompt = fixture.getCodexAcpAgent().prompt({sessionId, prompt: [{type: "text", text: "/compact"}]}); + await vi.waitFor(() => expect(compactStart).toHaveBeenCalledWith({threadId: sessionId})); + await sendNotifications(fixture, [compactionStarted(), compactionCompleted()]); + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-deduplicated.json", + ); + }); + + it("keeps equal compaction item IDs distinct in parent and native child sessions", async () => { + const fixture = await createNativeChildFixture(); + await sendNotifications(fixture, [ + compactionStarted(), + compactionStarted(compactionId, childSessionId, childTurnId), + compactionCompleted(), + compactionCompleted(compactionId, childSessionId, childTurnId), + { + method: "turn/completed", + params: {threadId: childSessionId, turn: {...createTurn("completed"), id: childTurnId}}, + }, + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-native-child-identities.json", + ); + }); + + it.each(["child", "parent"] as const)("cancels a native child compaction before the child closes when the %s is interrupted", async (interruptedSession) => { + const fixture = await createNativeChildFixture(); + await sendNotifications(fixture, [ + compactionStarted(compactionId, childSessionId, childTurnId), + { + method: "turn/completed", + params: { + threadId: interruptedSession === "child" ? childSessionId : sessionId, + turn: { + ...createTurn("interrupted"), + id: interruptedSession === "child" ? childTurnId : turnId, + }, + }, + }, + compactionCompleted(compactionId, childSessionId, childTurnId), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-native-child-interrupted.json", + ); + }); + + it("uses each child's collaboration terminal state to settle its compaction before closing", async () => { + const fixture = await createNativeChildFixture(); + const failedChildSessionId = "failed-compaction-child"; + await sendNotifications(fixture, [ + nativeChildActivityStarted(failedChildSessionId), + compactionStarted(compactionId, childSessionId, childTurnId), + compactionStarted(compactionId, failedChildSessionId, childTurnId), + ]); + fixture.clearAcpConnectionDump(); + await sendNotifications(fixture, [ + { + method: "item/completed", + params: { + threadId: sessionId, + turnId, + completedAtMs: 1, + item: { + type: "collabAgentToolCall", + id: "wait-compaction-children", + tool: "wait", + status: "completed", + senderThreadId: sessionId, + receiverThreadIds: [childSessionId, failedChildSessionId], + prompt: null, + model: null, + reasoningEffort: null, + agentsStates: { + [childSessionId]: {status: "interrupted", message: null}, + [failedChildSessionId]: {status: "errored", message: null}, + }, + }, + }, + }, + compactionCompleted(compactionId, childSessionId, childTurnId), + compactionCompleted(compactionId, failedChildSessionId, childTurnId), + ]); + + await expect(nativeCompactionLifecycleDump(fixture)).toMatchFileSnapshot( + "data/session-compaction-native-child-collaboration-terminal.json", + ); + }); + + it.each(["timer", "elapsed deadline"] as const)("settles a native child compaction before timeout closes the child via %s", async (timeoutPath) => { + vi.useFakeTimers(); + try { + const fixture = await createNativeChildFixture(); + const siblingSessionId = "timeout-wakeup-child"; + if (timeoutPath === "elapsed deadline") { + await sendNotifications(fixture, [nativeChildActivityStarted(siblingSessionId)]); + } + const prompt = fixture.getCodexAcpAgent().prompt({ + sessionId, + prompt: [{type: "text", text: "Wait for the delegated task."}], + }); + await vi.advanceTimersByTimeAsync(0); + fixture.clearAcpConnectionDump(); + await sendNotifications(fixture, [compactionStarted(compactionId, childSessionId, childTurnId)]); + + if (timeoutPath === "elapsed deadline") { + // Wake the pending wait after its deadline without firing the timeout timer. + vi.setSystemTime(Date.now() + 10 * 60 * 1000); + await sendNotifications(fixture, [{ + method: "turn/completed", + params: {threadId: siblingSessionId, turn: createTurn("completed")}, + }]); + } + else { + await vi.advanceTimersByTimeAsync(10 * 60 * 1000); + } + await expect(prompt).resolves.toMatchObject({stopReason: "end_turn"}); + await sendNotifications(fixture, [compactionCompleted(compactionId, childSessionId, childTurnId)]); + + await expect(nativeCompactionLifecycleDump(fixture, childSessionId)).toMatchFileSnapshot( + "data/session-compaction-native-child-timeout.json", + ); + } + finally { + vi.useRealTimers(); + } + }); + + it.each(["failed", "interrupted"] as const)("settles an unfinished compaction when its turn is %s", async (status) => { + const fixture = await createFixture(); + const turn = createTurn(status); + if (status === "failed") { + turn.error = { + message: "The compaction request failed.", + codexErrorInfo: null, + additionalDetails: null, + misalignment: null, + }; + } + await sendNotifications(fixture, [ + compactionStarted(), + {method: "turn/completed", params: {threadId: sessionId, turn}}, + {method: "turn/completed", params: {threadId: sessionId, turn}}, + compactionCompleted(), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + `data/session-compaction-${status}.json`, + ); + }); + + it("does not settle a compaction when a different turn finishes", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [ + compactionStarted(), + { + method: "turn/completed", + params: {threadId: sessionId, turn: {...createTurn("interrupted"), id: "previous-turn"}}, + }, + compactionCompleted(), + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-deduplicated.json", + ); + }); + + it.each([false, true])("settles a terminal error exactly once with typed failures %s", async (typedFailures) => { + const fixture = await createFixture({ + ...compactionCapabilities, + ...(typedFailures ? { + _meta: {jetbrains: {air: {version: 1, capabilities: ["sessionFailure"]}}}, + } : {}), + }); + await sendNotifications(fixture, [compactionStarted(), compactionError(false), compactionCompleted()]); + + await expect(compactionUpdatesDump(fixture)).toMatchFileSnapshot( + "data/session-compaction-error.json", + ); + }); + + it("keeps a compaction in progress while Codex retries its request", async () => { + const fixture = await createFixture(); + await sendNotifications(fixture, [compactionStarted(), compactionError(true), compactionCompleted()]); + + await expect(compactionUpdatesDump(fixture)).toMatchFileSnapshot( + "data/session-compaction-retried.json", + ); + }); + + it.each([ + ["missing session", {}], + ["null session", {session: null}], + ["missing compaction", {session: {}}], + ["null compaction", {session: {compaction: null}}], + ] as const)("preserves the existing v1 fallback with %s", async (_name, capabilities) => { + const fixture = await createFixture(capabilities); + await sendNotifications(fixture, [ + compactionStarted(), + compactionCompleted(), + {method: "thread/compacted", params: {threadId: sessionId, turnId}}, + ]); + + await expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot( + "data/session-compaction-legacy.json", + ); + }); + + it("replays a completed boundary between its neighboring messages", async () => { + const fixture = createCodexMockTestFixture(); + const agent = fixture.getCodexAcpAgent(); + const client = fixture.getCodexAcpClient(); + const appServer = fixture.getCodexAppServerClient(); + const model = createTestModel(); + vi.spyOn(client, "authRequired").mockResolvedValue(false); + vi.spyOn(client, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false}); + vi.spyOn(client, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(appServer, "listModels").mockResolvedValue({data: [model], nextCursor: null}); + const thread = createThread([ + agentMessage("before-message", "Before compaction."), + {type: "contextCompaction", id: compactionId}, + agentMessage("after-message", "After compaction."), + ]); + vi.spyOn(appServer, "threadResume").mockResolvedValue({ + thread, + model: model.id, + modelProvider: "openai", + serviceTier: null, + cwd: thread.cwd, + instructionSources: [], + approvalPolicy: "never", + approvalsReviewer: "user", + sandbox: {type: "dangerFullAccess"}, + reasoningEffort: model.defaultReasoningEffort, + turnsBackwardsCursor: null, + itemsBackwardsCursor: null, + }); + vi.spyOn(appServer, "threadReadWithHistory").mockResolvedValue({thread}); + await agent.initialize({protocolVersion: 1, clientCapabilities: compactionCapabilities}); + await agent.loadSession({sessionId, cwd: thread.cwd, mcpServers: []}); + + const timelineUpdates = fixture.getAcpConnectionEvents([]).filter(event => + event.method === "sessionUpdate" + && ["agent_message_chunk", "tool_call", "tool_call_update", "compaction_update", "compaction_summary_chunk"] + .includes(event.args[0].update.sessionUpdate), + ); + await expect(JSON.stringify(timelineUpdates, null, 2)).toMatchFileSnapshot( + "data/session-compaction-replay.json", + ); + }); + + it("negotiates the standard capability and delivers lifecycle updates through the ACP SDK", async () => { + const mocks = createMockConnections(); + const appServer = new CodexAppServerClient(mocks.mockCodexConnection); + const codexClient = new CodexAcpClient(appServer); + vi.spyOn(appServer, "initialize").mockResolvedValue({codexHome: null} as never); + vi.spyOn(appServer, "turnStart").mockResolvedValue({turn: createTurn("inProgress")}); + vi.spyOn(appServer, "awaitTurnCompleted").mockResolvedValue({ + threadId: sessionId, + turn: createTurn("completed"), + }); + const clientToAgent = new TransformStream(); + const agentToClient = new TransformStream(); + const updates: acp.SessionNotification[] = []; + let server!: CodexAcpServer; + const client = new acp.ClientSideConnection( + () => ({ + requestPermission: () => ({outcome: {outcome: "cancelled" as const}}), + sessionUpdate: (notification) => { updates.push(notification); }, + }), + acp.ndJsonStream(clientToAgent.writable, agentToClient.readable), + ); + new acp.AgentSideConnection( + connection => { + server = new CodexAcpServer(connection, codexClient, undefined, () => null); + return server; + }, + acp.ndJsonStream(agentToClient.writable, clientToAgent.readable), + ); + await client.initialize({protocolVersion: 1, clientCapabilities: compactionCapabilities}); + vi.spyOn(server, "getSessionState").mockReturnValue(createTestSessionState({sessionId})); + await client.prompt({sessionId, prompt: [{type: "text", text: "Continue."}]}); + updates.splice(0); + mocks.getUnhandledNotificationHandler()!(compactionStarted()); + mocks.getUnhandledNotificationHandler()!(compactionCompleted()); + await codexClient.waitForSessionNotifications(sessionId); + await vi.waitFor(() => expect(updates).toHaveLength(2)); + + await expect(JSON.stringify(updates, null, 2)).toMatchFileSnapshot( + "data/session-compaction-wire.json", + ); + }); +}); + +async function createFixture(clientCapabilities: acp.ClientCapabilities = compactionCapabilities, nativeSubagents = false) { + const fixture = createCodexMockTestFixture(); + const agent = fixture.getCodexAcpAgent(); + const appServer = fixture.getCodexAppServerClient(); + vi.spyOn(appServer, "turnStart").mockResolvedValue({turn: createTurn("inProgress")}); + vi.spyOn(appServer, "awaitTurnCompleted").mockResolvedValue({ + threadId: sessionId, + turn: createTurn("completed"), + }); + const sessionState = createTestSessionState({sessionId}); + if (nativeSubagents) { + sessionState.subagents = new CodexSubagentEventRouter( + sessionId, + true, + new ACPSessionConnection(fixture.getAcpConnection(), sessionId), + ); + } + vi.spyOn(agent, "getSessionState").mockReturnValue(sessionState); + await agent.initialize({protocolVersion: 1, clientCapabilities}); + await agent.prompt({sessionId, prompt: [{type: "text", text: "Continue."}]}); + fixture.clearAcpConnectionDump(); + return fixture; +} + +async function createNativeChildFixture() { + const fixture = await createFixture({ + ...compactionCapabilities, + _meta: {jetbrains: {air: {version: 1, capabilities: ["nativeSubagentSessions"]}}}, + }, true); + await sendNotifications(fixture, [ + { + method: "item/started", + params: { + threadId: sessionId, + turnId, + startedAtMs: 0, + item: { + type: "collabAgentToolCall", + id: "spawn-compaction-child", + tool: "spawnAgent", + status: "inProgress", + senderThreadId: sessionId, + receiverThreadIds: [childSessionId], + prompt: "Continue the delegated task.", + model: null, + reasoningEffort: null, + agentsStates: {[childSessionId]: {status: "running", message: null}}, + }, + }, + }, + { + method: "item/started", + params: { + threadId: sessionId, + turnId, + startedAtMs: 0, + item: { + type: "subAgentActivity", + id: "activity-compaction-child", + kind: "started", + agentThreadId: childSessionId, + agentPath: "/root/compaction_child", + }, + }, + }, + ]); + return fixture; +} + +async function sendNotifications(fixture: CodexMockTestFixture, notifications: ServerNotification[]) { + for (const notification of notifications) { + fixture.sendServerNotification(notification); + } + await fixture.getCodexAcpClient().waitForSessionNotifications(sessionId); +} + +function compactionStarted(id = compactionId, threadId = sessionId, activeTurnId = turnId): ServerNotification { + return { + method: "item/started", + params: {threadId, turnId: activeTurnId, startedAtMs: 0, item: {type: "contextCompaction", id}}, + }; +} + +function nativeChildActivityStarted(threadId: string): ServerNotification { + return { + method: "item/started", + params: { + threadId: sessionId, + turnId, + startedAtMs: 0, + item: { + type: "subAgentActivity", + id: `activity-${threadId}`, + kind: "started", + agentThreadId: threadId, + agentPath: `/root/${threadId}`, + }, + }, + }; +} + +function compactionCompleted(id = compactionId, threadId = sessionId, activeTurnId = turnId): ServerNotification { + return { + method: "item/completed", + params: {threadId, turnId: activeTurnId, completedAtMs: 1, item: {type: "contextCompaction", id}}, + }; +} + +function compactionError(willRetry: boolean): ServerNotification { + return { + method: "error", + params: { + threadId: sessionId, + turnId, + willRetry, + error: { + message: "The compaction request failed.", + codexErrorInfo: null, + additionalDetails: null, + misalignment: null, + }, + }, + }; +} + +function compactionUpdatesDump(fixture: CodexMockTestFixture): string { + return JSON.stringify(fixture.getAcpConnectionEvents([]).filter(event => + event.method === "sessionUpdate" && event.args[0].update.sessionUpdate === "compaction_update", + ), null, 2); +} + +function nativeCompactionLifecycleDump(fixture: CodexMockTestFixture, childId?: string): string { + return JSON.stringify(fixture.getAcpConnectionEvents([]).filter(event => { + if (event.method !== "sessionUpdate") return false; + const {sessionId: updateSessionId, update} = event.args[0]; + return (update.sessionUpdate === "compaction_update" && (!childId || updateSessionId === childId)) + || (update.sessionUpdate === "subagent_state_update" && (!childId || update.subagentSessionId === childId)); + }), null, 2); +} + +function createTurn(status: Turn["status"]): Turn { + return { + id: turnId, + items: [], + itemsView: "full", + status, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; +} + +function agentMessage(id: string, text: string): ThreadItem { + return {type: "agentMessage", id, text, phase: null, memoryCitation: null, delivery: null, questions: null}; +} + +function createThread(items: ThreadItem[]): Thread { + return { + id: sessionId, + sessionId, + parentThreadId: null, + threadSource: null, + originator: null, + forkedFromId: null, + preview: "Compaction history", + ephemeral: false, + modelProvider: "openai", + model: null, + reasoningEffort: null, + createdAt: 1, + updatedAt: 2, + recencyAt: null, + status: {type: "idle"}, + path: null, + cwd: "/test/cwd", + cliVersion: "0", + section: null, + sectionEnteredAt: null, + projectId: null, + historyMode: "legacy", + source: "cli", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [{...createTurn("completed"), items}], + }; +} diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index f358664c..d522b974 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -16,6 +16,7 @@ import {expect, vi} from "vitest"; import type {Model, ReasoningEffortOption} from "../app-server/v2"; import {CodexSubagentEventRouter} from "../subagents/CodexSubagentEventRouter"; import {CodexBackgroundTerminalTasks} from "../async-tasks/CodexBackgroundTerminalTasks"; +import {CodexSessionCompactions} from "../CodexSessionCompactions"; import {AUTH_STATUS_UPDATE_METHOD} from "../AuthStatusMeta"; export type MethodCallEvent = { method: string; args: any[] }; @@ -421,6 +422,7 @@ export function createTestSessionState(overrides?: Partial): Sessi goalRevision: 0, sessionTitle: null, sessionTitleSource: "unknown", + compactions: new CodexSessionCompactions(), subagents: new CodexSubagentEventRouter( sessionId, false, diff --git a/src/subagents/CodexSubagentEventRouter.ts b/src/subagents/CodexSubagentEventRouter.ts index 8d232f12..581b8bf3 100644 --- a/src/subagents/CodexSubagentEventRouter.ts +++ b/src/subagents/CodexSubagentEventRouter.ts @@ -32,6 +32,7 @@ type PendingSubagent = { export type ClosingChildSession = { threadId: string; sessionId: string; + state: SubagentState; }; /** Owns native lifecycle, child routing, waiting, and legacy activity deduplication. */ @@ -188,20 +189,25 @@ export class CodexSubagentEventRouter { closingChildSessions(notification: ServerNotification): ClosingChildSession[] { if (!this.supported) return []; if (notification.method === "turn/completed") { - if (terminalStateFromTurn(notification.params.turn.status) === undefined) return []; - return this.closingChildSession(notification.params.threadId); + const state = terminalStateFromTurn(notification.params.turn.status); + if (state === undefined) return []; + if (notification.params.threadId === this.rootSessionId && state !== "completed") { + return [...this.children.keys()].flatMap(threadId => this.closingChildSession(threadId, state)); + } + return this.closingChildSession(notification.params.threadId, state); } if (notification.method !== "item/started" && notification.method !== "item/completed") return []; const item = notification.params.item; if (item.type === "subAgentActivity") { - return item.kind === "interrupted" ? this.closingChildSession(item.agentThreadId) : []; + return item.kind === "interrupted" ? this.closingChildSession(item.agentThreadId, "cancelled") : []; } if (item.type !== "collabAgentToolCall") return []; const closing = new Map(); for (const [threadId, state] of Object.entries(item.agentsStates)) { - if (!state || terminalStateOf(state.status) === undefined) continue; - for (const child of this.closingChildSession(threadId)) closing.set(threadId, child); + const terminalState = state && terminalStateOf(state.status); + if (!terminalState) continue; + for (const child of this.closingChildSession(threadId, terminalState)) closing.set(threadId, child); } return [...closing.values()]; } @@ -242,18 +248,18 @@ export class CodexSubagentEventRouter { return createSubAgentActivityUpdate(item, "completed", sessionUpdate); } + /** The caller finalizes pending child updates before closing timed-out sessions. */ async wait( signal: AbortSignal, timeoutMs = CodexSubagentEventRouter.DEFAULT_WAIT_TIMEOUT_MS, - ): Promise { + ): Promise<"settled" | "aborted" | "timed_out"> { const deadline = Date.now() + timeoutMs; while (this.hasOutstanding()) { - if (signal.aborted) return; + if (signal.aborted) return "aborted"; const remainingMs = deadline - Date.now(); if (remainingMs <= 0) { logger.log(`Timed out waiting for subagents in session ${this.rootSessionId}; marking them failed`); - await this.finishOutstanding("failed"); - return; + return "timed_out"; } const changed = await new Promise((resolve) => { const timeout = setTimeout(() => { @@ -276,10 +282,10 @@ export class CodexSubagentEventRouter { }); if (!changed) { logger.log(`Timed out waiting for subagents in session ${this.rootSessionId}; marking them failed`); - await this.finishOutstanding("failed"); - return; + return "timed_out"; } } + return "settled"; } async finishOutstanding(state: SubagentState): Promise { @@ -298,10 +304,10 @@ export class CodexSubagentEventRouter { || this.terminalPendingSpawns.has(threadId)); } - private closingChildSession(threadId: string): ClosingChildSession[] { + private closingChildSession(threadId: string, state: SubagentState): ClosingChildSession[] { const child = this.children.get(threadId); return child && child.terminalState === undefined - ? [{threadId, sessionId: child.sessionId}] + ? [{threadId, sessionId: child.sessionId, state}] : []; }