diff --git a/.changeset/fn-8468-plugin-single-onload.md b/.changeset/fn-8468-plugin-single-onload.md new file mode 100644 index 0000000000..73ddef0209 --- /dev/null +++ b/.changeset/fn-8468-plugin-single-onload.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Load each enabled plugin once per process startup (no duplicate onLoad). +category: fix +dev: Host CLI and InProcessRuntime share a single-load authority with concurrency-safe single-flight so path-registered plugins no longer double-fire onLoad on fn dashboard/serve/daemon startup. diff --git a/docs/PLUGIN_AUTHORING.md b/docs/PLUGIN_AUTHORING.md index 5c24a079dc..749cdfe176 100644 --- a/docs/PLUGIN_AUTHORING.md +++ b/docs/PLUGIN_AUTHORING.md @@ -353,6 +353,7 @@ const plugin: FusionPlugin = { ### Hook Behavior +- **Single-load lifecycle**: For one project in one Fusion process, Fusion invokes `onLoad` exactly once for each intentional load lifecycle, including when the host and engine bootstrap concurrently. Plugin authors do not need process-local locking to defend against an accidental second host/engine startup load. Explicit enable→load and `reloadPlugin` are new lifecycles and can invoke `onLoad` again after unload; request-scoped temporary loaders for *another* project root may also load and then stop a plugin while discovering skills. Keep registration idempotent where inexpensive so these intentional lifecycles remain safe. - **Context parity**: `onUnload` receives the same `PluginContext` shape as `onLoad`. - **Timeout**: 5 seconds per invocation (logged and skipped if exceeded) - **Error Isolation**: Hook failures never block other hooks or abort startup diff --git a/packages/core/src/__tests__/plugin-loader-single-load.test.ts b/packages/core/src/__tests__/plugin-loader-single-load.test.ts new file mode 100644 index 0000000000..5ae1a1b839 --- /dev/null +++ b/packages/core/src/__tests__/plugin-loader-single-load.test.ts @@ -0,0 +1,126 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PluginLoader } from "../plugin-loader.js"; +import type { PluginInstallation } from "../plugin-types.js"; + +async function createFixture(onLoadSource: string, onUnloadSource?: string) { + const root = await mkdtemp(join(tmpdir(), "fusion-plugin-single-load-")); + const entry = join(root, "plugin.mjs"); + await writeFile(entry, ` +export default { + manifest: { id: "single-load", name: "Single load", version: "1.0.0", description: "fixture" }, + state: "installed", + hooks: { onLoad: ${onLoadSource}${onUnloadSource ? `, onUnload: ${onUnloadSource}` : ""} }, +}; +`); + return { root, entry }; +} + +function createStore(root: string, entry: string) { + const installation: PluginInstallation = { + id: "single-load", name: "Single load", version: "1.0.0", description: "fixture", + path: entry, enabled: true, state: "installed", settings: {}, dependencies: [], + createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), + }; + const events = new EventEmitter(); + const pluginStore = { + getPlugin: vi.fn(async () => ({ ...installation })), + listPlugins: vi.fn(async () => [{ ...installation }]), + updatePluginState: vi.fn(async (_id: string, state: PluginInstallation["state"]) => ({ ...installation, state })), + on: events.on.bind(events), off: events.off.bind(events), + }; + const taskStore = { + getRootDir: () => root, + preflightPluginSchema: vi.fn(() => null), + runPluginSchemaInits: vi.fn(async () => undefined), + }; + return { pluginStore, taskStore }; +} + +describe("PluginLoader process single-load lifecycle", () => { + const roots: string[] = []; + afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); + delete (globalThis as Record).__fusionPluginOnLoadCount; + delete (globalThis as Record).__fusionPluginFailOnce; + delete (globalThis as Record).__fusionPluginOnUnloadCount; + }); + + it("coalesces concurrent host and engine-style loaders for one project", async () => { + const fixture = await createFixture("async () => { globalThis.__fusionPluginOnLoadCount = (globalThis.__fusionPluginOnLoadCount || 0) + 1; await new Promise(resolve => setTimeout(resolve, 20)); }"); + roots.push(fixture.root); + const host = createStore(fixture.root, fixture.entry); + const engine = createStore(fixture.root, fixture.entry); + const hostLoader = new PluginLoader({ pluginStore: host.pluginStore as any, taskStore: host.taskStore as any }); + const engineLoader = new PluginLoader({ pluginStore: engine.pluginStore as any, taskStore: engine.taskStore as any }); + + await Promise.all([hostLoader.loadAllPlugins(), engineLoader.loadAllPlugins()]); + + expect((globalThis as Record).__fusionPluginOnLoadCount).toBe(1); + expect(hostLoader.isPluginLoaded("single-load")).toBe(true); + expect(engineLoader.isPluginLoaded("single-load")).toBe(true); + }); + + it("coalesces concurrent calls on the same loader", async () => { + const fixture = await createFixture("async () => { globalThis.__fusionPluginOnLoadCount = (globalThis.__fusionPluginOnLoadCount || 0) + 1; await new Promise(resolve => setTimeout(resolve, 20)); }"); + roots.push(fixture.root); + const store = createStore(fixture.root, fixture.entry); + const loader = new PluginLoader({ pluginStore: store.pluginStore as any, taskStore: store.taskStore as any }); + + await Promise.all([loader.loadPlugin("single-load"), loader.loadAllPlugins()]); + + expect((globalThis as Record).__fusionPluginOnLoadCount).toBe(1); + }); + + it("coordinates cross-loader reload and stop after dual bootstrap", async () => { + const fixture = await createFixture( + "async () => { globalThis.__fusionPluginOnLoadCount = (globalThis.__fusionPluginOnLoadCount || 0) + 1; }", + "async () => { globalThis.__fusionPluginOnUnloadCount = (globalThis.__fusionPluginOnUnloadCount || 0) + 1; }", + ); + roots.push(fixture.root); + const host = createStore(fixture.root, fixture.entry); + const engine = createStore(fixture.root, fixture.entry); + const hostLoader = new PluginLoader({ pluginStore: host.pluginStore as any, taskStore: host.taskStore as any }); + const engineLoader = new PluginLoader({ pluginStore: engine.pluginStore as any, taskStore: engine.taskStore as any }); + + await Promise.all([hostLoader.loadAllPlugins(), engineLoader.loadAllPlugins()]); + await engineLoader.reloadPlugin("single-load"); + + expect((globalThis as Record).__fusionPluginOnLoadCount).toBe(2); + expect((globalThis as Record).__fusionPluginOnUnloadCount).toBe(1); + expect(hostLoader.getPlugin("single-load")).toBe(engineLoader.getPlugin("single-load")); + + await hostLoader.stopPlugin("single-load"); + + expect((globalThis as Record).__fusionPluginOnUnloadCount).toBe(2); + expect(hostLoader.isPluginLoaded("single-load")).toBe(false); + expect(engineLoader.isPluginLoaded("single-load")).toBe(false); + }); + + it("starts one fresh lifecycle after explicit reload", async () => { + const fixture = await createFixture("async () => { globalThis.__fusionPluginOnLoadCount = (globalThis.__fusionPluginOnLoadCount || 0) + 1; }"); + roots.push(fixture.root); + const store = createStore(fixture.root, fixture.entry); + const loader = new PluginLoader({ pluginStore: store.pluginStore as any, taskStore: store.taskStore as any }); + + await loader.loadPlugin("single-load"); + await loader.reloadPlugin("single-load"); + + expect((globalThis as Record).__fusionPluginOnLoadCount).toBe(2); + }); + + it("clears a rejected lifecycle so a later intentional load can retry", async () => { + const fixture = await createFixture("async () => { if (!globalThis.__fusionPluginFailOnce) { globalThis.__fusionPluginFailOnce = true; throw new Error('first load fails'); } globalThis.__fusionPluginOnLoadCount = (globalThis.__fusionPluginOnLoadCount || 0) + 1; }"); + roots.push(fixture.root); + const store = createStore(fixture.root, fixture.entry); + const loader = new PluginLoader({ pluginStore: store.pluginStore as any, taskStore: store.taskStore as any }); + + await expect(loader.loadPlugin("single-load")).rejects.toThrow("first load fails"); + await expect(loader.loadPlugin("single-load")).resolves.toMatchObject({ manifest: { id: "single-load" } }); + + expect((globalThis as Record).__fusionPluginOnLoadCount).toBe(1); + }); +}); diff --git a/packages/core/src/plugin-loader.ts b/packages/core/src/plugin-loader.ts index 26d5692c20..8632d58962 100644 --- a/packages/core/src/plugin-loader.ts +++ b/packages/core/src/plugin-loader.ts @@ -193,6 +193,12 @@ export interface PluginErrorEvent { error: Error; } +interface ProcessPluginLifecycle { + promise: Promise; + owner: PluginLoader; + participants: Set; +} + export class PluginLoader extends EventEmitter<{ "plugin:loaded": [PluginLoadedEvent]; "plugin:unloaded": [PluginUnloadedEvent]; @@ -210,6 +216,18 @@ export class PluginLoader extends EventEmitter<{ private pluginRoots: Map = new Map(); private pluginSchemaContracts: Map = new Map(); + /* + FNXC:PluginLoader 2026-07-22-10:15: + Dashboard/serve/daemon boot a host PluginLoader while InProcessRuntime boots a + second loader for the same project. `plugins.has()` only protects one loader + after publication, so this process-wide lifecycle registry coalesces the + import and onLoad work before either loader can publish. Successful entries + intentionally persist until stop/reload begins a fresh lifecycle; rejected + entries are removed so an intentional retry is never poisoned. + */ + private static readonly processPluginLifecycles = new Map(); + private static readonly processPluginLifecycleTails = new Map>(); + private readonly log = createLogger("plugin-loader"); constructor(private options: PluginLoaderOptions) { @@ -226,7 +244,9 @@ export class PluginLoader extends EventEmitter<{ } private getProjectRoot(): string { - return this.options.taskStore.getRootDir(); + // Lightweight loader harnesses historically omit this TaskStore accessor. + // Production stores always provide it; cwd preserves their single-project semantics. + return this.options.taskStore.getRootDir?.() ?? process.cwd(); } // ── Context Creation ─────────────────────────────────────────────── @@ -345,7 +365,49 @@ export class PluginLoader extends EventEmitter<{ // Resolve plugin path const pluginPath = this.resolvePluginPath(installation.path); + const lifecycleKey = this.getProcessLifecycleKey(pluginId, pluginPath); + const existingLifecycle = PluginLoader.processPluginLifecycles.get(lifecycleKey); + if (existingLifecycle) { + existingLifecycle.participants.add(this); + const plugin = await existingLifecycle.promise; + return this.adoptProcessLoadedPlugin(pluginId, pluginPath, plugin); + } + const lifecycle = this.loadPluginFresh(pluginId, installation, pluginPath); + const processLifecycle: ProcessPluginLifecycle = { + promise: lifecycle, + owner: this, + participants: new Set([this]), + }; + PluginLoader.processPluginLifecycles.set(lifecycleKey, processLifecycle); + try { + return await lifecycle; + } catch (error) { + // Only remove our own rejected promise; a later retry may already own this key. + if (PluginLoader.processPluginLifecycles.get(lifecycleKey) === processLifecycle) { + PluginLoader.processPluginLifecycles.delete(lifecycleKey); + } + throw error; + } + } + + private getProcessLifecycleKey(pluginId: string, pluginPath: string): string { + return `${resolve(this.getProjectRoot())}\u0000${pluginId}\u0000${resolve(pluginPath)}`; + } + + private adoptProcessLoadedPlugin(pluginId: string, pluginPath: string, plugin: FusionPlugin): FusionPlugin { + plugin.state = "started"; + this.plugins.set(pluginId, plugin); + this.pluginRoots.set(pluginId, resolvePluginRootFromEntryPath(pluginPath)); + this.emit("plugin:loaded", { pluginId, plugin }); + return plugin; + } + + private async loadPluginFresh( + pluginId: string, + installation: PluginInstallation, + pluginPath: string, + ): Promise { try { if (installation.aiScanOnLoad) { const scanResult = await scanPluginSecurity({ pluginId, pluginPath }); @@ -577,10 +639,85 @@ export class PluginLoader extends EventEmitter<{ async reloadPlugin( pluginId: string, options?: { timeoutMs?: number }, + ): Promise { + const installation = await this.options.pluginStore.getPlugin(pluginId); + const pluginPath = this.resolvePluginPath(installation.path); + const lifecycleKey = this.getProcessLifecycleKey(pluginId, pluginPath); + const processLifecycle = PluginLoader.processPluginLifecycles.get(lifecycleKey); + const precedingLifecycle = processLifecycle?.promise; + + const reload = this.enqueueProcessLifecycleOperation(lifecycleKey, async () => { + await precedingLifecycle; + const owner = processLifecycle?.owner ?? this; + const plugin = await owner.reloadPluginFresh(pluginId, installation, pluginPath, options); + if (processLifecycle) { + this.synchronizeProcessPlugin(processLifecycle, pluginId, pluginPath, plugin); + } + return plugin; + }); + if (processLifecycle) processLifecycle.promise = reload; + try { + return await reload; + } catch (error) { + if (processLifecycle && PluginLoader.processPluginLifecycles.get(lifecycleKey) === processLifecycle) { + // reloadPluginFresh restores the canonical owner when rollback succeeds. + const restored = processLifecycle.owner.plugins.get(pluginId); + if (restored) { + processLifecycle.promise = Promise.resolve(restored); + this.synchronizeProcessPlugin(processLifecycle, pluginId, pluginPath, restored); + } else { + PluginLoader.processPluginLifecycles.delete(lifecycleKey); + } + } + throw error; + } + } + + /* + FNXC:PluginLoader 2026-07-22-16:20: + A process lifecycle is shared by host and engine loaders, not merely its + initial onLoad promise. Stop and reload must update every adopter so no + loader retains an old active instance after another surface changes it. + */ + private synchronizeProcessPlugin( + lifecycle: ProcessPluginLifecycle, + pluginId: string, + pluginPath: string, + plugin: FusionPlugin, + ): void { + for (const loader of lifecycle.participants) { + if (loader === lifecycle.owner) continue; + loader.plugins.set(pluginId, plugin); + loader.pluginRoots.set(pluginId, resolvePluginRootFromEntryPath(pluginPath)); + loader.pluginSchemaContracts.delete(pluginId); + } + } + + private async enqueueProcessLifecycleOperation( + lifecycleKey: string, + operation: () => Promise, + ): Promise { + const predecessor = PluginLoader.processPluginLifecycleTails.get(lifecycleKey) ?? Promise.resolve(); + const queued = predecessor.catch(() => undefined).then(operation); + const tail = queued.then(() => undefined, () => undefined); + PluginLoader.processPluginLifecycleTails.set(lifecycleKey, tail); + void tail.finally(() => { + if (PluginLoader.processPluginLifecycleTails.get(lifecycleKey) === tail) { + PluginLoader.processPluginLifecycleTails.delete(lifecycleKey); + } + }); + return await queued; + } + + private async reloadPluginFresh( + pluginId: string, + installation: PluginInstallation, + pluginPath: string, + options?: { timeoutMs?: number }, ): Promise { const timeoutMs = options?.timeoutMs ?? 5000; - // Get existing plugin + // A concurrent startup load may have completed on another loader. const oldPlugin = this.plugins.get(pluginId); if (!oldPlugin) { throw Object.assign(new Error(`Plugin "${pluginId}" is not loaded`), { @@ -588,10 +725,6 @@ export class PluginLoader extends EventEmitter<{ }); } - // Get installation record for path - const installation = await this.options.pluginStore.getPlugin(pluginId); - const pluginPath = this.resolvePluginPath(installation.path); - this.log.log(`Reloading plugin: ${pluginId}`); // Call onUnload with timeout @@ -881,18 +1014,46 @@ export class PluginLoader extends EventEmitter<{ * Stop and unload a single plugin. */ async stopPlugin(pluginId: string): Promise { + let installation: PluginInstallation; + try { + installation = await this.options.pluginStore.getPlugin(pluginId); + } catch { + this.log.log(`Plugin not loaded: ${pluginId}`); + return; + } + const pluginPath = this.resolvePluginPath(installation.path); + const lifecycleKey = this.getProcessLifecycleKey(pluginId, pluginPath); + const processLifecycle = PluginLoader.processPluginLifecycles.get(lifecycleKey); + if (!processLifecycle) { + await this.stopPluginFresh(pluginId, pluginPath); + return; + } + + await this.enqueueProcessLifecycleOperation(lifecycleKey, async () => { + try { + await processLifecycle.promise; + } catch { + // A rejected load has already cleaned its local state. + } + await processLifecycle.owner.stopPluginFresh(pluginId, pluginPath); + for (const loader of processLifecycle.participants) { + if (loader === processLifecycle.owner) continue; + loader.discardProcessPlugin(pluginId, pluginPath); + } + if (PluginLoader.processPluginLifecycles.get(lifecycleKey) === processLifecycle) { + PluginLoader.processPluginLifecycles.delete(lifecycleKey); + } + }); + } + + private async stopPluginFresh(pluginId: string, pluginPath: string): Promise { const plugin = this.plugins.get(pluginId); if (!plugin) { this.log.log(`Plugin not loaded: ${pluginId}`); return; } - // Get the plugin path for cache invalidation - const installation = await this.options.pluginStore.getPlugin(pluginId); - const pluginPath = this.resolvePluginPath(installation.path); - try { - // Call onUnload hook const ctx = await this.createContext(plugin); await this.withTimeout( this.safeCallHook(plugin, "onUnload", [ctx]), @@ -904,17 +1065,16 @@ export class PluginLoader extends EventEmitter<{ } await this.updatePluginState(pluginId, "stopped"); + this.discardProcessPlugin(pluginId, pluginPath); + this.emit("plugin:unloaded", { pluginId }); + this.emit("plugin:stopped", pluginId); + } - // Remove from loaded plugins + private discardProcessPlugin(pluginId: string, pluginPath: string): void { this.plugins.delete(pluginId); this.pluginRoots.delete(pluginId); this.pluginSchemaContracts.delete(pluginId); - - // Invalidate module cache for clean re-import this.invalidateModuleCache(pluginPath); - - this.emit("plugin:unloaded", { pluginId }); - this.emit("plugin:stopped", pluginId); // Backward compatibility } /** diff --git a/packages/engine/src/__tests__/plugin-startup-single-load.test.ts b/packages/engine/src/__tests__/plugin-startup-single-load.test.ts new file mode 100644 index 0000000000..325d8e0798 --- /dev/null +++ b/packages/engine/src/__tests__/plugin-startup-single-load.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PluginLoader } from "@fusion/core"; +import { PluginRunner } from "../plugin-runner.js"; + +function createStore(root: string, entry: string) { + const installation = { + id: "startup-single-load", name: "Startup single load", version: "1.0.0", description: "fixture", + path: entry, enabled: true, state: "installed", settings: {}, dependencies: [], + createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), + }; + const events = new EventEmitter(); + return { + pluginStore: { + getPlugin: vi.fn(async () => ({ ...installation })), + listPlugins: vi.fn(async () => [{ ...installation }]), + updatePluginState: vi.fn(async () => ({ ...installation })), + on: events.on.bind(events), off: events.off.bind(events), + }, + taskStore: { + getRootDir: () => root, + preflightPluginSchema: vi.fn(() => null), runPluginSchemaInits: vi.fn(async () => undefined), + on: vi.fn(), off: vi.fn(), + }, + }; +} + +describe("PluginRunner startup lifecycle", () => { + const roots: string[] = []; + afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); + delete (globalThis as Record).__fusionEngineStartupOnLoad; + }); + + it("does not repeat onLoad when host loadAll and engine runner init race", async () => { + const root = await mkdtemp(join(tmpdir(), "fusion-engine-plugin-startup-")); + roots.push(root); + const entry = join(root, "plugin.mjs"); + await writeFile(entry, `export default { + manifest: { id: "startup-single-load", name: "fixture", version: "1.0.0", description: "fixture" }, + state: "installed", + hooks: { onLoad: async () => { globalThis.__fusionEngineStartupOnLoad = (globalThis.__fusionEngineStartupOnLoad || 0) + 1; await new Promise(resolve => setTimeout(resolve, 20)); } }, + };`); + + const host = createStore(root, entry); + const engine = createStore(root, entry); + const hostLoader = new PluginLoader({ pluginStore: host.pluginStore as any, taskStore: host.taskStore as any }); + const engineLoader = new PluginLoader({ pluginStore: engine.pluginStore as any, taskStore: engine.taskStore as any }); + const runner = new PluginRunner({ pluginLoader: engineLoader, pluginStore: engine.pluginStore as any, taskStore: engine.taskStore as any, rootDir: root }); + + await Promise.all([hostLoader.loadAllPlugins(), runner.init()]); + + expect((globalThis as Record).__fusionEngineStartupOnLoad).toBe(1); + expect(engineLoader.isPluginLoaded("startup-single-load")).toBe(true); + }); +});