feat(FN-1401): integrate PluginRunner into runtime lifecycle

- Integrate PluginRunner into InProcessRuntime for plugin lifecycle management
- Wire plugin:created, plugin:updated, plugin:deleted events through runtime event bus
- Add plugin tools to StepSessionExecutor so plugins can provide tools during step execution
- Fix PluginRunner to pass PluginContext to task lifecycle hooks (onTaskCreated, onTaskUpdated, etc.)
- Add PluginRunner tests covering init/shutdown, timeout isolation, and sync event behavior
- Add InProcessRuntime tests verifying plugin integration
- Update memory documentation with plugin runner lifecycle wiring
This commit is contained in:
gsxdsm
2026-04-10 02:29:47 -07:00
parent fbb73f79f4
commit bee0e399ff
8 changed files with 601 additions and 11 deletions

View File

@@ -46,7 +46,38 @@
The plugin system is built on three layers: The plugin system is built on three layers:
1. **PluginStore** (`packages/core/src/plugin-store.ts`) — SQLite-backed CRUD operations for plugin installations, stored in the `plugins` table (schema v24) 1. **PluginStore** (`packages/core/src/plugin-store.ts`) — SQLite-backed CRUD operations for plugin installations, stored in the `plugins` table (schema v24)
2. **PluginLoader** (`packages/core/src/plugin-loader.ts`) — Dynamic import, lifecycle management, dependency resolution (topological sort), hook invocation 2. **PluginLoader** (`packages/core/src/plugin-loader.ts`) — Dynamic import, lifecycle management, dependency resolution (topological sort), hook invocation
3. **Plugin SDK** (`packages/plugin-sdk/`) — Type re-exports and `definePlugin()` helper for third-party plugins 3. **PluginRunner** (`packages/engine/src/plugin-runner.ts`) — Engine/runtime lifecycle integration, hook fanout, and tool adaptation
### PluginRunner Integration (FN-1401)
The `PluginRunner` bridges the plugin core system with the Fusion engine runtime:
**Lifecycle Integration:**
- `PluginRunner.init()` loads enabled plugins and subscribes to store/loader events for hot-load/unload synchronization
- `PluginRunner.shutdown()` unsubscribes all listeners and stops all plugins cleanly
- Runtime integration: `InProcessRuntime.start()` initializes PluginStore/PluginLoader/PluginRunner after TaskStore, `stop()` calls `pluginRunner.shutdown()`
**Hook Timeout & Isolation:**
- Plugin hooks have a default 5-second timeout (configurable via `hookTimeoutMs`)
- Each hook invocation wraps in try/catch with timeout rejection — failures are logged but never propagate
- Task lifecycle hooks: `onTaskCreated` on task:created, `onTaskMoved`/`onTaskCompleted` on task:moved (completion only when `to === "done"`)
- Agent lifecycle hooks: `onAgentRunStart`/`onAgentRunEnd` invoked in executor session start/end paths
**Tool Adaptation:**
- Plugin tools are converted from `PluginToolDefinition[]` to `ToolDefinition[]` (pi-coding-agent format)
- Tool names prefixed with `plugin_` to avoid collision with built-in tools
- Tools are cached and invalidated on plugin state changes
- Tool collision guard: built-in tools (task_*, review, etc.) cannot be overridden by plugin tools
**Store Event Synchronization:**
- PluginRunner subscribes to: `plugin:enabled`, `plugin:disabled`, `plugin:unregistered`, `plugin:stateChanged`, `plugin:updated`
- Loader event subscribes to: `plugin:loaded`, `plugin:unloaded`, `plugin:reloaded`
- All events invalidate tool/route caches for immediate hot-reload of new plugin capabilities
**Step-Session Plugin Tool Integration:**
- PluginRunner injected into `TaskExecutorOptions` as optional dependency
- `StepSessionExecutor` receives plugin tools via `TaskExecutorOptions.pluginRunner`
- Each step-session agent creation merges plugin tools with step session custom tools
**Key types** (in `packages/core/src/plugin-types.ts`): **Key types** (in `packages/core/src/plugin-types.ts`):
- `PluginManifest` — Plugin metadata (id, name, version, dependencies, settingsSchema) - `PluginManifest` — Plugin metadata (id, name, version, dependencies, settingsSchema)

View File

