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 1d113fb27a..2169a059f2 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 @@ -16,6 +16,7 @@ import type { AgentSessionConfig, AgentSlashCommand, AgentStreamEvent, + AgentTimelineItem, } from "../agent-sdk-types.js"; import { buildCodexAppServerEnv, @@ -114,7 +115,6 @@ interface CodexSessionTestAccess { ensureThreadLoaded(): Promise; handleToolApprovalRequest(params: unknown): Promise; handleNotification(method: string, params: unknown): Promise; - loadPersistedHistory(client: CodexClientLike | null): Promise; refreshResolvedCollaborationMode(): void; serviceTier: "fast" | null; planModeEnabled: boolean; @@ -503,6 +503,81 @@ function archivedThreadErrorMessage(threadId: string): string { ); } +function codexUserItem(id: string, text: string) { + return { type: "userMessage", id, content: [{ type: "text", text }] }; +} + +function codexAgentItem(text: string) { + return { type: "agentMessage", id: `agent-${text}`, text }; +} + +// An AgentManager over a real Codex provider whose every spawn gets a fresh fake +// app-server serving `threads` from thread/read. +function createReopenableCodexAgent(threads: Record) { + const appServers: FakeCodexAppServer[] = []; + const provider = new CodexAppServerAgentClient(createTestLogger()); + const internals = castInternals<{ + goalsEnabledPromise: Promise | null; + autoReviewEnabledPromise: Promise | null; + spawnAppServer: () => Promise; + }>(provider); + internals.goalsEnabledPromise = Promise.resolve(false); + internals.autoReviewEnabledPromise = Promise.resolve(false); + internals.spawnAppServer = async () => { + const appServer = createFakeCodexAppServer({ + "thread/read": (params) => { + const { threadId } = params as { threadId: string }; + return { thread: { id: threadId, turns: threads[threadId] ?? [] } }; + }, + }); + appServers.push(appServer); + return appServer.child; + }; + // The fake app-server stands in for the Codex CLI, which test machines may not have. + const client: AgentClient = { + provider: "codex", + capabilities: provider.capabilities, + isAvailable: async () => true, + fetchCatalog: async () => ({ models: [], modes: [] }), + createSession: async () => { + throw new Error("Unexpected create"); + }, + resumeSession: (...args) => provider.resumeSession(...args), + }; + const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() }); + const agentId = "00000000-0000-4000-8000-000000000001"; + const handle = { provider: "codex", sessionId: "saved-thread" }; + const config = createConfig({ cwd: process.cwd() }); + + async function open(): Promise { + await manager.resumeAgentFromPersistence(handle, config, agentId); + await manager.hydrateTimelineFromProvider(agentId); + } + + return { + manager, + agentId, + latestAppServer: () => appServers[appServers.length - 1], + threadMethods: (index: number) => + appServers[index] + .requests() + .map((request) => request.method) + .filter((method) => typeof method === "string" && method.startsWith("thread/")), + // Cold-loads the agent, closes its runtime, and reopens it while Paseo keeps its timeline. + async loadCloseAndReopen(): Promise { + await open(); + const projected = manager.getTimeline(agentId); + await manager.closeAgent(agentId); + await open(); + return projected; + }, + async close(): Promise { + await manager.closeAgent(agentId); + for (const appServer of appServers) appServer.assertNoErrors(); + }, + }; +} + function asInternals(session: CodexTestSession): CodexSessionTestAccess { return castInternals(session); } @@ -1897,12 +1972,147 @@ describe("Codex app-server provider", () => { "thread/resume", "thread/unarchive", "thread/resume", - "thread/read", ]); await session.close(); appServer.assertNoErrors(); }); + test("resumes a Codex thread without asking for its turns", async () => { + const appServer = createFakeCodexAppServer(); + const provider = createProviderWithFakeAppServer(appServer); + + const session = await provider.resumeSession({ sessionId: "saved-thread" }); + + expect( + appServer + .requests() + .filter((request) => request.method === "thread/resume") + .map((request) => request.params), + ).toEqual([expect.objectContaining({ threadId: "saved-thread", excludeTurns: true })]); + await session.close(); + appServer.assertNoErrors(); + }); + + test("reopening an agent whose timeline Paseo still holds does not read Codex history", async () => { + const codex = createReopenableCodexAgent({ + "saved-thread": [ + { id: "turn-1", items: [codexUserItem("user-1", "Hello"), codexAgentItem("Hi there.")] }, + ], + }); + try { + const projected = await codex.loadCloseAndReopen(); + + expect(projected).toEqual([ + expect.objectContaining({ type: "user_message", messageId: "user-1", text: "Hello" }), + expect.objectContaining({ type: "assistant_message", text: "Hi there." }), + ]); + expect(codex.threadMethods(0)).toEqual([ + "thread/loaded/list", + "thread/resume", + "thread/read", + ]); + expect(codex.threadMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]); + expect(codex.manager.getTimeline(codex.agentId)).toEqual(projected); + } finally { + await codex.close(); + } + }); + + test("rewinds an agent reopened without reading its Codex history", async () => { + const firstTurn = { + id: "turn-1", + items: [codexUserItem("user-1", "First"), codexAgentItem("First answer.")], + }; + const codex = createReopenableCodexAgent({ + "saved-thread": [ + firstTurn, + { id: "turn-2", items: [codexUserItem("user-2", "Second"), codexAgentItem("Second.")] }, + ], + "forked-thread": [firstTurn], + }); + try { + await codex.loadCloseAndReopen(); + + await codex.manager.rewind(codex.agentId, "user-2", "conversation"); + + expect(codex.latestAppServer().recordedRollbacks).toEqual([ + { threadId: "forked-thread", numTurns: 1 }, + ]); + expect(codex.manager.getTimeline(codex.agentId)).toEqual([ + expect.objectContaining({ type: "user_message", messageId: "user-1", text: "First" }), + expect.objectContaining({ type: "assistant_message", text: "First answer." }), + ]); + } finally { + await codex.close(); + } + }); + + test("keeps a reopened agent's interaction with an earlier sub-agent on its original card", async () => { + const codex = createReopenableCodexAgent({ + "saved-thread": [ + { + id: "turn-1", + items: [ + codexUserItem("user-1", "Delegate"), + { + type: "subAgentActivity", + id: "spawn-child", + kind: "started", + agentThreadId: "child-thread", + agentPath: "/root/child", + }, + ], + }, + ], + "child-thread": [{ id: "child-turn-1", items: [codexAgentItem("Child started.")] }], + }); + try { + await codex.loadCloseAndReopen(); + const appServer = codex.latestAppServer(); + + const run = codex.manager.runAgent(codex.agentId, "Ask the child again"); + await appServer.waitForTurnStart(); + appServer.startsTurn({ threadId: "saved-thread", turnId: "turn-2" }); + appServer.beginsSubAgentActivity({ + callId: "interact-child", + threadId: "child-thread", + parentThreadId: "saved-thread", + kind: "interacted", + agentPath: "/root/child", + }); + appServer.says({ threadId: "child-thread", itemId: "child-reply", text: "Child replied." }); + appServer.completesSubAgentActivity({ + callId: "interact-child", + threadId: "child-thread", + parentThreadId: "saved-thread", + kind: "interacted", + agentPath: "/root/child", + }); + appServer.completeTurn({ threadId: "saved-thread" }); + await run; + + const toolCalls = codex.manager + .getTimeline(codex.agentId) + .filter((item) => item.type === "tool_call"); + expect(new Set(toolCalls.map((item) => item.callId))).toEqual(new Set(["spawn-child"])); + expect(toolCalls.at(-1)).toMatchObject({ callId: "spawn-child", status: "completed" }); + expect(codex.manager.getProviderSubagent(codex.agentId, "child-thread")).toMatchObject({ + toolCallId: "spawn-child", + status: "completed", + }); + expect( + codex.manager + .fetchProviderSubagentTimeline(codex.agentId, "child-thread") + .rows.map((row) => row.item), + ).toEqual([ + expect.objectContaining({ type: "assistant_message", text: "Child started." }), + expect.objectContaining({ type: "assistant_message", text: "Child replied." }), + ]); + } finally { + await codex.close(); + } + }); + test("closes Codex app-server when an interactive resume fails", async () => { const appServer = createFakeCodexAppServer({ "thread/resume": () => Promise.reject(new Error("thread resume failed")), @@ -4673,8 +4883,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -4727,7 +4935,7 @@ describe("Codex app-server provider", () => { })), }; - await asInternals(session).loadPersistedHistory(session.client); + await Array.fromAsync(session.streamHistory()); expect(asInternals(session).codexUserMessageTurns().resolve("message-history")).toEqual({ index: 0, @@ -4796,8 +5004,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -4958,8 +5164,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -5023,8 +5227,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -5200,8 +5402,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -5261,8 +5461,6 @@ describe("Codex app-server provider", () => { }), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); @@ -5564,7 +5762,10 @@ describe("Codex app-server provider", () => { expect(session.currentThreadId).toBe("archived-thread-id"); expect(requests).toEqual([ { method: "thread/loaded/list", params: {} }, - { method: "thread/resume", params: { threadId: "archived-thread-id" } }, + { + method: "thread/resume", + params: { threadId: "archived-thread-id", excludeTurns: true }, + }, ]); }); @@ -6399,8 +6600,6 @@ describe("Codex app-server provider", () => { })), }; - await asInternals(session).loadPersistedHistory(session.client); - const history: AgentStreamEvent[] = []; for await (const event of session.streamHistory()) { history.push(event); 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 291841c730..cdad017d45 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 @@ -496,6 +496,11 @@ interface PersistedTimelineEntry { providerTurnId?: string; } +interface CodexThreadHistory { + timeline: PersistedTimelineEntry[]; + providerSubagentEvents: AgentStreamEvent[]; +} + interface PersistedSubAgentRoute { childThreadId: string; toolCall: ToolCallTimelineItem; @@ -3464,10 +3469,12 @@ export class CodexAppServerAgentSession implements AgentSession { private serviceTier: string | null = null; private speedModels: CodexModel[] = []; private planModeEnabled = false; - private historyPending = false; - private persistedHistory: PersistedTimelineEntry[] = []; - private loadingPersistedHistory = false; - private persistedProviderSubagentEvents: AgentStreamEvent[] = []; + // History read by rewind or an archived resume, held until streamHistory consumes it. + private stagedHistory: CodexThreadHistory | null = null; + // Collects provider subagent events emitted while a thread history read rebuilds routes. + private historySubagentEvents: AgentStreamEvent[] | null = null; + // A resumed thread's earlier turns and sub-agents are known only after reading its history. + private historyIndexed = true; private pendingPermissions = new Map(); private mcpElicitationPermissionIds = new Map(); private pendingPermissionHandlers = new Map(); @@ -3574,7 +3581,7 @@ export class CodexAppServerAgentSession implements AgentSession { if (this.resumeHandle?.sessionId) { this.currentThreadId = this.resumeHandle.sessionId; - this.historyPending = true; + this.historyIndexed = false; // Resuming can start a goal before the caller receives the session. if (this.initialResumePurpose === "interactive") this.startupEvents = []; } @@ -3622,7 +3629,7 @@ export class CodexAppServerAgentSession implements AgentSession { private async establishConnection(): Promise { if (this.initialResumePurpose === "history") { - await this.readArchivedHistory(); + this.stagedHistory = await this.readArchivedHistory(); this.connectionState = "history-ready"; return; } @@ -3655,7 +3662,6 @@ export class CodexAppServerAgentSession implements AgentSession { if (this.currentThreadId) { await this.ensureThreadLoaded(); - await this.loadPersistedHistory(this.client); await this.applyDefaultModelAndThinking(); } @@ -3970,39 +3976,40 @@ export class CodexAppServerAgentSession implements AgentSession { ); } - private async readArchivedHistory(): Promise { + private async readArchivedHistory(): Promise { const child = await this.spawnAppServer(); const client = new CodexAppServerClient(child, this.logger, () => this.traceContext()); try { await client.request("initialize", buildCodexAppServerInitializeParams()); client.notify("initialized", {}); - await this.loadPersistedHistory(client); + return await this.readThreadHistory(client); } finally { await client.dispose(); } } - private async loadPersistedHistory(client: CodexAppServerClientLike | null): Promise { - if (!client || !this.currentThreadId) return; - const threadId = this.currentThreadId; - - const history = await loadCodexThreadHistoryTimeline({ - threadId, + // Reads the thread and its sub-agent threads, rebuilding the turn index and + // sub-agent routes. The caller owns the returned history; nothing retains it. + private async readThreadHistory( + client: CodexAppServerClientLike | null, + ): Promise { + if (!client || !this.currentThreadId) { + return { timeline: [], providerSubagentEvents: [] }; + } + const { timeline, subAgentRoutes } = await loadCodexThreadHistoryTimeline({ + threadId: this.currentThreadId, cwd: this.config.cwd ?? null, - requestThread: (threadIdToRead) => { - return readCodexThread(client, threadIdToRead); - }, + requestThread: (threadIdToRead) => readCodexThread(client, threadIdToRead), }); - const { timeline, subAgentRoutes } = history; this.subAgentCallsByCallId.clear(); this.subAgentCallIdByChildThreadId.clear(); this.pendingSubAgentNotificationsByThreadId.clear(); - this.persistedProviderSubagentEvents = []; - this.loadingPersistedHistory = true; + const providerSubagentEvents: AgentStreamEvent[] = []; + this.historySubagentEvents = providerSubagentEvents; try { await this.loadPersistedSubAgentHistories(client, subAgentRoutes); } finally { - this.loadingPersistedHistory = false; + this.historySubagentEvents = null; } this.resetCodexUserMessageTurns(); for (const entry of timeline) { @@ -4013,8 +4020,15 @@ export class CodexAppServerAgentSession implements AgentSession { this.rememberCodexUserMessageTurn(entry.item.messageId, entry.providerTurnId); } } - this.persistedHistory = timeline; - this.historyPending = timeline.length > 0; + this.historyIndexed = true; + return { timeline, providerSubagentEvents }; + } + + // Turns and rewind route by the thread's earlier turns and sub-agents. A resumed + // session whose projection Paseo already holds reads them here and drops the rest. + private async ensureHistoryIndexed(): Promise { + if (this.historyIndexed) return; + await this.readThreadHistory(this.client); } private async loadPersistedSubAgentHistories( @@ -4066,7 +4080,8 @@ export class CodexAppServerAgentSession implements AgentSession { private async ensureThreadLoaded(): Promise { if (!this.client || !this.currentThreadId) return; - const params: Record = { threadId: this.currentThreadId }; + // Paseo reads history only when it replays it; the resume response needs no turns. + const params: Record = { threadId: this.currentThreadId, excludeTurns: true }; const developerInstructions = composeSystemPromptParts( this.config.systemPrompt, this.config.daemonAppendSystemPrompt, @@ -4386,6 +4401,7 @@ export class CodexAppServerAgentSession implements AgentSession { if (this.currentThreadId) { await this.ensureThreadLoaded(); + await this.ensureHistoryIndexed(); } else { await this.ensureThread(); } @@ -4559,21 +4575,12 @@ export class CodexAppServerAgentSession implements AgentSession { } async *streamHistory(): AsyncGenerator { - if ( - (!this.historyPending || this.persistedHistory.length === 0) && - this.persistedProviderSubagentEvents.length === 0 - ) { - return; - } - const history = this.persistedHistory; - const providerSubagents = this.persistedProviderSubagentEvents; - this.persistedHistory = []; - this.persistedProviderSubagentEvents = []; - this.historyPending = false; - for (const event of providerSubagents) { + const history = this.stagedHistory ?? (await this.readThreadHistory(this.client)); + this.stagedHistory = null; + for (const event of history.providerSubagentEvents) { yield event; } - for (const entry of history) { + for (const entry of history.timeline) { yield { type: "timeline", provider: CODEX_PROVIDER, @@ -4956,6 +4963,7 @@ export class CodexAppServerAgentSession implements AgentSession { } else { await this.ensureThread(); } + await this.ensureHistoryIndexed(); await revertCodexConversation({ client: this.client, @@ -4970,10 +4978,8 @@ export class CodexAppServerAgentSession implements AgentSession { setThreadId: async (threadId) => { this.currentThreadId = threadId; this.cachedRuntimeInfo = null; - this.persistedHistory = []; - this.historyPending = false; - await this.loadPersistedHistory(this.client); - this.reconcileAsyncQuestionsAfterRewind(); + this.stagedHistory = await this.readThreadHistory(this.client); + this.reconcileAsyncQuestionsAfterRewind(this.stagedHistory.timeline); }, }); } @@ -5211,6 +5217,8 @@ export class CodexAppServerAgentSession implements AgentSession { await this.connect(); if (this.currentThreadId) { await this.ensureThreadLoaded(); + // A goal runs turns of its own. + await this.ensureHistoryIndexed(); } else { await this.ensureThread(); } @@ -5404,8 +5412,8 @@ export class CodexAppServerAgentSession implements AgentSession { } private emitEvent(event: AgentStreamEvent): void { - if (this.loadingPersistedHistory && event.type === "provider_subagent") { - this.persistedProviderSubagentEvents.push(event); + if (this.historySubagentEvents && event.type === "provider_subagent") { + this.historySubagentEvents.push(event); return; } this.notifySubscribers(event); @@ -6583,9 +6591,9 @@ export class CodexAppServerAgentSession implements AgentSession { } } - private reconcileAsyncQuestionsAfterRewind(): void { + private reconcileAsyncQuestionsAfterRewind(timeline: readonly PersistedTimelineEntry[]): void { const retainedIds = new Set( - this.persistedHistory.flatMap(({ item }) => + timeline.flatMap(({ item }) => item.type === "tool_call" && item.name === "request_user_input_async" ? [item.callId] : [], ), );