diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 88090fdf14..c90da96eef 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/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 { 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"); } diff --git a/packages/server/src/server/plugins/lifecycle.e2e.test.ts b/packages/server/src/server/plugins/lifecycle.e2e.test.ts index a6a27ba5bc..05157e3816 100644 --- a/packages/server/src/server/plugins/lifecycle.e2e.test.ts +++ b/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 { + 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);