feat(plugins): attach tools to individual agent sessions

This commit is contained in:
Mohamed Boudra
2026-10-26 14:01:06 +01:00
parent 678fe308db
commit f5e311ffd8
18 changed files with 1197 additions and 40 deletions
+38 -1
View File
@@ -92,7 +92,12 @@ function typeErrors(code: string): string[] {
}
describe("plugin SDK import boundaries", () => {
it("publishes process capabilities without their Windows implementation helpers", () => {
expect(Object.keys(server).sort()).toEqual(["execCommand", "spawnProcess", "terminateProcess"]);
expect(Object.keys(server).sort()).toEqual([
"defineAgentTool",
"execCommand",
"spawnProcess",
"terminateProcess",
]);
});
it("classifies every published entry", () => {
@@ -209,3 +214,35 @@ describe("plugin panel SDK boundary", () => {
).toEqual([]);
});
});
describe("plugin agent tools SDK boundary", () => {
it("lets a session hook attach typed tools without exposing their transport", () => {
expect(
typeErrors(`
import { z } from "zod";
import { defineAgentTool, type PluginServerContext, type AgentToolResult, type AgentToolContext, type AgentToolDefinition } from "@getpaseo/plugin/server";
const tool: AgentToolDefinition = defineAgentTool({
name: "create_issue",
description: "Create an issue",
input: z.object({ title: z.string(), count: z.string().transform(Number) }),
async execute(input, { agent, paseo, signal }) {
const count: number = input.count;
// @ts-expect-error input is inferred from the schema
input.missing;
return { text: agent.id + input.title, structuredContent: { count } };
},
});
export default function contribute(server: PluginServerContext) {
server.before("agent.session_open", ({ request }) => {
const title: string | null = request.title;
// @ts-expect-error labels are read-only
request.labels.x = "changed";
if (request.labels.x !== "y") return;
return { ...request, tools: [...request.tools, tool] };
});
return () => {};
}
`),
).toEqual([]);
});
});
+39
View File
@@ -0,0 +1,39 @@
import type { PaseoApi } from "@getpaseo/client";
import type { z } from "zod";
import type { PluginHookAgent } from "./lifecycle.js";
export interface AgentToolContext {
agent: PluginHookAgent;
paseo: PaseoApi;
signal: AbortSignal;
}
export interface AgentToolResult {
text?: string;
structuredContent?: unknown;
isError?: boolean;
}
export interface AgentToolDefinition {
name: string;
title?: string;
description: string;
input: z.ZodType;
execute(input: unknown, context: AgentToolContext): AgentToolResult | Promise<AgentToolResult>;
}
export function defineAgentTool<Input extends z.ZodType>(tool: {
name: string;
title?: string;
description: string;
input: Input;
execute(
input: z.output<Input>,
context: AgentToolContext,
): AgentToolResult | Promise<AgentToolResult>;
}): AgentToolDefinition {
if (!/^[a-z][a-z0-9_]*$/.test(tool.name) || tool.name.length > 64) {
throw new Error(`Invalid agent tool name: ${tool.name}`);
}
return tool;
}
+3
View File
@@ -24,3 +24,6 @@ export type {
} from "./lifecycle.js";
export { spawnProcess, execCommand, terminateProcess } from "./process.js";
export { defineAgentTool } from "./agent-tools.js";
export type { AgentToolDefinition, AgentToolResult, AgentToolContext } from "./agent-tools.js";
+4
View File
@@ -1,3 +1,4 @@
import type { AgentToolDefinition } from "./agent-tools.js";
import type {
AgentPermissionRequest,
AgentPermissionResponse,
@@ -36,6 +37,9 @@ export interface PluginSessionOpenRequest {
cwd: string;
reason: "create" | "resume" | "refresh" | "import";
purpose: "interactive" | "history";
readonly title: string | null;
readonly labels: Readonly<Record<string, string>>;
tools: AgentToolDefinition[];
env: Record<string, string>;
}
@@ -1,3 +1,5 @@
import type { AttachedAgentTool, PluginAgentToolsRuntime } from "../plugins/agent-tools.js";
import type { AgentToolResult } from "@getpaseo/plugin/server";
import type { RuntimeCleanupCandidate } from "./idle-runtime-cleanup/index.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { projectTimelineRows } from "./timeline-projection.js";
@@ -333,6 +335,7 @@ export interface AgentManagerOptions {
terminalManager?: TerminalManager | null;
mcpBaseUrl?: string;
mcpAuthToken?: string;
pluginAgentTools?: PluginAgentToolsRuntime;
paseoToolsEnabled?: boolean;
paseoToolCatalogFactory?: PaseoToolCatalogFactory;
resolvePaseoToolPolicy?: (provider: AgentProvider) => ProviderPaseoToolsPolicy | undefined;
@@ -766,6 +769,8 @@ export class AgentManager {
private readonly agentStreamCoalescer: AgentStreamCoalescer;
private mcpBaseUrl: string | null;
private readonly mcpAuthToken: string | null;
private readonly attachedPluginTools = new Map<string, AttachedAgentTool[]>();
private readonly pluginAgentTools: PluginAgentToolsRuntime | undefined;
private paseoToolsEnabled = true;
private paseoToolCatalogFactory: PaseoToolCatalogFactory | null = null;
private readonly paseoToolPolicies = new Map<string, ProviderPaseoToolsPolicy | undefined>();
@@ -783,6 +788,7 @@ export class AgentManager {
constructor(options: AgentManagerOptions) {
this.pluginLifecycle = options.pluginLifecycle;
this.pluginAgentTools = options.pluginAgentTools;
this.timelineStore = options.timelineStore ?? new InMemoryAgentTimelineStore();
this.idFactory = options.idFactory ?? (() => randomUUID());
this.registry = options?.registry;
@@ -887,6 +893,37 @@ export class AgentManager {
this.paseoToolCatalogFactory = factory;
}
getPluginTools(agentId: string): readonly AttachedAgentTool[] {
return (this.attachedPluginTools.get(agentId) ?? []).filter((tool) =>
this.pluginAgentTools?.isToolPluginAvailable(tool.pluginId, tool.pluginInstanceId),
);
}
async executePluginTool(
tool: AttachedAgentTool,
agentId: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult> {
const agent = this.getAgent(agentId);
if (!agent || agent.lifecycle === "closed") throw new Error(`Agent is not live: ${agentId}`);
if (!this.pluginAgentTools) throw new Error("Plugin tools are unavailable");
const record = await this.registry?.get(agentId);
return this.pluginAgentTools.callTool(
tool.pluginId,
tool.pluginInstanceId,
describeHookAgent({ ...agent, title: record?.title ?? agent.config.title }),
tool.name,
input,
signal,
);
}
private detachPluginTools(agentId: string): void {
this.attachedPluginTools.delete(agentId);
this.pluginAgentTools?.detachTools(agentId);
}
getPaseoToolPolicy(agentId: string): ProviderPaseoToolsPolicy | undefined {
return this.paseoToolPolicies.get(agentId);
}
@@ -1312,7 +1349,13 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
options?.env,
{ reason: "create", purpose: "interactive", workspaceId: options.workspaceId ?? null },
{
reason: "create",
purpose: "interactive",
workspaceId: options.workspaceId ?? null,
title: options.initialTitle ?? storedConfig.title ?? null,
labels: options.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
const createOptions = this.buildCreateSessionOptions(options);
@@ -1434,6 +1477,8 @@ export class AgentManager {
reason: "resume",
purpose,
workspaceId: options?.workspaceId ?? null,
title: record?.title ?? mergedConfig.title ?? null,
labels: record?.labels ?? options?.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
@@ -1491,7 +1536,12 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
undefined,
{ reason: "import", purpose: "interactive", workspaceId: input.workspaceId },
{
reason: "import",
purpose: "interactive",
workspaceId: input.workspaceId,
labels: input.labels ?? {},
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
const imported = await client.importSession(
@@ -1586,7 +1636,13 @@ export class AgentManager {
storedConfig.cwd,
paseoToolPolicy,
undefined,
{ reason: "refresh", purpose: "interactive", workspaceId: existing.workspaceId },
{
reason: "refresh",
purpose: "interactive",
workspaceId: existing.workspaceId,
title: await this.resolveInitialPersistedTitle(agentId, existing.config, null),
labels: existing.labels,
},
);
const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext);
if (
@@ -1650,6 +1706,7 @@ export class AgentManager {
throw error;
} finally {
if (!handedToRegistration) {
this.detachPluginTools(agentId);
if (hadPreviousPaseoToolPolicy) {
this.paseoToolPolicies.set(agentId, previousPaseoToolPolicy);
} else {
@@ -1857,6 +1914,7 @@ export class AgentManager {
} catch (error) {
persistError = error;
}
this.detachPluginTools(agentId);
this.emitClosedAgent(closedAgent, { persist: false });
this.logger.trace(
{
@@ -3940,6 +3998,7 @@ export class AgentManager {
private discardRetainedAgentState(agentId: string): void {
this.timelineStore.delete(agentId);
this.paseoToolPolicies.delete(agentId);
this.detachPluginTools(agentId);
for (const event of this.providerSubagents.deleteParent(agentId)) {
this.dispatch({ type: "provider_subagent", event });
}
@@ -5426,10 +5485,13 @@ export class AgentManager {
reason: PluginSessionOpenRequest["reason"];
purpose: PluginSessionOpenRequest["purpose"];
workspaceId?: string | null;
title?: string | null;
labels?: Record<string, string>;
},
): Promise<AgentLaunchContext> {
this.detachPluginTools(agentId);
if (this.pluginLifecycle) {
const request: PluginSessionOpenRequest = {
const request = {
agentId,
provider: client.provider,
cwd,
@@ -5437,9 +5499,14 @@ export class AgentManager {
reason: opening?.reason ?? "resume",
purpose: opening?.purpose ?? "interactive",
env: { ...env },
title: opening?.title ?? null,
labels: opening?.labels ?? {},
tools: [],
};
const transformed = await this.pluginLifecycle.before("agent.session_open", request);
env = transformed.env;
if (request.purpose === "interactive")
this.attachedPluginTools.set(agentId, transformed.tools as AttachedAgentTool[]);
}
const context: AgentLaunchContext = {
agentId,
@@ -5449,16 +5516,24 @@ export class AgentManager {
PASEO_AGENT_CWD: cwd,
},
};
if (
const useNativeTools =
this.paseoToolsEnabled &&
isPaseoToolPolicyEnabled(paseoToolPolicy) &&
client.capabilities.supportsNativePaseoTools &&
this.paseoToolCatalogFactory
client.capabilities.supportsNativePaseoTools;
if (
this.paseoToolCatalogFactory &&
(useNativeTools || this.getPluginTools(agentId).length > 0)
) {
context.paseoTools = await this.paseoToolCatalogFactory({
callerAgentId: agentId,
paseoToolPolicy,
});
try {
const catalog = await this.paseoToolCatalogFactory({
callerAgentId: agentId,
paseoToolPolicy,
});
if (useNativeTools) context.paseoTools = catalog;
} catch (error) {
this.detachPluginTools(agentId);
throw error;
}
}
return context;
}
@@ -270,6 +270,7 @@ function buildAgentManagerSpies() {
hydrateTimelineFromProvider: vi.fn().mockResolvedValue(undefined),
appendTimelineItem: vi.fn().mockResolvedValue(undefined),
emitLiveTimelineItem: vi.fn().mockResolvedValue(undefined),
getPluginTools: () => [],
hasInFlightRun: vi.fn().mockReturnValue(false),
steerOrReplaceActiveTurn: async () => ({ status: "inactive" as const }),
tryRunOutOfBand: vi.fn().mockReturnValue(false),
@@ -617,6 +618,10 @@ class FakeBrowserToolsBroker {
}
class BoundaryAgentManagerFake {
public getPluginTools(): [] {
return [];
}
private readonly agent = createManagedAgent({
id: "agent-1",
cwd: REPO_CWD,
@@ -1,3 +1,4 @@
import { registerPluginTools } from "./plugin-tools.js";
import type { FinishSubscriptions, FinishRegistration } from "../finish-subscriptions/index.js";
import type { AgentMessageDelivery } from "../agent-messages/index.js";
import {
@@ -704,6 +705,7 @@ function createToolCatalog(
}
const localHostOperations = new Map<string, HostOperation>();
const tools = new Map<string, PaseoToolDefinition>();
const builtinNames = new Set<string>();
const hostTools = new Map([
["create_agent", "createAgent"],
["create_workspace", "createWorkspace"],
@@ -738,6 +740,7 @@ function createToolCatalog(
// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Tool handlers are schema-validated at registration boundaries.
handler: (input: any, context: PaseoToolExecutionContext) => Promise<PaseoToolResult>,
) => {
builtinNames.add(name);
if (!isPaseoToolEnabled(options.paseoToolPolicy, name)) {
return;
}
@@ -3272,6 +3275,14 @@ function createToolCatalog(
},
);
registerPluginTools({
agentManager,
callerAgentId,
scope,
tools,
builtinNames,
policy: options.paseoToolPolicy,
});
return toCatalog();
}
@@ -0,0 +1,40 @@
import { z } from "zod";
import type { ProviderPaseoToolsPolicy } from "@getpaseo/protocol/provider-config";
import type { AgentManager } from "../agent-manager.js";
import { isPaseoToolEnabled } from "../paseo-tool-policy.js";
import type { PaseoToolDefinition } from "./types.js";
import { addModelVisibleStructuredContent } from "./paseo-tool-serialization.js";
export function registerPluginTools(options: {
agentManager: AgentManager;
callerAgentId?: string;
scope: "agent" | "top-level";
tools: Map<string, PaseoToolDefinition>;
builtinNames: Set<string>;
policy?: ProviderPaseoToolsPolicy;
}): void {
const { agentManager, callerAgentId, scope, tools, builtinNames, policy } = options;
if (scope !== "agent" || !callerAgentId) return;
for (const tool of agentManager.getPluginTools(callerAgentId)) {
if (builtinNames.has(tool.name)) {
throw new Error(
`Agent tool ${tool.name} from plugin ${tool.pluginId} collides with Paseo tool ${tool.name}`,
);
}
if (!isPaseoToolEnabled(policy, tool.name)) continue;
tools.set(tool.name, {
name: tool.name,
title: tool.title,
description: tool.description,
inputSchema: z.fromJSONSchema(tool.inputSchema),
async handler(input, { signal }) {
const result = await agentManager.executePluginTool(tool, callerAgentId, input, signal);
return addModelVisibleStructuredContent({
content: result.text === undefined ? [] : [{ type: "text", text: result.text }],
structuredContent: result.structuredContent,
isError: result.isError,
});
},
});
}
}
+1
View File
@@ -995,6 +995,7 @@ export async function createPaseoDaemon(
const initialAgentManagerState = providerSnapshotManager.getAgentManagerProviderState();
const agentManager = new AgentManager({
pluginLifecycle: pluginRuntime,
pluginAgentTools: pluginRuntime,
clients: initialAgentManagerState.clients,
providerDefinitions: initialAgentManagerState.providerDefinitions,
registry: agentStorage,
@@ -0,0 +1,512 @@
import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
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 { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { BuiltinPluginLoader } from "./builtin/index.js";
import { serializePaseoToolInputParameters } from "../agent/tools/paseo-tool-serialization.js";
import type {
AgentClient,
AgentLaunchContext,
AgentSessionConfig,
} from "../agent/agent-sdk-types.js";
function recordingProvider(native: boolean) {
const inner = createTestAgentClient("codex", { supportsMcpServers: true });
const launches: { config: AgentSessionConfig; context?: AgentLaunchContext }[] = [];
const provider: AgentClient = {
provider: inner.provider,
capabilities: { ...inner.capabilities, supportsNativePaseoTools: native },
createSession: async (config, context, options) => {
launches.push({ config, context });
return inner.createSession(config, context, options);
},
resumeSession: async (handle, config, context, options) => {
launches.push({ config: config as AgentSessionConfig, context });
return inner.resumeSession(handle, config, context, options);
},
isAvailable: () => inner.isAvailable(),
fetchCatalog: (options) => inner.fetchCatalog(options),
};
return { provider, launches };
}
const pluginSource = `
import { z } from "zod";
import { defineAgentTool } from "@getpaseo/plugin/server";
let calls = 0;
const tool = defineAgentTool({
name: "plugin_echo", description: "Echo for the chosen agent",
input: z.object({ text: z.string().min(1).refine(value => value !== "forbidden", "Forbidden by plugin").describe("Message") }),
async execute(input, { agent, paseo, signal }) {
calls++;
if (input.text === "throw") throw new Error("handler failed");
return { text: JSON.stringify({ agent, input, calls, sdk: typeof paseo.agents.ref, aborted: signal.aborted }) };
},
});
export default function contribute(server) {
server.before("agent.session_open", ({ request }) => {
if (request.labels.x !== "y") return;
return { ...request, env: { ...request.env, OPEN_INFO: JSON.stringify({ title: request.title, labels: request.labels, reason: request.reason }) }, tools: [...request.tools, tool] };
});
return () => {};
}
`;
async function setup(
native = false,
source = pluginSource,
policy?: { enabled?: boolean; disabledTools?: string[] },
) {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-agent-tools-"));
const recorder = recordingProvider(native);
const daemon = await createTestPaseoDaemon({
daemonVersion: "0.13.2",
agentClients: { codex: recorder.provider },
mcpEnabled: true,
providerOverrides: { codex: { paseoTools: policy } },
});
const client = new DaemonClient({
url: `ws://127.0.0.1:${daemon.port}/ws`,
appVersion: "0.13.2",
});
const mcps: Client[] = [];
async function install(id: string, code: string) {
const pluginDirectory = path.join(directory, id);
await mkdir(pluginDirectory);
await writeFile(
path.join(pluginDirectory, "paseo-plugin.json"),
JSON.stringify({ id, requirements: { paseo: ">=0.13.2" } }),
);
await writeFile(path.join(pluginDirectory, "index.server.ts"), code);
await client.installDirectoryPlugin(pluginDirectory);
}
await client.connect();
await client.fetchAgents({ subscribe: {} });
await client.patchDaemonConfig({ pluginsEnabled: true });
try {
await install("tools-test", source);
} catch (error) {
await client.close();
await daemon.close();
await rm(directory, { recursive: true, force: true });
throw error;
}
return {
...recorder,
daemon,
client,
directory,
install,
async create(labels: Record<string, string> = {}) {
return daemon.daemon.agentManager.createAgent(
{ provider: "codex", cwd: directory },
undefined,
{ labels, initialTitle: "Chosen title" },
);
},
async mcp(agentId?: string) {
const server = recorder.launches.at(-1)!.config.mcpServers?.paseo;
let url: string;
if (agentId) url = `http://127.0.0.1:${daemon.port}/mcp/agents?callerAgentId=${agentId}`;
else {
if (server?.type !== "http") throw new Error("Expected HTTP server");
url = server.url;
}
const mcp = new Client({ name: "plugin-tools-test", version: "1" });
mcps.push(mcp);
await mcp.connect(new StreamableHTTPClientTransport(new URL(url)));
return mcp;
},
async close() {
await Promise.all(mcps.map((mcp) => mcp.close()));
await client.close();
await daemon.close();
await rm(directory, { recursive: true, force: true });
},
};
}
test("session hooks attach tools only to matching agents over MCP, with validated input and caller context", async () => {
const h = await setup();
try {
const a = await h.create({ x: "y" });
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO)).toEqual({
title: "Chosen title",
labels: { x: "y" },
reason: "create",
});
const mcpA = await h.mcp();
const listed = (await mcpA.listTools()).tools.find((tool) => tool.name === "plugin_echo");
expect(listed).toMatchObject({
name: "plugin_echo",
inputSchema: {
type: "object",
properties: { text: { type: "string", minLength: 1, description: "Message" } },
required: ["text"],
},
});
const b = await h.create();
const mcpB = await h.mcp();
expect((await mcpB.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
const invalid = await mcpA.callTool({ name: "plugin_echo", arguments: { text: "" } });
expect(invalid.isError).toBe(true);
expect(JSON.stringify(invalid)).toContain("text");
expect(
await mcpA.callTool({ name: "plugin_echo", arguments: { text: "forbidden" } }),
).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("Forbidden by plugin") }],
});
const result = await mcpA.callTool({ name: "plugin_echo", arguments: { text: "hello" } });
expect(result.isError).not.toBe(true);
expect(JSON.parse((result.content as { text: string }[])[0].text)).toMatchObject({
agent: { id: a.id, cwd: h.directory, title: "Chosen title" },
input: { text: "hello" },
calls: 1,
sdk: "function",
aborted: false,
});
expect(a.id).not.toBe(b.id);
expect(
await mcpA.callTool({ name: "plugin_echo", arguments: { text: "throw" } }),
).toMatchObject({ isError: true, content: [{ type: "text", text: "handler failed" }] });
expect(
(await mcpA.callTool({ name: "plugin_echo", arguments: { text: "still alive" } })).isError,
).not.toBe(true);
} finally {
await h.close();
}
}, 60_000);
test("native providers get the same attached schema and execution", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
expect(catalog.getTool("plugin_echo")?.description).toBe("Echo for the chosen agent");
const mcp = await h.mcp(a.id);
const advertised = (await mcp.listTools()).tools.find((tool) => tool.name === "plugin_echo")!;
expect(serializePaseoToolInputParameters(catalog.getTool("plugin_echo")!)).toEqual(
advertised.inputSchema,
);
const result = await catalog.executeTool("plugin_echo", { text: "native" });
expect(JSON.parse(result.content[0].text!)).toMatchObject({
agent: { id: a.id },
input: { text: "native" },
});
} finally {
await h.close();
}
}, 60_000);
test("resume uses the stored title and labels, and close removes the attachment", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
const handle = a.persistence!;
await h.daemon.daemon.agentManager.setTitle(a.id, "Stored title");
await h.daemon.daemon.agentManager.closeAgent(a.id);
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
await expect(catalog.executeTool("plugin_echo", { text: "closed" })).rejects.toThrow(
"Agent is not live",
);
await h.daemon.daemon.agentManager.resumeAgentFromPersistence(handle, a.config, a.id);
expect(JSON.parse(h.launches.at(-1)!.context!.env!.OPEN_INFO)).toEqual({
title: "Stored title",
labels: { x: "y" },
reason: "resume",
});
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")?.name).toBe(
"plugin_echo",
);
} finally {
await h.close();
}
}, 60_000);
test.each([
{
name: "Paseo",
source: pluginSource.replace('name: "plugin_echo"', 'name: "list_agents"'),
other: false,
message: /plugin tools-test.*Paseo tool list_agents/,
},
{
name: "another plugin",
source: pluginSource,
other: true,
message: /plugin tools-test.*plugin other-tools/,
},
{
name: "the same plugin",
source: pluginSource.replace("tools: [...request.tools, tool]", "tools: [tool, tool]"),
other: false,
message: /Duplicate agent tool: plugin_echo/,
},
])(
"collisions with $name fail only targeted session opens",
async ({ source, other, message }) => {
const h = await setup(false, source);
try {
if (other) await h.install("other-tools", pluginSource);
await expect(h.create({ x: "y" })).rejects.toThrow(message);
expect(h.launches).toHaveLength(0);
await h.create();
expect(h.launches).toHaveLength(1);
expect((await (await h.mcp()).listTools()).tools.map((tool) => tool.name)).not.toContain(
"plugin_echo",
);
} finally {
await h.close();
}
},
60_000,
);
test("each plugin owns its tool array and callbacks can remove earlier attachments", async () => {
const source = pluginSource.replace(
" return () => {};",
`
server.before("agent.session_open", ({ request }) => {
if (request.labels.x === "y" && request.tools[0]?.name !== "plugin_echo") throw new Error("Lost plugin-local definitions");
return { ...request, tools: [] };
});
return () => {};`,
);
const h = await setup(false, source);
try {
await h.install(
"other-tools",
pluginSource
.replace('name: "plugin_echo"', 'name: "other_echo"')
.replace(
' if (request.labels.x !== "y") return;',
' if (request.tools.length !== 0) throw new Error("Saw another plugin tools");\n if (request.labels.x !== "y") return;',
),
);
await h.create({ x: "y" });
const names = (await (await h.mcp()).listTools()).tools.map((tool) => tool.name);
expect(names).toContain("other_echo");
expect(names).not.toContain("plugin_echo");
} finally {
await h.close();
}
}, 60_000);
test("disabled plugins remove tools from existing and new MCP catalogs", async () => {
const h = await setup();
try {
await h.create({ x: "y" });
const mcp = await h.mcp();
expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo");
await h.client.patchDaemonConfig({ pluginsEnabled: false });
expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
await h.create({ x: "y" });
expect((await (await h.mcp()).listTools()).tools.map((tool) => tool.name)).not.toContain(
"plugin_echo",
);
} finally {
await h.close();
}
}, 60_000);
test.each([false, true])(
"provider tools disabled gates plugin tools (native=%s)",
async (native) => {
const h = await setup(native, pluginSource, { enabled: false });
try {
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools).toBeUndefined();
expect(h.launches.at(-1)!.config.mcpServers?.paseo).toBeUndefined();
} finally {
await h.close();
}
},
60_000,
);
test("disabledTools filters plugin tool names", async () => {
const h = await setup(true, pluginSource, { disabledTools: ["plugin_echo"] });
try {
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
const waitingSource = pluginSource
.replace("import { z }", 'import { writeFile } from "node:fs/promises";\nimport { z }')
.replace(
" calls++;",
`
calls++;
if (input.text === "wait") {
await writeFile(agent.cwd + "/started", agent.id);
await new Promise(resolve => signal.addEventListener("abort", resolve, { once: true }));
await writeFile(agent.cwd + "/aborted", agent.id);
return { text: "discard this result" };
}
`,
);
test("caller cancellation aborts the plugin handler and discards its result", async () => {
const h = await setup(true, waitingSource);
try {
const a = await h.create({ x: "y" });
const catalog = h.launches.at(-1)!.context!.paseoTools!;
const controller = new AbortController();
const call = catalog.executeTool(
"plugin_echo",
{ text: "wait" },
{ signal: controller.signal },
);
const rejected = expect(call).rejects.toThrow("Canceled by caller");
await expect.poll(() => readFile(path.join(h.directory, "started"), "utf8")).toBe(a.id);
controller.abort(new Error("Canceled by caller"));
await rejected;
await expect.poll(() => readFile(path.join(h.directory, "aborted"), "utf8")).toBe(a.id);
} finally {
await h.close();
}
}, 60_000);
test("a plugin stopping mid-call returns a tool error", async () => {
const h = await setup(true, waitingSource);
try {
const a = await h.create({ x: "y" });
const call = h.launches
.at(-1)!
.context!.paseoTools!.executeTool("plugin_echo", { text: "wait" });
await expect.poll(() => readFile(path.join(h.directory, "started"), "utf8")).toBe(a.id);
await h.client.patchDaemonConfig({ pluginsEnabled: false });
expect(await call).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("Plugin stopped") }],
});
} finally {
await h.close();
}
}, 60_000);
test("history sessions ignore attached tools", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const handle = a.persistence!;
await h.client.archiveAgent(a.id);
await h.daemon.daemon.agentManager.resumeAgentFromPersistence(handle, a.config, a.id);
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
expect(h.launches.at(-1)!.context!.paseoTools?.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
test("disabled plugins attach no tools to a new native session", async () => {
const h = await setup(true);
try {
await h.client.patchDaemonConfig({ pluginsEnabled: false });
await h.create({ x: "y" });
expect(h.launches.at(-1)!.context!.paseoTools!.getTool("plugin_echo")).toBeUndefined();
} finally {
await h.close();
}
}, 60_000);
test("structured tool results get model-visible text when text is omitted", async () => {
const h = await setup(
true,
pluginSource.replace(
" calls++;",
' return { structuredContent: { status: "created" } };',
),
);
try {
await h.create({ x: "y" });
expect(
await h.launches
.at(-1)!
.context!.paseoTools!.executeTool("plugin_echo", { text: "structured" }),
).toEqual({
content: [{ type: "text", text: JSON.stringify({ status: "created" }, null, 2) }],
structuredContent: { status: "created" },
isError: undefined,
});
} finally {
await h.close();
}
}, 60_000);
test("reloading a plugin retires its old attachments before an agent reopens", async () => {
const h = await setup(true);
try {
const a = await h.create({ x: "y" });
const mcp = await h.mcp(a.id);
const oldCatalog = h.launches.at(-1)!.context!.paseoTools!;
expect((await mcp.listTools()).tools.map((tool) => tool.name)).toContain("plugin_echo");
await h.client.reloadPlugin("tools-test");
expect((await mcp.listTools()).tools.map((tool) => tool.name)).not.toContain("plugin_echo");
expect(h.daemon.daemon.agentManager.getPluginTools(a.id)).toEqual([]);
expect(await oldCatalog.executeTool("plugin_echo", { text: "retired" })).toMatchObject({
isError: true,
content: [{ type: "text", text: expect.stringContaining("not available") }],
});
} finally {
await h.close();
}
}, 60_000);
test("built-in plugins attach and execute tools while configured plugins are disabled", async () => {
const root = await mkdtemp(path.join(tmpdir(), "paseo-builtin-tools-"));
const directory = path.join(root, "builtin-tools");
await mkdir(directory);
await writeFile(
path.join(directory, "paseo-plugin.json"),
JSON.stringify({ id: "builtin-tools", requirements: { paseo: ">=0.13.2" } }),
);
await writeFile(path.join(directory, "index.server.ts"), pluginSource);
const recorder = recordingProvider(true);
const daemon = await createTestPaseoDaemon({
daemonVersion: "0.13.2",
agentClients: { codex: recorder.provider },
pluginsEnabled: false,
builtinPlugins: new BuiltinPluginLoader(root, ["builtin-tools"]),
});
try {
const a = await daemon.daemon.agentManager.createAgent(
{ provider: "codex", cwd: root },
undefined,
{ labels: { x: "y" } },
);
const catalog = recorder.launches.at(-1)!.context!.paseoTools!;
expect(catalog.getTool("plugin_echo")?.name).toBe("plugin_echo");
const result = await catalog.executeTool("plugin_echo", { text: "builtin" });
expect(result.isError).not.toBe(true);
expect(JSON.parse(result.content[0].text!)).toMatchObject({
agent: { id: a.id },
input: { text: "builtin" },
});
} finally {
await daemon.close();
await rm(root, { recursive: true, force: true });
}
}, 60_000);
test("a session hook missing tools reports a schema issue", async () => {
const h = await setup(
false,
pluginSource.replace("tools: [...request.tools, tool]", "tools: undefined"),
);
try {
await expect(h.create({ x: "y" })).rejects.toThrow('"tools"');
} finally {
await h.close();
}
}, 60_000);
@@ -0,0 +1,46 @@
import { expect, test } from "vitest";
import { z } from "zod";
import { defineAgentTool } from "@getpaseo/plugin/server";
import { serializeAgentTool } from "./agent-tools.js";
import { serializePaseoToolInputParameters } from "../agent/tools/paseo-tool-serialization.js";
test("agent tool schemas survive the process boundary and catalog round trip", () => {
const input = z
.object({
title: z.string().min(1).describe("Issue title"),
state: z.enum(["open", "closed"]),
tags: z.array(z.string()).optional(),
nested: z.object({ enabled: z.boolean() }),
})
.describe("Create an issue");
const tool = defineAgentTool({
name: "create_issue",
description: "Create",
input,
execute: () => ({ text: "done" }),
});
const metadata = JSON.parse(JSON.stringify(serializeAgentTool(tool)));
const restored = z.fromJSONSchema(metadata.inputSchema);
expect(
serializePaseoToolInputParameters({
...metadata,
inputSchema: restored,
handler: async () => ({ content: [] }),
}),
).toEqual({
...metadata.inputSchema,
additionalProperties: {},
properties: {
...metadata.inputSchema.properties,
nested: { ...metadata.inputSchema.properties.nested, additionalProperties: {} },
},
});
expect(restored.parse({ title: "Bug", state: "open", nested: { enabled: true } })).toEqual({
title: "Bug",
state: "open",
nested: { enabled: true },
});
expect(
restored.safeParse({ title: "", state: "unknown", nested: { enabled: true } }).success,
).toBe(false);
});
@@ -0,0 +1,108 @@
import { z } from "zod";
import type {
AgentToolDefinition,
AgentToolResult,
PluginHookContext,
PluginSessionOpenRequest,
PluginHookAgent,
} from "@getpaseo/plugin/server";
import { toJsonSchemaCompat } from "@modelcontextprotocol/sdk/server/zod-json-schema-compat.js";
export const agentToolMetadataSchema = z
.object({
name: z
.string()
.regex(/^[a-z][a-z0-9_]*$/)
.max(64),
title: z.string().optional(),
description: z.string(),
inputSchema: z.record(z.string(), z.unknown()),
pluginId: z.string().optional(),
pluginInstanceId: z.string().optional(),
})
.strict();
export type AgentToolMetadata = z.infer<typeof agentToolMetadataSchema>;
export type AttachedAgentTool = AgentToolMetadata & { pluginId: string; pluginInstanceId: string };
export type SessionOpenWireRequest = Omit<PluginSessionOpenRequest, "tools"> & {
tools: AgentToolMetadata[];
};
export interface PluginAgentToolsRuntime {
isToolPluginAvailable(pluginId: string, pluginInstanceId: string): boolean;
detachTools(agentId: string): void;
callTool(
pluginId: string,
pluginInstanceId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult>;
}
export function serializeAgentTool(tool: AgentToolDefinition): AgentToolMetadata {
return agentToolMetadataSchema.parse({
name: tool.name,
title: tool.title,
description: tool.description,
inputSchema: toJsonSchemaCompat(tool.input, { strictUnions: true, pipeStrategy: "input" }),
});
}
// This store lives only in the plugin process. Definitions and parsed values never cross IPC.
export class PluginAgentTools {
private readonly agents = new Map<string, AgentToolDefinition[]>();
private readonly calls = new Map<string, { agentId: string; controller: AbortController }>();
attach(agentId: string, tools: AgentToolDefinition[]): AgentToolMetadata[] {
const names = new Set<string>();
const metadata = tools.map((tool) => {
const serialized = serializeAgentTool(tool);
if (names.has(tool.name)) throw new Error(`Duplicate agent tool: ${tool.name}`);
names.add(tool.name);
return serialized;
});
this.agents.set(agentId, tools);
return metadata;
}
detach(agentId: string): void {
this.agents.delete(agentId);
for (const call of this.calls.values()) {
if (call.agentId === agentId) call.controller.abort();
}
}
cancel(requestId: string): void {
this.calls.get(requestId)?.controller.abort();
}
close(): void {
for (const call of this.calls.values()) call.controller.abort();
this.agents.clear();
}
async call(
requestId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
paseo: PluginHookContext["paseo"],
): Promise<AgentToolResult> {
const controller = new AbortController();
this.calls.set(requestId, { agentId: agent.id, controller });
try {
const tool = this.agents.get(agent.id)?.find((definition) => definition.name === name);
if (!tool) throw new Error(`Agent tool not attached: ${name}`);
const parsed = await tool.input.parseAsync(input);
controller.signal.throwIfAborted();
const result = await tool.execute(parsed, { agent, paseo, signal: controller.signal });
controller.signal.throwIfAborted();
return result;
} catch (error) {
return { isError: true, text: error instanceof Error ? error.message : String(error) };
} finally {
this.calls.delete(requestId);
}
}
}
@@ -1,3 +1,4 @@
import type { PluginAgentToolsRuntime } from "./agent-tools.js";
import type { PluginLifecycle } from "./lifecycle/index.js";
import path from "node:path";
import { stat, rm } from "node:fs/promises";
@@ -37,6 +38,9 @@ import type { PluginUsageSourceMetadata } from "./plugin-process-protocol.js";
const BUILTIN_PROVIDER_ID_SET: ReadonlySet<string> = new Set(BUILTIN_PROVIDER_IDS);
interface PluginRuntimePort {
isToolPluginAvailable?: PluginRuntime["isToolPluginAvailable"];
callTool?: PluginRuntime["callTool"];
detachTools?: PluginRuntime["detachTools"];
emit?: PluginLifecycle["emit"];
drainEvents?: PluginRuntime["drainEvents"];
before?: PluginLifecycle["before"];
@@ -131,6 +135,30 @@ export class PluginService {
this.runtime.emit?.(name, event);
};
readonly isToolPluginAvailable: PluginAgentToolsRuntime["isToolPluginAvailable"] = (
pluginId,
pluginInstanceId,
) => {
return this.runtime.isToolPluginAvailable?.(pluginId, pluginInstanceId) ?? false;
};
readonly detachTools: PluginAgentToolsRuntime["detachTools"] = (agentId) => {
this.runtime.detachTools?.(agentId);
};
readonly callTool: PluginAgentToolsRuntime["callTool"] = async (
pluginId,
pluginInstanceId,
agent,
name,
input,
signal,
) => {
if (!this.isToolPluginAvailable(pluginId, pluginInstanceId) || !this.runtime.callTool)
return { isError: true, text: `Plugin is not available: ${pluginId}` };
return this.runtime.callTool(pluginId, pluginInstanceId, agent, name, input, signal);
};
async drainEvents(): Promise<void> {
await this.runtime.drainEvents?.();
}
@@ -93,8 +93,39 @@ test("session-open hooks reject changes to session identity instead of silently
reason: "resume",
purpose: "interactive",
env: {},
title: null,
labels: {},
tools: [],
},
paseo,
),
).rejects.toThrow("agent.session_open hooks can only change env");
});
test.each([{ title: "Changed" }, { labels: { changed: "true" } }])(
"session-open targeting metadata is read-only: %j",
async (patch) => {
const hooks = new PluginHookHandlers(() => {});
hooks.before("agent.session_open", ({ request }) => ({ ...request, ...patch }));
await expect(
hooks.invoke(
"operation",
"before",
"agent.session_open",
{
agentId: "agent",
workspaceId: null,
provider: "codex",
cwd: "/project",
title: "Original",
labels: {},
tools: [],
env: {},
reason: "create",
purpose: "interactive",
},
paseo,
),
).rejects.toThrow("agent.session_open hooks can only change env and tools");
},
);
@@ -1,3 +1,11 @@
import { isDeepStrictEqual } from "node:util";
import {
serializeAgentTool,
agentToolMetadataSchema,
type SessionOpenWireRequest,
type PluginAgentTools,
} from "../agent-tools.js";
import type { AgentToolDefinition } from "@getpaseo/plugin/server";
import type { AgentStreamEvent } from "../../agent/agent-sdk-types.js";
import { z } from "zod";
import { CreateAgentRequestMessageSchema } from "@getpaseo/protocol/messages";
@@ -15,6 +23,10 @@ import type {
import { WorkspaceCreateRequestSchema } from "@getpaseo/protocol/messages";
import type { PersistedWorkspaceRecord } from "../../workspace-registry.js";
export type PluginWireBeforeRequests = Omit<PluginBeforeRequests, "agent.session_open"> & {
"agent.session_open": SessionOpenWireRequest;
};
export const lifecycleEventNames = [
"agent.created",
"agent.closed",
@@ -38,28 +50,44 @@ const beforeSchemas = {
cwd: z.string(),
reason: z.enum(["create", "resume", "refresh", "import"]),
purpose: z.enum(["interactive", "history"]),
title: z.string().nullable(),
labels: z.record(z.string(), z.string()),
tools: z.array(agentToolMetadataSchema),
env: z.record(z.string(), z.string()),
})
.strict(),
"workspace.create": WorkspaceCreateRequestSchema.omit({ type: true, requestId: true }).strict(),
};
const sessionOpenLocalResultSchema = beforeSchemas["agent.session_open"].extend({
tools: z.array(
z.custom<AgentToolDefinition>(
(tool) =>
typeof tool === "object" &&
tool !== null &&
"input" in tool &&
"execute" in tool &&
typeof tool.execute === "function",
),
),
});
export interface PluginLifecycle {
emit<Name extends keyof PluginLifecycleEvents>(
name: Name,
event: PluginLifecycleEvents[Name],
): void;
before<Name extends keyof PluginBeforeRequests>(
before<Name extends keyof PluginWireBeforeRequests>(
name: Name,
request: PluginBeforeRequests[Name],
): Promise<PluginBeforeRequests[Name]>;
request: PluginWireBeforeRequests[Name],
): Promise<PluginWireBeforeRequests[Name]>;
}
export function validateBeforeRequest<Name extends keyof PluginBeforeRequests>(
export function validateBeforeRequest<Name extends keyof PluginWireBeforeRequests>(
name: Name,
value: unknown,
): PluginBeforeRequests[Name] {
return beforeSchemas[name].parse(value) as PluginBeforeRequests[Name];
): PluginWireBeforeRequests[Name] {
return beforeSchemas[name].parse(value) as PluginWireBeforeRequests[Name];
}
export function describeHookWorkspace(workspace: PersistedWorkspaceRecord): PluginHookWorkspace {
@@ -129,11 +157,11 @@ export function publishAgentStream(
}
}
export function validateBeforeResult<Name extends keyof PluginBeforeRequests>(
export function validateBeforeResult<Name extends keyof PluginWireBeforeRequests>(
name: Name,
input: PluginBeforeRequests[Name],
input: PluginWireBeforeRequests[Name],
output: unknown,
): PluginBeforeRequests[Name] {
): PluginWireBeforeRequests[Name] {
const result = validateBeforeRequest(name, output);
if (name === "agent.session_open") {
const previous = beforeSchemas["agent.session_open"].parse(input);
@@ -144,9 +172,11 @@ export function validateBeforeResult<Name extends keyof PluginBeforeRequests>(
previous.provider !== next.provider ||
previous.cwd !== next.cwd ||
previous.reason !== next.reason ||
previous.purpose !== next.purpose
previous.purpose !== next.purpose ||
previous.title !== next.title ||
!isDeepStrictEqual(previous.labels, next.labels)
) {
throw new Error("agent.session_open hooks can only change env");
throw new Error("agent.session_open hooks can only change env and tools");
}
}
if (name === "agent.create") {
@@ -168,7 +198,10 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
private readonly changed: () => void;
constructor(changed: () => void) {
private readonly agentTools: PluginAgentTools | undefined;
constructor(changed: () => void, agentTools?: PluginAgentTools) {
this.agentTools = agentTools;
this.changed = changed;
}
@@ -209,6 +242,9 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
throw new Error(`Unknown before hook: ${name}`);
}
const hookName = name as keyof import("@getpaseo/plugin/server").PluginBeforeRequests;
if (hookName === "agent.session_open") {
return await this.openSession(input, { paseo, signal: controller.signal });
}
let request = validateBeforeRequest(hookName, input);
for (const handler of this.transforms.get(name) ?? []) {
controller.signal.throwIfAborted();
@@ -216,9 +252,7 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
{ request: structuredClone(request) },
{ paseo, signal: controller.signal },
);
if (result !== undefined) {
request = validateBeforeResult(hookName, request, result);
}
if (result !== undefined) request = validateBeforeResult(hookName, request, result);
}
return request;
}
@@ -236,6 +270,31 @@ export class PluginHookHandlers implements PluginLifecycleRegistration {
}
}
private async openSession(
input: unknown,
context: PluginHookContext,
): Promise<SessionOpenWireRequest> {
let request = validateBeforeRequest("agent.session_open", input);
this.agentTools?.detach(request.agentId);
let tools: AgentToolDefinition[] = [];
for (const handler of this.transforms.get("agent.session_open") ?? []) {
context.signal.throwIfAborted();
const result = await handler(
{ request: { ...structuredClone(request), tools: [...tools] } },
context,
);
if (result === undefined) continue;
const local = sessionOpenLocalResultSchema.parse(result);
const metadata = local.tools.map(serializeAgentTool);
request = validateBeforeResult("agent.session_open", request, { ...local, tools: metadata });
tools = local.tools;
}
context.signal.throwIfAborted();
if (request.purpose === "history") request.tools = [];
else this.agentTools?.attach(request.agentId, tools);
return request;
}
cancel(id: string): void {
this.active.get(id)?.abort();
}
@@ -1,3 +1,4 @@
import type { PluginHookAgent } from "@getpaseo/plugin/server";
import { UsageScopeSchema, type UsageScope } from "@getpaseo/plugin/server/usage";
import type {
ProviderConnectRequest,
@@ -31,6 +32,9 @@ export interface PluginUsageSourceMetadata {
}
export type PluginProcessRequest =
| { type: "tool.call"; requestId: string; agent: PluginHookAgent; name: string; input: unknown }
| { type: "tool.cancel"; requestId: string }
| { type: "tools.detach"; agentId: string }
| {
type: "provider.status";
requestId: string;
@@ -141,6 +145,26 @@ const frameFields = {
export const PluginProcessRequestSchema: z.ZodType<PluginProcessRequest> = z.discriminatedUnion(
"type",
[
z
.object({
type: z.literal("tool.call"),
requestId: z.string(),
agent: z
.object({
id: z.string(),
workspaceId: z.string().nullable(),
parentAgentId: z.string().nullable(),
provider: z.string(),
cwd: z.string(),
title: z.string().nullable(),
})
.strict(),
name: z.string(),
input: z.unknown(),
})
.strict(),
z.object({ type: z.literal("tool.cancel"), requestId: z.string() }).strict(),
z.object({ type: z.literal("tools.detach"), agentId: z.string() }).strict(),
z
.object({
type: z.literal("provider.status"),
@@ -1,3 +1,4 @@
import { PluginAgentTools } from "./agent-tools.js";
import { PluginHookHandlers } from "./lifecycle/index.js";
import { evaluateBundle } from "./bundle-evaluator.js";
import {
@@ -59,9 +60,10 @@ export function createPluginWorker(options: {
handler: RpcHandler;
}
const agentTools = new PluginAgentTools();
const hooks = new PluginHookHandlers(() => {
send({ type: "hooks.changed", hooks: hooks.catalog() });
});
}, agentTools);
const handlers = new Map<string, RegisteredRpc>();
const providers = new Map<string, ProviderRegistration>();
const usageSources = new Map<string, UsageSourceRegistration>();
@@ -322,6 +324,7 @@ export function createPluginWorker(options: {
?.dispose()
.catch((error) => console.error("Plugin API cleanup failed", error));
hooks.close();
agentTools.close();
for (const pending of pendingProviderConnections.values()) pending.tombstoned = true;
const currentCleanup = cleanup;
cleanup = null;
@@ -381,6 +384,42 @@ export function createPluginWorker(options: {
}
}
function handleToolMessage(
message: PluginProcessRequest,
): message is Extract<
PluginProcessRequest,
{ type: "tool.call" | "tool.cancel" | "tools.detach" }
> {
if (message.type === "tools.detach") {
agentTools.detach(message.agentId);
return true;
}
if (message.type === "tool.cancel") {
agentTools.cancel(message.requestId);
return true;
}
if (message.type === "tool.call") {
if (!paseo) {
send({ type: "error", requestId: message.requestId, error: "Plugin API unavailable" });
return true;
}
void agentTools
.call(message.requestId, message.agent, message.name, message.input, paseo)
.then((output) =>
send({
type: "result",
requestId: message.requestId,
output: jsonTransportValue(output),
}),
)
.catch((error) =>
send({ type: "error", requestId: message.requestId, error: describeError(error) }),
);
return true;
}
return false;
}
function handleMessage(rawMessage: unknown): void {
const parsed = PluginProcessRequestSchema.safeParse(rawMessage);
if (!parsed.success) {
@@ -422,6 +461,7 @@ export function createPluginWorker(options: {
rejectWhileStopping(message);
return;
}
if (handleToolMessage(message)) return;
if (message.type === "provider.status") {
void (async () => {
const provider = providers.get(message.providerId);
+106 -12
View File
@@ -1,10 +1,16 @@
import type { PluginWireBeforeRequests } from "./lifecycle/index.js";
import type { SessionOpenWireRequest, AttachedAgentTool } from "./agent-tools.js";
import type { UsageScope } from "@getpaseo/plugin/server/usage";
import {
ProviderStatusSchema,
type ProviderStatus,
type ProviderStatusRequest,
} from "@getpaseo/plugin/server/provider";
import type { PluginBeforeRequests, PluginLifecycleEvents } from "@getpaseo/plugin/server";
import type {
PluginLifecycleEvents,
PluginHookAgent,
AgentToolResult,
} from "@getpaseo/plugin/server";
import { validateBeforeRequest, validateBeforeResult } from "./lifecycle/index.js";
import { fork } from "node:child_process";
import { stat } from "node:fs/promises";
@@ -40,6 +46,7 @@ import { evaluateBundle } from "./bundle-evaluator.js";
const CLIENT_ENTRY_FILENAMES = ["index.client.ts", "index.client.tsx"] as const;
const SERVER_ENTRY_FILENAMES = ["index.server.ts", "index.server.tsx"] as const;
const REQUEST_TIMEOUT_MS = 30_000;
const TOOL_CALL_TIMEOUT_MS = 10 * 60_000;
const MAX_LOG_ENTRIES = 500;
const MAX_LOG_BYTES = 256 * 1024;
const MAX_LOG_LINE_BYTES = 16 * 1024;
@@ -69,6 +76,7 @@ interface PendingInvocation {
interface LoadedPlugin {
id: string;
instanceId: string;
requirements: PluginRequirements | undefined;
clientBundle: string;
methods: ReadonlySet<string>;
@@ -546,14 +554,15 @@ export class PluginRuntime {
await Promise.all(this.pendingEvents);
}
async before<Name extends keyof PluginBeforeRequests>(
async before<Name extends keyof PluginWireBeforeRequests>(
name: Name,
input: PluginBeforeRequests[Name],
): Promise<PluginBeforeRequests[Name]> {
input: PluginWireBeforeRequests[Name],
): Promise<PluginWireBeforeRequests[Name]> {
let request = validateBeforeRequest(name, input);
const plugins = [...this.plugins.values()].sort((left, right) => {
return left.id.localeCompare(right.id);
});
const attached: AttachedAgentTool[] = [];
for (const loaded of plugins) {
if (!loaded.hooks.before.includes(name)) {
continue;
@@ -564,15 +573,21 @@ export class PluginRuntime {
requestId: randomUUID(),
kind: "before",
name,
input: request,
input: name === "agent.session_open" ? { ...request, tools: [] } : request,
});
request = validateBeforeResult(name, request, output);
if (name === "agent.session_open") {
this.collectAgentTools(attached, loaded, (request as SessionOpenWireRequest).tools);
}
} catch (error) {
if (name === "agent.session_open")
this.detachTools((request as SessionOpenWireRequest).agentId);
throw new Error(`Plugin ${loaded.id} before ${name} failed: ${describeError(error)}`, {
cause: error,
});
}
}
if (name === "agent.session_open") (request as SessionOpenWireRequest).tools = attached;
return request;
}
@@ -594,25 +609,102 @@ export class PluginRuntime {
return output;
}
private collectAgentTools(
attached: AttachedAgentTool[],
loaded: LoadedPlugin,
tools: SessionOpenWireRequest["tools"],
): void {
for (const tool of tools) {
const collision = attached.find((entry) => entry.name === tool.name);
if (collision)
throw new Error(
`Agent tool ${tool.name} from plugin ${loaded.id} collides with plugin ${collision.pluginId}`,
);
attached.push({ ...tool, pluginId: loaded.id, pluginInstanceId: loaded.instanceId });
}
}
isToolPluginAvailable(pluginId: string, pluginInstanceId: string): boolean {
return this.plugins.get(pluginId)?.instanceId === pluginInstanceId;
}
detachTools(agentId: string): void {
for (const loaded of this.plugins.values()) {
if (loaded.child) void send(loaded.child, { type: "tools.detach", agentId }).catch(() => {});
}
}
async callTool(
pluginId: string,
pluginInstanceId: string,
agent: PluginHookAgent,
name: string,
input: unknown,
signal?: AbortSignal,
): Promise<AgentToolResult> {
const loaded = this.plugins.get(pluginId);
if (!loaded || loaded.instanceId !== pluginInstanceId)
return { isError: true, text: `Plugin is not available: ${pluginId}` };
const requestId = randomUUID();
try {
return (await this.request(
loaded,
{ type: "tool.call", requestId, agent, name, input },
signal,
)) as AgentToolResult;
} catch (error) {
if (signal?.aborted) throw error;
return { isError: true, text: describeError(error) };
}
}
private request(
loaded: LoadedPlugin,
message: Extract<PluginProcessRequest, { requestId: string }>,
signal?: AbortSignal,
): Promise<unknown> {
const child = loaded.child;
const pluginId = loaded.id;
if (!child) throw new Error(`Plugin ${pluginId} has no server entry`);
const requestId = message.requestId;
signal?.throwIfAborted();
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
const cancel = () => {
clearTimeout(timeout);
loaded.pending.delete(requestId);
if (message.type === "hook") {
void send(child, { type: "hook.cancel", requestId }).catch(() => {});
}
reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
}, REQUEST_TIMEOUT_MS);
loaded.pending.set(requestId, { resolve, reject, timeout });
if (message.type === "tool.call")
void send(child, { type: "tool.cancel", requestId }).catch(() => {});
signal?.removeEventListener("abort", cancel);
reject(signal?.reason ?? new Error("Tool call canceled"));
};
const timeout = setTimeout(
() => {
loaded.pending.delete(requestId);
signal?.removeEventListener("abort", cancel);
if (message.type === "tool.call")
void send(child, { type: "tool.cancel", requestId }).catch(() => {});
if (message.type === "hook") {
void send(child, { type: "hook.cancel", requestId }).catch(() => {});
}
reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
},
message.type === "tool.call" ? TOOL_CALL_TIMEOUT_MS : REQUEST_TIMEOUT_MS,
);
signal?.addEventListener("abort", cancel, { once: true });
loaded.pending.set(requestId, {
resolve: (value) => {
signal?.removeEventListener("abort", cancel);
resolve(value);
},
reject: (error) => {
signal?.removeEventListener("abort", cancel);
reject(error);
},
timeout,
});
void send(child, message).catch((error) => {
clearTimeout(timeout);
signal?.removeEventListener("abort", cancel);
loaded.pending.delete(requestId);
reject(error);
});
@@ -641,6 +733,7 @@ export class PluginRuntime {
if (!serverBundle) {
return {
id: pluginId,
instanceId: randomUUID(),
clientBundle: bundles.clientBundle ?? "",
requirements: manifest.requirements,
methods: new Set(),
@@ -769,6 +862,7 @@ export class PluginRuntime {
}
loaded = {
id: pluginId,
instanceId: randomUUID(),
clientBundle,
requirements,
methods: new Set(ready.methods),