diff --git a/packages/server/src/server/agent/plugin-provider.test.ts b/packages/server/src/server/agent/plugin-provider.test.ts index 6aa32e2731..8774b8784c 100644 --- a/packages/server/src/server/agent/plugin-provider.test.ts +++ b/packages/server/src/server/agent/plugin-provider.test.ts @@ -449,6 +449,63 @@ describe("PluginAgentClientRegistry", () => { expect(eventsOfType(events, "turn_failed")).toHaveLength(1); }); + test("leaves a completed session without a failure when its plugin provider is removed", async () => { + const harness = createProviderHarness(); + const registry = new PluginAgentClientRegistry(createTestLogger()); + registry.replace([harness.registration]); + const session = await registry.clients()[harness.registration.id]!.createSession({ + provider: harness.registration.id, + cwd: "/workspace", + }); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + await session.run("hello", { clientMessageId: "completed-message" }); + expect(eventsOfType(events, "turn_completed")).toHaveLength(1); + + registry.replace([]); + await harness.waitForClose(); + + expect(eventsOfType(events, "turn_failed")).toEqual([]); + await expect( + session.startTurn("after reload", { clientMessageId: "after-reload" }), + ).rejects.toBeInstanceOf(StaleProviderSessionError); + }); + + test("fails an accepted turn that has not started when its plugin provider is removed", async () => { + const harness = createProviderHarness({ + handleInput: async (input, emit) => { + if (input.type !== "session.prompt") return false; + emit({ + type: "session.prompt_result", + sessionId: input.sessionId, + clientMessageId: input.prompt.clientMessageId, + result: { type: "turn", turnId: "accepted-turn" }, + }); + return true; + }, + }); + const registry = new PluginAgentClientRegistry(createTestLogger()); + registry.replace([harness.registration]); + const session = await registry.clients()[harness.registration.id]!.createSession({ + provider: harness.registration.id, + cwd: "/workspace", + }); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + await expect( + session.startTurn("hello", { clientMessageId: "accepted-message" }), + ).resolves.toEqual({ turnId: "accepted-turn" }); + + registry.replace([]); + await harness.waitForClose(); + + expect(eventsOfType(events, "turn_failed")).toEqual([ + expect.objectContaining({ error: "Provider connection closed" }), + ]); + }); + test("closes a stale session after its plugin provider is replaced", async () => { const old = createProviderHarness(); const next = createProviderHarness(); diff --git a/packages/server/src/server/agent/plugin-provider.ts b/packages/server/src/server/agent/plugin-provider.ts index 2ebeb3c797..e6712d3c3b 100644 --- a/packages/server/src/server/agent/plugin-provider.ts +++ b/packages/server/src/server/agent/plugin-provider.ts @@ -547,6 +547,7 @@ class ProviderRuntimeSession { string, Deferred> >(); + private readonly activeTurnIds = new Set(); private terminal = false; config: ProviderConfigState = { models: [], modes: [], thinkingOptions: [], settings: [] }; commands: Array<{ name: string; description: string; argumentHint?: string }> = []; @@ -702,6 +703,7 @@ class ProviderRuntimeSession { return; } if (event.type === "session.prompt_result") { + if (event.result.type === "turn") this.activeTurnIds.add(event.result.turnId); this.prompts.get(event.clientMessageId)?.resolve(event); return; } @@ -720,14 +722,16 @@ class ProviderRuntimeSession { } connectionClosed(error = new Error("Provider connection closed")): void { - if (!this.terminal) { - this.terminal = true; + // Closing fails only an interrupted turn. An idle session goes stale, and its next prompt + // reopens it from persistence. + if (!this.terminal && this.activeTurnIds.size > 0) { this.publish({ type: "session.runtime_failed", sessionId: this.id, error: { message: error.message }, }); } + this.terminal = true; this.rejectPending(error); } @@ -751,6 +755,10 @@ class ProviderRuntimeSession { } private publish(event: ProviderEvent): void { + if (event.type === "session.turn") { + if (event.state === "started") this.activeTurnIds.add(event.turnId); + else this.activeTurnIds.delete(event.turnId); + } if (event.type === "session.config") this.config = event.config; if (event.type === "session.commands") this.commands = [...event.commands]; if (event.type === "session.persistence" && this.restoration === "core") { diff --git a/packages/server/src/server/plugins/lifecycle.e2e.test.ts b/packages/server/src/server/plugins/lifecycle.e2e.test.ts index aeaa1c79bc..fd408da004 100644 --- a/packages/server/src/server/plugins/lifecycle.e2e.test.ts +++ b/packages/server/src/server/plugins/lifecycle.e2e.test.ts @@ -423,6 +423,55 @@ test("a clean daemon shutdown leaves a completed plugin-provider agent without a expect(persisted?.attentionReason ?? null).not.toBe("error"); }, 60_000); +test("reloading a plugin leaves a completed plugin-provider agent without an error", async () => { + const root = await mkdtemp(path.join(tmpdir(), "plugin-provider-reload-")); + onTestFinished(() => rm(root, { recursive: true, force: true })); + const pluginDirectory = await createProviderPlugin(root); + const workspace = path.join(root, "workspace"); + await mkdir(workspace, { recursive: true }); + + const daemon = await createTestPaseoDaemon({ + paseoHomeRoot: path.join(root, "daemon"), + staticDir: path.join(root, "static"), + pluginsEnabled: true, + plugins: { + "shutdown-provider-plugin": { source: "directory", path: pluginDirectory, enabled: true }, + }, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.9.1" }); + onTestFinished(async () => { + await client.close(); + await daemon.close(); + }); + await client.connect(); + await client.fetchAgents({ subscribe: {} }); + + const agent = await client.createAgent({ + provider: PROVIDER_ID, + model: "shutdown-model", + cwd: workspace, + title: "Reload provider agent", + }); + await client.sendMessage(agent.id, "hello"); + expect((await client.waitForFinish(agent.id, 30_000)).status).toBe("idle"); + + const reloaded = await client.reloadPlugin("shutdown-provider-plugin"); + expect(reloaded.status).toBe("running"); + + const afterReload = await daemon.daemon.agentStorage.get(agent.id); + expect(afterReload?.lastError ?? null).toBeNull(); + expect(afterReload?.attentionReason ?? null).not.toBe("error"); + const timeline = await client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + expect(JSON.stringify(timeline.entries)).not.toContain("Provider connection closed"); + + await client.sendMessage(agent.id, "after reload"); + expect((await client.waitForFinish(agent.id, 30_000)).status).toBe("idle"); +}, 60_000); + test("a provider that never acknowledges session.close does not hold the daemon open", async () => { const root = await mkdtemp(path.join(tmpdir(), "plugin-provider-stuck-close-")); onTestFinished(() => rm(root, { recursive: true, force: true }));