From c03853f4e1d207286de693f99fa914713bcbbca7 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 26 Sep 2026 18:08:16 +0200 Subject: [PATCH] feat(server): run internal plugins in-process through the plugin runtime --- packages/server/src/plugins/index.ts | 9 + packages/server/src/server/bootstrap.ts | 3 + packages/server/src/server/plugins/index.ts | 29 +- .../server/plugins/internal-boundary.test.ts | 37 + .../src/server/plugins/internal-child.ts | 101 +++ .../server/plugins/internal-seam.e2e.test.ts | 58 ++ .../src/server/plugins/plugin-process.ts | 855 +++++++++--------- packages/server/src/server/plugins/runtime.ts | 31 +- .../internal-seam/index.server.ts | 19 + .../internal-seam/paseo-plugin.json | 1 + .../src/server/test-utils/paseo-daemon.ts | 3 + 11 files changed, 730 insertions(+), 416 deletions(-) create mode 100644 packages/server/src/plugins/index.ts create mode 100644 packages/server/src/server/plugins/internal-boundary.test.ts create mode 100644 packages/server/src/server/plugins/internal-child.ts create mode 100644 packages/server/src/server/plugins/internal-seam.e2e.test.ts create mode 100644 packages/server/src/server/plugins/test-fixtures/internal-seam/index.server.ts create mode 100644 packages/server/src/server/plugins/test-fixtures/internal-seam/paseo-plugin.json diff --git a/packages/server/src/plugins/index.ts b/packages/server/src/plugins/index.ts new file mode 100644 index 0000000000..6ccd5ee954 --- /dev/null +++ b/packages/server/src/plugins/index.ts @@ -0,0 +1,9 @@ +import type { PluginServerContribution } from "@getpaseo/plugin/server"; + +export interface InternalPlugin { + id: string; + directory: string; + contribute: PluginServerContribution; +} + +export const INTERNAL_PLUGINS: InternalPlugin[] = []; diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index cfa955eac5..1481384391 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -231,6 +231,7 @@ import { } from "./hub/relationship-remote.js"; import { DaemonExecutions } from "./hub/daemon-executions.js"; import { PluginService } from "./plugins/index.js"; +import { INTERNAL_PLUGINS, type InternalPlugin } from "../plugins/index.js"; import { ManagedPluginSources } from "./plugins/managed-source.js"; const MCP_DEBUG_BATCH_LIMIT = 10; @@ -476,6 +477,7 @@ export interface PaseoDaemon { } export interface PaseoDaemonDependencies { + internalPlugins?: readonly InternalPlugin[]; hubRelationshipRemote?: HubRelationshipRemote; hubRelationshipClock?: HubRelationshipClock; hubRelationshipRetryPolicy?: HubRelationshipRetryPolicy; @@ -611,6 +613,7 @@ export async function createPaseoDaemon( const browserToolsBroker = new BrowserToolsBroker({}); const pluginRuntime = new PluginService(logger, daemonConfigStore, daemonVersion, { managedSources: new ManagedPluginSources(config.paseoHome), + internalPlugins: dependencies.internalPlugins ?? INTERNAL_PLUGINS, settingsDirectory: path.join(config.paseoHome, "plugin-settings"), }); diff --git a/packages/server/src/server/plugins/index.ts b/packages/server/src/server/plugins/index.ts index c1e404bb07..1ce3a0c2fd 100644 --- a/packages/server/src/server/plugins/index.ts +++ b/packages/server/src/server/plugins/index.ts @@ -23,6 +23,7 @@ import { type ManagedPluginCandidate, ManagedPluginSources } from "./managed-sou import { readPluginManifest } from "./manifest.js"; import { runPluginBuild } from "./preparation.js"; import { PluginRuntime } from "./runtime.js"; +import type { InternalPlugin } from "../../plugins/index.js"; import type { PluginProviderMetadata } from "./plugin-process-protocol.js"; import { readPluginProviderIcon } from "./provider-icon.js"; @@ -40,6 +41,7 @@ interface PluginRuntimePort { getProviderCatalogCacheKey?: PluginRuntime["getProviderCatalogCacheKey"]; validatePlugin?(path: string): Promise; startPlugin(pluginId: string, path: string, canPublish: () => boolean): Promise; + startInternalPlugin?(plugin: InternalPlugin): Promise; stopPluginById(pluginId: string): Promise; stopAll(): Promise; subscribe(listener: (pluginId: string, error?: string) => void): () => void; @@ -50,6 +52,7 @@ interface PluginServiceDependencies { settingsDirectory?: string; runtime?: PluginRuntimePort; managedSources?: ManagedPluginSources; + internalPlugins?: readonly InternalPlugin[]; } function resolvePluginStatus(input: { @@ -137,6 +140,10 @@ export class PluginService { this.started = true; const config = this.configStore.get(); this.globalStartsBlocked = config.pluginsEnabled !== true; + for (const plugin of this.dependencies.internalPlugins ?? []) { + await this.runtime.startInternalPlugin?.(plugin); + await this.publishProviderRegistrations(plugin.id, plugin.directory); + } if (config.pluginsEnabled === true) { for (const [pluginId, source] of Object.entries(config.plugins ?? {})) { if (source.enabled === false) continue; @@ -186,12 +193,16 @@ export class PluginService { } getLogs(pluginId: string): PluginLogEntry[] { - this.requireSource(pluginId); + if (!this.dependencies.internalPlugins?.some((plugin) => plugin.id === pluginId)) { + this.requireSource(pluginId); + } return this.runtime.getLogs(pluginId); } catalog(): ReturnType { - return this.runtime.catalog(); + return this.runtime + .catalog() + .filter(({ id }) => !this.dependencies.internalPlugins?.some((plugin) => plugin.id === id)); } async installDirectory(input: { path: string; id?: string }): Promise { @@ -200,7 +211,10 @@ export class PluginService { const manifest = await readPluginManifest(directory); assertPluginCompatibility({ ...manifest, version: this.daemonVersion, runtime: "daemon" }); const pluginId = PluginIdSchema.parse(input.id ?? manifest.id); - if (this.configStore.get().plugins?.[pluginId]) { + if ( + this.configStore.get().plugins?.[pluginId] || + this.dependencies.internalPlugins?.some((plugin) => plugin.id === pluginId) + ) { throw new Error( `Plugin ID "${pluginId}" is already configured; choose another ID with --id`, ); @@ -428,7 +442,7 @@ export class PluginService { private handleGlobalSwitch(enabled: boolean): void { if (!enabled) { this.globalStartsBlocked = true; - const stopping = this.stopAll(); + const stopping = this.stopConfiguredPlugins(); for (const id of Object.keys(this.configStore.get().plugins ?? {})) this.notify(id); void this.enqueue(async () => { await stopping; @@ -496,6 +510,13 @@ export class PluginService { return this.runtime.stopPluginById(pluginId); } + private async stopConfiguredPlugins(): Promise { + const stopping = Object.keys(this.configStore.get().plugins ?? {}).map((id) => + this.stopPlugin(id), + ); + await Promise.all(stopping); + } + private async stopAll(): Promise { for (const pluginId of this.providerIdsByPlugin.keys()) { this.removeProviderRegistrations(pluginId); diff --git a/packages/server/src/server/plugins/internal-boundary.test.ts b/packages/server/src/server/plugins/internal-boundary.test.ts new file mode 100644 index 0000000000..73b2a96818 --- /dev/null +++ b/packages/server/src/server/plugins/internal-boundary.test.ts @@ -0,0 +1,37 @@ +import { readdir } from "node:fs/promises"; +import { isBuiltin } from "node:module"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, test } from "vitest"; +import { createPluginImportReader } from "./compiler-imports.js"; +import { compilePlugin } from "./compiler.js"; + +const root = fileURLToPath(new URL("../../plugins/", import.meta.url)); + +test("internal plugins keep every import inside the public plugin boundary", async () => { + const entries = await readdir(root, { withFileTypes: true }); + for (const entry of entries) { + if (!entry.isDirectory()) continue; + const directory = path.join(root, entry.name); + const server = path.join(directory, "index.server.ts"); + await compilePlugin({ server, client: null }); + const reader = createPluginImportReader(directory); + const pending = [server]; + const visited = new Set(); + while (pending.length > 0) { + const file = pending.pop()!; + if (visited.has(file)) continue; + visited.add(file); + for (const dependency of reader.read(file)) { + const specifier = dependency.specifier; + if (specifier === "@getpaseo/plugin" || specifier.startsWith("@getpaseo/plugin/")) continue; + if (isBuiltin(specifier)) continue; + const resolved = reader.resolve(specifier, file, dependency.kind); + expect(resolved, `${file}: ${specifier}`).toBeDefined(); + const relative = path.relative(directory, resolved!); + expect(relative, `${file}: ${specifier}`).toMatch(/^(server|shared)\//); + pending.push(resolved!); + } + } + } +}); diff --git a/packages/server/src/server/plugins/internal-child.ts b/packages/server/src/server/plugins/internal-child.ts new file mode 100644 index 0000000000..15ad4c5362 --- /dev/null +++ b/packages/server/src/server/plugins/internal-child.ts @@ -0,0 +1,101 @@ +import { EventEmitter } from "node:events"; +import { AsyncLocalStorage } from "node:async_hooks"; +import { format } from "node:util"; +import { PassThrough } from "node:stream"; +import type { PluginServerContribution } from "@getpaseo/plugin/server"; +import { createPluginWorker, type PluginWorkerChannel } from "./plugin-process.js"; +import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js"; + +const outputContext = new AsyncLocalStorage(); +const originalConsole = { + log: console.log, + info: console.info, + debug: console.debug, + warn: console.warn, + error: console.error, +}; +let activeChildren = 0; + +function captureOutput(): void { + if (activeChildren++ > 0) return; + for (const method of Object.keys(originalConsole) as Array) { + console[method] = (...args: unknown[]) => { + const child = outputContext.getStore(); + if (!child) return originalConsole[method](...args); + const stream = method === "warn" || method === "error" ? child.stderr : child.stdout; + stream.write(`${format(...args)}\n`); + }; + } +} + +function releaseOutput(): void { + if (--activeChildren > 0) return; + for (const method of Object.keys(originalConsole) as Array) { + console[method] = originalConsole[method]; + } +} + +export class InternalPluginChild extends EventEmitter { + connected = true; + killed = false; + readonly stdout = new PassThrough(); + readonly stderr = new PassThrough(); + private readonly worker: ReturnType; + private readonly workerMessages = new Set<(message: PluginProcessRequest) => void>(); + private closing: Promise | null = null; + + constructor(contribute: PluginServerContribution) { + super(); + captureOutput(); + const channel: PluginWorkerChannel = { + send: (message: PluginProcessMessage, callback?: () => void) => { + const copy = structuredClone(message); + queueMicrotask(() => { + if (this.connected) this.emit("message", copy); + callback?.(); + }); + }, + onMessage: (handler) => { + this.workerMessages.add(handler); + return () => this.workerMessages.delete(handler); + }, + disconnect: () => this.disconnect(), + }; + this.worker = createPluginWorker({ channel, contribute }); + } + + send(message: PluginProcessRequest, callback?: (error: Error | null) => void): boolean { + if (!this.connected) { + callback?.(new Error("Plugin child is disconnected")); + return false; + } + const copy = structuredClone(message); + queueMicrotask(() => { + outputContext.run(this, () => { + for (const handler of this.workerMessages) handler(copy); + }); + callback?.(null); + }); + return true; + } + + kill(): boolean { + if (this.killed) return false; + this.killed = true; + this.disconnect(); + return true; + } + + disconnect(): void { + if (this.closing) return; + this.closing = Promise.resolve() + .then(() => this.worker.shutdown()) + .finally(() => { + this.connected = false; + releaseOutput(); + this.stdout.end(); + this.stderr.end(); + this.emit("close", 0, null); + }); + } +} diff --git a/packages/server/src/server/plugins/internal-seam.e2e.test.ts b/packages/server/src/server/plugins/internal-seam.e2e.test.ts new file mode 100644 index 0000000000..fe36cd5e86 --- /dev/null +++ b/packages/server/src/server/plugins/internal-seam.e2e.test.ts @@ -0,0 +1,58 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, test } from "vitest"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import contribute from "./test-fixtures/internal-seam/index.server.js"; + +const directory = fileURLToPath(new URL("./test-fixtures/internal-seam/", import.meta.url)); + +test("internal and directory plugins share RPC and lifecycle behavior while internal plugins stay hidden and enabled", async () => { + const workspaceDirectory = await mkdtemp(path.join(tmpdir(), "paseo-internal-seam-")); + const daemon = await createTestPaseoDaemon({ + daemonVersion: "0.8.0", + pluginsEnabled: false, + internalPlugins: [{ id: "internal-seam", directory, contribute }], + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" }); + try { + await client.connect(); + expect(await client.listPlugins()).toEqual([]); + expect(await client.invokePluginRpc("internal-seam", "state", {})).toEqual({ workspaces: 0 }); + expect((await client.getPluginLogs("internal-seam")).map(({ message }) => message)).toContain( + "[paseo] Plugin ready", + ); + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(directory, "directory-seam"); + expect((await client.listPlugins()).map(({ id }) => id)).toEqual(["directory-seam"]); + expect(await client.invokePluginRpc("directory-seam", "state", {})).toEqual({ workspaces: 0 }); + await client.createWorkspace({ + idempotencyKey: "internal-seam-workspace", + source: { kind: "directory", path: workspaceDirectory }, + }); + await expect + .poll(() => client.invokePluginRpc("internal-seam", "state", {})) + .toEqual({ workspaces: 1 }); + await expect + .poll(() => client.invokePluginRpc("directory-seam", "state", {})) + .toEqual({ workspaces: 1 }); + await expect + .poll(async () => + (await client.getPluginLogs("internal-seam")).some( + ({ message }) => message === "workspace created by seam fixture", + ), + ) + .toBe(true); + await client.patchDaemonConfig({ pluginsEnabled: false }); + expect(await client.invokePluginRpc("internal-seam", "state", {})).toEqual({ workspaces: 1 }); + expect(await client.listPlugins()).toMatchObject([ + { id: "directory-seam", status: "disabled" }, + ]); + } finally { + await client.close(); + await daemon.close(); + await rm(workspaceDirectory, { recursive: true, force: true }); + } +}, 60_000); diff --git a/packages/server/src/server/plugins/plugin-process.ts b/packages/server/src/server/plugins/plugin-process.ts index 1cc0f571a1..6839b60665 100644 --- a/packages/server/src/server/plugins/plugin-process.ts +++ b/packages/server/src/server/plugins/plugin-process.ts @@ -9,7 +9,8 @@ import * as pluginSharedRuntime from "@getpaseo/plugin"; import * as pluginProviderRuntime from "@getpaseo/plugin/server/provider"; import * as pluginAcpRuntime from "@getpaseo/plugin/server/acp"; import type { SettingsDefinition, PluginRpcContract } from "@getpaseo/plugin"; -import type { PluginHandlerContext } from "@getpaseo/plugin/server"; +import type { PluginHandlerContext, PluginServerContribution } from "@getpaseo/plugin/server"; +import { fileURLToPath } from "node:url"; import type { ZodType } from "zod"; import { ProviderEventSchema, @@ -23,206 +24,8 @@ import { isPluginClientOnlySdkSpecifier } from "./plugin-sdk-specifiers.js"; import { createPluginClientId } from "./plugin-session-identity.js"; import { PluginSettingsStore } from "./settings/index.js"; -let settingsStore: PluginSettingsStore | null = null; -function registerSettings(definition: SettingsDefinition) { - if (!settingsStore) throw new Error("Plugin settings storage is unavailable"); - const handlers = settingsStore.register(definition); - register(handlers.read.contract, handlers.read.handle); - register(handlers.write.contract, (input) => - handlers.write.handle(handlers.write.contract.input.parse(input)), - ); - register(handlers.reset.contract, (input) => - handlers.reset.handle(handlers.reset.contract.input.parse(input)), - ); - return handlers.settings; -} - -type RpcHandler = (input: unknown, context: PluginHandlerContext) => unknown | Promise; - -interface RegisteredRpc { - contract: PluginRpcContract; - handler: RpcHandler; -} - -const hooks = new PluginHookHandlers(() => { - send({ type: "hooks.changed", hooks: hooks.catalog() }); -}); -const handlers = new Map(); -const providers = new Map(); -const providerConnections = new Map< - string, - { connection: ProviderConnection; unsubscribe: () => void; closing?: Promise } ->(); -const pendingProviderConnections = new Map(); -let cleanup: (() => void | Promise) | null = null; -let daemonClient: DaemonClient | null = null; -let paseo: PaseoApi | null = null; -let stopping = false; const nodeRequire = createRequire(import.meta.url); -function send(message: PluginProcessMessage): void { - process.send?.(message); -} - -function sendAndWait(message: PluginProcessMessage): Promise { - return new Promise((resolve) => { - if (!process.send) { - resolve(); - return; - } - process.send(message, () => resolve()); - }); -} - -function describeError(error: unknown): string { - return error instanceof Error ? error.message : String(error); -} - -function jsonTransportValue(value: Value): Value { - const encoded = JSON.stringify(value); - if (encoded === undefined) throw new Error("Provider value is not JSON-serializable"); - return JSON.parse(encoded) as Value; -} - -function validateMethod(method: string): string { - const normalized = method.trim(); - if (!/^[a-z][a-z0-9._-]*$/.test(normalized)) { - throw new Error(`Invalid plugin RPC method: ${method}`); - } - if (handlers.has(normalized)) { - throw new Error(`Duplicate plugin RPC method: ${normalized}`); - } - return normalized; -} - -function register(contract: PluginRpcContract, handler: RpcHandler): void { - if (typeof handler !== "function") { - throw new Error(`Plugin RPC ${contract.name} must provide a handler`); - } - const method = validateMethod(contract.name); - handlers.set(method, { contract: { ...contract, name: method }, handler }); -} - -function registerProvider(provider: ProviderRegistration): void { - const id = provider.id.trim(); - if (!/^[a-z][a-z0-9._-]*$/.test(id)) { - throw new Error(`Invalid plugin provider ID: ${provider.id}`); - } - if (!provider.label.trim()) throw new Error(`Plugin provider ${id} requires a label`); - if (typeof provider.connect !== "function") { - throw new Error(`Plugin provider ${id} must implement connect()`); - } - if ( - provider.getCatalogCacheKey !== undefined && - typeof provider.getCatalogCacheKey !== "function" - ) { - throw new Error(`Invalid catalogue key callback for plugin provider ${id}`); - } - if (providers.has(id)) throw new Error(`Duplicate plugin provider ID: ${id}`); - providers.set(id, { ...provider, id }); -} - -function providerMetadata(provider: ProviderRegistration) { - return { - id: provider.id, - label: provider.label, - description: provider.description, - iconPath: provider.icon, - hasCatalogCacheKey: provider.getCatalogCacheKey !== undefined, - }; -} - -async function connectProvider( - message: Extract, -): Promise { - if (stopping) throw new Error("Plugin is stopping"); - const provider = providers.get(message.providerId); - if (!provider) throw new Error(`Unknown plugin provider: ${message.providerId}`); - if ( - providerConnections.has(message.connectionId) || - pendingProviderConnections.has(message.connectionId) - ) { - throw new Error(`Duplicate provider connection: ${message.connectionId}`); - } - const pending = { tombstoned: false }; - pendingProviderConnections.set(message.connectionId, pending); - let connection: ProviderConnection; - try { - connection = await provider.connect(message.request); - } catch (error) { - pendingProviderConnections.delete(message.connectionId); - if (pending.tombstoned || stopping) return; - throw error; - } - pendingProviderConnections.delete(message.connectionId); - if (pending.tombstoned || stopping) { - await connection.close().catch(() => undefined); - return; - } - let unsubscribe = () => {}; - unsubscribe = connection.onEvent((event) => { - try { - send({ - type: "provider.event", - connectionId: message.connectionId, - event: ProviderEventSchema.parse(jsonTransportValue(event)), - }); - } catch (error) { - providerConnections.delete(message.connectionId); - unsubscribe(); - void connection.close().catch(() => undefined); - send({ - type: "provider.closed", - connectionId: message.connectionId, - error: describeError(error), - }); - } - }); - providerConnections.set(message.connectionId, { connection, unsubscribe }); - send({ - type: "provider.connected", - connectionId: message.connectionId, - version: connection.version, - capabilities: connection.capabilities, - }); -} - -async function sendProviderInput( - message: Extract, -): Promise { - if (stopping) throw new Error("Plugin is stopping"); - const current = providerConnections.get(message.connectionId); - if (!current) throw new Error(`Unknown provider connection: ${message.connectionId}`); - if (current.closing) throw new Error("Provider connection is closing"); - await current.connection.send(message.input); - send({ - type: "provider.accepted", - connectionId: message.connectionId, - acceptanceId: message.acceptanceId, - }); -} - -// The connection stays registered until its close has reported, so shutdown -// waits for a close already in flight instead of disconnecting underneath it. -async function closeProviderConnection(connectionId: string): Promise { - const current = providerConnections.get(connectionId); - if (!current) return; - if (current.closing) return current.closing; - const closing = (async () => { - current.unsubscribe(); - try { - await current.connection.close(); - send({ type: "provider.closed", connectionId }); - } catch (error) { - send({ type: "provider.closed", connectionId, error: describeError(error) }); - } finally { - providerConnections.delete(connectionId); - } - })(); - current.closing = closing; - return closing; -} - function runtimeRequire(name: string): unknown { if (isPluginClientOnlySdkSpecifier(name)) { throw new Error(`${name} is available only in plugin client code`); @@ -236,7 +39,7 @@ function runtimeRequire(name: string): unknown { return nodeRequire(name); } -function evaluateBundle(bundle: string): void { +function evaluateBundle(bundle: string): PluginServerContribution { const evaluate: (source: string) => unknown = globalThis.eval; const factory = evaluate(bundle); if (typeof factory !== "function") throw new Error("Plugin server bundle is not executable"); @@ -246,230 +49,474 @@ function evaluateBundle(bundle: string): void { if (typeof setup !== "function") { throw new Error("Plugin server bundle must default export a function"); } - const contributedCleanup = setup({ - handle: register, - registerProvider, - registerSettings, - on: hooks.on, - before: hooks.before, - }); - if (typeof contributedCleanup !== "function") { - throw new Error("Plugin contribution must return a cleanup function"); + return setup as PluginServerContribution; +} + +function describeError(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +export interface PluginWorkerChannel { + send(message: PluginProcessMessage, callback?: () => void): void; + onMessage(handler: (message: PluginProcessRequest) => void): () => void; + disconnect(): void; +} + +export function createPluginWorker(options: { + channel: PluginWorkerChannel; + contribute: PluginServerContribution; +}): { shutdown(): Promise } { + const { channel, contribute } = options; + let settingsStore: PluginSettingsStore | null = null; + function registerSettings(definition: SettingsDefinition) { + if (!settingsStore) throw new Error("Plugin settings storage is unavailable"); + const handlers = settingsStore.register(definition); + register(handlers.read.contract, handlers.read.handle); + register(handlers.write.contract, (input) => + handlers.write.handle(handlers.write.contract.input.parse(input)), + ); + register(handlers.reset.contract, (input) => + handlers.reset.handle(handlers.reset.contract.input.parse(input)), + ); + return handlers.settings; } - cleanup = contributedCleanup; -} -const transportFactory = createPluginDaemonTransportFactory({ - send, - onMessage(handler) { - process.on("message", handler); - return () => process.off("message", handler); - }, -}); + type RpcHandler = (input: unknown, context: PluginHandlerContext) => unknown | Promise; -async function initialize(message: Extract) { - daemonClient = new DaemonClient({ - url: `ipc://plugin/${encodeURIComponent(message.pluginId)}`, - clientId: createPluginClientId(message.pluginId), - clientType: "cli", - appVersion: message.appVersion, - // The runtime re-attaches a session when the daemon drops this socket. - reconnect: { enabled: true }, - transportFactory, - }); - paseo = createPaseoApi(daemonClient); - await daemonClient.connect(); - settingsStore = message.settingsDirectory - ? new PluginSettingsStore(message.settingsDirectory, (settingsId) => - send({ type: "settings.changed", settingsId }), - ) - : null; - evaluateBundle(message.bundle); - send({ - type: "ready", - methods: [...handlers.keys()].sort(), - hooks: hooks.catalog(), - providers: [...providers.values()] - .sort((left, right) => left.id.localeCompare(right.id)) - .map(providerMetadata), - }); -} - -async function shutdown(): Promise { - if (stopping) return; - stopping = true; - const releaseApi = paseo - ?.dispose() - .catch((error) => console.error("Plugin API cleanup failed", error)); - hooks.close(); - for (const pending of pendingProviderConnections.values()) pending.tombstoned = true; - const currentCleanup = cleanup; - cleanup = null; - try { - await currentCleanup?.(); - } catch (error) { - console.error("Plugin cleanup failed", error); + interface RegisteredRpc { + contract: PluginRpcContract; + handler: RpcHandler; } - await Promise.all([...providerConnections.keys()].map(closeProviderConnection)); - await releaseApi; - await daemonClient?.close().catch(() => undefined); - await sendAndWait({ type: "paseo_close" }); - daemonClient = null; - paseo = null; - process.disconnect(); -} -process.on("message", (rawMessage: unknown) => { - const parsed = PluginProcessRequestSchema.safeParse(rawMessage); - if (!parsed.success) { - const value = rawMessage as { connectionId?: unknown; acceptanceId?: unknown } | null; - if (value && typeof value.connectionId === "string" && typeof value.acceptanceId === "string") { - send({ - type: "provider.rejected", - connectionId: value.connectionId, - acceptanceId: value.acceptanceId, - error: `Invalid provider input: ${parsed.error.message}`, - }); - void closeProviderConnection(value.connectionId); + const hooks = new PluginHookHandlers(() => { + send({ type: "hooks.changed", hooks: hooks.catalog() }); + }); + const handlers = new Map(); + const providers = new Map(); + const providerConnections = new Map< + string, + { connection: ProviderConnection; unsubscribe: () => void; closing?: Promise } + >(); + const pendingProviderConnections = new Map(); + let cleanup: (() => void | Promise) | null = null; + let daemonClient: DaemonClient | null = null; + let paseo: PaseoApi | null = null; + let stopping = false; + function send(message: PluginProcessMessage): void { + channel.send(message); + } + + function sendAndWait(message: PluginProcessMessage): Promise { + return new Promise((resolve) => { + channel.send(message, () => resolve()); + }); + } + + function jsonTransportValue(value: Value): Value { + const encoded = JSON.stringify(value); + if (encoded === undefined) throw new Error("Provider value is not JSON-serializable"); + return JSON.parse(encoded) as Value; + } + + function validateMethod(method: string): string { + const normalized = method.trim(); + if (!/^[a-z][a-z0-9._-]*$/.test(normalized)) { + throw new Error(`Invalid plugin RPC method: ${method}`); + } + if (handlers.has(normalized)) { + throw new Error(`Duplicate plugin RPC method: ${normalized}`); + } + return normalized; + } + + function register(contract: PluginRpcContract, handler: RpcHandler): void { + if (typeof handler !== "function") { + throw new Error(`Plugin RPC ${contract.name} must provide a handler`); + } + const method = validateMethod(contract.name); + handlers.set(method, { contract: { ...contract, name: method }, handler }); + } + + function registerProvider(provider: ProviderRegistration): void { + const id = provider.id.trim(); + if (!/^[a-z][a-z0-9._-]*$/.test(id)) { + throw new Error(`Invalid plugin provider ID: ${provider.id}`); + } + if (!provider.label.trim()) throw new Error(`Plugin provider ${id} requires a label`); + if (typeof provider.connect !== "function") { + throw new Error(`Plugin provider ${id} must implement connect()`); + } + if ( + provider.getCatalogCacheKey !== undefined && + typeof provider.getCatalogCacheKey !== "function" + ) { + throw new Error(`Invalid catalogue key callback for plugin provider ${id}`); + } + if (providers.has(id)) throw new Error(`Duplicate plugin provider ID: ${id}`); + providers.set(id, { ...provider, id }); + } + + function providerMetadata(provider: ProviderRegistration) { + return { + id: provider.id, + label: provider.label, + description: provider.description, + iconPath: provider.icon, + hasCatalogCacheKey: provider.getCatalogCacheKey !== undefined, + }; + } + + async function connectProvider( + message: Extract, + ): Promise { + if (stopping) throw new Error("Plugin is stopping"); + const provider = providers.get(message.providerId); + if (!provider) throw new Error(`Unknown plugin provider: ${message.providerId}`); + if ( + providerConnections.has(message.connectionId) || + pendingProviderConnections.has(message.connectionId) + ) { + throw new Error(`Duplicate provider connection: ${message.connectionId}`); + } + const pending = { tombstoned: false }; + pendingProviderConnections.set(message.connectionId, pending); + let connection: ProviderConnection; + try { + connection = await provider.connect(message.request); + } catch (error) { + pendingProviderConnections.delete(message.connectionId); + if (pending.tombstoned || stopping) return; + throw error; + } + pendingProviderConnections.delete(message.connectionId); + if (pending.tombstoned || stopping) { + await connection.close().catch(() => undefined); return; } - send({ type: "fatal", error: `Invalid plugin process request: ${parsed.error.message}` }); - void shutdown(); - return; - } - const message = parsed.data; - if (message.type === "initialize") { - void initialize(message).catch(async (error) => { - send({ type: "fatal", error: describeError(error) }); - await paseo - ?.dispose() - .catch((failure) => console.error("Plugin API cleanup failed", failure)); - await daemonClient?.close().catch(() => undefined); + let unsubscribe = () => {}; + unsubscribe = connection.onEvent((event) => { + try { + send({ + type: "provider.event", + connectionId: message.connectionId, + event: ProviderEventSchema.parse(jsonTransportValue(event)), + }); + } catch (error) { + providerConnections.delete(message.connectionId); + unsubscribe(); + void connection.close().catch(() => undefined); + send({ + type: "provider.closed", + connectionId: message.connectionId, + error: describeError(error), + }); + } }); - return; - } - if (message.type === "shutdown") { - void shutdown(); - return; - } - if (stopping) { - if (message.type === "provider.catalog_key") { - send({ type: "error", requestId: message.requestId, error: "Plugin is stopping" }); - } else if (message.type === "provider.connect") { - send({ - type: "provider.connect_failed", - connectionId: message.connectionId, - error: "Plugin is stopping", - }); - } else if (message.type === "provider.send") { - send({ - type: "provider.rejected", - connectionId: message.connectionId, - acceptanceId: message.acceptanceId, - error: "Plugin is stopping", - }); - } else if (message.type === "provider.close") { - send({ type: "provider.closed", connectionId: message.connectionId }); - } - return; - } - if (message.type === "provider.catalog_key") { - void (async () => { - const provider = providers.get(message.providerId); - if (!provider) throw new Error(`Unknown provider: ${message.providerId}`); - const output = await provider.getCatalogCacheKey?.(message.options); - if (output !== undefined && typeof output !== "string") - throw new Error("Invalid catalogue key"); - send({ type: "result", requestId: message.requestId, output }); - })().catch((error) => - send({ type: "error", requestId: message.requestId, error: describeError(error) }), - ); - return; - } - if (message.type === "provider.connect") { - void connectProvider(message).catch((error) => { - if (stopping) return; - send({ - type: "provider.connect_failed", - connectionId: message.connectionId, - error: describeError(error), - }); - }); - return; - } - if (message.type === "provider.send") { - void sendProviderInput(message).catch((error) => { - if (stopping) return; - send({ - type: "provider.rejected", - connectionId: message.connectionId, - acceptanceId: message.acceptanceId, - error: describeError(error), - }); - }); - return; - } - if (message.type === "provider.close") { - void closeProviderConnection(message.connectionId); - return; - } - if (message.type === "paseo_frame" || message.type === "paseo_close") return; - if (isHookMessage(message)) { - handleHookMessage(message); - return; - } - const registered = handlers.get(message.method); - if (!registered) { + providerConnections.set(message.connectionId, { connection, unsubscribe }); send({ - type: "error", - requestId: message.requestId, - error: `Unknown RPC method: ${message.method}`, + type: "provider.connected", + connectionId: message.connectionId, + version: connection.version, + capabilities: connection.capabilities, }); - return; } - void registered.contract.input - .parseAsync(message.input) - .then((input) => { - if (!paseo) throw new Error("Plugin Paseo API is unavailable"); - return registered.handler(input, { paseo }); - }) - .then((output) => registered.contract.output.parseAsync(output)) - .then( - (output) => send({ type: "result", requestId: message.requestId, output }), - (error) => send({ type: "error", requestId: message.requestId, error: describeError(error) }), - ); -}); -function handleHookMessage( - message: Extract, -): void { - if (message.type === "hook.cancel") { - hooks.cancel(message.requestId); - return; + async function sendProviderInput( + message: Extract, + ): Promise { + if (stopping) throw new Error("Plugin is stopping"); + const current = providerConnections.get(message.connectionId); + if (!current) throw new Error(`Unknown provider connection: ${message.connectionId}`); + if (current.closing) throw new Error("Provider connection is closing"); + await current.connection.send(message.input); + send({ + type: "provider.accepted", + connectionId: message.connectionId, + acceptanceId: message.acceptanceId, + }); } - if (message.type === "hook") { - if (!paseo) { + + // The connection stays registered until its close has reported, so shutdown + // waits for a close already in flight instead of disconnecting underneath it. + async function closeProviderConnection(connectionId: string): Promise { + const current = providerConnections.get(connectionId); + if (!current) return; + if (current.closing) return current.closing; + const closing = (async () => { + current.unsubscribe(); + try { + await current.connection.close(); + send({ type: "provider.closed", connectionId }); + } catch (error) { + send({ type: "provider.closed", connectionId, error: describeError(error) }); + } finally { + providerConnections.delete(connectionId); + } + })(); + current.closing = closing; + return closing; + } + + const transportFactory = createPluginDaemonTransportFactory({ + send, + onMessage(handler) { + return channel.onMessage(handler); + }, + }); + + async function initialize(message: Extract) { + daemonClient = new DaemonClient({ + url: `ipc://plugin/${encodeURIComponent(message.pluginId)}`, + clientId: createPluginClientId(message.pluginId), + clientType: "cli", + appVersion: message.appVersion, + // The runtime re-attaches a session when the daemon drops this socket. + reconnect: { enabled: true }, + transportFactory, + }); + paseo = createPaseoApi(daemonClient); + await daemonClient.connect(); + settingsStore = message.settingsDirectory + ? new PluginSettingsStore(message.settingsDirectory, (settingsId) => + send({ type: "settings.changed", settingsId }), + ) + : null; + const contributedCleanup = contribute({ + handle: (contract, handler) => + register(contract, (input, context) => handler(contract.input.parse(input), context)), + registerProvider, + registerSettings, + on: hooks.on, + before: hooks.before, + }); + if (typeof contributedCleanup !== "function") { + throw new Error("Plugin contribution must return a cleanup function"); + } + cleanup = contributedCleanup; + send({ + type: "ready", + methods: [...handlers.keys()].sort(), + hooks: hooks.catalog(), + providers: [...providers.values()] + .sort((left, right) => left.id.localeCompare(right.id)) + .map(providerMetadata), + }); + } + + async function shutdown(): Promise { + if (stopping) return; + stopping = true; + const releaseApi = paseo + ?.dispose() + .catch((error) => console.error("Plugin API cleanup failed", error)); + hooks.close(); + for (const pending of pendingProviderConnections.values()) pending.tombstoned = true; + const currentCleanup = cleanup; + cleanup = null; + try { + await currentCleanup?.(); + } catch (error) { + console.error("Plugin cleanup failed", error); + } + await Promise.all([...providerConnections.keys()].map(closeProviderConnection)); + await releaseApi; + await daemonClient?.close().catch(() => undefined); + await sendAndWait({ type: "paseo_close" }); + daemonClient = null; + paseo = null; + channel.disconnect(); + } + + channel.onMessage((rawMessage: unknown) => { + const parsed = PluginProcessRequestSchema.safeParse(rawMessage); + if (!parsed.success) { + const value = rawMessage as { connectionId?: unknown; acceptanceId?: unknown } | null; + if ( + value && + typeof value.connectionId === "string" && + typeof value.acceptanceId === "string" + ) { + send({ + type: "provider.rejected", + connectionId: value.connectionId, + acceptanceId: value.acceptanceId, + error: `Invalid provider input: ${parsed.error.message}`, + }); + void closeProviderConnection(value.connectionId); + return; + } + send({ type: "fatal", error: `Invalid plugin process request: ${parsed.error.message}` }); + void shutdown(); + return; + } + const message = parsed.data; + if (message.type === "initialize") { + void initialize(message).catch(async (error) => { + send({ type: "fatal", error: describeError(error) }); + await paseo + ?.dispose() + .catch((failure) => console.error("Plugin API cleanup failed", failure)); + await daemonClient?.close().catch(() => undefined); + }); + return; + } + if (message.type === "shutdown") { + void shutdown(); + return; + } + if (stopping) { + if (message.type === "provider.catalog_key") { + send({ type: "error", requestId: message.requestId, error: "Plugin is stopping" }); + } else if (message.type === "provider.connect") { + send({ + type: "provider.connect_failed", + connectionId: message.connectionId, + error: "Plugin is stopping", + }); + } else if (message.type === "provider.send") { + send({ + type: "provider.rejected", + connectionId: message.connectionId, + acceptanceId: message.acceptanceId, + error: "Plugin is stopping", + }); + } else if (message.type === "provider.close") { + send({ type: "provider.closed", connectionId: message.connectionId }); + } + return; + } + if (message.type === "provider.catalog_key") { + void (async () => { + const provider = providers.get(message.providerId); + if (!provider) throw new Error(`Unknown provider: ${message.providerId}`); + const output = await provider.getCatalogCacheKey?.(message.options); + if (output !== undefined && typeof output !== "string") + throw new Error("Invalid catalogue key"); + send({ type: "result", requestId: message.requestId, output }); + })().catch((error) => + send({ type: "error", requestId: message.requestId, error: describeError(error) }), + ); + return; + } + if (message.type === "provider.connect") { + void connectProvider(message).catch((error) => { + if (stopping) return; + send({ + type: "provider.connect_failed", + connectionId: message.connectionId, + error: describeError(error), + }); + }); + return; + } + if (message.type === "provider.send") { + void sendProviderInput(message).catch((error) => { + if (stopping) return; + send({ + type: "provider.rejected", + connectionId: message.connectionId, + acceptanceId: message.acceptanceId, + error: describeError(error), + }); + }); + return; + } + if (message.type === "provider.close") { + void closeProviderConnection(message.connectionId); + return; + } + if (message.type === "paseo_frame" || message.type === "paseo_close") return; + if (isHookMessage(message)) { + handleHookMessage(message); + return; + } + const registered = handlers.get(message.method); + if (!registered) { send({ type: "error", requestId: message.requestId, - error: "Plugin Paseo API is unavailable", + error: `Unknown RPC method: ${message.method}`, }); return; } - void hooks.invoke(message.requestId, message.kind, message.name, message.input, paseo).then( - (output) => { - return send({ type: "result", requestId: message.requestId, output }); - }, - (error) => { - return send({ type: "error", requestId: message.requestId, error: describeError(error) }); - }, - ); - return; + void registered.contract.input + .parseAsync(message.input) + .then((input) => { + if (!paseo) throw new Error("Plugin Paseo API is unavailable"); + return registered.handler(input, { paseo }); + }) + .then((output) => registered.contract.output.parseAsync(output)) + .then( + (output) => send({ type: "result", requestId: message.requestId, output }), + (error) => + send({ type: "error", requestId: message.requestId, error: describeError(error) }), + ); + }); + + function handleHookMessage( + message: Extract, + ): void { + if (message.type === "hook.cancel") { + hooks.cancel(message.requestId); + return; + } + if (message.type === "hook") { + if (!paseo) { + send({ + type: "error", + requestId: message.requestId, + error: "Plugin Paseo API is unavailable", + }); + return; + } + void hooks.invoke(message.requestId, message.kind, message.name, message.input, paseo).then( + (output) => { + return send({ type: "result", requestId: message.requestId, output }); + }, + (error) => { + return send({ type: "error", requestId: message.requestId, error: describeError(error) }); + }, + ); + return; + } } + + function isHookMessage( + message: PluginProcessRequest, + ): message is Extract { + return message.type === "hook" || message.type === "hook.cancel"; + } + return { shutdown }; } -function isHookMessage( - message: PluginProcessRequest, -): message is Extract { - return message.type === "hook" || message.type === "hook.cancel"; +if (process.argv[1] === fileURLToPath(import.meta.url)) { + let worker: ReturnType | null = null; + const handlers = new Set<(message: PluginProcessRequest) => void>(); + const channel: PluginWorkerChannel = { + send(message, callback) { + if (callback) process.send?.(message, () => callback()); + else process.send?.(message); + }, + onMessage(handler) { + handlers.add(handler); + return () => { + handlers.delete(handler); + }; + }, + disconnect() { + process.disconnect(); + }, + }; + process.on("message", (raw) => { + const parsed = PluginProcessRequestSchema.safeParse(raw); + if (!worker && parsed.success && parsed.data.type === "initialize") { + try { + worker = createPluginWorker({ channel, contribute: evaluateBundle(parsed.data.bundle) }); + } catch (error) { + channel.send({ type: "fatal", error: describeError(error) }); + return; + } + } + if (worker) for (const handler of handlers) handler(raw as PluginProcessRequest); + }); } diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index 473cc6f359..f5a917ebcb 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -27,6 +27,8 @@ import type { } from "./plugin-process-protocol.js"; import { PluginProcessMessageSchema } from "./plugin-process-protocol.js"; import { PluginSessionSocket } from "./session-socket.js"; +import { InternalPluginChild } from "./internal-child.js"; +import type { PluginServerContribution } from "@getpaseo/plugin/server"; const CLIENT_ENTRY_FILENAMES = ["index.client.ts", "index.client.tsx"] as const; const SERVER_ENTRY_FILENAMES = ["index.server.ts", "index.server.tsx"] as const; @@ -332,6 +334,18 @@ export class PluginRuntime { this.appendLog(pluginId, "stdout", "[paseo] Plugin ready"); } + async startInternalPlugin(input: { + id: string; + directory: string; + contribute: PluginServerContribution; + }): Promise { + if (this.plugins.has(input.id)) throw new Error(`Plugin is already running: ${input.id}`); + this.appendLog(input.id, "stdout", "[paseo] Loading plugin"); + const loaded = await this.loadDirectoryPlugin(input.id, input.directory, input.contribute); + this.plugins.set(input.id, loaded); + this.appendLog(input.id, "stdout", "[paseo] Plugin ready"); + } + async validatePlugin(configuredPath: string): Promise { const directory = path.resolve(configuredPath); const manifest = await readPluginManifest(directory); @@ -551,17 +565,18 @@ export class PluginRuntime { private async loadDirectoryPlugin( pluginId: string, configuredPath: string, + contribute?: PluginServerContribution, ): Promise { const directory = path.resolve(configuredPath); const manifest = await readPluginManifest(directory); assertPluginCompatibility({ ...manifest, version: this.daemonVersion, runtime: "daemon" }); - const entryPaths = await resolveEntryPaths(directory); - const bundles = await compilePlugin(entryPaths); - const serverBundle = bundles.serverBundle; - if (!serverBundle) { + const entryPaths = contribute ? null : await resolveEntryPaths(directory); + const bundles = entryPaths ? await compilePlugin(entryPaths) : null; + const serverBundle = contribute ? "" : bundles?.serverBundle; + if (!contribute && !serverBundle) { return { id: pluginId, - clientBundle: bundles.clientBundle ?? "", + clientBundle: bundles?.clientBundle ?? "", requirements: manifest.requirements, methods: new Set(), hooks: { events: [], before: [] }, @@ -577,7 +592,7 @@ export class PluginRuntime { } const sessionHost = this.sessionHost; if (!sessionHost) throw new Error("Plugin Paseo session host is not attached"); - const child = this.spawnChild(); + const child = contribute ? new InternalPluginChild(contribute) : this.spawnChild(); const outputCapture = new PluginOutputCapture(child, (stream, message) => { this.appendLog(pluginId, stream, message); }); @@ -654,7 +669,7 @@ export class PluginRuntime { type: "initialize", pluginId, appVersion: this.daemonVersion, - bundle: serverBundle, + bundle: serverBundle ?? "", settingsDirectory: this.dependencies.settingsDirectory ? path.join(this.dependencies.settingsDirectory, pluginId) : undefined, @@ -669,7 +684,7 @@ export class PluginRuntime { } loaded = { id: pluginId, - clientBundle: bundles.clientBundle ?? "", + clientBundle: bundles?.clientBundle ?? "", requirements: manifest.requirements, methods: new Set(ready.methods), hooks: ready.hooks ?? { events: [], before: [] }, diff --git a/packages/server/src/server/plugins/test-fixtures/internal-seam/index.server.ts b/packages/server/src/server/plugins/test-fixtures/internal-seam/index.server.ts new file mode 100644 index 0000000000..291cbd705a --- /dev/null +++ b/packages/server/src/server/plugins/test-fixtures/internal-seam/index.server.ts @@ -0,0 +1,19 @@ +import { defineRpc } from "@getpaseo/plugin"; +import type { PluginServerContext } from "@getpaseo/plugin/server"; +import { z } from "zod"; + +const state = defineRpc({ + name: "state", + input: z.object({}), + output: z.object({ workspaces: z.number() }), +}); + +export default function contribute(server: PluginServerContext) { + let workspaces = 0; + server.handle(state, () => ({ workspaces })); + server.on("workspace.created", () => { + workspaces += 1; + console.log("workspace created by seam fixture"); + }); + return () => {}; +} diff --git a/packages/server/src/server/plugins/test-fixtures/internal-seam/paseo-plugin.json b/packages/server/src/server/plugins/test-fixtures/internal-seam/paseo-plugin.json new file mode 100644 index 0000000000..e84dff55d8 --- /dev/null +++ b/packages/server/src/server/plugins/test-fixtures/internal-seam/paseo-plugin.json @@ -0,0 +1 @@ +{ "id": "internal-seam", "requirements": { "paseo": ">=0.8.0" } } diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index 9cdce46865..8a72b975e5 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -1,3 +1,4 @@ +import type { InternalPlugin } from "../../plugins/index.js"; import os from "node:os"; import path from "node:path"; import { mkdir, mkdtemp, rm } from "node:fs/promises"; @@ -49,6 +50,7 @@ interface TestPaseoDaemonOptions { agentProfiles?: AgentProfile[]; autoArchiveAfterMerge?: boolean; pluginsEnabled?: PaseoDaemonConfig["pluginsEnabled"]; + internalPlugins?: readonly InternalPlugin[]; plugins?: PaseoDaemonConfig["plugins"]; } @@ -100,6 +102,7 @@ export async function createTestPaseoDaemon( const { config, paseoHomeRoot, paseoHome, staticDir } = await prepareTestDaemonConfig(options); const logger = options.logger ?? pino({ level: "silent" }); const daemon = await createPaseoDaemon(config, logger, { + internalPlugins: options.internalPlugins, serverFeatureOverrides: { daemonStatusRpc: options.daemonStatusRpcCapability, relayConfig: options.relayConfigCapability,