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, }); }