diff --git a/packages/plugin/src/server/lifecycle.ts b/packages/plugin/src/server/lifecycle.ts index 9115145cde..acebde32f1 100644 --- a/packages/plugin/src/server/lifecycle.ts +++ b/packages/plugin/src/server/lifecycle.ts @@ -60,6 +60,7 @@ export interface PluginLifecycleEvents { }; "agent.archived": { agent: PluginHookAgent; archivedAt: string }; "agent.created": { agent: PluginHookAgent }; + "agent.closed": { agent: PluginHookAgent }; "workspace.created": { workspace: PluginHookWorkspace }; "workspace.archived": { workspace: PluginHookWorkspace }; } diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index aebede5fa1..7c39876b30 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -3730,6 +3730,11 @@ export class AgentManager { private emitClosedAgent(agent: ManagedAgentClosed, options?: { persist?: boolean }): void { this.emitState(agent, options); + if (!agent.internal) { + this.pluginLifecycle?.emit("agent.closed", { + agent: describeHookAgent({ ...agent, title: agent.config.title }), + }); + } } private subscribeToSession(agent: ActiveManagedAgent): void { if (agent.unsubscribeSession) { diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 659ac54d02..ebf6e0bc81 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1807,6 +1807,11 @@ export async function createPaseoDaemon( wsServer?.prepareForShutdown(); agentManager.prepareForShutdown(); await closeAllAgents(logger, agentManager); + await withTimeout({ + promise: pluginRuntime.drainEvents(), + timeoutMs: AGENT_CLOSE_TIMEOUT_MS, + label: "drain plugin lifecycle events", + }).catch((error) => logger.warn({ err: error }, "Plugin lifecycle events did not finish")); await agentManager.flushForShutdown().catch(() => undefined); detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); diff --git a/packages/server/src/server/plugins/index.ts b/packages/server/src/server/plugins/index.ts index 167216dbc1..3940ef033f 100644 --- a/packages/server/src/server/plugins/index.ts +++ b/packages/server/src/server/plugins/index.ts @@ -33,6 +33,7 @@ const BUILTIN_PROVIDER_ID_SET: ReadonlySet = new Set(BUILTIN_PROVIDER_ID interface PluginRuntimePort { emit?: PluginLifecycle["emit"]; + drainEvents?: PluginRuntime["drainEvents"]; before?: PluginLifecycle["before"]; catalog: PluginRuntime["catalog"]; invoke(pluginId: string, method: string, input: unknown): Promise; @@ -119,6 +120,10 @@ export class PluginService { this.runtime.emit?.(name, event); }; + async drainEvents(): Promise { + await this.runtime.drainEvents?.(); + } + readonly before: PluginLifecycle["before"] = async (name, request) => { if (this.runtime.before) { return this.runtime.before(name, request); diff --git a/packages/server/src/server/plugins/lifecycle-archive.e2e.test.ts b/packages/server/src/server/plugins/lifecycle-archive.e2e.test.ts index 8457c94711..6f3d8661e7 100644 --- a/packages/server/src/server/plugins/lifecycle-archive.e2e.test.ts +++ b/packages/server/src/server/plugins/lifecycle-archive.e2e.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -6,7 +6,7 @@ import { expect, test } from "vitest"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; -test("workspace archive publishes agent archive hooks for both live and closed agents", async () => { +test("closing and archiving agents publish their distinct lifecycle hooks once", async () => { const directory = await mkdtemp(path.join(tmpdir(), "paseo-archive-hooks-")); const daemon = await createTestPaseoDaemon({ daemonVersion: "0.8.0" }); const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" }); @@ -33,7 +33,16 @@ test("workspace archive publishes agent archive hooks for both live and closed a workspaceId: workspace.id, }); await daemon.daemon.agentManager.closeAgent(closed.id); + await daemon.daemon.agentManager.closeAgent(closed.id); expect(daemon.daemon.agentManager.getAgent(closed.id)).toBeNull(); + await expect + .poll(async () => { + const logs = await client.getPluginLogs("lifecycle-logger"); + return logs + .filter((entry) => entry.message.startsWith('{"hook":"agent.closed"')) + .map((entry) => JSON.parse(entry.message).data.agent.id); + }) + .toEqual([closed.id]); await client.archiveWorkspace(workspace.id); await expect .poll(async () => { @@ -48,6 +57,50 @@ test("workspace archive publishes agent archive hooks for both live and closed a .sort(); }) .toEqual([live.id, closed.id].sort()); + await expect + .poll(async () => { + const logs = await client.getPluginLogs("lifecycle-logger"); + return logs + .filter((entry) => entry.message.startsWith('{"hook":"agent.closed"')) + .map((entry) => JSON.parse(entry.message).data.agent.id) + .sort(); + }) + .toEqual([live.id, closed.id].sort()); + } finally { + await client.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } +}, 60_000); + +test("daemon shutdown lets agent close hooks finish before stopping plugins", async () => { + const directory = await mkdtemp(path.join(tmpdir(), "paseo-close-hook-shutdown-")); + const daemon = await createTestPaseoDaemon({ daemonVersion: "0.8.0" }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" }); + const completedFile = path.join(directory, "closed.txt"); + try { + await writeFile( + path.join(directory, "paseo-plugin.json"), + JSON.stringify({ id: "close-hook-shutdown", requirements: { paseo: ">=0.8.0" } }), + ); + await writeFile( + path.join(directory, "index.server.ts"), + `import { writeFile } from "node:fs/promises"; +import { setTimeout } from "node:timers/promises"; +export default function contribute(server) { + server.on("agent.closed", async (event, context) => { + await setTimeout(200, undefined, { signal: context.signal }); + await writeFile(${JSON.stringify(completedFile)}, event.agent.id); + }); + return () => {}; +}`, + ); + await client.connect(); + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(directory); + const agent = await client.createAgent({ provider: "claude", cwd: directory }); + await daemon.close(); + expect(await readFile(completedFile, "utf8")).toBe(agent.id); } finally { await client.close(); await daemon.close(); diff --git a/packages/server/src/server/plugins/lifecycle.real.e2e.test.ts b/packages/server/src/server/plugins/lifecycle.real.e2e.test.ts index bd71f0ce7e..fc9cfbd84d 100644 --- a/packages/server/src/server/plugins/lifecycle.real.e2e.test.ts +++ b/packages/server/src/server/plugins/lifecycle.real.e2e.test.ts @@ -280,6 +280,7 @@ test.skipIf(process.platform !== "linux")( }) .toEqual([ "agent.archived", + "agent.closed", "agent.created", "agent.permission_requested", "agent.permission_resolved", diff --git a/packages/server/src/server/plugins/lifecycle/index.ts b/packages/server/src/server/plugins/lifecycle/index.ts index 766ccc0bce..447c325ad4 100644 --- a/packages/server/src/server/plugins/lifecycle/index.ts +++ b/packages/server/src/server/plugins/lifecycle/index.ts @@ -17,6 +17,7 @@ import type { PersistedWorkspaceRecord } from "../../workspace-registry.js"; export const lifecycleEventNames = [ "agent.created", + "agent.closed", "agent.turn_started", "agent.turn_ended", "agent.permission_requested", diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index a26e7cbfea..66d71d277d 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -295,6 +295,7 @@ async function resolveEntryPaths(directory: string): Promise<{ export class PluginRuntime { private readonly plugins = new Map(); + private readonly pendingEvents = new Set>(); private readonly logTails = new Map(); private readonly logger: pino.Logger; private readonly spawnChild: () => PluginChild; @@ -515,22 +516,30 @@ export class PluginRuntime { if (!loaded.hooks.events.includes(name)) { continue; } - void this.request(loaded, { + const request = this.request(loaded, { type: "hook", requestId: randomUUID(), kind: "event", name, input: event, - }).catch((error) => { - this.appendLog( - loaded.id, - "stderr", - `Lifecycle hook ${name} failed: ${describeError(error)}`, - ); - }); + }) + .then(() => undefined) + .catch((error) => { + this.appendLog( + loaded.id, + "stderr", + `Lifecycle hook ${name} failed: ${describeError(error)}`, + ); + }); + this.pendingEvents.add(request); + void request.finally(() => this.pendingEvents.delete(request)); } } + async drainEvents(): Promise { + await Promise.all(this.pendingEvents); + } + async before( name: Name, input: PluginBeforeRequests[Name], diff --git a/plugin-examples/lifecycle-logger/index.server.ts b/plugin-examples/lifecycle-logger/index.server.ts index 04d967949c..0f5fd4ed95 100644 --- a/plugin-examples/lifecycle-logger/index.server.ts +++ b/plugin-examples/lifecycle-logger/index.server.ts @@ -29,6 +29,10 @@ export default function contribute(server: PluginServerContext) { logHook("agent.created", event); }); + server.on("agent.closed", (event) => { + logHook("agent.closed", event); + }); + server.on("agent.turn_started", (event) => { logHook("agent.turn_started", event); }); diff --git a/public-docs/plugins/reference.md b/public-docs/plugins/reference.md index 188e88c154..85e04bbdc3 100644 --- a/public-docs/plugins/reference.md +++ b/public-docs/plugins/reference.md @@ -654,6 +654,7 @@ plans, and mode changes; requesting permission does not end the turn. | Name | Event fields | Trigger | | ---------------------------- | ---------------------------------------- | -------------------------------------------------- | | `agent.created` | `agent` | Ordinary creation finishes; excludes import/resume | +| `agent.closed` | `agent` | A live agent runtime closes | | `agent.turn_started` | `agent`, `turnId` | Live turn starts | | `agent.turn_ended` | `agent`, `turnId`, `outcome`, `timeline` | Live turn completes, fails, or is canceled | | `agent.permission_requested` | `agent`, `request` | Permission or question becomes pending | @@ -663,6 +664,8 @@ plans, and mode changes; requesting permission does not end the turn. | `workspace.archived` | `workspace` | Archive state is saved | Agent events exclude internal utility agents. Archive events can precede runtime/worktree cleanup; +closing an agent that is already closed does not emit another `agent.closed` event. +During daemon shutdown, pending event hooks have up to five seconds to finish before plugins stop. `workspace.created` is not a setup barrier before agent startup. **Shared payload shapes** (`@getpaseo/plugin/server`):