From 052004a97d7c41677b74c6d53ffa9afc457efcd9 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 14:03:30 +0100 Subject: [PATCH] feat(plugins): reopen idle agent sessions after hook changes Defer automatic plugin changes until the next idle prompt so running turns keep steering. Expose session reload separately from snapshot refresh in the SDK. --- packages/client/src/index.test.ts | 63 ++++++ packages/client/src/index.ts | 11 + packages/plugin/src/boundaries.test.ts | 19 ++ .../server/src/server/agent/agent-manager.ts | 30 +++ .../server/src/server/agent/agent-prompt.ts | 2 + .../server/agent/create-agent/create.test.ts | 1 + .../src/server/agent/mcp-server.test.ts | 1 + packages/server/src/server/bootstrap.ts | 6 + .../server/plugins/agent-tools.e2e.test.ts | 207 +++++++++++++++++- packages/server/src/server/plugins/index.ts | 5 + packages/server/src/server/plugins/runtime.ts | 23 ++ .../server/src/server/schedule/service.ts | 1 + .../server/test-utils/fake-agent-client.ts | 13 ++ 13 files changed, 377 insertions(+), 5 deletions(-) diff --git a/packages/client/src/index.test.ts b/packages/client/src/index.test.ts index a7bb17121c..1b11a3032f 100644 --- a/packages/client/src/index.test.ts +++ b/packages/client/src/index.test.ts @@ -1770,3 +1770,66 @@ test("qualified parents preserve foreign identity without resolving an equal loc await client.close(); expect(localParent.address).toEqual({ serverId: "srv_sdk_test", agentId: "same-parent" }); }); + +test("agent reload reopens through the daemon RPC then returns the observed snapshot; refresh remains a read", async () => { + const { client, ws } = await connectClient(); + try { + const agent = client.agents.ref("agent_sdk"); + const snapshot = createAgent(); + const reloaded = agent.reload({ requestId: "reload-sdk" }); + expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({ + type: "refresh_agent_request", + agentId: "agent_sdk", + requestId: "reload-sdk", + }); + ws.message( + sessionMessage({ + type: "status", + payload: { + status: "agent_refreshed", + requestId: "reload-sdk", + agentId: "agent_sdk", + timelineSize: 0, + }, + }), + ); + await expect + .poll(() => parseSentSessionMessage(ws.sent.at(-1)).type) + .toBe("fetch_agent_request"); + const fetched = parseSentSessionMessage(ws.sent.at(-1)); + ws.message( + sessionMessage({ + type: "fetch_agent_response", + payload: { + requestId: fetched.requestId, + agentId: "agent_sdk", + agent: snapshot, + project: null, + error: null, + }, + }), + ); + expect(await reloaded).toEqual({ agent: snapshot, project: null }); + expect(agent.current()).toEqual(snapshot); + const read = agent.refresh("snapshot-read"); + expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({ + type: "fetch_agent_request", + requestId: "snapshot-read", + }); + ws.message( + sessionMessage({ + type: "fetch_agent_response", + payload: { + requestId: "snapshot-read", + agentId: "agent_sdk", + agent: snapshot, + project: null, + error: null, + }, + }), + ); + expect(await read).toEqual({ agent: snapshot, project: null }); + } finally { + await client.close(); + } +}); diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 455d377634..1437068f95 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -369,6 +369,11 @@ export interface PaseoAgentHandle { readonly timeline: PaseoAgentTimelineHandle; current(): PaseoAgent | null; refresh(requestId?: string): Promise; + /** + * Reopens the provider session with current configuration, rerunning session-open hooks. + * Interrupts a running turn and returns the refreshed snapshot. + */ + reload(options?: { requestId?: string }): Promise; send(text: string, options?: PaseoAgentSendOptions): Promise; respondToPermission(options: PaseoAgentRespondToPermissionOptions): Promise; /** Sends a prompt and resolves when that turn finishes or needs attention. */ @@ -981,6 +986,12 @@ function createAgentHandleFactory( current = result?.agent ?? null; return result; }, + reload: async (options) => { + await daemonClient.refreshAgent(id, options?.requestId); + const result = await daemonClient.fetchAgent({ agentId: id }); + current = result?.agent ?? null; + return result; + }, send: async (text, options) => { await daemonClient.sendAgentMessage(id, text, options); }, diff --git a/packages/plugin/src/boundaries.test.ts b/packages/plugin/src/boundaries.test.ts index a312cfd71c..017b69bda5 100644 --- a/packages/plugin/src/boundaries.test.ts +++ b/packages/plugin/src/boundaries.test.ts @@ -246,3 +246,22 @@ describe("plugin agent tools SDK boundary", () => { ).toEqual([]); }); }); + +describe("agent session reload SDK boundary", () => { + it("exposes reload separately from snapshot refresh", () => { + expect( + typeErrors(` + import { type PluginServerContext } from "@getpaseo/plugin/server"; + import { type PaseoAgentRefetchResult } from "@getpaseo/client"; + async function reload(server: PluginServerContext, paseo: import("@getpaseo/client").PaseoApi) { + const agent = paseo.agents.ref("chosen-agent"); + const read: PaseoAgentRefetchResult | null = await agent.refresh("read-snapshot"); + const reopened: PaseoAgentRefetchResult | null = await agent.reload({ requestId: "reload-session" }); + await agent.reload(); + // @ts-expect-error reload options can grow without a positional signature change + await agent.reload("reload-session"); + } + `), + ).toEqual([]); + }); +}); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 9cc8534d60..1342576f7f 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -769,6 +769,8 @@ export class AgentManager { private readonly agentStreamCoalescer: AgentStreamCoalescer; private mcpBaseUrl: string | null; private readonly mcpAuthToken: string | null; + private sessionOpenGeneration = 0; + private readonly sessionOpenGenerations = new Map(); private readonly attachedPluginTools = new Map(); private readonly pluginAgentTools: PluginAgentToolsRuntime | undefined; private paseoToolsEnabled = true; @@ -893,6 +895,30 @@ export class AgentManager { this.paseoToolCatalogFactory = factory; } + invalidatePluginSessions(): void { + this.sessionOpenGeneration++; + } + + async reloadStalePluginSession(agentId: string): Promise { + if (!this.needsPluginSessionReload(agentId)) return; + await this.trackAgentRegistrationOperation( + this.runLifecycleMutation(agentId, async () => { + if (!this.needsPluginSessionReload(agentId)) return; + await this.reloadAgentSessionInternal(agentId); + }), + ); + } + + private needsPluginSessionReload(agentId: string): boolean { + const agent = this.getAgent(agentId); + return ( + agent !== null && + agent.lifecycle !== "closed" && + this.sessionOpenGenerations.get(agentId) !== this.sessionOpenGeneration && + !this.hasInFlightRun(agentId) + ); + } + getPluginTools(agentId: string): readonly AttachedAgentTool[] { return (this.attachedPluginTools.get(agentId) ?? []).filter((tool) => this.pluginAgentTools?.isToolPluginAvailable(tool.pluginId, tool.pluginInstanceId), @@ -921,6 +947,7 @@ export class AgentManager { private detachPluginTools(agentId: string): void { this.attachedPluginTools.delete(agentId); + this.sessionOpenGenerations.delete(agentId); this.pluginAgentTools?.detachTools(agentId); } @@ -5489,6 +5516,8 @@ export class AgentManager { labels?: Record; }, ): Promise { + // A plugin changing during this open must still make the resulting session stale. + const generation = this.sessionOpenGeneration; this.detachPluginTools(agentId); if (this.pluginLifecycle) { const request = { @@ -5535,6 +5564,7 @@ export class AgentManager { throw error; } } + this.sessionOpenGenerations.set(agentId, generation); return context; } diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 08aca8e626..dde5c02bbd 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -15,6 +15,7 @@ export type AgentRunController = Pick< | "getAgent" | "tryRunOutOfBand" | "hasInFlightRun" + | "reloadStalePluginSession" | "replaceAgentRun" | "steerOrReplaceActiveTurn" | "streamAgent" @@ -115,6 +116,7 @@ export async function startAgentRun( return { disposition: "out_of_band" }; } try { + await agentManager.reloadStalePluginSession(agentId); return await startAgentRunInner(agentManager, agentId, prompt, logger, options); } catch (error) { if (!isStaleProviderSessionError(error)) throw error; diff --git a/packages/server/src/server/agent/create-agent/create.test.ts b/packages/server/src/server/agent/create-agent/create.test.ts index 5d2c9fddc1..daa5ba9ec1 100644 --- a/packages/server/src/server/agent/create-agent/create.test.ts +++ b/packages/server/src/server/agent/create-agent/create.test.ts @@ -69,6 +69,7 @@ test("session create forwards clientMessageId to the initial prompt run options" getAgent: vi.fn(() => snapshot), tryRunOutOfBand: vi.fn(() => false), hasInFlightRun: vi.fn(() => false), + reloadStalePluginSession: async () => {}, streamAgent, waitForAgentRunStart: vi.fn(async () => undefined), } as unknown as Parameters[0]["agentManager"], diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index 663f195f5a..f2c7da8574 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -271,6 +271,7 @@ function buildAgentManagerSpies() { appendTimelineItem: vi.fn().mockResolvedValue(undefined), emitLiveTimelineItem: vi.fn().mockResolvedValue(undefined), getPluginTools: () => [], + reloadStalePluginSession: async () => {}, hasInFlightRun: vi.fn().mockReturnValue(false), steerOrReplaceActiveTurn: async () => ({ status: "inactive" as const }), tryRunOutOfBand: vi.fn().mockReturnValue(false), diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index f0b2005bdc..2bbde7de34 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1016,6 +1016,10 @@ export async function createPaseoDaemon( const unsubscribePluginProviders = pluginRuntime.subscribeProviderRegistrations(syncPluginProviders); + const unsubscribePluginSessionChanges = pluginRuntime.subscribeSessionOpenChanges(() => { + agentManager.invalidatePluginSessions(); + }); + const detachAgentStoragePersistence = attachAgentStoragePersistence( logger, agentManager, @@ -1915,6 +1919,7 @@ export async function createPaseoDaemon( localCredential = null; await deleteLocalCredential(config.paseoHome); unsubscribePluginProviders(); + unsubscribePluginSessionChanges(); await pluginRuntime.stopAllPlugins().catch(() => undefined); await serviceProxy.stopStandalone().catch(() => undefined); await agentProviderRuntime.shutdown().catch(() => undefined); @@ -1934,6 +1939,7 @@ export async function createPaseoDaemon( // that is still open. Plugins themselves are stopped once every session // they serve has been closed, further down. unsubscribePluginProviders(); + unsubscribePluginSessionChanges(); await hubRelationships.stop(); workspaceReconciliation.dispose(); scriptHealthMonitor.stop(); diff --git a/packages/server/src/server/plugins/agent-tools.e2e.test.ts b/packages/server/src/server/plugins/agent-tools.e2e.test.ts index 8d78c210ed..48f4ffc490 100644 --- a/packages/server/src/server/plugins/agent-tools.e2e.test.ts +++ b/packages/server/src/server/plugins/agent-tools.e2e.test.ts @@ -5,7 +5,10 @@ import { expect, test } from "vitest"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { DaemonClient } from "../test-utils/daemon-client.js"; -import { createTestAgentClient } from "../test-utils/fake-agent-client.js"; +import { + createTestAgentClient, + type TestAgentClientOptions, +} from "../test-utils/fake-agent-client.js"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import { BuiltinPluginLoader } from "./builtin/index.js"; import { serializePaseoToolInputParameters } from "../agent/tools/paseo-tool-serialization.js"; @@ -13,11 +16,30 @@ import type { AgentClient, AgentLaunchContext, AgentSessionConfig, + AgentPromptInput, } from "../agent/agent-sdk-types.js"; -function recordingProvider(native: boolean) { - const inner = createTestAgentClient("codex", { supportsMcpServers: true }); +function recordingProvider(native: boolean, callbacks: TestAgentClientOptions = {}) { const launches: { config: AgentSessionConfig; context?: AgentLaunchContext }[] = []; + const turns: { prompt: AgentPromptInput; launchCount: number }[] = []; + const steers: AgentPromptInput[] = []; + const events = { interrupts: 0 }; + const inner = createTestAgentClient("codex", { + ...callbacks, + supportsMcpServers: true, + onStartTurn: (prompt, config) => { + turns.push({ prompt, launchCount: launches.length }); + callbacks.onStartTurn?.(prompt, config); + }, + steerActiveTurn: async (prompt) => { + steers.push(prompt); + return { status: "accepted" }; + }, + onInterrupt: () => { + events.interrupts++; + callbacks.onInterrupt?.(); + }, + }); const provider: AgentClient = { provider: inner.provider, capabilities: { ...inner.capabilities, supportsNativePaseoTools: native }, @@ -32,7 +54,7 @@ function recordingProvider(native: boolean) { isAvailable: () => inner.isAvailable(), fetchCatalog: (options) => inner.fetchCatalog(options), }; - return { provider, launches }; + return { provider, launches, turns, steers, events }; } const pluginSource = ` @@ -61,9 +83,10 @@ async function setup( native = false, source = pluginSource, policy?: { enabled?: boolean; disabledTools?: string[] }, + providerOptions?: TestAgentClientOptions, ) { const directory = await mkdtemp(path.join(tmpdir(), "paseo-agent-tools-")); - const recorder = recordingProvider(native); + const recorder = recordingProvider(native, providerOptions); const daemon = await createTestPaseoDaemon({ daemonVersion: "0.13.2", agentClients: { codex: recorder.provider }, @@ -510,3 +533,177 @@ test("a session hook missing tools reports a schema issue", async () => { await h.close(); } }, 60_000); + +async function sendAndFinish(h: Awaited>, agentId: string, text: string) { + const previous = h.turns.length; + await h.client.sendMessage(agentId, text, { activeTurnBehavior: "steer" }); + await expect.poll(() => h.turns.length).toBe(previous + 1); + await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(agentId)).toBe(false); +} + +test("an idle agent reopens once before its next prompt after a session-hook plugin reload", async () => { + const h = await setup(); + try { + const b = await h.create(); + const mcp = await h.mcp(); + expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo"); + await writeFile( + path.join(h.directory, "tools-test", "index.server.ts"), + pluginSource.replace(' if (request.labels.x !== "y") return;', ""), + ); + await h.client.reloadPlugin("tools-test"); + expect(h.launches).toHaveLength(1); + await sendAndFinish(h, b.id, "first prompt"); + expect(h.launches).toHaveLength(2); + expect(h.turns.map((turn) => turn.launchCount)).toEqual([2]); + expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo"); + await sendAndFinish(h, b.id, "second prompt"); + expect(h.launches).toHaveLength(2); + expect(h.turns.map((turn) => turn.launchCount)).toEqual([2, 2]); + expect(h.events.interrupts).toBe(0); + } finally { + await h.close(); + } +}, 60_000); + +test("a busy agent is steered after plugin reload and reopens only at its next idle prompt", async () => { + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + const h = await setup(true, pluginSource, undefined, { beforeTurnComplete: async () => blocked }); + try { + const a = await h.create({ x: "y" }); + await h.client.sendMessage(a.id, "busy prompt", { activeTurnBehavior: "steer" }); + await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(a.id)).toBe(true); + await expect + .poll(() => h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId) + .not.toBeNull(); + const turnId = h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId; + await h.client.reloadPlugin("tools-test"); + await h.client.sendMessage(a.id, "steering prompt", { activeTurnBehavior: "steer" }); + expect(h.steers).toHaveLength(1); + expect(h.launches).toHaveLength(1); + expect(h.events.interrupts).toBe(0); + expect(h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId).toBe(turnId); + release(); + await expect.poll(() => h.daemon.daemon.agentManager.hasInFlightRun(a.id)).toBe(false); + await sendAndFinish(h, a.id, "idle prompt"); + expect(h.launches).toHaveLength(2); + expect(h.events.interrupts).toBe(0); + expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe( + "plugin_echo", + ); + await sendAndFinish(h, a.id, "stable prompt"); + expect(h.launches).toHaveLength(2); + } finally { + release(); + await h.close(); + } +}, 60_000); + +const reloadRpcSource = pluginSource + .replace("import { z }", 'import { defineRpc } from "@getpaseo/plugin";\nimport { z }') + .replace( + " return () => {};", + ` + server.handle(defineRpc({ name: "reload_agent", input: z.object({ id: z.string() }), output: z.object({ id: z.string() }) }), async ({ id }, { paseo }) => { + const agent = paseo.agents.ref(id); + await agent.refresh(); + await agent.reload(); + return { id: agent.current().id }; + }); + return () => {}; +`, + ); + +test("a plugin can reload an agent through the SDK and rerun session hooks with reason refresh", async () => { + const h = await setup(true, reloadRpcSource); + try { + const a = await h.create({ x: "y" }); + expect(h.launches).toHaveLength(1); + expect(await h.client.invokePluginRpc("tools-test", "reload_agent", { id: a.id })).toEqual({ + id: a.id, + }); + expect(h.launches).toHaveLength(2); + expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO).reason).toBe("refresh"); + expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe( + "plugin_echo", + ); + } finally { + await h.close(); + } +}, 60_000); + +test("starting and stopping a session-hook plugin each causes one idle reopen", async () => { + const h = await setup(true); + try { + const a = await h.create({ x: "y" }); + await h.install( + "added-tools", + pluginSource.replace('name: "plugin_echo"', 'name: "added_echo"'), + ); + await sendAndFinish(h, a.id, "after start"); + expect(h.launches).toHaveLength(2); + expect(h.launches.at(-1)!.context!.paseoTools!.getTool("added_echo")?.name).toBe("added_echo"); + await h.client.disablePlugin("added-tools"); + await sendAndFinish(h, a.id, "after stop"); + expect(h.launches).toHaveLength(3); + expect(h.launches.at(-1)!.context!.paseoTools!.getTool("added_echo")).toBeUndefined(); + await sendAndFinish(h, a.id, "stable after stop"); + expect(h.launches).toHaveLength(3); + expect(h.events.interrupts).toBe(0); + } finally { + await h.close(); + } +}, 60_000); + +test("plugins without a session-open hook do not make agent sessions stale", async () => { + const h = await setup(true); + try { + const a = await h.create({ x: "y" }); + await h.install( + "no-session-hook", + "export default function contribute(server) { server.on('agent.created', () => {}); return () => {}; }", + ); + await sendAndFinish(h, a.id, "after unrelated start"); + expect(h.launches).toHaveLength(1); + await h.client.reloadPlugin("no-session-hook"); + await sendAndFinish(h, a.id, "after unrelated reload"); + expect(h.launches).toHaveLength(1); + await h.client.disablePlugin("no-session-hook"); + await sendAndFinish(h, a.id, "after unrelated stop"); + expect(h.launches).toHaveLength(1); + } finally { + await h.close(); + } +}, 60_000); + +test("explicit SDK reload interrupts a running turn before reopening its session", async () => { + let release!: () => void; + const blocked = new Promise((resolve) => { + release = resolve; + }); + const h = await setup(true, reloadRpcSource, undefined, { + beforeTurnComplete: async () => blocked, + onInterrupt: () => release(), + }); + try { + const a = await h.create({ x: "y" }); + await h.client.sendMessage(a.id, "running before explicit reload", { + activeTurnBehavior: "steer", + }); + await expect + .poll(() => h.daemon.daemon.agentManager.getAgent(a.id)!.activeTurnId) + .not.toBeNull(); + expect(await h.client.invokePluginRpc("tools-test", "reload_agent", { id: a.id })).toEqual({ + id: a.id, + }); + expect(h.events.interrupts).toBe(1); + expect(h.launches).toHaveLength(2); + expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO).reason).toBe("refresh"); + } finally { + release(); + await h.close(); + } +}, 60_000); diff --git a/packages/server/src/server/plugins/index.ts b/packages/server/src/server/plugins/index.ts index 6b785b4e51..8385dc2122 100644 --- a/packages/server/src/server/plugins/index.ts +++ b/packages/server/src/server/plugins/index.ts @@ -38,6 +38,7 @@ import type { PluginUsageSourceMetadata } from "./plugin-process-protocol.js"; const BUILTIN_PROVIDER_ID_SET: ReadonlySet = new Set(BUILTIN_PROVIDER_IDS); interface PluginRuntimePort { + subscribeSessionOpenChanges?: PluginRuntime["subscribeSessionOpenChanges"]; isToolPluginAvailable?: PluginRuntime["isToolPluginAvailable"]; callTool?: PluginRuntime["callTool"]; detachTools?: PluginRuntime["detachTools"]; @@ -135,6 +136,10 @@ export class PluginService { this.runtime.emit?.(name, event); }; + subscribeSessionOpenChanges(listener: () => void): () => void { + return this.runtime.subscribeSessionOpenChanges?.(listener) ?? (() => {}); + } + readonly isToolPluginAvailable: PluginAgentToolsRuntime["isToolPluginAvailable"] = ( pluginId, pluginInstanceId, diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index dbea63c7af..f5a644ef9d 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -310,6 +310,7 @@ export class PluginRuntime { private readonly spawnChild: () => PluginChild; private sessionHost: PluginPaseoSessionHost | null; private readonly listeners = new Set<(pluginId: string, error?: string) => void>(); + private readonly sessionOpenListeners = new Set<() => void>(); constructor( logger: pino.Logger, @@ -332,6 +333,16 @@ export class PluginRuntime { return () => this.listeners.delete(listener); } + subscribeSessionOpenChanges(listener: () => void): () => void { + this.sessionOpenListeners.add(listener); + return () => this.sessionOpenListeners.delete(listener); + } + + private notifySessionOpenChange(loaded: LoadedPlugin): void { + if (!loaded.hooks.before.includes("agent.session_open")) return; + for (const listener of this.sessionOpenListeners) listener(); + } + async startPlugin( pluginId: string, configuredPath: string, @@ -348,6 +359,7 @@ export class PluginRuntime { throw new Error(`Plugin start cancelled: ${pluginId}`); } this.plugins.set(pluginId, loaded); + this.notifySessionOpenChange(loaded); this.appendLog(pluginId, "stdout", "[paseo] Plugin ready"); } @@ -371,6 +383,7 @@ export class PluginRuntime { clientBundle: bundles.clientBundle ?? "", }); this.plugins.set(input.id, loaded); + this.notifySessionOpenChange(loaded); this.appendLog(input.id, "stdout", "[paseo] Plugin ready"); } @@ -386,6 +399,7 @@ export class PluginRuntime { const loaded = this.plugins.get(pluginId); if (!loaded) return false; this.plugins.delete(pluginId); + this.notifySessionOpenChange(loaded); this.rejectPending(loaded, `Plugin stopped: ${pluginId}`); await this.stopPlugin(loaded); return true; @@ -715,6 +729,7 @@ export class PluginRuntime { const loaded = [...this.plugins.values()]; this.plugins.clear(); for (const plugin of loaded) { + this.notifySessionOpenChange(plugin); this.rejectPending(plugin, `Plugin stopped: ${plugin.id}`); } await Promise.all(loaded.map((plugin) => this.stopPlugin(plugin))); @@ -946,7 +961,14 @@ export class PluginRuntime { private handleChildMessage(loaded: LoadedPlugin, message: PluginProcessMessage): void { if (message.type === "hooks.changed") { + const hadSessionOpen = loaded.hooks.before.includes("agent.session_open"); loaded.hooks = message.hooks; + if ( + this.plugins.get(loaded.id) === loaded && + hadSessionOpen !== loaded.hooks.before.includes("agent.session_open") + ) { + for (const listener of this.sessionOpenListeners) listener(); + } return; } if (message.type === "settings.changed") { @@ -1260,6 +1282,7 @@ export class PluginRuntime { const wasPublished = this.plugins.get(loaded.id) === loaded; if (wasPublished) { this.plugins.delete(loaded.id); + this.notifySessionOpenChange(loaded); } this.rejectPending(loaded, `Plugin process exited: ${loaded.id}`); for (const state of loaded.providerConnections.values()) { diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 3b462cc5b1..6121b7e6bd 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -226,6 +226,7 @@ type ScheduleAgentManager = Pick< AgentRunController, | "getAgent" | "reloadAgentSession" + | "reloadStalePluginSession" | "tryRunOutOfBand" | "hasInFlightRun" | "replaceAgentRun" diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index 1c7127d144..42ac0c7c79 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -60,6 +60,8 @@ interface FakeAgentSessionOptions { closeSession?: () => Promise; onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void; beforeTurnComplete?: (prompt: AgentPromptInput) => Promise; + steerActiveTurn?: AgentSession["steerActiveTurn"]; + onInterrupt?: () => void; historyEvents?: readonly AgentStreamEvent[]; } @@ -68,6 +70,8 @@ export interface TestAgentClientOptions { closeSession?: () => Promise; onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void; beforeTurnComplete?: (prompt: AgentPromptInput) => Promise; + steerActiveTurn?: AgentSession["steerActiveTurn"]; + onInterrupt?: () => void; supportsMcpServers?: boolean; historyEvents?: readonly AgentStreamEvent[]; } @@ -340,6 +344,8 @@ class FakeAgentSession implements AgentSession { private nextTurnOrdinal = 0; private activeForegroundTurnId: string | null = null; + readonly steerActiveTurn: AgentSession["steerActiveTurn"]; + private readonly onInterrupt: (() => void) | undefined; private readonly closeSession: (() => Promise) | undefined; private readonly onStartTurn: TestAgentClientOptions["onStartTurn"]; private readonly beforeTurnComplete: ((prompt: AgentPromptInput) => Promise) | undefined; @@ -354,6 +360,8 @@ class FakeAgentSession implements AgentSession { this.config = options.config; this.id = options.sessionId ?? randomUUID(); this.memoryMarker = options.memoryMarker ?? null; + this.steerActiveTurn = options.steerActiveTurn; + this.onInterrupt = options.onInterrupt; this.closeSession = options.closeSession; this.onStartTurn = options.onStartTurn; this.beforeTurnComplete = options.beforeTurnComplete; @@ -896,6 +904,7 @@ class FakeAgentSession implements AgentSession { } async interrupt(): Promise { + this.onInterrupt?.(); this.interruptSignal.resolve(); } @@ -1234,6 +1243,8 @@ class FakeAgentClient implements AgentClient { closeSession: this.options.closeSession, onStartTurn: this.options.onStartTurn, beforeTurnComplete: this.options.beforeTurnComplete, + steerActiveTurn: this.options.steerActiveTurn, + onInterrupt: this.options.onInterrupt, historyEvents: this.options.historyEvents, }); } @@ -1261,6 +1272,8 @@ class FakeAgentClient implements AgentClient { closeSession: this.options.closeSession, onStartTurn: this.options.onStartTurn, beforeTurnComplete: this.options.beforeTurnComplete, + steerActiveTurn: this.options.steerActiveTurn, + onInterrupt: this.options.onInterrupt, historyEvents: this.options.historyEvents, }); }