feat(server): run internal plugins in-process through the plugin runtime

This commit is contained in:
Mohamed Boudra
2026-09-26 20:52:37 +02:00
parent d7b7016cc9
commit c03853f4e1
11 changed files with 730 additions and 416 deletions
+9
View File
@@ -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[] = [];
+3
View File
@@ -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"),
});
+25 -4
View File
@@ -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<void>;
startPlugin(pluginId: string, path: string, canPublish: () => boolean): Promise<void>;
startInternalPlugin?(plugin: InternalPlugin): Promise<void>;
stopPluginById(pluginId: string): Promise<boolean>;
stopAll(): Promise<void>;
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<PluginRuntime["catalog"]> {
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<PluginListItem> {
@@ -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<void> {
const stopping = Object.keys(this.configStore.get().plugins ?? {}).map((id) =>
this.stopPlugin(id),
);
await Promise.all(stopping);
}
private async stopAll(): Promise<void> {
for (const pluginId of this.providerIdsByPlugin.keys()) {
this.removeProviderRegistrations(pluginId);
@@ -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<string>();
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!);
}
}
}
});
@@ -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<InternalPluginChild>();
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<keyof typeof originalConsole>) {
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<keyof typeof originalConsole>) {
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<typeof createPluginWorker>;
private readonly workerMessages = new Set<(message: PluginProcessRequest) => void>();
private closing: Promise<void> | 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);
});
}
}
@@ -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);
@@ -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<Schema extends ZodType>(definition: SettingsDefinition<Schema>) {
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<unknown>;
interface RegisteredRpc {
contract: PluginRpcContract;
handler: RpcHandler;
}
const hooks = new PluginHookHandlers(() => {
send({ type: "hooks.changed", hooks: hooks.catalog() });
});
const handlers = new Map<string, RegisteredRpc>();
const providers = new Map<string, ProviderRegistration>();
const providerConnections = new Map<
string,
{ connection: ProviderConnection; unsubscribe: () => void; closing?: Promise<void> }
>();
const pendingProviderConnections = new Map<string, { tombstoned: boolean }>();
let cleanup: (() => void | Promise<void>) | 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<void> {
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): 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<PluginProcessRequest, { type: "provider.connect" }>,
): Promise<void> {
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<PluginProcessRequest, { type: "provider.send" }>,
): Promise<void> {
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<void> {
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<void> } {
const { channel, contribute } = options;
let settingsStore: PluginSettingsStore | null = null;
function registerSettings<Schema extends ZodType>(definition: SettingsDefinition<Schema>) {
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<unknown>;
async function initialize(message: Extract<PluginProcessRequest, { type: "initialize" }>) {
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<void> {
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<string, RegisteredRpc>();
const providers = new Map<string, ProviderRegistration>();
const providerConnections = new Map<
string,
{ connection: ProviderConnection; unsubscribe: () => void; closing?: Promise<void> }
>();
const pendingProviderConnections = new Map<string, { tombstoned: boolean }>();
let cleanup: (() => void | Promise<void>) | 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<void> {
return new Promise((resolve) => {
channel.send(message, () => resolve());
});
}
function jsonTransportValue<Value>(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<PluginProcessRequest, { type: "provider.connect" }>,
): Promise<void> {
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<PluginProcessRequest, { type: "hook" | "hook.cancel" }>,
): void {
if (message.type === "hook.cancel") {
hooks.cancel(message.requestId);
return;
async function sendProviderInput(
message: Extract<PluginProcessRequest, { type: "provider.send" }>,
): Promise<void> {
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<void> {
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<PluginProcessRequest, { type: "initialize" }>) {
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<void> {
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<PluginProcessRequest, { type: "hook" | "hook.cancel" }>,
): 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<PluginProcessRequest, { type: "hook" | "hook.cancel" }> {
return message.type === "hook" || message.type === "hook.cancel";
}
return { shutdown };
}
function isHookMessage(
message: PluginProcessRequest,
): message is Extract<PluginProcessRequest, { type: "hook" | "hook.cancel" }> {
return message.type === "hook" || message.type === "hook.cancel";
if (process.argv[1] === fileURLToPath(import.meta.url)) {
let worker: ReturnType<typeof createPluginWorker> | 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);
});
}
+23 -8
View File
@@ -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<void> {
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<void> {
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<LoadedPlugin> {
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: [] },
@@ -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 () => {};
}
@@ -0,0 +1 @@
{ "id": "internal-seam", "requirements": { "paseo": ">=0.8.0" } }
@@ -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,