diff --git a/docs/providers.md b/docs/providers.md index a8e05895f..b76aa3d89 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -85,6 +85,8 @@ Rewind accepts the canonical wire `messageId` and resolves it to the provider id Submitted user-message wire items carry the same Paseo ID in `messageId` and `clientMessageId`. Provider adapters attach `clientMessageId` only to the echo for that foreground submission; provider history and externally initiated user rows do not have a Paseo client ID. +Provider adapters must terminalize every transient timeline row before emitting the turn's terminal event. Codex may omit the completed `contextCompaction` item when a turn ends during compaction, so its adapter closes any pending root compaction before forwarding `turn_completed`, `turn_failed`, or `turn_canceled`. A terminal turn must never leave the client showing an operation as still loading. + Draft metadata lookups should avoid creating provider sessions when the upstream provider has top-level APIs for that metadata. Prefer `AgentClient.fetchCatalog`, `listCommands`, or `listFeatures` over creating a scratch `AgentSession`; scratch sessions can show up as empty native sessions in provider import/history UIs. `fetchCatalog` is the single discovery API for models and modes — provider implementations may use one process, separate upstream calls, or static data internally, but callers outside the provider do not get separate runtime model/mode probes. Draft command listing and scratch-session feature listing require an explicit draft model. Do not resolve a default model through catalog discovery. A client-level `listFeatures` implementation may return features from an incomplete, model-less draft and owns which features are valid in that state. Provider session import has its own contract. The picker calls `listImportableSessions` and receives rows only: provider handle, cwd, title, prompt previews, and last activity. Import calls `importSession({ providerHandleId, cwd })` for the selected row and must not call listing again. The provider returns the resumed session, storage config, persistence handle, and hydrated timeline for that one native session; `AgentManager.importProviderSession` seeds the daemon timeline and publishes the Paseo agent only after it is ready. diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 3b0229e2a..c3594b9d7 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -1055,7 +1055,7 @@ describe("stream reducer canonical tool calls", () => { ]); }); - it("preserves compaction trigger when completed update replaces loading marker", () => { + it("terminalizes the loading compaction before a completed turn", () => { const state = hydrateStreamState([ { event: compactionTimeline("loading", "auto"), @@ -1065,6 +1065,10 @@ describe("stream reducer canonical tool calls", () => { event: compactionTimeline("completed"), timestamp: new Date("2025-01-01T10:50:01Z"), }, + { + event: { type: "turn_completed", provider: "codex" }, + timestamp: new Date("2025-01-01T10:50:02Z"), + }, ]); const compactions = state.filter( @@ -1074,6 +1078,10 @@ describe("stream reducer canonical tool calls", () => { assert.strictEqual(compactions.length, 1); assert.strictEqual(compactions[0].status, "completed"); assert.strictEqual(compactions[0].trigger, "auto"); + assert.strictEqual( + state.some((item) => item.kind === "compaction" && item.status === "loading"), + false, + ); }); it("renders Claude TodoWrite as todo_list and suppresses tool call badge", () => { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index c0d16c70c..1c62182dd 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -90,6 +90,11 @@ type CodexTestSession = AgentSession & { client: CodexClientLike | null; }; +type TurnTerminalEvent = Extract< + AgentStreamEvent, + { type: "turn_completed" | "turn_failed" | "turn_canceled" } +>; + const ONE_BY_ONE_PNG_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+X1r0AAAAASUVORK5CYII="; const CODEX_PROVIDER = "codex"; @@ -139,6 +144,41 @@ function createProviderWithFakeAppServer(appServer: FakeCodexAppServer): CodexAp return provider; } +async function startCompactionTurnTest(): Promise<{ + appServer: FakeCodexAppServer; + session: AgentSession; + events: AgentStreamEvent[]; + terminalEvent: Promise; +}> { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + const events: AgentStreamEvent[] = []; + const terminalEvent = new Promise((resolve) => { + session.subscribe((event) => { + const isCompaction = event.type === "timeline" && event.item.type === "compaction"; + const isTerminal = + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled"; + if (isCompaction || isTerminal) { + events.push(event); + } + if (isTerminal) { + resolve(event); + } + }); + }); + + await session.startTurn("exercise compaction lifecycle"); + appServer.startsTurn({ threadId: "thread-1", turnId: "codex-turn-1" }); + return { appServer, session, events, terminalEvent }; +} + function archivedThreadHandle() { return { sessionId: "archived-thread-id", @@ -3843,6 +3883,99 @@ describe("Codex app-server provider", () => { ]); }); + test("completes a pending Codex compaction when its turn ends", async () => { + const { appServer, session, events, terminalEvent } = await startCompactionTurnTest(); + + try { + appServer.startsCompaction({ + threadId: "thread-1", + itemId: "compact-without-completion", + }); + appServer.completeTurn(); + await terminalEvent; + + expect( + events.map((event) => + event.type === "timeline" ? `${event.item.type}:${event.item.status}` : event.type, + ), + ).toEqual(["compaction:loading", "compaction:completed", "turn_completed"]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("does not complete a Codex compaction twice when its item finishes before the turn", async () => { + const { appServer, session, events, terminalEvent } = await startCompactionTurnTest(); + + try { + appServer.startsCompaction({ + threadId: "thread-1", + itemId: "compact-completed-normally", + }); + appServer.completesCompaction({ + threadId: "thread-1", + itemId: "compact-completed-normally", + }); + appServer.completeTurn(); + await terminalEvent; + + expect( + events.map((event) => + event.type === "timeline" ? `${event.item.type}:${event.item.status}` : event.type, + ), + ).toEqual(["compaction:loading", "compaction:completed", "turn_completed"]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("does not let a late compaction completion consume the current pending item", async () => { + const { appServer, session, events, terminalEvent } = await startCompactionTurnTest(); + + try { + appServer.startsCompaction({ threadId: "thread-1", itemId: "current-compaction" }); + appServer.completesCompaction({ threadId: "thread-1", itemId: "older-compaction" }); + appServer.completeTurn(); + await terminalEvent; + + expect( + events.map((event) => + event.type === "timeline" ? `${event.item.type}:${event.item.status}` : event.type, + ), + ).toEqual(["compaction:loading", "compaction:completed", "turn_completed"]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test.each([ + { status: "failed", terminalType: "turn_failed" }, + { status: "interrupted", terminalType: "turn_canceled" }, + ])("completes a pending compaction before a $status turn", async ({ status, terminalType }) => { + const { appServer, session, events, terminalEvent } = await startCompactionTurnTest(); + + try { + appServer.startsCompaction({ threadId: "thread-1", itemId: `compact-${status}` }); + appServer.completeTurn({ + status, + error: status === "failed" ? { message: "Compaction failed" } : null, + }); + await terminalEvent; + + expect( + events.map((event) => + event.type === "timeline" ? `${event.item.type}:${event.item.status}` : event.type, + ), + ).toEqual(["compaction:loading", "compaction:completed", terminalType]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + test("emits and dedupes Codex thread/compacted notifications", () => { const session = createSession(); session.activeForegroundTurnId = null; diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index ae119ef89..0a628498a 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -3166,6 +3166,10 @@ interface CodexPendingPermissionHandler { planText?: string; } +interface ConsumedRootCompaction { + itemId?: string; +} + export class CodexAppServerAgentSession implements AgentSession { readonly provider = CODEX_PROVIDER; readonly capabilities = CODEX_APP_SERVER_CAPABILITIES; @@ -3231,6 +3235,8 @@ export class CodexAppServerAgentSession implements AgentSession { private readonly userMessageTurnIds: string[] = []; private pendingManualCompactionStarts = 0; private compactionTriggerByItemId = new Map(); + private pendingRootCompactionItemIds = new Set(); + private pendingAnonymousRootCompactions = 0; // Codex can report one completed compaction through both channels: // `thread/compacted` and a completed `contextCompaction` item. private unpairedCompactionNotificationCompletions = 0; @@ -5183,6 +5189,16 @@ export class CodexAppServerAgentSession implements AgentSession { if (!this.isContextCompactionItem(item)) { return false; } + const consumedPendingCompaction = this.consumePendingRootCompaction(item.id); + const hasDifferentPendingCompaction = + this.pendingRootCompactionItemIds.size > 0 || this.pendingAnonymousRootCompactions > 0; + const isLateCompletionForOlderItem = + item.id !== undefined && + consumedPendingCompaction === undefined && + hasDifferentPendingCompaction; + if (isLateCompletionForOlderItem) { + return true; + } if (this.unpairedCompactionNotificationCompletions > 0) { this.unpairedCompactionNotificationCompletions -= 1; return true; @@ -5570,6 +5586,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.emitSubAgentActivityUpdate(subAgentCallId, status); return; } + this.completePendingRootCompactions(); if (parsed.status === "failed") { this.emitEvent({ type: "turn_failed", @@ -5610,6 +5627,8 @@ export class CodexAppServerAgentSession implements AgentSession { this.pendingFileChangeOutputDeltas.clear(); this.pendingAssistantMessageBoundary = false; this.warnedIncompleteEditToolCallIds.clear(); + this.pendingRootCompactionItemIds.clear(); + this.pendingAnonymousRootCompactions = 0; this.unpairedCompactionNotificationCompletions = 0; this.unpairedCompactionItemCompletions = 0; } @@ -5669,6 +5688,61 @@ export class CodexAppServerAgentSession implements AgentSession { return undefined; } + private trackPendingRootCompaction(itemId?: string): void { + if (itemId) { + this.pendingRootCompactionItemIds.add(itemId); + return; + } + this.pendingAnonymousRootCompactions += 1; + } + + private consumePendingRootCompaction(itemId?: string): ConsumedRootCompaction | undefined { + if (itemId) { + if (this.pendingRootCompactionItemIds.delete(itemId)) { + return { itemId }; + } + if ( + this.pendingRootCompactionItemIds.size === 0 && + this.pendingAnonymousRootCompactions > 0 + ) { + this.pendingAnonymousRootCompactions -= 1; + return {}; + } + return undefined; + } + const pendingItemId = this.pendingRootCompactionItemIds.values().next().value; + if (typeof pendingItemId === "string") { + this.pendingRootCompactionItemIds.delete(pendingItemId); + return { itemId: pendingItemId }; + } + if (this.pendingAnonymousRootCompactions > 0) { + this.pendingAnonymousRootCompactions -= 1; + return {}; + } + return undefined; + } + + private completePendingRootCompactions(): void { + // Some Codex builds end a turn without completing the contextCompaction + // item. Close every loading timeline row before emitting the terminal turn. + for (const itemId of this.pendingRootCompactionItemIds) { + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("completed", itemId), + }); + } + for (let index = 0; index < this.pendingAnonymousRootCompactions; index += 1) { + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("completed"), + }); + } + this.pendingRootCompactionItemIds.clear(); + this.pendingAnonymousRootCompactions = 0; + } + private createContextCompactionTimelineItem( status: "loading" | "completed", itemId?: string, @@ -5718,11 +5792,12 @@ export class CodexAppServerAgentSession implements AgentSession { this.unpairedCompactionItemCompletions -= 1; return; } + const pendingItemId = this.consumePendingRootCompaction()?.itemId; this.unpairedCompactionNotificationCompletions += 1; this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, - item: this.createContextCompactionTimelineItem("completed"), + item: this.createContextCompactionTimelineItem("completed", pendingItemId), ...(parsed.turnId ? { turnId: parsed.turnId } : {}), }); } @@ -6047,6 +6122,7 @@ export class CodexAppServerAgentSession implements AgentSession { return; } if (this.isContextCompactionItem(parsed.item)) { + this.trackPendingRootCompaction(parsed.item.id); this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index dc04a31f0..fc0031818 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -51,8 +51,13 @@ export interface FakeCodexAppServer { waitForTurnStart(): Promise; nextResponse(): Promise; startsTurn(params: { threadId: string; turnId?: string }): void; + startsCompaction(params: { threadId: string; itemId: string }): void; updatesPlan(params: { threadId: string; steps: string[] }): void; - completeTurn(params?: { threadId?: string }): void; + completeTurn(params?: { + threadId?: string; + status?: "completed" | "failed" | "interrupted"; + error?: { message: string } | null; + }): void; startsSubAgent(params: { callId: string; threadId: string; @@ -320,6 +325,12 @@ export function createFakeCodexAppServer( })}\n`, ); }, + startsCompaction(params) { + writeNotification("item/started", { + threadId: params.threadId, + item: { type: "contextCompaction", id: params.itemId }, + }); + }, updatesPlan(params) { child.stdout.write( `${JSON.stringify({ @@ -337,7 +348,10 @@ export function createFakeCodexAppServer( method: "turn/completed", params: { threadId: params.threadId ?? "thread-1", - turn: { status: "completed" }, + turn: { + status: params.status ?? "completed", + error: params.error ?? null, + }, }, })}\n`, );