Keep completed plugin-provider sessions error-free across a plugin reload (#5579)

Removing a plugin provider closed its runtime and published a runtime
failure for every open session, so an idle session whose turn had already
completed gained a "Provider connection closed" error. Only a turn that
is still in flight is interrupted now; an idle session goes stale and its
next prompt reopens it from persistence.

Closes #5535

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

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();

packages/server/src/server/agent/plugin-provider.ts

@@ -547,6 +547,7 @@ class ProviderRuntimeSession {
     string,
     Deferred<Extract<ProviderEvent, { type: "session.prompt_result" }>>
   >();
+  private readonly activeTurnIds = new Set<string>();
   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") {

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 }));