diff --git a/packages/server/src/server/plugins/index.posix.test.ts b/packages/server/src/server/plugins/index.posix.test.ts index 1f8b3f124b..60216ff0fa 100644 --- a/packages/server/src/server/plugins/index.posix.test.ts +++ b/packages/server/src/server/plugins/index.posix.test.ts @@ -20,6 +20,7 @@ import { DaemonConfigStore } from "../daemon-config-store.js"; import { PluginService } from "./index.js"; import { BuiltinPluginLoader } from "./builtin/index.js"; import { ManagedPluginSources } from "./managed-source.js"; +import type { PluginProviderMetadata } from "./plugin-process-protocol.js"; import { startNpmRegistry, npmPluginPackages, @@ -179,6 +180,55 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) { return { runtime, started, releaseStart, starts }; } +function createGatedRuntime(providers: Record = {}) { + const gates = new Map>(); + const gate = (pluginId: string) => { + const existing = gates.get(pluginId); + if (existing) return existing; + const created = Promise.withResolvers(); + gates.set(pluginId, created); + return created; + }; + const begun: string[] = []; + const running = new Set(); + const runtime: TestPluginRuntime = { + ...emptyUsageRuntime, + catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })), + invoke: async () => undefined, + getLogs: () => [], + clearLogs: () => undefined, + getProviderRegistrations: (pluginId) => + running.has(pluginId) ? (providers[pluginId] ?? []) : [], + startPlugin: async (pluginId) => { + begun.push(pluginId); + await gate(pluginId).promise; + running.add(pluginId); + }, + stopPluginById: async (pluginId) => running.delete(pluginId), + stopAll: async () => { + running.clear(); + }, + subscribe: () => () => undefined, + bindPaseoSessionHost: () => undefined, + }; + return { + runtime, + begun, + finish: (pluginId: string) => gate(pluginId).resolve(), + fail: (pluginId: string, error: Error) => gate(pluginId).reject(error), + }; +} + +function nextNotification(service: PluginService, pluginId: string): Promise { + return new Promise((resolve) => { + const unsubscribe = service.subscribe((notified) => { + if (notified !== pluginId) return; + unsubscribe(); + resolve(); + }); + }); +} + describe("PluginService", () => { it.each([ { shadow: false, runtime: "subprocess" }, @@ -372,15 +422,127 @@ describe("PluginService", () => { "startup-first": { source: "directory", path: first }, "startup-second": { source: "directory", path: second }, }); - const snapshots: string[][] = []; - service.subscribe(() => snapshots.push(catalogIds(service))); + const notifications: { pluginId: string; running: boolean }[] = []; + service.subscribe((pluginId) => + notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }), + ); await service.start(); - expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]); + expect(notifications).toEqual([ + { pluginId: "startup-first", running: true }, + { pluginId: "startup-second", running: true }, + ]); + expect(catalogIds(service)).toEqual(["startup-first", "startup-second"]); await service.stopAllPlugins(); }, 20_000); + it("starts configured plugins concurrently and publishes them in config order", async () => { + const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-")); + roots.push(home); + const gated = createGatedRuntime(); + const service = createService( + home, + { + alpha: { source: "directory", path: "/plugins/alpha" }, + beta: { source: "directory", path: "/plugins/beta" }, + gamma: { source: "directory", path: "/plugins/gamma" }, + }, + { runtime: gated.runtime }, + ); + const notified: string[] = []; + service.subscribe((pluginId) => notified.push(pluginId)); + let started = false; + const starting = (async () => { + await service.start(); + started = true; + })(); + + await expect.poll(() => gated.begun).toEqual(["alpha", "beta", "gamma"]); + + gated.finish("gamma"); + const alphaPublished = nextNotification(service, "alpha"); + gated.finish("alpha"); + await alphaPublished; + expect(notified).toEqual(["alpha"]); + expect(started).toBe(false); + + gated.finish("beta"); + await starting; + expect(notified).toEqual(["alpha", "beta", "gamma"]); + expect(catalogIds(service)).toEqual(["alpha", "beta", "gamma"]); + await service.stopAllPlugins(); + }); + + it("starts the other configured plugins when one fails while they are still starting", async () => { + const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-")); + roots.push(home); + const gated = createGatedRuntime(); + const service = createService( + home, + { + broken: { source: "directory", path: "/plugins/broken" }, + healthy: { source: "directory", path: "/plugins/healthy" }, + }, + { runtime: gated.runtime }, + ); + const notified: string[] = []; + service.subscribe((pluginId) => notified.push(pluginId)); + const starting = service.start(); + + await expect.poll(() => gated.begun).toEqual(["broken", "healthy"]); + const brokenPublished = nextNotification(service, "broken"); + gated.fail("broken", new Error("startup exploded")); + await brokenPublished; + gated.finish("healthy"); + await starting; + + expect(notified).toEqual(["broken", "healthy"]); + expect(catalogIds(service)).toEqual(["healthy"]); + expect(await service.listPlugins()).toMatchObject([ + { id: "broken", status: "failed", error: "startup exploded" }, + { id: "healthy", status: "running" }, + ]); + await service.stopAllPlugins(); + }); + + it("keeps a provider ID with the plugin listed first when both start at once", async () => { + const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-")); + roots.push(home); + const gated = createGatedRuntime({ + first: [{ id: "shared-agent", label: "First agent" }], + second: [{ id: "shared-agent", label: "Second agent" }], + }); + const service = createService( + home, + { + first: { source: "directory", path: "/plugins/first" }, + second: { source: "directory", path: "/plugins/second" }, + }, + { runtime: gated.runtime }, + ); + const starting = service.start(); + + await expect.poll(() => gated.begun).toEqual(["first", "second"]); + gated.finish("second"); + gated.finish("first"); + await starting; + + expect(service.getProviderRegistrations()).toMatchObject([ + { id: "shared-agent", label: "First agent" }, + ]); + expect(catalogIds(service)).toEqual(["first"]); + expect(await service.listPlugins()).toMatchObject([ + { id: "first", status: "running" }, + { + id: "second", + status: "failed", + error: 'Plugin second cannot register provider ID "shared-agent" twice', + }, + ]); + await service.stopAllPlugins(); + }); + it("uses an explicit config key, exposes reload failure, and retries from disk", async () => { const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-")); roots.push(home); diff --git a/packages/server/src/server/plugins/index.ts b/packages/server/src/server/plugins/index.ts index 5dad547d5c..65faa39f75 100644 --- a/packages/server/src/server/plugins/index.ts +++ b/packages/server/src/server/plugins/index.ts @@ -230,11 +230,18 @@ export class PluginService { } }); if (config.pluginsEnabled === true) { - for (const [pluginId, source] of Object.entries(config.plugins ?? {})) { - if (source.enabled === false) continue; - await this.startConfigured(pluginId); - this.notify(pluginId); - } + let earlierPublished: Promise = Promise.resolve(); + await Promise.all( + Object.entries(config.plugins ?? {}) + .filter(([, source]) => source.enabled !== false) + .map(async ([pluginId]) => { + const publishTurn = earlierPublished; + const starting = this.startConfigured(pluginId, publishTurn); + earlierPublished = Promise.all([publishTurn, starting]); + await starting; + this.notify(pluginId); + }), + ); } this.configStore.onFieldChange("pluginsEnabled", (value) => { this.handleGlobalSwitch(value === true); @@ -560,12 +567,12 @@ export class PluginService { }).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins")); } - private async startConfigured(pluginId: string): Promise { + private async startConfigured(pluginId: string, publishTurn?: Promise): Promise { const source = this.configStore.get().plugins?.[pluginId]; if (!source || source.enabled === false || !this.canPublish(pluginId)) return; this.errors.delete(pluginId); try { - await this.startPlugin(pluginId, source.path); + await this.startPlugin(pluginId, source.path, publishTurn); } catch (error) { if (this.canPublish(pluginId)) this.recordFailure(pluginId, error); } @@ -594,8 +601,13 @@ export class PluginService { ); } - private async startPlugin(pluginId: string, sourcePath: string): Promise { + private async startPlugin( + pluginId: string, + sourcePath: string, + publishTurn?: Promise, + ): Promise { await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId)); + await publishTurn; try { await this.publishProviderRegistrations(pluginId, sourcePath); this.publishUsageSources(pluginId);