From 08853039fb2e2bbfd47accabaf3913e0d3e81726 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:55:29 +0200 Subject: [PATCH] fix(server): keep resumed goal turns visible and stoppable (#7339) A Codex goal can start while thread/resume is still connecting, before the manager can subscribe. Retain those startup events until the first subscriber and apply them before registration returns, preserving run state and cancellation ownership. Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> Co-authored-by: Hemanth --- .../server/src/server/agent/agent-manager.ts | 2 + .../providers/codex-app-server-agent.test.ts | 94 +++++++++++++++++++ .../agent/providers/codex-app-server-agent.ts | 11 +++ 3 files changed, 107 insertions(+) diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index ff91ef85ef..a15d94389f 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -3654,6 +3654,8 @@ export class AgentManager { this.assertAgentRegistrationActive(managed); this.emitState(managed, { persist: false }); this.subscribeToSession(managed); + await this.drainSessionEvents(managed.id); + this.assertAgentRegistrationActive(managed); return { ...managed }; } catch (error) { if (!registered) { 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 8238a80131..8fca5f24c0 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 @@ -9,6 +9,7 @@ import { PassThrough } from "node:stream"; import { fileURLToPath } from "node:url"; import type { + AgentClient, AgentLaunchContext, AgentPermissionRequest, AgentSession, @@ -94,6 +95,7 @@ import { } from "./codex/test-utils/fake-app-server.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { asInternals as castInternals, createStub } from "../../test-utils/class-mocks.js"; +import { AgentManager } from "../agent-manager.js"; import { buildProviderRegistry } from "../provider-registry.js"; interface CollaborationModeRecord { @@ -7073,3 +7075,95 @@ describe("Codex denied plan approvals", () => { }); }); }); + +describe("Codex interactive resume startup", () => { + test("delivers an autonomous turn that starts during resume to the first subscriber", async () => { + const appServer = createFakeCodexAppServer({ + "turn/interrupt": () => ({}), + "thread/resume": () => { + appServer.startsTurn({ threadId: "saved-thread", turnId: "autonomous-turn" }); + appServer.says({ threadId: "saved-thread", text: "Working on the goal" }); + return { thread: { id: "saved-thread" } }; + }, + "thread/read": () => ({ thread: { turns: [] } }), + }); + const session = new CodexAppServerAgentSession( + createConfig({ modeId: "full-access" }), + { sessionId: "saved-thread" }, + createTestLogger(), + async () => appServer.child, + ); + try { + await session.connect(); + const events: AgentStreamEvent[] = []; + const unsubscribe = session.subscribe((event) => events.push(event)); + expect(events.some((event) => event.type === "turn_started")).toBe(true); + expect( + events.some( + (event) => + event.type === "timeline" && + event.item.type === "assistant_message" && + event.item.text.includes("Working on the goal"), + ), + ).toBe(true); + await session.interrupt(); + expect( + appServer.requests().find((request) => request.method === "turn/interrupt"), + ).toMatchObject({ params: { threadId: "saved-thread", turnId: "autonomous-turn" } }); + unsubscribe(); + const laterEvents: AgentStreamEvent[] = []; + session.subscribe((event) => laterEvents.push(event)); + expect(laterEvents).toEqual([]); + } finally { + await session.close(); + appServer.assertNoErrors(); + } + }); + test.each([false, true])( + "applies resumed goal events before returning (completed during resume: %s)", + async (completedDuringResume) => { + const appServer = createFakeCodexAppServer({ + "thread/resume": () => { + appServer.startsTurn({ threadId: "saved-thread", turnId: "autonomous-turn" }); + if (completedDuringResume) appServer.completeTurn({ threadId: "saved-thread" }); + return { thread: { id: "saved-thread" } }; + }, + "thread/read": () => ({ thread: { turns: [] } }), + }); + const session = new CodexAppServerAgentSession( + createConfig({ title: "Resumed goal", cwd: process.cwd() }), + { sessionId: "saved-thread" }, + createTestLogger(), + async () => appServer.child, + ); + const client: AgentClient = { + provider: "codex", + capabilities: session.capabilities, + isAvailable: async () => true, + fetchCatalog: async () => ({ models: [], modes: [] }), + createSession: async () => { + throw new Error("Unexpected create"); + }, + resumeSession: async () => { + await session.connect(); + return session; + }, + }; + const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() }); + let id: string | undefined; + try { + const resumed = await manager.resumeAgentFromPersistence( + { provider: "codex", sessionId: "saved-thread" }, + createConfig({ title: "Resumed goal", cwd: process.cwd() }), + ); + id = resumed.id; + expect(resumed.lifecycle).toBe(completedDuringResume ? "idle" : "running"); + expect(manager.getAgent(id)?.lifecycle).toBe(completedDuringResume ? "idle" : "running"); + } finally { + if (id) await manager.closeAgent(id); + await session.close(); + appServer.assertNoErrors(); + } + }, + ); +}); 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 36267592f8..62ceb14c70 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 @@ -3453,6 +3453,7 @@ export class CodexAppServerAgentSession implements AgentSession { private client: CodexAppServerClient | null = null; private threadRollbackAvailable = true; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); + private startupEvents: AgentStreamEvent[] | null = null; private nextTurnOrdinal = 0; private activeForegroundTurnId: string | null = null; private activeClientMessageId: string | null = null; @@ -3571,6 +3572,8 @@ export class CodexAppServerAgentSession implements AgentSession { if (this.resumeHandle?.sessionId) { this.currentThreadId = this.resumeHandle.sessionId; this.historyPending = true; + // Resuming can start a goal before the caller receives the session. + if (this.initialResumePurpose === "interactive") this.startupEvents = []; } } @@ -4543,6 +4546,9 @@ export class CodexAppServerAgentSession implements AgentSession { subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); + const startupEvents = this.startupEvents; + this.startupEvents = null; + for (const event of startupEvents ?? []) this.notifySubscribers(event); return () => { this.subscribers.delete(callback); }; @@ -5028,6 +5034,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.clearPendingPermissions(); this.pendingSubAgentNotificationsByThreadId.clear(); this.subscribers.clear(); + this.startupEvents = null; this.activeForegroundTurnId = null; this.activeClientMessageId = null; this.pendingForegroundTurnIdentification?.resolve(null); @@ -5375,6 +5382,10 @@ export class CodexAppServerAgentSession implements AgentSession { private notifySubscribers(event: AgentStreamEvent): void { const turnId = getAgentStreamEventTurnId(event) ?? this.activeForegroundTurnId; const tagged = turnId ? { ...event, turnId } : event; + if (this.startupEvents) { + this.startupEvents.push(tagged); + return; + } this.logger.trace( { agentId: this.agentId,