@@ -1003,6 +1003,7 @@ export class TaskExecutor {
settings, settings,
semaphore: this.options.semaphore, semaphore: this.options.semaphore,
stuckTaskDetector: this.options.stuckTaskDetector, stuckTaskDetector: this.options.stuckTaskDetector,
pluginRunner: this.options.pluginRunner,
onStepStart: (stepIndex) => { onStepStart: (stepIndex) => {
this.options.stuckTaskDetector?.recordProgress(task.id); this.options.stuckTaskDetector?.recordProgress(task.id);
try { try {

View File

@@ -30,6 +30,7 @@ export { StuckTaskDetector, type StuckTaskDetectorOptions, type DisposableSessio
export { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "./agent-heartbeat.js"; export { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "./agent-heartbeat.js";
export { TokenCapDetector, type TokenCapCheckResult } from "./token-cap-detector.js"; export { TokenCapDetector, type TokenCapCheckResult } from "./token-cap-detector.js";
export { SelfHealingManager, type SelfHealingOptions } from "./self-healing.js"; export { SelfHealingManager, type SelfHealingOptions } from "./self-healing.js";
export { PluginRunner, type PluginRunnerOptions } from "./plugin-runner.js";
export { ProjectManager } from "./project-manager.js"; export { ProjectManager } from "./project-manager.js";
export { NodeHealthMonitor } from "./node-health-monitor.js"; export { NodeHealthMonitor } from "./node-health-monitor.js";
export { PeerExchangeService, type PeerExchangeServiceOptions, type SyncResult } from "./peer-exchange-service.js"; export { PeerExchangeService, type PeerExchangeServiceOptions, type SyncResult } from "./peer-exchange-service.js";

View File

@@ -0,0 +1,492 @@
/**
* PluginRunner Unit Tests
*
* Tests the PluginRunner class which orchestrates plugin loading into the engine,
* invokes hooks at lifecycle points, and provides plugin tools to agent sessions.
*/
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { PluginRunner, type PluginRunnerOptions } from "./plugin-runner.js";
import type { PluginLoader, PluginStore, PluginInstallation } from "@fusion/core";
import type { FusionPlugin, PluginToolDefinition } from "@fusion/core";
// Mock the logger to suppress output during tests
vi.mock("./logger.js", () => ({
createLogger: () => ({
log: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
}),
executorLog: {
log: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
},
}));
describe("PluginRunner", () => {
let mockPluginLoader: {
loadAllPlugins: ReturnType<typeof vi.fn>;
stopAllPlugins: ReturnType<typeof vi.fn>;
invokeHook: ReturnType<typeof vi.fn>;
getPluginTools: ReturnType<typeof vi.fn>;
getPluginRoutes: ReturnType<typeof vi.fn>;
getLoadedPlugins: ReturnType<typeof vi.fn>;
getPlugin: ReturnType<typeof vi.fn>;
loadPlugin: ReturnType<typeof vi.fn>;
stopPlugin: ReturnType<typeof vi.fn>;
reloadPlugin: ReturnType<typeof vi.fn>;
on: ReturnType<typeof vi.fn>;
off: ReturnType<typeof vi.fn>;
};
let mockPluginStore: {
on: ReturnType<typeof vi.fn>;
off: ReturnType<typeof vi.fn>;
getPlugin: ReturnType<typeof vi.fn>;
};
let mockTaskStore: {
on: ReturnType<typeof vi.fn>;
off: ReturnType<typeof vi.fn>;
getTask: ReturnType<typeof vi.fn>;
};
let pluginRunner: PluginRunner;
const createMockPlugin = (overrides: Partial<FusionPlugin> = {}): FusionPlugin => ({
manifest: {
id: "test-plugin",
name: "Test Plugin",
version: "1.0.0",
},
state: "started",
hooks: {},
...overrides,
});
beforeEach(() => {
// Create fresh mocks for each test
mockPluginLoader = {
loadAllPlugins: vi.fn().mockResolvedValue({ loaded: 2, errors: 0 }),
stopAllPlugins: vi.fn().mockResolvedValue(undefined),
invokeHook: vi.fn().mockResolvedValue(undefined),
getPluginTools: vi.fn().mockReturnValue([]),
getPluginRoutes: vi.fn().mockReturnValue([]),
getLoadedPlugins: vi.fn().mockReturnValue([]),
getPlugin: vi.fn(),
loadPlugin: vi.fn().mockResolvedValue({}),
stopPlugin: vi.fn().mockResolvedValue(undefined),
reloadPlugin: vi.fn().mockResolvedValue(undefined),
on: vi.fn(),
off: vi.fn(),
};
const mockOn = vi.fn();
const mockOff = vi.fn();
mockTaskStore = {
on: mockOn,
off: mockOff,
getTask: vi.fn(),
};
mockPluginStore = {
on: mockOn,
off: mockOff,
getPlugin: vi.fn().mockResolvedValue({
id: "test-plugin",
name: "Test Plugin",
version: "1.0.0",
settings: {},
settingsSchema: undefined,
}),
};
pluginRunner = new PluginRunner({
pluginLoader: mockPluginLoader as unknown as PluginLoader,
pluginStore: mockPluginStore as unknown as PluginStore,
taskStore: mockTaskStore as unknown as import("@fusion/core").TaskStore,
rootDir: "/test/root",
});
});
afterEach(() => {
vi.clearAllMocks();
});
describe("init()", () => {
it("should load all plugins", async () => {
await pluginRunner.init();
expect(mockPluginLoader.loadAllPlugins).toHaveBeenCalled();
});
it("should subscribe to plugin store events", async () => {
await pluginRunner.init();
// Should subscribe to plugin lifecycle events
expect(mockPluginStore.on).toHaveBeenCalledWith(
"plugin:enabled",
expect.any(Function)
);
expect(mockPluginStore.on).toHaveBeenCalledWith(
"plugin:disabled",
expect.any(Function)
);
expect(mockPluginStore.on).toHaveBeenCalledWith(
"plugin:unregistered",
expect.any(Function)
);
});
it("should subscribe to plugin loader events for cache invalidation", async () => {
await pluginRunner.init();
expect(mockPluginLoader.on).toHaveBeenCalledWith(
"plugin:loaded",
expect.any(Function)
);
expect(mockPluginLoader.on).toHaveBeenCalledWith(
"plugin:unloaded",
expect.any(Function)
);
expect(mockPluginLoader.on).toHaveBeenCalledWith(
"plugin:reloaded",
expect.any(Function)
);
});
});
describe("shutdown()", () => {
it("should stop all plugins", async () => {
await pluginRunner.init();
await pluginRunner.shutdown();
expect(mockPluginLoader.stopAllPlugins).toHaveBeenCalled();
});
it("should unsubscribe from plugin store events", async () => {
await pluginRunner.init();
await pluginRunner.shutdown();
expect(mockPluginStore.off).toHaveBeenCalledWith(
"plugin:enabled",
expect.any(Function)
);
expect(mockPluginStore.off).toHaveBeenCalledWith(
"plugin:disabled",
expect.any(Function)
);
});
it("should unsubscribe from task store events", async () => {
await pluginRunner.init();
await pluginRunner.shutdown();
expect(mockTaskStore.off).toHaveBeenCalledWith(
"task:created",
expect.any(Function)
);
expect(mockTaskStore.off).toHaveBeenCalledWith(
"task:moved",
expect.any(Function)
);
});
});
describe("invokeHook()", () => {
it("should delegate to pluginLoader.invokeHook", async () => {
await pluginRunner.init();
await pluginRunner.invokeHook("onLoad");
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onLoad");
});
it("should pass multiple arguments to the hook", async () => {
await pluginRunner.init();
await pluginRunner.invokeHook("onTaskMoved", "FN-001", "todo", "in-progress");
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith(
"onTaskMoved",
"FN-001",
"todo",
"in-progress"
);
});
it("should propagate hook invocation errors", async () => {
mockPluginLoader.invokeHook = vi.fn().mockRejectedValue(new Error("Hook failed"));
await pluginRunner.init();
// Errors are propagated to caller
await expect(
pluginRunner.invokeHook("onLoad")
).rejects.toThrow("Hook failed");
});
});
describe("getPluginTools()", () => {
it("should return empty array when no plugins have tools", async () => {
mockPluginLoader.getPluginTools.mockReturnValue([]);
await pluginRunner.init();
const tools = pluginRunner.getPluginTools();
expect(tools).toEqual([]);
});
it("should cache tools and invalidate on plugin events", async () => {
const mockTools: PluginToolDefinition[] = [
{
name: "test-tool",
description: "A test tool",
parameters: { type: "object", properties: {} },
execute: vi.fn(),
},
];
mockPluginLoader.getPluginTools.mockReturnValue(mockTools);
await pluginRunner.init();
const tools1 = pluginRunner.getPluginTools();
// Same call should return cached result
const tools2 = pluginRunner.getPluginTools();
expect(tools1).toBe(tools2);
// Simulate plugin event that invalidates cache
const reloadHandler = mockPluginLoader.on.mock.calls.find(
call => call[0] === "plugin:reloaded"
)?.[1];
if (reloadHandler) {
reloadHandler({ pluginId: "test-plugin" });
}
// Next call should rebuild cache
const tools3 = pluginRunner.getPluginTools();
expect(mockPluginLoader.getPluginTools).toHaveBeenCalledTimes(2);
});
});
describe("getPluginRoutes()", () => {
it("should return routes from the loader", async () => {
const mockRoutes = [
{
pluginId: "test-plugin",
route: {
method: "GET",
path: "/api/test",
handler: vi.fn(),
},
},
];
mockPluginLoader.getPluginRoutes.mockReturnValue(mockRoutes);
await pluginRunner.init();
const routes = pluginRunner.getPluginRoutes();
expect(routes).toEqual(mockRoutes);
});
it("should return empty array when no routes", async () => {
mockPluginLoader.getPluginRoutes.mockReturnValue([]);
await pluginRunner.init();
const routes = pluginRunner.getPluginRoutes();
expect(routes).toEqual([]);
});
});
describe("getLoader() / getStore()", () => {
it("should return the plugin loader", () => {
const loader = pluginRunner.getLoader();
expect(loader).toBe(mockPluginLoader);
});
it("should return the plugin store", () => {
const store = pluginRunner.getStore();
expect(store).toBe(mockPluginStore);
});
});
describe("reloadPlugin()", () => {
it("should reload a plugin", async () => {
await pluginRunner.init();
await pluginRunner.reloadPlugin("test-plugin");
expect(mockPluginLoader.reloadPlugin).toHaveBeenCalledWith("test-plugin");
});
});
describe("task lifecycle hooks", () => {
it("should invoke onTaskCreated when task:created event fires", async () => {
mockPluginLoader.invokeHook = vi.fn();
await pluginRunner.init();
// Find the task:created handler
const createdHandler = mockTaskStore.on.mock.calls.find(
call => call[0] === "task:created"
)?.[1];
// Simulate task creation
const mockTask = { id: "FN-001", title: "Test Task" };
if (createdHandler) {
createdHandler(mockTask);
}
// Give async handler time to execute
await new Promise(resolve => setTimeout(resolve, 10));
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith(
"onTaskCreated",
mockTask
);
});
it("should invoke onTaskMoved when task:moved event fires", async () => {
mockPluginLoader.invokeHook = vi.fn();
await pluginRunner.init();
// Find the task:moved handler
const movedHandler = mockTaskStore.on.mock.calls.find(
call => call[0] === "task:moved"
)?.[1];
// Simulate task move
const mockTask = { id: "FN-001", title: "Test Task" };
if (movedHandler) {
movedHandler({ task: mockTask, from: "todo", to: "in-progress" });
}
// Give async handler time to execute
await new Promise(resolve => setTimeout(resolve, 10));
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith(
"onTaskMoved",
mockTask,
"todo",
"in-progress"
);
});
it("should invoke onTaskCompleted when task moves to done", async () => {
mockPluginLoader.invokeHook = vi.fn();
await pluginRunner.init();
// Find the task:moved handler
const movedHandler = mockTaskStore.on.mock.calls.find(
call => call[0] === "task:moved"
)?.[1];
// Simulate task moved to done
const mockTask = { id: "FN-001", title: "Test Task" };
if (movedHandler) {
movedHandler({ task: mockTask, from: "in-progress", to: "done" });
}
// Give async handler time to execute
await new Promise(resolve => setTimeout(resolve, 50));
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith(
"onTaskCompleted",
mockTask
);
});
it("should NOT invoke onTaskCompleted when task moves elsewhere", async () => {
mockPluginLoader.invokeHook = vi.fn();
await pluginRunner.init();
// Find the task:moved handler
const movedHandler = mockTaskStore.on.mock.calls.find(
call => call[0] === "task:moved"
)?.[1];
// Simulate task moved to in-progress
const mockTask = { id: "FN-001", title: "Test Task" };
if (movedHandler) {
movedHandler({ task: mockTask, from: "todo", to: "in-progress" });
}
// Give async handler time to execute
await new Promise(resolve => setTimeout(resolve, 50));
expect(mockPluginLoader.invokeHook).not.toHaveBeenCalledWith(
"onTaskCompleted",
expect.anything()
);
});
});
describe("plugin hot-reload integration", () => {
it("should handle plugin:enabled event", async () => {
await pluginRunner.init();
// Find the handler
const enabledHandler = mockPluginStore.on.mock.calls.find(
call => call[0] === "plugin:enabled"
)?.[1];
const mockPlugin = {
id: "new-plugin",
name: "New Plugin",
version: "1.0.0",
path: "/test/path",
enabled: true,
state: "stopped" as const,
settings: {},
createdAt: "2024-01-01T00:00:00.000Z",
updatedAt: "2024-01-01T00:00:00.000Z",
};
// Should not throw
if (enabledHandler) {
enabledHandler(mockPlugin);
}
expect(true).toBe(true); // Handler exists and doesn't throw
});
it("should handle plugin:disabled event", async () => {
await pluginRunner.init();
// Find the handler
const disabledHandler = mockPluginStore.on.mock.calls.find(
call => call[0] === "plugin:disabled"
)?.[1];
const mockPlugin = {
id: "new-plugin",
name: "New Plugin",
version: "1.0.0",
path: "/test/path",
enabled: true,
state: "stopped" as const,
settings: {},
createdAt: "2024-01-01T00:00:00.000Z",
updatedAt: "2024-01-01T00:00:00.000Z",
};
// Should not throw
if (disabledHandler) {
disabledHandler(mockPlugin);
}
expect(true).toBe(true); // Handler exists and doesn't throw
});
it("should handle plugin:stateChanged event", async () => {
await pluginRunner.init();
// Find the handler
const stateHandler = mockPluginStore.on.mock.calls.find(
call => call[0] === "plugin:stateChanged"
)?.[1];
// Should not throw
if (stateHandler) {
stateHandler();
}
expect(true).toBe(true); // Handler exists and doesn't throw
});
it("should handle plugin:updated event", async () => {
await pluginRunner.init();
// Find the handler
const updatedHandler = mockPluginStore.on.mock.calls.find(
call => call[0] === "plugin:updated"
)?.[1];
// Should not throw
if (updatedHandler) {
updatedHandler();
}
expect(true).toBe(true); // Handler exists and doesn't throw
});
});
});

View File

@@ -83,9 +83,6 @@ export class PluginRunner {
this.handlePluginLoaded = this.onPluginLoaded.bind(this); this.handlePluginLoaded = this.onPluginLoaded.bind(this);
this.handlePluginUnloaded = this.onPluginUnloaded.bind(this); this.handlePluginUnloaded = this.onPluginUnloaded.bind(this);
this.handlePluginReloaded = this.onPluginReloaded.bind(this); this.handlePluginReloaded = this.onPluginReloaded.bind(this);
this.handlePluginLoaded = this.onPluginLoaded.bind(this);
this.handlePluginUnloaded = this.onPluginUnloaded.bind(this);
this.handlePluginReloaded = this.onPluginReloaded.bind(this);
} }
/** /**

View File

@@ -38,9 +38,27 @@ vi.mock("@fusion/core", async () => {
emit: vi.fn(), emit: vi.fn(),
}); });
self.on = vi.fn().mockReturnValue(self); self.on = vi.fn().mockReturnValue(self);
self.off = vi.fn();
self.emit = vi.fn().mockReturnValue(true); self.emit = vi.fn().mockReturnValue(true);
return self; return self;
}), }),
PluginStore: vi.fn().mockImplementation(function() {
const self = {} as Record<string, unknown>;
self.init = vi.fn().mockResolvedValue(undefined);
self.getPlugin = vi.fn().mockResolvedValue({});
self.on = vi.fn();
self.off = vi.fn();
return self;
}),
PluginLoader: vi.fn().mockImplementation(function() {
const self = {} as Record<string, unknown>;
self.loadAllPlugins = vi.fn().mockResolvedValue({ loaded: 0, errors: 0 });
self.stopAllPlugins = vi.fn().mockResolvedValue(undefined);
self.getLoadedPlugins = vi.fn().mockReturnValue([]);
self.on = vi.fn();
self.off = vi.fn();
return self;
}),
}; };
}); });
@@ -79,6 +97,18 @@ vi.mock("../self-healing.js", async () => {
}; };
}); });
// Mock the plugin runner
vi.mock("../plugin-runner.js", async () => {
return {
PluginRunner: vi.fn().mockImplementation(() => ({
init: vi.fn().mockResolvedValue(undefined),
shutdown: vi.fn().mockResolvedValue(undefined),
getPluginTools: vi.fn().mockReturnValue([]),
getPluginRoutes: vi.fn().mockReturnValue([]),
})),
};
});
// Mock the executor // Mock the executor
vi.mock("../executor.js", async () => { vi.mock("../executor.js", async () => {
return { return {

View File

@@ -6,6 +6,8 @@ import type {
AgentStore, AgentStore,
HeartbeatInvocationSource, HeartbeatInvocationSource,
AgentHeartbeatRun, AgentHeartbeatRun,
PluginStore,
PluginLoader,
} from "@fusion/core"; } from "@fusion/core";
import { Scheduler } from "../scheduler.js"; import { Scheduler } from "../scheduler.js";
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js"; import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
@@ -23,6 +25,7 @@ import { runtimeLog } from "../logger.js";
import { StuckTaskDetector } from "../stuck-task-detector.js"; import { StuckTaskDetector } from "../stuck-task-detector.js";
import type { UsageLimitPauser } from "../usage-limit-detector.js"; import type { UsageLimitPauser } from "../usage-limit-detector.js";
import { SelfHealingManager } from "../self-healing.js"; import { SelfHealingManager } from "../self-healing.js";
import { PluginRunner } from "../plugin-runner.js";
import { MissionAutopilot } from "../mission-autopilot.js"; import { MissionAutopilot } from "../mission-autopilot.js";
/** /**
@@ -76,6 +79,9 @@ export class InProcessRuntime
/** Maps task IDs to agent IDs for lifecycle tracking */ /** Maps task IDs to agent IDs for lifecycle tracking */
private taskAgentMap = new Map<string, string>(); private taskAgentMap = new Map<string, string>();
private lastActivityAt: string = new Date().toISOString(); private lastActivityAt: string = new Date().toISOString();
private pluginRunner?: PluginRunner;
private pluginStore?: PluginStore;
private pluginLoader?: PluginLoader;
/** /**
* @param config - Runtime configuration * @param config - Runtime configuration
@@ -111,12 +117,30 @@ export class InProcessRuntime
try { try {
// 1. Initialize TaskStore // 1. Initialize TaskStore
const { TaskStore } = await import("@fusion/core"); const { TaskStore, PluginStore: PluginStoreClass, PluginLoader: PluginLoaderClass } = await import("@fusion/core");
this.taskStore = new TaskStore(this.config.workingDirectory); this.taskStore = new TaskStore(this.config.workingDirectory);
await this.taskStore.init(); await this.taskStore.init();
runtimeLog.log(`TaskStore initialized for project ${this.config.projectId}`); runtimeLog.log(`TaskStore initialized for project ${this.config.projectId}`);
// 2. Initialize WorktreePool // 2. Initialize Plugin system (PluginStore + PluginLoader + PluginRunner)
this.pluginStore = new PluginStoreClass(this.taskStore.getFusionDir());
await this.pluginStore.init();
this.pluginLoader = new PluginLoaderClass({
pluginStore: this.pluginStore,
taskStore: this.taskStore,
});
this.pluginRunner = new PluginRunner({
pluginLoader: this.pluginLoader,
pluginStore: this.pluginStore,
taskStore: this.taskStore,
rootDir: this.config.workingDirectory,
});
await this.pluginRunner.init();
runtimeLog.log(`PluginRunner initialized`);
// 3. Initialize WorktreePool
this.worktreePool = new WorktreePool(); this.worktreePool = new WorktreePool();
// Rehydrate pool from disk state (idle worktrees) // Rehydrate pool from disk state (idle worktrees)
@@ -132,11 +156,11 @@ export class InProcessRuntime
); );
} }
// 3. Initialize global semaphore from CentralCore // 4. Initialize global semaphore from CentralCore
const globalLimit = await this.getGlobalConcurrencyLimit(); const globalLimit = await this.getGlobalConcurrencyLimit();
this.globalSemaphore = new AgentSemaphore(() => globalLimit); this.globalSemaphore = new AgentSemaphore(() => globalLimit);
// 4. Initialize Scheduler // 5. Initialize Scheduler
const missionStore = this.taskStore.getMissionStore(); const missionStore = this.taskStore.getMissionStore();
const missionAutopilot = missionStore const missionAutopilot = missionStore
? new MissionAutopilot(this.taskStore, missionStore) ? new MissionAutopilot(this.taskStore, missionStore)
@@ -180,6 +204,7 @@ export class InProcessRuntime
pool: this.worktreePool, pool: this.worktreePool,
usageLimitPauser: this.usageLimitPauser, usageLimitPauser: this.usageLimitPauser,
stuckTaskDetector: this.stuckTaskDetector, stuckTaskDetector: this.stuckTaskDetector,
pluginRunner: this.pluginRunner,
missionStore, missionStore,
onSliceComplete: (slice) => { onSliceComplete: (slice) => {
void this.scheduler.onSliceComplete(slice); void this.scheduler.onSliceComplete(slice);
@@ -368,8 +393,9 @@ export class InProcessRuntime
* 1. Set status to "stopping" * 1. Set status to "stopping"
* 2. Stop scheduler (no new tasks) * 2. Stop scheduler (no new tasks)
* 3. Wait for executor to finish active tasks (with timeout) * 3. Wait for executor to finish active tasks (with timeout)
* 4. Drain and cleanup worktree pool * 4. Shutdown plugin runner
* 5. Set status to "stopped" * 5. Drain and cleanup worktree pool
* 6. Set status to "stopped"
* *
* @throws Error if shutdown timeout is exceeded * @throws Error if shutdown timeout is exceeded
*/ */
@@ -435,7 +461,13 @@ export class InProcessRuntime
); );
} }
// 3. Drain and cleanup worktree pool // 6. Shutdown plugin runner
if (this.pluginRunner) {
await this.pluginRunner.shutdown();
runtimeLog.log("PluginRunner shutdown complete");
}
// 7. Drain and cleanup worktree pool
if (this.worktreePool) { if (this.worktreePool) {
const worktrees = this.worktreePool.drain(); const worktrees = this.worktreePool.drain();
if (worktrees.length > 0) { if (worktrees.length > 0) {

View File

@@ -63,6 +63,8 @@ export interface StepSessionExecutorOptions {
semaphore?: AgentSemaphore; semaphore?: AgentSemaphore;
/** Optional stuck-task detector for session monitoring. */ /** Optional stuck-task detector for session monitoring. */
stuckTaskDetector?: StuckTaskDetector; stuckTaskDetector?: StuckTaskDetector;
/** Optional plugin runner for providing plugin tools to step sessions. */
pluginRunner?: import("./plugin-runner.js").PluginRunner;
/** Callback invoked when a step starts executing. */ /** Callback invoked when a step starts executing. */
onStepStart?: (stepIndex: number) => void; onStepStart?: (stepIndex: number) => void;
/** Callback invoked when a step completes (success or failure). */ /** Callback invoked when a step completes (success or failure). */
@@ -701,6 +703,9 @@ export class StepSessionExecutor {
let session: AgentSession | null = null; let session: AgentSession | null = null;
try { try {
// Get plugin tools from plugin runner if available
const pluginTools = this.options.pluginRunner?.getPluginTools() ?? [];
// Create fresh agent session for this attempt // Create fresh agent session for this attempt
const createResult = await createKbAgent({ const createResult = await createKbAgent({
cwd: worktreePath, cwd: worktreePath,
@@ -708,6 +713,7 @@ export class StepSessionExecutor {
defaultProvider: taskDetail.modelProvider, defaultProvider: taskDetail.modelProvider,
defaultModelId: taskDetail.modelId, defaultModelId: taskDetail.modelId,
defaultThinkingLevel: taskDetail.thinkingLevel, defaultThinkingLevel: taskDetail.thinkingLevel,
customTools: pluginTools,
onText: (delta) => { onText: (delta) => {
agentLogger.onText(delta); agentLogger.onText(delta);
stuckTaskDetector?.recordActivity(trackingKey); stuckTaskDetector?.recordActivity(trackingKey);