Stop plugins after the agents they serve are closed (#5253)

* Close agents before stopping the plugins that serve them

A clean daemon shutdown stopped every plugin before it closed any agent.
Stopping a plugin withdraws its provider registrations, which pushes an
empty registration set through the agent provider registry, shuts the
plugin provider client down, and closes each still-open session with
`connectionClosed()`. That is reported as `session.runtime_failed` and
translated to `turn_failed`, so an agent whose last turn had completed
successfully was persisted with `lastError: "Provider connection closed"`
and came back after a restart carrying a failure it never had.

Stop tracking plugin provider registrations first so plugin teardown can
no longer withdraw a provider from under an open agent, and stop the
plugins themselves after `closeAllAgents()` and the provider runtime
shutdown have already made every session terminal. Closing the sessions
while their plugin is still running also means `session.close` now
reaches the provider.

Fixes #4877

* Bound each agent close so a silent provider cannot stall shutdown

Closing an agent asks its provider to close the session and waits for the
answer, and that request is only rejected when the provider runtime shuts
down. With agents now closing before the runtime does, a provider that
never answers `session.close` would leave `closeAllAgents()` waiting with
nothing left to unblock it.

Give each close a deadline. Whatever outlives it is abandoned and logged,
and the provider runtime shutdown that runs next rejects the request that
was still pending.

Also type the test fixture's `send` input as `ProviderInput`.

* Move the shutdown tests into the plugin lifecycle suite

Both tests exercise plugin lifecycle against a real daemon, which
lifecycle.e2e.test.ts already covers, so they go there rather than into a
suite of their own.

Drop the wall-clock assertion on how long stop() took. What the test
establishes is that stop() returns at all, and the suite timeout is what
enforces that; measuring elapsed time only added a second bound that a
loaded machine could trip.

---------

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

packages/server/src/server/bootstrap.ts

@@ -199,6 +199,7 @@ import {
   type ManagedProcessRegistry,
 } from "./managed-processes/managed-processes.js";
 import { terminateWithTreeKill } from "../utils/tree-kill.js";
+import { withTimeout } from "../utils/promise-timeout.js";
 import { isHostnameAllowed, type HostnamesConfig } from "./hostnames.js";
 import {
   createRequireBearerMiddleware,
@@ -1777,7 +1778,10 @@ export async function createPaseoDaemon(
   };
 
   const stop = async () => {
-    await pluginRuntime.stopAllPlugins();
+    // Stop tracking plugin provider registrations before anything tears plugins
+    // down, so plugin shutdown cannot withdraw a provider from under an agent
+    // that is still open. Plugins themselves are stopped once every session
+    // they serve has been closed, further down.
     unsubscribePluginProviders();
     await hubRelationships.stop();
     workspaceReconciliation.dispose();
@@ -1790,6 +1794,7 @@ export async function createPaseoDaemon(
     detachAgentStoragePersistence();
     await agentStorage.flush().catch(() => undefined);
     await agentProviderRuntime.shutdown();
+    await pluginRuntime.stopAllPlugins();
     terminalManager.killAll();
     await speechService.stop();
     await scheduleService.stop().catch(() => undefined);
@@ -1829,12 +1834,24 @@ export async function createPaseoDaemon(
   };
 }
 
+/**
+ * Closing an agent asks its provider to close the session and waits for the
+ * answer. A provider that never answers must not hold the daemon open, so a
+ * close that outlives this deadline is abandoned; `agentProviderRuntime`
+ * shutdown runs next and rejects the request that was still pending.
+ */
+const AGENT_CLOSE_TIMEOUT_MS = 5_000;
+
 async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise<void> {
   const agents = agentManager.listAgents();
   await Promise.all(
     agents.map(async (agent) => {
       try {
-        await agentManager.closeAgent(agent.id);
+        await withTimeout({
+          promise: agentManager.closeAgent(agent.id),
+          timeoutMs: AGENT_CLOSE_TIMEOUT_MS,
+          label: `close agent ${agent.id}`,
+        });
       } catch (err) {
         logger.error({ err, agentId: agent.id }, "Failed to close agent");
       }

packages/server/src/server/plugins/lifecycle.e2e.test.ts

@@ -1,9 +1,10 @@
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
-import { expect, test } from "vitest";
+import { expect, onTestFinished, test } from "vitest";
 import { DaemonClient } from "../test-utils/daemon-client.js";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { resolveDaemonVersion } from "../daemon-version.js";
 
 test("a plugin transforms workspace creation once across receipt replays and observes its committed lifecycle", async () => {
   const directory = await mkdtemp(path.join(tmpdir(), "paseo-lifecycle-"));
@@ -269,3 +270,190 @@ export default function contribute(server) {
     await rm(directory, { recursive: true, force: true });
   }
 }, 60_000);
+
+const PROVIDER_ID = "shutdown-provider";
+
+async function createProviderPlugin(
+  root: string,
+  options: { acknowledgeClose: boolean } = { acknowledgeClose: true },
+): Promise<string> {
+  const directory = path.join(root, "plugin");
+  await mkdir(directory, { recursive: true });
+  await writeFile(
+    path.join(directory, "paseo-plugin.json"),
+    JSON.stringify({
+      id: "shutdown-provider-plugin",
+      requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` },
+    }),
+  );
+  await writeFile(
+    path.join(directory, "index.server.ts"),
+    `import type {
+  ProviderEvent,
+  ProviderInput,
+  ProviderRegistration,
+} from "@getpaseo/plugin/server/provider";
+
+const CAPABILITIES = ["prompt.message", "session.persistence"];
+
+const provider: ProviderRegistration = {
+  id: ${JSON.stringify(PROVIDER_ID)},
+  label: "Shutdown provider",
+  async connect() {
+    let listener: ((event: ProviderEvent) => void) | null = null;
+    const emit = (event: ProviderEvent) => listener?.(event);
+    let turn = 0;
+    return {
+      version: 1,
+      capabilities: CAPABILITIES,
+      async send(input: ProviderInput) {
+        if (input.type === "catalog") {
+          emit({
+            type: "catalog",
+            requestId: input.requestId,
+            catalog: {
+              models: [{ id: "shutdown-model", label: "Shutdown model" }],
+              modes: [],
+              thinkingOptions: [],
+              defaultModel: "shutdown-model",
+            },
+          });
+          return;
+        }
+        if (input.type === "session.open") {
+          emit({
+            type: "session.opened",
+            requestId: input.requestId,
+            sessionId: input.sessionId,
+            capabilities: CAPABILITIES,
+            restoration: "core",
+            persistence: { version: 1, data: { token: "root" } },
+            cwd: input.config.cwd,
+          });
+          emit({ type: "session.ready", requestId: input.requestId, sessionId: input.sessionId });
+          return;
+        }
+        if (input.type === "session.prompt") {
+          turn += 1;
+          const turnId = "turn-" + turn;
+          emit({
+            type: "session.prompt_result",
+            sessionId: input.sessionId,
+            clientMessageId: input.prompt.clientMessageId,
+            result: { type: "turn", turnId },
+          });
+          emit({ type: "session.turn", sessionId: input.sessionId, turnId, state: "started" });
+          emit({
+            type: "timeline.item",
+            sessionId: input.sessionId,
+            item: { type: "assistant_message", id: "answer-" + turn, text: "Done" },
+          });
+          emit({ type: "session.turn", sessionId: input.sessionId, turnId, state: "completed" });
+          return;
+        }
+        if (input.type === "session.close") {
+          if (!${JSON.stringify(options.acknowledgeClose)}) return;
+          emit({ type: "session.closed", sessionId: input.sessionId });
+        }
+        if ("requestId" in input) {
+          emit({ type: "request.completed", requestId: input.requestId });
+        }
+      },
+      onEvent(next: (event: ProviderEvent) => void) {
+        listener = next;
+        return () => {
+          if (listener === next) listener = null;
+        };
+      },
+      async close() {},
+    };
+  },
+};
+
+export default function contribute(server: { registerProvider(p: ProviderRegistration): void }) {
+  server.registerProvider(provider);
+  return () => undefined;
+}`,
+  );
+  return directory;
+}
+
+test("a clean daemon shutdown leaves a completed plugin-provider agent without an error", async () => {
+  const root = await mkdtemp(path.join(tmpdir(), "plugin-provider-shutdown-"));
+  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"),
+    cleanup: false,
+    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" });
+  await client.connect();
+  await client.fetchAgents({ subscribe: {} });
+
+  const agent = await client.createAgent({
+    provider: PROVIDER_ID,
+    model: "shutdown-model",
+    cwd: workspace,
+    title: "Shutdown provider agent",
+  });
+  await client.sendMessage(agent.id, "hello");
+  const finished = await client.waitForFinish(agent.id, 30_000);
+  expect(finished.status).toBe("idle");
+
+  const beforeShutdown = await daemon.daemon.agentStorage.get(agent.id);
+  expect(beforeShutdown?.lastError ?? null).toBeNull();
+
+  await client.close();
+  await daemon.daemon.stop();
+  await daemon.daemon.agentStorage.flush().catch(() => undefined);
+
+  const persisted = await daemon.daemon.agentStorage.get(agent.id);
+  expect(persisted).not.toBeNull();
+  expect(persisted?.lastError ?? null).toBeNull();
+  expect(persisted?.attentionReason ?? null).not.toBe("error");
+}, 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 }));
+  const pluginDirectory = await createProviderPlugin(root, { acknowledgeClose: false });
+  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"),
+    cleanup: false,
+    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" });
+  await client.connect();
+  await client.fetchAgents({ subscribe: {} });
+
+  const agent = await client.createAgent({
+    provider: PROVIDER_ID,
+    model: "shutdown-model",
+    cwd: workspace,
+    title: "Stuck close agent",
+  });
+  await client.sendMessage(agent.id, "hello");
+  await client.waitForFinish(agent.id, 30_000);
+  await client.close();
+
+  // Without a deadline on each agent close, stop() never returns here and the
+  // test fails by timing out.
+  await expect(daemon.daemon.stop()).resolves.toBeUndefined();
+}, 90_000);