From 78b285059f6ebd0b257c98bd191df4626721270a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 5 Sep 2026 14:55:19 +0200 Subject: [PATCH] Fix agent reloads blocked by an active session writer (#4353) * fix(server): keep agent reloads from overlapping session writers Idle runtimes can still own their persisted session writer. Release the original runtime before resuming and retain failed cleanup for retry; a timeout must not authorize another writer. * fix(server): honor closes queued during agent reload Serialize public close with reload so a completed close cannot be undone by replacement registration. Keep archive cleanup inside its existing lifecycle operation. --- docs/agent-lifecycle.md | 4 + .../src/server/agent/agent-manager.test.ts | 294 ++++++++++++++---- .../server/src/server/agent/agent-manager.ts | 94 +++--- .../src/server/agent/agent-prompt.test.ts | 4 +- .../codex-app-server-agent.local.e2e.test.ts | 54 +++- .../providers/codex-app-server-agent.test.ts | 8 +- .../agent/providers/codex-app-server-agent.ts | 2 +- .../codex/app-server-transport.test.ts | 25 +- .../providers/codex/app-server-transport.ts | 6 +- .../agent-reload.native.real.e2e.test.ts | 128 ++++++++ 10 files changed, 513 insertions(+), 106 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/agent-reload.native.real.e2e.test.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 20a823de80..418e820979 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -21,6 +21,10 @@ the agent runs through `ensureAgentLoaded()`, which resumes the durable provider same Paseo agent ID. Provider history is not appended again when the canonical timeline is already primed. +Reload releases the old runtime before resuming its durable session: an idle provider process can +still own an exclusive writer. A close failure retains that runtime for cleanup and blocks the +replacement. Once closure succeeds, a failed resume leaves the durable agent closed and retryable. + Idle agents remain resident indefinitely. Runtime closure happens only through an explicit lifecycle action such as archive, replacement, reload, workspace teardown, or daemon shutdown. diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index f2c8fc7d05..bc38784e48 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1623,7 +1623,7 @@ test("reload leaves a closed durable snapshot when shutdown starts during the sw }).toMatchObject({ agents: [], record: { lastStatus: "closed" }, - replacementSessionClosed: true, + replacementSessionClosed: false, }); } finally { client.finishClosing(); @@ -1633,7 +1633,7 @@ test("reload leaves a closed durable snapshot when shutdown starts during the sw } }); -test("reload closes both sessions when the closed snapshot cannot be persisted", async () => { +test("reload does not create a replacement when the closed snapshot cannot be persisted", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-persist-failure-test-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -1672,7 +1672,7 @@ test("reload closes both sessions when the closed snapshot cannot be persisted", }).toEqual({ agents: [], originalSessionClosed: true, - replacementSessionClosed: true, + replacementSessionClosed: false, }); } finally { client.finishClosing(); @@ -2244,74 +2244,254 @@ test("setAgentMode persists the selected mode across session reload", async () = expect(reloaded.currentModeId).toBe("full-access"); }); -test("reloadAgentSession completes when the previous session close hangs", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-close-timeout-")); - const storagePath = join(workdir, "agents"); - const storage = new AgentStorage(storagePath, logger); - - class HangingCloseSession extends TestAgentSession { - closeCalled = false; - - override async close(): Promise { - this.closeCalled = true; - await new Promise(() => {}); +test("reload releases the original writer before resuming the same session", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-writer-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + class ExclusiveWriterClient extends TestAgentClient { + current: CloseRecordingTestAgentSession | undefined; + override async createSession(config: AgentSessionConfig): Promise { + this.current = new CloseRecordingTestAgentSession(config); + return this.current; } - } - - class HangingCloseClient extends TestAgentClient { - readonly firstSession = new HangingCloseSession({ - provider: "codex", - cwd: workdir, - }); - resumeSessionCalls = 0; - - override async createSession(): Promise { - return this.firstSession; - } - override async resumeSession( - _handle: AgentPersistenceHandle, + handle: AgentPersistenceHandle, config?: Partial, ): Promise { - this.resumeSessionCalls += 1; - return new TestAgentSession({ + if (this.current && !this.current.closed) { + throw new Error("thread already has an active writer"); + } + this.current = new CloseRecordingTestAgentSession({ provider: "codex", cwd: config?.cwd ?? workdir, }); + this.current.describePersistence = () => ({ provider: "codex", sessionId: handle.sessionId }); + return this.current; } } - - const client = new HangingCloseClient(); - const manager = new AgentManager({ - clients: { - codex: client, - }, - registry: storage, - logger, - rescueTimeouts: { reloadSessionCloseMs: 10 }, - idFactory: () => "00000000-0000-4000-8000-000000000302", - }); - + const client = new ExclusiveWriterClient(); + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); try { - const snapshot = await manager.createAgent( - { - provider: "codex", - cwd: workdir, - }, - undefined, - { workspaceId: undefined }, - ); - - const reloaded = await manager.reloadAgentSession(snapshot.id); - - expect(reloaded.id).toBe(snapshot.id); - expect(client.firstSession.closeCalled).toBe(true); - expect(client.resumeSessionCalls).toBe(1); + const created = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + for (let i = 0; i < 3; i++) { + const reloaded = await manager.reloadAgentSession(created.id, undefined, { + rehydrateFromDisk: true, + }); + expect(reloaded.id).toBe(created.id); + expect(reloaded.persistence?.sessionId).toBe(created.persistence?.sessionId); + } + await manager.closeAgent(created.id); } finally { + await client.current?.close(); + await storage.flush(); rmSync(workdir, { recursive: true, force: true }); } }); +test("opening during reload waits for the replacement", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-open-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const client = new HeldReloadCloseClient(); + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); + try { + const created = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + const reloading = manager.reloadAgentSession(created.id); + await client.waitForCloseToStart(); + let opened = false; + const opening = ensureAgentLoaded(created.id, { + agentManager: manager, + agentStorage: storage, + logger, + }).then((agent) => { + opened = true; + return agent; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(opened).toBe(false); + client.finishClosing(); + const replacement = await reloading; + expect((await opening).session).toBe(replacement.session); + await manager.closeAgent(created.id); + } finally { + client.finishClosing(); + await storage.flush(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("closing during reload leaves the replacement closed", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-close-race-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const client = new HeldReloadCloseClient(); + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); + try { + const created = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + const reloading = manager.reloadAgentSession(created.id); + await client.waitForCloseToStart(); + const closing = manager.closeAgent(created.id); + client.finishClosing(); + await Promise.all([reloading, closing]); + expect(manager.getAgent(created.id)).toBeNull(); + expect(client.replacementSessionClosed).toBe(true); + expect(await storage.get(created.id)).toMatchObject({ lastStatus: "closed" }); + } finally { + client.finishClosing(); + await storage.flush(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("retrying a timed-out reload waits for the original close to finish", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-late-close-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const client = new HeldReloadCloseClient(); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + rescueTimeouts: { reloadSessionCloseMs: 10 }, + }); + try { + const created = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + await expect(manager.reloadAgentSession(created.id)).rejects.toThrow("Timed out closing"); + expect(manager.getAgent(created.id)?.lifecycle).toBe("error"); + const retry = manager.reloadAgentSession(created.id); + client.finishClosing(); + expect((await retry).id).toBe(created.id); + expect(client.originalSessionClosed).toBe(true); + await manager.closeAgent(created.id); + } finally { + client.finishClosing(); + await storage.flush(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("failed reload retains the closed agent for a later resume", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-recovery-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + class FailingResumeClient extends TestAgentClient { + fail = true; + override async resumeSession( + handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + if (this.fail) throw new Error("resume unavailable"); + return super.resumeSession(handle, config); + } + } + const client = new FailingResumeClient(); + const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger }); + try { + const created = await manager.createAgent( + { provider: "codex", cwd: workdir, title: "Keep me" }, + undefined, + { workspaceId: undefined }, + ); + await expect(manager.reloadAgentSession(created.id)).rejects.toThrow("resume unavailable"); + expect(manager.getAgent(created.id)).toBeNull(); + expect(await storage.get(created.id)).toMatchObject({ + lastStatus: "closed", + title: "Keep me", + persistence: created.persistence, + }); + client.fail = false; + const recovered = await ensureAgentLoaded(created.id, { + agentManager: manager, + agentStorage: storage, + logger, + }); + expect(recovered.id).toBe(created.id); + expect((await storage.get(created.id))?.title).toBe("Keep me"); + await manager.closeAgent(created.id); + } finally { + await storage.flush(); + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test.each(["hang", "reject"])( + "reload does not resume when the previous session close fails: %s", + async (failure) => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-close-timeout-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + class HangingCloseSession extends TestAgentSession { + closeCalled = false; + + override async close(): Promise { + this.closeCalled = true; + if (failure === "reject") throw new Error("close rejected"); + await new Promise(() => {}); + } + } + + class HangingCloseClient extends TestAgentClient { + readonly firstSession = new HangingCloseSession({ + provider: "codex", + cwd: workdir, + }); + resumeSessionCalls = 0; + + override async createSession(): Promise { + return this.firstSession; + } + + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + this.resumeSessionCalls += 1; + return new TestAgentSession({ + provider: "codex", + cwd: config?.cwd ?? workdir, + }); + } + } + + const client = new HangingCloseClient(); + const manager = new AgentManager({ + clients: { + codex: client, + }, + registry: storage, + logger, + rescueTimeouts: { reloadSessionCloseMs: 10 }, + idFactory: () => "00000000-0000-4000-8000-000000000302", + }); + + try { + const snapshot = await manager.createAgent( + { + provider: "codex", + cwd: workdir, + }, + undefined, + { workspaceId: undefined }, + ); + + for (let i = 0; i < 2; i++) { + await expect(manager.reloadAgentSession(snapshot.id)).rejects.toThrow( + failure === "hang" ? "Timed out closing previous session" : "close rejected", + ); + } + expect(client.firstSession.closeCalled).toBe(true); + expect(client.resumeSessionCalls).toBe(0); + expect(manager.getAgent(snapshot.id)?.session).toBe(client.firstSession); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } + }, +); + test("cancelAgentRun preserves running state when the provider interrupt hangs", async () => { const fixture = await createControlledInterruptFixture({ name: "interrupt-timeout", @@ -2809,7 +2989,7 @@ test("reloadAgentSession preserves the live session when its replacement cannot }), ).rejects.toThrow("Provider 'codex' does not support MCP servers"); - expect(replacement.closed).toBe(true); + expect(replacement.closed).toBe(false); expect(original.closed).toBe(false); expect(manager.getAgent(created.id)?.session).toBe(original); expect(manager.getAgent(created.id)?.lifecycle).toBe("idle"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index edda3204e4..4d3c1447f7 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -704,6 +704,7 @@ export class AgentManager { private readonly backgroundTasks = new Set>(); private readonly agentRegistrationTasks = new Set>(); private readonly inFlightAgentCloses = new Map>(); + private readonly reloadedSessionCloses = new WeakMap>(); private readonly lifecycleMutationTails = new Map>(); private readonly agentStreamCoalescer: AgentStreamCoalescer; private mcpBaseUrl: string | null; @@ -1120,6 +1121,8 @@ export class AgentManager { } async waitForAgentClose(agentId: string): Promise { + // Loading during reload must wait for the replacement, not resume another writer. + await this.lifecycleMutationTails.get(agentId); await this.inFlightAgentCloses?.get(agentId)?.catch(() => undefined); } @@ -1398,7 +1401,9 @@ export class AgentManager { options?: { rehydrateFromDisk?: boolean }, ): Promise { return this.trackAgentRegistrationOperation( - this.reloadAgentSessionInternal(agentId, overrides, options), + this.runLifecycleMutation(agentId, () => + this.reloadAgentSessionInternal(agentId, overrides, options), + ), ); } @@ -1439,28 +1444,32 @@ export class AgentManager { paseoToolPolicy, ); const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); - this.paseoToolPolicies.set(agentId, paseoToolPolicy); + if ( + Object.keys(storedConfig.mcpServers ?? {}).length > 0 && + existing.session.capabilities.supportsMcpServers !== true + ) { + throw new Error(`Provider '${provider}' does not support MCP servers`); + } let session: AgentSession | undefined; + let closedExisting: ManagedAgentClosed | undefined; let handedToRegistration = false; - let replacedExisting = false; try { + // A persisted thread can have only one writer, even when its turn is idle. + await this.closeReloadedSession(existing.session, agentId); + await this.drainSessionEvents(agentId); + this.cancelRunningProviderSubagents(agentId); + closedExisting = this.prepareAgentForClosure(existing, "agent reloaded"); + await this.persistSnapshot(closedExisting); + this.assertAcceptingAgentRegistrations(); + + this.paseoToolPolicies.set(agentId, paseoToolPolicy); session = handle ? await client.resumeSession(handle, providerLaunchConfig, launchContext) : await client.createSession(providerLaunchConfig, launchContext); await this.requireExternalMcpSupport(session, storedConfig); - this.assertAcceptingAgentRegistrations(); - this.cancelRunningProviderSubagents(agentId); - const closedExisting = this.prepareAgentForClosure(existing, "agent reloaded"); - replacedExisting = true; - try { - await this.persistSnapshot(closedExisting); - } finally { - await this.closeReloadedSession(existing.session, agentId); - } - if (rehydrateFromDisk) { // Wipe the in-memory timeline so registerSession mints a new epoch and // hydrateTimelineFromProvider re-streams the freshly read provider history. @@ -1484,41 +1493,49 @@ export class AgentManager { lastError: preservedLastError, attention: preservedAttention, }); + } catch (error) { + if (closedExisting) { + this.emitClosedAgent(closedExisting, { persist: false }); + } else if (this.agents.get(agentId) === existing) { + existing.lifecycle = "error"; + existing.lastError = error instanceof Error ? error.message : String(error); + this.emitState(existing); + } + throw error; } finally { - if (!replacedExisting) { + if (!handedToRegistration) { if (hadPreviousPaseoToolPolicy) { this.paseoToolPolicies.set(agentId, previousPaseoToolPolicy); } else { this.paseoToolPolicies.delete(agentId); } - } - if (session && !handedToRegistration) { - await this.closeUnregisteredSession(session); + if (session) { + await this.closeUnregisteredSession(session); + } } } } private async closeReloadedSession(session: AgentSession, agentId: string): Promise { - try { - const result = await this.waitWithTimeout({ - operation: session.close(), - timeoutMs: this.rescueTimeouts.reloadSessionCloseMs, - onLateError: (error) => { - this.logger.warn( - { err: error, agentId }, - "Previous session close failed after refresh timeout", - ); - }, - }); - - if (result === "timed_out") { + let operation = this.reloadedSessionCloses.get(session); + if (!operation) { + operation = session.close(); + this.reloadedSessionCloses.set(session, operation); + // Keep pending closes across request timeouts; a retry must await the same release. + void operation.catch(() => this.reloadedSessionCloses.delete(session)); + } + const result = await this.waitWithTimeout({ + operation, + timeoutMs: this.rescueTimeouts.reloadSessionCloseMs, + onLateError: (error) => { this.logger.warn( - { agentId, timeoutMs: this.rescueTimeouts.reloadSessionCloseMs }, - "Timed out closing previous session during refresh", + { err: error, agentId }, + "Previous session close failed after refresh timeout", ); - } - } catch (error) { - this.logger.warn({ err: error, agentId }, "Failed to close previous session during refresh"); + }, + }); + if (result === "timed_out") { + throw new Error("Timed out closing previous session during refresh"); } } @@ -1558,7 +1575,10 @@ export class AgentManager { return existing; } - const close = this.closeAgentRuntime(agentId); + const close = this.runLifecycleMutation(agentId, async () => { + // A preceding reload or archive may already have closed the durable agent. + if (this.agents.has(agentId)) await this.closeAgentRuntime(agentId); + }); this.inFlightAgentCloses.set(agentId, close); const clearClose = () => { if (this.inFlightAgentCloses.get(agentId) === close) { @@ -1651,7 +1671,7 @@ export class AgentManager { const { archivedAt } = await this.markRecordArchived(stored); agent.updatedAt = new Date(archivedAt); - await this.closeAgent(agentId); + await this.closeAgentRuntime(agentId); this.discardRetainedAgentState(agentId); await this.cascadeArchiveChildren(agentId); diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index 864813a140..08bd66a79c 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -88,7 +88,7 @@ function createFinishNotificationScenario( Reflect.set(callerAgent, "lifecycle", "idle"); Reflect.set(callerAgent, "config", { title: "Caller Agent" }); - const agentManager: AgentManager = Object.create(AgentManager.prototype); + const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() }); Reflect.set(agentManager, "getAgent", (agentId: string) => { if (agentId === "child-agent") { return childAgent; @@ -468,7 +468,7 @@ it("does not notify archived callers", async () => { const streamAgentSpy = vi.fn(() => (async function* noop() {})()); const replaceAgentRunSpy = vi.fn(() => (async function* noop() {})()); - const agentManager: AgentManager = Object.create(AgentManager.prototype); + const agentManager = new AgentManager({ clients: {}, logger: createTestLogger() }); Reflect.set( agentManager, "getAgent", diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts index 28d0252cf5..0469d9fbe1 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { execFileSync } from "node:child_process"; import os from "node:os"; import path from "node:path"; @@ -6,6 +6,9 @@ import { createServer } from "node:http"; import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { once } from "node:events"; +import { AgentManager } from "../agent-manager.js"; +import { AgentStorage } from "../agent-storage.js"; + import { CodexAppServerAgentClient } from "./codex-app-server-agent.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js"; @@ -173,7 +176,7 @@ function writeMockCodexConfig(codexHome: string, serverUrl: string): void { path.join(codexHome, "config.toml"), ` model = "mock-model" -approval_policy = "untrusted" +approval_policy = "on-request" sandbox_mode = "read-only" model_provider = "mock_provider" @@ -214,6 +217,53 @@ function waitForEvent(params: { } describe("Codex app-server provider (local e2e)", () => { + test.runIf(isCodexInstalled())( + "reloads a persisted idle thread repeatedly without overlapping writers", + async () => { + const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-reload-cwd-")); + const codexHome = mkdtempSync(path.join(os.tmpdir(), "codex-reload-home-")); + const mockServer = await startMockResponsesServer([assistantMessageSse("saved")]); + const logger = createTestLogger(); + const storage = new AgentStorage(path.join(cwd, "agents"), logger); + const manager = new AgentManager({ + clients: { codex: new CodexAppServerAgentClient(logger) }, + registry: storage, + logger, + }); + let agentId: string | undefined; + try { + writeMockCodexConfig(codexHome, mockServer.url); + vi.stubEnv("CODEX_HOME", codexHome); + const created = await manager.createAgent( + { provider: "codex", cwd, modeId: "auto", model: "mock-model" }, + undefined, + { workspaceId: undefined }, + ); + agentId = created.id; + await manager.runAgent(agentId, "save this turn"); + const threadId = manager.getAgent(agentId)?.persistence?.sessionId; + expect(threadId).toBeTruthy(); + for (let i = 0; i < 3; i++) { + const reloaded = await manager.reloadAgentSession(agentId, undefined, { + rehydrateFromDisk: true, + }); + expect(reloaded.persistence?.sessionId).toBe(threadId); + expect(reloaded.lifecycle).toBe("idle"); + } + await manager.runAgent(agentId, "continue after reload"); + expect(manager.getAgent(agentId)?.lifecycle).toBe("idle"); + } finally { + if (agentId && manager.getAgent(agentId)) await manager.closeAgent(agentId); + await storage.flush(); + vi.unstubAllEnvs(); + await mockServer.close(); + rmSync(cwd, { recursive: true, force: true }); + rmSync(codexHome, { recursive: true, force: true }); + } + }, + 30_000, + ); + test.runIf(isCodexInstalled())( "surfaces request_user_input from the app-server as question permissions and timeline tool calls", async () => { 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 c1b2c6bfde..15e7cfdf54 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 @@ -1022,7 +1022,13 @@ describe("Codex app-server provider", () => { child.stderr = new PassThrough() as ChildProcessWithoutNullStreams["stderr"]; child.exitCode = null; child.signalCode = null; - child.kill = vi.fn(() => true) as ChildProcessWithoutNullStreams["kill"]; + child.kill = vi.fn((signal) => { + if (signal === "SIGKILL") { + child.signalCode = "SIGKILL"; + child.emit("exit", null, "SIGKILL"); + } + return true; + }) as ChildProcessWithoutNullStreams["kill"]; const client = new CodexAppServerClient(child, createTestLogger()); try { 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 87a463a11f..0cc71bf1b8 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 @@ -4816,12 +4816,12 @@ export class CodexAppServerAgentSession implements AgentSession { private async disposeClient(): Promise { const client = this.client; - this.client = null; this.connected = false; this.currentTurnId = null; if (client) { await client.dispose(); } + this.client = null; } async listCommands(): Promise { diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts index 1d7417510b..5c9dbe956c 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import { @@ -32,6 +32,29 @@ describe("Codex app-server transport", () => { await expect(request).rejects.toThrow("Codex app-server client is closed"); }); + test("dispose rejects until the child has actually exited", async () => { + vi.useFakeTimers(); + const child = createCodexAppServerChildProcess(); + child.kill = () => true; + const client = new CodexAppServerClient(child, createTestLogger()); + try { + for (let i = 0; i < 2; i++) { + const closing = expect(client.dispose()).rejects.toThrow( + "did not report exit after SIGKILL", + ); + await vi.advanceTimersByTimeAsync(3_000); + await closing; + } + child.exitCode = 0; + child.emit("exit", 0, null); + await expect(client.dispose()).resolves.toBeUndefined(); + } finally { + vi.useRealTimers(); + child.stdout.end(); + child.stderr.end(); + } + }); + test.each([ "item/commandExecution/requestApproval", "item/fileChange/requestApproval", diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts index 32152e7ea5..1720118c63 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -257,7 +257,6 @@ export class CodexAppServerClient { } async dispose(): Promise { - if (this.disposed) return; this.disposed = true; this.unexpectedTerminationHandler = null; this.rl.close(); @@ -278,10 +277,7 @@ export class CodexAppServerClient { }, }); if (result === "kill-timeout") { - this.logger.warn( - { timeoutMs: APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS }, - "Codex app-server did not report exit after SIGKILL", - ); + throw new Error("Codex app-server did not report exit after SIGKILL"); } } diff --git a/packages/server/src/server/daemon-e2e/agent-reload.native.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-reload.native.real.e2e.test.ts new file mode 100644 index 0000000000..ac9d2f83cb --- /dev/null +++ b/packages/server/src/server/daemon-e2e/agent-reload.native.real.e2e.test.ts @@ -0,0 +1,128 @@ +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { expect, test } from "vitest"; + +import type { AgentClient, AgentSessionConfig } from "../agent/agent-sdk-types.js"; +import { ClaudeAgentClient } from "../agent/providers/claude/agent.js"; +import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js"; +import { OpenCodeServerManager } from "../agent/providers/opencode/server-manager.js"; +import { OpenCodeAgentClient } from "../agent/providers/opencode-agent.js"; +import { + getRealProviderConfig, + getRealProviderRuntimeSettings, +} from "./real-provider-test-config.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; + +type NativeProvider = "claude" | "codex" | "opencode"; + +interface ReloadConversation { + rememberToken: () => Promise; + reloadAndRecallToken: (sessionId: string) => Promise; +} + +async function withNativeConversation( + provider: NativeProvider, + run: (conversation: ReloadConversation) => Promise, +): Promise { + const secret = `RELOAD_${provider.toUpperCase()}_7F31`; + const root = mkdtempSync(path.join(tmpdir(), `paseo-reload-${provider}-`)); + const cwd = path.join(root, "workspace"); + mkdirSync(cwd); + const logger = pino({ level: "warn" }); + let daemon: TestPaseoDaemon | undefined; + let client: DaemonClient | undefined; + let openCode: OpenCodeAgentClient | undefined; + let openCodeRuntimeRoot: string | undefined; + try { + let providerClient: AgentClient; + let config: Partial; + if (provider === "claude") { + providerClient = new ClaudeAgentClient({ logger }); + config = { model: "haiku", modeId: "bypassPermissions" }; + } else if (provider === "codex") { + providerClient = new CodexAppServerAgentClient(logger); + config = { modeId: "full-access", thinkingOptionId: "low" }; + } else { + const settings = getRealProviderRuntimeSettings("opencode"); + openCodeRuntimeRoot = path.dirname(settings.env!.XDG_CONFIG_HOME!); + openCode = new OpenCodeAgentClient(logger, settings, { + serverManager: new OpenCodeServerManager({ logger, runtimeSettings: settings }), + }); + providerClient = openCode; + config = getRealProviderConfig("opencode"); + } + daemon = await createTestPaseoDaemon({ + agentClients: { [provider]: providerClient }, + logger, + pluginsEnabled: false, + }); + client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.7.2" }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "native-reload-qa" } }); + const agent = await client.createAgent({ provider, cwd, ...config }); + process.stdout.write(`${provider}: model=${agent.model ?? "default"}\n`); + const manager = daemon.daemon.agentManager; + const connectedClient = client; + await run({ + rememberToken: async () => { + const first = await manager.runAgent( + agent.id, + `Remember this token for this conversation: ${secret}. Reply with exactly that token. Do not use tools.`, + ); + expect(first.finalText).toContain(secret); + const handle = manager.getAgent(agent.id)?.persistence; + expect(handle?.sessionId).toBeTruthy(); + process.stdout.write(`${provider}: first real turn passed; session=${handle!.sessionId}\n`); + return handle!.sessionId; + }, + reloadAndRecallToken: async (sessionId: string) => { + for (let round = 1; round <= 3; round++) { + await connectedClient.refreshAgent(agent.id); + expect(manager.getAgent(agent.id)?.persistence?.sessionId).toBe(sessionId); + const response = await manager.runAgent( + agent.id, + "What token did I ask you to remember? Reply with only that token. Do not use tools.", + ); + expect(response.finalText).toContain(secret); + const timeline = await connectedClient.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect( + timeline.entries.some( + ({ item }) => item.type === "assistant_message" && item.text.includes(secret), + ), + ).toBe(true); + process.stdout.write( + `${provider}: refresh ${round}/3 passed; same session; memory recalled; timeline present\n`, + ); + } + }, + }); + await manager.closeAgent(agent.id); + } finally { + await client?.close(); + await daemon?.close(); + await openCode?.shutdown(); + if (openCodeRuntimeRoot) rmSync(openCodeRuntimeRoot, { recursive: true, force: true }); + rmSync(root, { recursive: true, force: true }); + } +} + +// Opt in: real requests using local Claude/Codex auth and the OpenRouter test setup. +test + .runIf(process.env.PASEO_NATIVE_RELOAD_QA === "1") + .each(["claude", "codex", "opencode"] as const)( + "%s retains its session and conversation through repeated refresh RPCs", + async (provider) => { + await withNativeConversation(provider, async (conversation) => { + const sessionId = await conversation.rememberToken(); + await conversation.reloadAndRecallToken(sessionId); + }); + }, + 180_000, +);