mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-14 13:22:37 +00:00
feat(plugins): expose agent closed lifecycle event (#5971)
* feat(plugins): expose agent closed lifecycle event Emit the event when a live agent reaches its closed state so plugins can release agent-scoped resources. Keep archive and close notifications distinct. Co-authored-by: ilteoood <6383527+ilteoood@users.noreply.github.com> * fix(plugins): finish close hooks during daemon shutdown Wait briefly for pending lifecycle events before stopping plugins so agent close hooks can release their resources during shutdown. Co-authored-by: ilteoood <6383527+ilteoood@users.noreply.github.com> --------- Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> Co-authored-by: ilteoood <6383527+ilteoood@users.noreply.github.com>
This commit is contained in:
co-authored by
ilteoood
parent
8aec0787d2
commit
5375f43a05
@@ -60,6 +60,7 @@ export interface PluginLifecycleEvents {
|
|||||||
};
|
};
|
||||||
"agent.archived": { agent: PluginHookAgent; archivedAt: string };
|
"agent.archived": { agent: PluginHookAgent; archivedAt: string };
|
||||||
"agent.created": { agent: PluginHookAgent };
|
"agent.created": { agent: PluginHookAgent };
|
||||||
|
"agent.closed": { agent: PluginHookAgent };
|
||||||
"workspace.created": { workspace: PluginHookWorkspace };
|
"workspace.created": { workspace: PluginHookWorkspace };
|
||||||
"workspace.archived": { workspace: PluginHookWorkspace };
|
"workspace.archived": { workspace: PluginHookWorkspace };
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3730,6 +3730,11 @@ export class AgentManager {
|
|||||||
|
|
||||||
private emitClosedAgent(agent: ManagedAgentClosed, options?: { persist?: boolean }): void {
|
private emitClosedAgent(agent: ManagedAgentClosed, options?: { persist?: boolean }): void {
|
||||||
this.emitState(agent, options);
|
this.emitState(agent, options);
|
||||||
|
if (!agent.internal) {
|
||||||
|
this.pluginLifecycle?.emit("agent.closed", {
|
||||||
|
agent: describeHookAgent({ ...agent, title: agent.config.title }),
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
private subscribeToSession(agent: ActiveManagedAgent): void {
|
private subscribeToSession(agent: ActiveManagedAgent): void {
|
||||||
if (agent.unsubscribeSession) {
|
if (agent.unsubscribeSession) {
|
||||||
|
|||||||
@@ -1807,6 +1807,11 @@ export async function createPaseoDaemon(
|
|||||||
wsServer?.prepareForShutdown();
|
wsServer?.prepareForShutdown();
|
||||||
agentManager.prepareForShutdown();
|
agentManager.prepareForShutdown();
|
||||||
await closeAllAgents(logger, agentManager);
|
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);
|
await agentManager.flushForShutdown().catch(() => undefined);
|
||||||
detachAgentStoragePersistence();
|
detachAgentStoragePersistence();
|
||||||
await agentStorage.flush().catch(() => undefined);
|
await agentStorage.flush().catch(() => undefined);
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ const BUILTIN_PROVIDER_ID_SET: ReadonlySet<string> = new Set(BUILTIN_PROVIDER_ID
|
|||||||
|
|
||||||
interface PluginRuntimePort {
|
interface PluginRuntimePort {
|
||||||
emit?: PluginLifecycle["emit"];
|
emit?: PluginLifecycle["emit"];
|
||||||
|
drainEvents?: PluginRuntime["drainEvents"];
|
||||||
before?: PluginLifecycle["before"];
|
before?: PluginLifecycle["before"];
|
||||||
catalog: PluginRuntime["catalog"];
|
catalog: PluginRuntime["catalog"];
|
||||||
invoke(pluginId: string, method: string, input: unknown): Promise<unknown>;
|
invoke(pluginId: string, method: string, input: unknown): Promise<unknown>;
|
||||||
@@ -119,6 +120,10 @@ export class PluginService {
|
|||||||
this.runtime.emit?.(name, event);
|
this.runtime.emit?.(name, event);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
async drainEvents(): Promise<void> {
|
||||||
|
await this.runtime.drainEvents?.();
|
||||||
|
}
|
||||||
|
|
||||||
readonly before: PluginLifecycle["before"] = async (name, request) => {
|
readonly before: PluginLifecycle["before"] = async (name, request) => {
|
||||||
if (this.runtime.before) {
|
if (this.runtime.before) {
|
||||||
return this.runtime.before(name, request);
|
return this.runtime.before(name, request);
|
||||||
|
|||||||
@@ -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 { tmpdir } from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { fileURLToPath } from "node:url";
|
import { fileURLToPath } from "node:url";
|
||||||
@@ -6,7 +6,7 @@ import { expect, test } from "vitest";
|
|||||||
import { DaemonClient } from "../test-utils/daemon-client.js";
|
import { DaemonClient } from "../test-utils/daemon-client.js";
|
||||||
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.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 directory = await mkdtemp(path.join(tmpdir(), "paseo-archive-hooks-"));
|
||||||
const daemon = await createTestPaseoDaemon({ daemonVersion: "0.8.0" });
|
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 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,
|
workspaceId: workspace.id,
|
||||||
});
|
});
|
||||||
await daemon.daemon.agentManager.closeAgent(closed.id);
|
await daemon.daemon.agentManager.closeAgent(closed.id);
|
||||||
|
await daemon.daemon.agentManager.closeAgent(closed.id);
|
||||||
expect(daemon.daemon.agentManager.getAgent(closed.id)).toBeNull();
|
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 client.archiveWorkspace(workspace.id);
|
||||||
await expect
|
await expect
|
||||||
.poll(async () => {
|
.poll(async () => {
|
||||||
@@ -48,6 +57,50 @@ test("workspace archive publishes agent archive hooks for both live and closed a
|
|||||||
.sort();
|
.sort();
|
||||||
})
|
})
|
||||||
.toEqual([live.id, closed.id].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 {
|
} finally {
|
||||||
await client.close();
|
await client.close();
|
||||||
await daemon.close();
|
await daemon.close();
|
||||||
|
|||||||
@@ -280,6 +280,7 @@ test.skipIf(process.platform !== "linux")(
|
|||||||
})
|
})
|
||||||
.toEqual([
|
.toEqual([
|
||||||
"agent.archived",
|
"agent.archived",
|
||||||
|
"agent.closed",
|
||||||
"agent.created",
|
"agent.created",
|
||||||
"agent.permission_requested",
|
"agent.permission_requested",
|
||||||
"agent.permission_resolved",
|
"agent.permission_resolved",
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ import type { PersistedWorkspaceRecord } from "../../workspace-registry.js";
|
|||||||
|
|
||||||
export const lifecycleEventNames = [
|
export const lifecycleEventNames = [
|
||||||
"agent.created",
|
"agent.created",
|
||||||
|
"agent.closed",
|
||||||
"agent.turn_started",
|
"agent.turn_started",
|
||||||
"agent.turn_ended",
|
"agent.turn_ended",
|
||||||
"agent.permission_requested",
|
"agent.permission_requested",
|
||||||
|
|||||||
@@ -295,6 +295,7 @@ async function resolveEntryPaths(directory: string): Promise<{
|
|||||||
|
|
||||||
export class PluginRuntime {
|
export class PluginRuntime {
|
||||||
private readonly plugins = new Map<string, LoadedPlugin>();
|
private readonly plugins = new Map<string, LoadedPlugin>();
|
||||||
|
private readonly pendingEvents = new Set<Promise<void>>();
|
||||||
private readonly logTails = new Map<string, PluginLogTail>();
|
private readonly logTails = new Map<string, PluginLogTail>();
|
||||||
private readonly logger: pino.Logger;
|
private readonly logger: pino.Logger;
|
||||||
private readonly spawnChild: () => PluginChild;
|
private readonly spawnChild: () => PluginChild;
|
||||||
@@ -515,22 +516,30 @@ export class PluginRuntime {
|
|||||||
if (!loaded.hooks.events.includes(name)) {
|
if (!loaded.hooks.events.includes(name)) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
void this.request(loaded, {
|
const request = this.request(loaded, {
|
||||||
type: "hook",
|
type: "hook",
|
||||||
requestId: randomUUID(),
|
requestId: randomUUID(),
|
||||||
kind: "event",
|
kind: "event",
|
||||||
name,
|
name,
|
||||||
input: event,
|
input: event,
|
||||||
}).catch((error) => {
|
})
|
||||||
this.appendLog(
|
.then(() => undefined)
|
||||||
loaded.id,
|
.catch((error) => {
|
||||||
"stderr",
|
this.appendLog(
|
||||||
`Lifecycle hook ${name} failed: ${describeError(error)}`,
|
loaded.id,
|
||||||
);
|
"stderr",
|
||||||
});
|
`Lifecycle hook ${name} failed: ${describeError(error)}`,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
this.pendingEvents.add(request);
|
||||||
|
void request.finally(() => this.pendingEvents.delete(request));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async drainEvents(): Promise<void> {
|
||||||
|
await Promise.all(this.pendingEvents);
|
||||||
|
}
|
||||||
|
|
||||||
async before<Name extends keyof PluginBeforeRequests>(
|
async before<Name extends keyof PluginBeforeRequests>(
|
||||||
name: Name,
|
name: Name,
|
||||||
input: PluginBeforeRequests[Name],
|
input: PluginBeforeRequests[Name],
|
||||||
|
|||||||
@@ -29,6 +29,10 @@ export default function contribute(server: PluginServerContext) {
|
|||||||
logHook("agent.created", event);
|
logHook("agent.created", event);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
server.on("agent.closed", (event) => {
|
||||||
|
logHook("agent.closed", event);
|
||||||
|
});
|
||||||
|
|
||||||
server.on("agent.turn_started", (event) => {
|
server.on("agent.turn_started", (event) => {
|
||||||
logHook("agent.turn_started", event);
|
logHook("agent.turn_started", event);
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -654,6 +654,7 @@ plans, and mode changes; requesting permission does not end the turn.
|
|||||||
| Name | Event fields | Trigger |
|
| Name | Event fields | Trigger |
|
||||||
| ---------------------------- | ---------------------------------------- | -------------------------------------------------- |
|
| ---------------------------- | ---------------------------------------- | -------------------------------------------------- |
|
||||||
| `agent.created` | `agent` | Ordinary creation finishes; excludes import/resume |
|
| `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_started` | `agent`, `turnId` | Live turn starts |
|
||||||
| `agent.turn_ended` | `agent`, `turnId`, `outcome`, `timeline` | Live turn completes, fails, or is canceled |
|
| `agent.turn_ended` | `agent`, `turnId`, `outcome`, `timeline` | Live turn completes, fails, or is canceled |
|
||||||
| `agent.permission_requested` | `agent`, `request` | Permission or question becomes pending |
|
| `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 |
|
| `workspace.archived` | `workspace` | Archive state is saved |
|
||||||
|
|
||||||
Agent events exclude internal utility agents. Archive events can precede runtime/worktree cleanup;
|
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.
|
`workspace.created` is not a setup barrier before agent startup.
|
||||||
|
|
||||||
**Shared payload shapes** (`@getpaseo/plugin/server`):
|
**Shared payload shapes** (`@getpaseo/plugin/server`):
|
||||||
|
|||||||
Reference in New Issue
Block a user