feat(FN-1113): complete Step 2 — executor integration with PluginRunner
This commit is contained in:
gsxdsm
397
packages/engine/src/__tests__/plugin-runner.test.ts
Normal file
397
packages/engine/src/__tests__/plugin-runner.test.ts
Normal file
@@ -0,0 +1,397 @@
|
||||
/**
|
||||
* PluginRunner Unit Tests
|
||||
*/
|
||||
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { PluginRunner, type PluginRunnerOptions } from "../plugin-runner.js";
|
||||
import type { PluginLoader, PluginStore } from "@fusion/core";
|
||||
import type { FusionPlugin, PluginToolDefinition, PluginRouteDefinition } 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>;
|
||||
};
|
||||
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(),
|
||||
};
|
||||
|
||||
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 any,
|
||||
rootDir: "/test/project",
|
||||
hookTimeoutMs: 5000,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("init()", () => {
|
||||
it("should call loadAllPlugins on the loader", async () => {
|
||||
await pluginRunner.init();
|
||||
expect(mockPluginLoader.loadAllPlugins).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("should subscribe to task store events", async () => {
|
||||
await pluginRunner.init();
|
||||
// Should have subscribed to task:created and task:moved
|
||||
expect(mockTaskStore.on).toHaveBeenCalledWith("task:created", expect.any(Function));
|
||||
expect(mockTaskStore.on).toHaveBeenCalledWith("task:moved", expect.any(Function));
|
||||
});
|
||||
|
||||
it("should subscribe to plugin store events for cache invalidation", async () => {
|
||||
await pluginRunner.init();
|
||||
expect(mockPluginStore.on).toHaveBeenCalledWith("plugin:stateChanged", expect.any(Function));
|
||||
expect(mockPluginStore.on).toHaveBeenCalledWith("plugin:updated", expect.any(Function));
|
||||
});
|
||||
});
|
||||
|
||||
describe("shutdown()", () => {
|
||||
it("should call stopAllPlugins on the loader", async () => {
|
||||
await pluginRunner.shutdown();
|
||||
expect(mockPluginLoader.stopAllPlugins).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("should unsubscribe from store events", async () => {
|
||||
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("onTaskCreated", { id: "FN-001" } as any);
|
||||
|
||||
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onTaskCreated", { id: "FN-001" });
|
||||
});
|
||||
|
||||
it("should pass all arguments to the hook", async () => {
|
||||
await pluginRunner.init();
|
||||
const task = { id: "FN-001" } as any;
|
||||
const from = "todo";
|
||||
const to = "in-progress";
|
||||
|
||||
await pluginRunner.invokeHook("onTaskMoved", task, from, to);
|
||||
|
||||
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onTaskMoved", task, from, to);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getPluginTools()", () => {
|
||||
it("should return empty array when no plugins have tools", async () => {
|
||||
await pluginRunner.init();
|
||||
const tools = pluginRunner.getPluginTools();
|
||||
expect(tools).toEqual([]);
|
||||
});
|
||||
|
||||
it("should return converted tools from loaded plugins", async () => {
|
||||
const executeFn = vi.fn().mockResolvedValue({
|
||||
content: [{ type: "text", text: "result" }],
|
||||
});
|
||||
|
||||
const pluginTool: PluginToolDefinition = {
|
||||
name: "testTool",
|
||||
description: "A test tool",
|
||||
parameters: {
|
||||
type: "object",
|
||||
properties: {
|
||||
input: { type: "string" },
|
||||
},
|
||||
},
|
||||
execute: executeFn,
|
||||
};
|
||||
|
||||
const plugin = createMockPlugin({
|
||||
manifest: { id: "test-plugin", name: "Test Plugin", version: "1.0.0" },
|
||||
tools: [pluginTool],
|
||||
});
|
||||
|
||||
mockPluginLoader.getLoadedPlugins.mockReturnValue([plugin]);
|
||||
mockPluginLoader.getPluginTools.mockReturnValue([pluginTool]);
|
||||
mockPluginLoader.getPlugin.mockReturnValue(plugin);
|
||||
|
||||
await pluginRunner.init();
|
||||
const tools = pluginRunner.getPluginTools();
|
||||
|
||||
expect(tools.length).toBe(1);
|
||||
expect(tools[0].name).toBe("plugin_testTool");
|
||||
expect(tools[0].label).toBe("testTool");
|
||||
expect(tools[0].description).toBe("A test tool");
|
||||
});
|
||||
|
||||
it("should wrap execute function correctly", async () => {
|
||||
const executeFn = vi.fn().mockResolvedValue({
|
||||
content: [{ type: "text", text: "test result" }],
|
||||
isError: false,
|
||||
});
|
||||
|
||||
const pluginTool: PluginToolDefinition = {
|
||||
name: "testTool",
|
||||
description: "A test tool",
|
||||
parameters: { type: "object", properties: {} },
|
||||
execute: executeFn,
|
||||
};
|
||||
|
||||
const plugin = createMockPlugin({
|
||||
manifest: { id: "test-plugin", name: "Test Plugin", version: "1.0.0" },
|
||||
tools: [pluginTool],
|
||||
});
|
||||
|
||||
mockPluginLoader.getLoadedPlugins.mockReturnValue([plugin]);
|
||||
mockPluginLoader.getPluginTools.mockReturnValue([pluginTool]);
|
||||
mockPluginLoader.getPlugin.mockReturnValue(plugin);
|
||||
|
||||
await pluginRunner.init();
|
||||
const tools = pluginRunner.getPluginTools();
|
||||
|
||||
// Call the wrapped execute
|
||||
const result = await tools[0].execute(
|
||||
"tool-call-1",
|
||||
{ input: "test" },
|
||||
undefined,
|
||||
undefined,
|
||||
{} as any,
|
||||
);
|
||||
|
||||
expect(executeFn).toHaveBeenCalledWith(
|
||||
{ input: "test" },
|
||||
expect.objectContaining({
|
||||
pluginId: "test-plugin",
|
||||
taskStore: mockTaskStore,
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result).toEqual({
|
||||
content: [{ type: "text", text: "test result" }],
|
||||
isError: false,
|
||||
details: {},
|
||||
});
|
||||
});
|
||||
|
||||
it("should invalidate cache when plugin state changes", async () => {
|
||||
const pluginTool: PluginToolDefinition = {
|
||||
name: "testTool",
|
||||
description: "A test tool",
|
||||
parameters: { type: "object", properties: {} },
|
||||
execute: vi.fn().mockResolvedValue({ content: [] }),
|
||||
};
|
||||
|
||||
const plugin = createMockPlugin({
|
||||
manifest: { id: "test-plugin", name: "Test Plugin", version: "1.0.0" },
|
||||
tools: [pluginTool],
|
||||
});
|
||||
|
||||
mockPluginLoader.getLoadedPlugins.mockReturnValue([plugin]);
|
||||
mockPluginLoader.getPluginTools.mockReturnValue([pluginTool]);
|
||||
mockPluginLoader.getPlugin.mockReturnValue(plugin);
|
||||
|
||||
await pluginRunner.init();
|
||||
|
||||
// First call caches the tools
|
||||
const tools1 = pluginRunner.getPluginTools();
|
||||
|
||||
// Simulate plugin state change
|
||||
const stateChangeHandler = mockPluginStore.on.mock.calls.find(
|
||||
(call) => call[0] === "plugin:stateChanged",
|
||||
)?.[1];
|
||||
stateChangeHandler?.();
|
||||
|
||||
// Second call should rebuild cache
|
||||
mockPluginLoader.getPluginTools.mockReturnValue([pluginTool]);
|
||||
const tools2 = pluginRunner.getPluginTools();
|
||||
|
||||
// Both should return tools (cache was rebuilt)
|
||||
expect(tools1.length).toBe(1);
|
||||
expect(tools2.length).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getPluginRoutes()", () => {
|
||||
it("should return routes from the loader", async () => {
|
||||
const routes: Array<{ pluginId: string; route: PluginRouteDefinition }> = [
|
||||
{
|
||||
pluginId: "test-plugin",
|
||||
route: {
|
||||
method: "GET",
|
||||
path: "/status",
|
||||
handler: vi.fn(),
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
mockPluginLoader.getPluginRoutes.mockReturnValue(routes);
|
||||
|
||||
await pluginRunner.init();
|
||||
const result = pluginRunner.getPluginRoutes();
|
||||
|
||||
expect(result).toEqual(routes);
|
||||
expect(mockPluginLoader.getPluginRoutes).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("should return empty array when no routes", async () => {
|
||||
await pluginRunner.init();
|
||||
const result = pluginRunner.getPluginRoutes();
|
||||
expect(result).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getLoader() / getStore()", () => {
|
||||
it("should return the plugin loader", () => {
|
||||
expect(pluginRunner.getLoader()).toBe(mockPluginLoader);
|
||||
});
|
||||
|
||||
it("should return the plugin store", () => {
|
||||
expect(pluginRunner.getStore()).toBe(mockPluginStore);
|
||||
});
|
||||
});
|
||||
|
||||
describe("task lifecycle hooks", () => {
|
||||
it("should invoke onTaskCreated when task:created event fires", async () => {
|
||||
await pluginRunner.init();
|
||||
|
||||
// Find the task:created handler
|
||||
const createdHandler = mockTaskStore.on.mock.calls.find(
|
||||
(call) => call[0] === "task:created",
|
||||
)?.[1] as (task: any) => void;
|
||||
|
||||
const mockTask = { id: "FN-001", title: "Test Task" };
|
||||
createdHandler?.(mockTask);
|
||||
|
||||
// Give async handler time to run
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
|
||||
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onTaskCreated", mockTask);
|
||||
});
|
||||
|
||||
it("should invoke onTaskMoved when task:moved event fires", async () => {
|
||||
await pluginRunner.init();
|
||||
|
||||
const movedHandler = mockTaskStore.on.mock.calls.find(
|
||||
(call) => call[0] === "task:moved",
|
||||
)?.[1] as (event: any) => void;
|
||||
|
||||
const event = { task: { id: "FN-001" }, from: "todo", to: "in-progress" };
|
||||
movedHandler?.(event);
|
||||
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
|
||||
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onTaskMoved", event.task, event.from, event.to);
|
||||
});
|
||||
|
||||
it("should invoke onTaskCompleted when task moves to done", async () => {
|
||||
await pluginRunner.init();
|
||||
|
||||
const movedHandler = mockTaskStore.on.mock.calls.find(
|
||||
(call) => call[0] === "task:moved",
|
||||
)?.[1] as (event: any) => void;
|
||||
|
||||
const event = { task: { id: "FN-001" }, from: "in-progress", to: "done" };
|
||||
movedHandler?.(event);
|
||||
|
||||
await new Promise((r) => setTimeout(r, 10));
|
||||
|
||||
expect(mockPluginLoader.invokeHook).toHaveBeenCalledWith("onTaskCompleted", event.task);
|
||||
});
|
||||
});
|
||||
|
||||
describe("hook timeout", () => {
|
||||
it("should handle slow hooks with timeout", async () => {
|
||||
// Create a runner with a short timeout
|
||||
const slowMockLoader = {
|
||||
...mockPluginLoader,
|
||||
invokeHook: vi.fn().mockImplementation(async () => {
|
||||
// Simulate slow hook
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
}),
|
||||
};
|
||||
|
||||
const runner = new PluginRunner({
|
||||
pluginLoader: slowMockLoader as unknown as PluginLoader,
|
||||
pluginStore: mockPluginStore as unknown as PluginStore,
|
||||
taskStore: mockTaskStore as any,
|
||||
rootDir: "/test/project",
|
||||
hookTimeoutMs: 50, // Very short timeout
|
||||
});
|
||||
|
||||
await runner.init();
|
||||
|
||||
// The invokeHook should complete (the slow plugin's error is logged but not thrown)
|
||||
await expect(runner.invokeHook("onTaskCreated", {})).resolves.not.toThrow();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -3,7 +3,6 @@ import { join } from "node:path";
|
||||
import { existsSync } from "node:fs";
|
||||
import type { TaskStore, Task, TaskDetail, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext } from "@fusion/core";
|
||||
import type { AgentStore } from "@fusion/core";
|
||||
import type { PluginRunner } from "@fusion/core";
|
||||
import { buildExecutionMemoryInstructions, resolveAgentPrompt } from "@fusion/core";
|
||||
import { findWorktreeUser } from "./merger.js";
|
||||
import { generateWorktreeName, slugify } from "./worktree-names.js";
|
||||
@@ -21,6 +20,7 @@ import { isTransientError, isSilentTransientError } from "./transient-error-dete
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./recovery-policy.js";
|
||||
import type { StuckTaskDetector, StuckTaskEvent } from "./stuck-task-detector.js";
|
||||
import type { PluginRunner } from "./plugin-runner.js";
|
||||
import { isContextLimitError } from "./context-limit-detector.js";
|
||||
import { StepSessionExecutor, type StepSessionExecutorOptions, type StepResult } from "./step-session-executor.js";
|
||||
import { resolveAgentInstructions, buildSystemPromptWithInstructions } from "./agent-instructions.js";
|
||||
@@ -1210,6 +1210,9 @@ export class TaskExecutor {
|
||||
stuckDetector?.trackTask(task.id, session);
|
||||
executorLog.log(`${task.id}: session registered (model=${describeModel(session)}, stuckDetector=${!!stuckDetector})`);
|
||||
|
||||
// Invoke plugin onAgentRunStart hook (fire-and-forget)
|
||||
void (this.options.pluginRunner as any)?.invokeHook("onAgentRunStart", task.id);
|
||||
|
||||
try {
|
||||
// Record activity on prompt start (heartbeat for stuck detection)
|
||||
stuckDetector?.recordActivity(task.id);
|
||||
@@ -1469,6 +1472,8 @@ export class TaskExecutor {
|
||||
if (!wasPaused && !this.pausedAborted.has(task.id)) {
|
||||
this.store.updateTask(task.id, { sessionFile: null }).catch(() => {});
|
||||
}
|
||||
// Invoke plugin onAgentRunEnd hook (fire-and-forget)
|
||||
void (this.options.pluginRunner as any)?.invokeHook("onAgentRunEnd", task.id);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
389
packages/engine/src/plugin-runner.ts
Normal file
389
packages/engine/src/plugin-runner.ts
Normal file
@@ -0,0 +1,389 @@
|
||||
/**
|
||||
* PluginRunner - Bridge between PluginLoader and Fusion Engine
|
||||
*
|
||||
* Orchestrates plugin loading into the engine, invokes hooks at lifecycle points,
|
||||
* and provides plugin tools to agent sessions.
|
||||
*/
|
||||
|
||||
import type { TaskStore, Task } from "@fusion/core";
|
||||
import type {
|
||||
PluginLoader,
|
||||
PluginStore,
|
||||
} from "@fusion/core";
|
||||
import type {
|
||||
FusionPlugin,
|
||||
PluginToolDefinition,
|
||||
PluginRouteDefinition,
|
||||
PluginContext,
|
||||
} from "@fusion/core";
|
||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||
import { Type } from "@mariozechner/pi-ai";
|
||||
import { createLogger, executorLog } from "./logger.js";
|
||||
|
||||
// Type for the task store's event data
|
||||
interface TaskMovedEvent {
|
||||
task: Task;
|
||||
from: string;
|
||||
to: string;
|
||||
}
|
||||
|
||||
export interface PluginRunnerOptions {
|
||||
pluginLoader: PluginLoader;
|
||||
pluginStore: PluginStore;
|
||||
taskStore: TaskStore;
|
||||
rootDir: string;
|
||||
hookTimeoutMs?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cached converted tools - rebuilt when plugin state changes
|
||||
*/
|
||||
interface CachedTools {
|
||||
tools: ToolDefinition[];
|
||||
version: number;
|
||||
}
|
||||
|
||||
const DEFAULT_HOOK_TIMEOUT_MS = 5000;
|
||||
|
||||
export class PluginRunner {
|
||||
private readonly log = createLogger("plugin-runner");
|
||||
private cachedTools: CachedTools | null = null;
|
||||
private toolsCacheVersion = 0;
|
||||
private hookTimeoutMs: number;
|
||||
|
||||
constructor(private options: PluginRunnerOptions) {
|
||||
this.hookTimeoutMs = options.hookTimeoutMs ?? DEFAULT_HOOK_TIMEOUT_MS;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the plugin runner.
|
||||
* Loads all plugins and subscribes to store events.
|
||||
*/
|
||||
async init(): Promise<void> {
|
||||
executorLog.log("Initializing PluginRunner...");
|
||||
|
||||
// Load all enabled plugins
|
||||
const result = await this.options.pluginLoader.loadAllPlugins();
|
||||
executorLog.log(`PluginRunner loaded ${result.loaded} plugins (${result.errors} errors)`);
|
||||
|
||||
// Subscribe to store events for task lifecycle hooks
|
||||
this.subscribeToStoreEvents();
|
||||
|
||||
// Subscribe to plugin state changes to invalidate tools cache
|
||||
this.options.pluginStore.on("plugin:stateChanged", this.handlePluginStateChanged);
|
||||
this.options.pluginStore.on("plugin:updated", this.handlePluginUpdated);
|
||||
|
||||
// Build initial tools cache
|
||||
this.invalidateToolsCache();
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown the plugin runner.
|
||||
* Stops all plugins and unsubscribes from events.
|
||||
*/
|
||||
async shutdown(): Promise<void> {
|
||||
executorLog.log("Shutting down PluginRunner...");
|
||||
|
||||
// Unsubscribe from store events
|
||||
this.unsubscribeFromStoreEvents();
|
||||
|
||||
// Unsubscribe from plugin store events
|
||||
this.options.pluginStore.off("plugin:stateChanged", this.handlePluginStateChanged);
|
||||
this.options.pluginStore.off("plugin:updated", this.handlePluginUpdated);
|
||||
|
||||
// Stop all plugins
|
||||
await this.options.pluginLoader.stopAllPlugins();
|
||||
|
||||
executorLog.log("PluginRunner shutdown complete");
|
||||
}
|
||||
|
||||
/**
|
||||
* Invoke a named hook on all loaded plugins.
|
||||
* Errors are isolated - one plugin's failure doesn't affect others.
|
||||
* Each hook call has a timeout (default 5 seconds).
|
||||
*/
|
||||
async invokeHook(hookName: keyof FusionPlugin["hooks"], ...args: unknown[]): Promise<void> {
|
||||
await this.options.pluginLoader.invokeHook(hookName, ...args);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all plugin tools converted to the engine's ToolDefinition format.
|
||||
* Tools are cached and only rebuilt when plugin state changes.
|
||||
*/
|
||||
getPluginTools(): ToolDefinition[] {
|
||||
if (!this.cachedTools || this.cachedTools.version !== this.toolsCacheVersion) {
|
||||
const pluginTools = this.options.pluginLoader.getPluginTools();
|
||||
this.cachedTools = {
|
||||
tools: this.convertPluginTools(pluginTools),
|
||||
version: this.toolsCacheVersion,
|
||||
};
|
||||
}
|
||||
return this.cachedTools.tools;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all plugin routes with their plugin IDs.
|
||||
*/
|
||||
getPluginRoutes(): Array<{ pluginId: string; route: PluginRouteDefinition }> {
|
||||
return this.options.pluginLoader.getPluginRoutes();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the underlying plugin loader.
|
||||
*/
|
||||
getLoader(): PluginLoader {
|
||||
return this.options.pluginLoader;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the underlying plugin store.
|
||||
*/
|
||||
getStore(): PluginStore {
|
||||
return this.options.pluginStore;
|
||||
}
|
||||
|
||||
// ── Tool Conversion ───────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Convert PluginToolDefinition[] to ToolDefinition[] for the pi-coding-agent.
|
||||
*
|
||||
* Plugin tools have this signature:
|
||||
* execute(params: Record<string, unknown>, ctx: PluginContext): Promise<PluginToolResult>
|
||||
*
|
||||
* Engine ToolDefinition has this signature:
|
||||
* execute(toolCallId, params, signal, onUpdate, ctx): Promise<AgentToolResult>
|
||||
*
|
||||
* The conversion:
|
||||
* 1. Prefixes the tool name with "plugin_"
|
||||
* 2. Maps name/description directly (use name as label)
|
||||
* 3. Wraps execute to extract params and call plugin's execute
|
||||
* 4. Returns { content: result.content } format
|
||||
*/
|
||||
private convertPluginTools(pluginTools: PluginToolDefinition[]): ToolDefinition[] {
|
||||
return pluginTools.map((pluginTool) => {
|
||||
// Get the plugin context for this tool
|
||||
const pluginId = this.getPluginIdForTool(pluginTool);
|
||||
const plugin = pluginId ? this.options.pluginLoader.getPlugin(pluginId) : undefined;
|
||||
|
||||
// Store the timeout for use in the closure
|
||||
const timeout = this.hookTimeoutMs;
|
||||
|
||||
// Create wrapper that extracts params and uses stored context
|
||||
const wrappedExecute = async (
|
||||
_toolCallId: string,
|
||||
params: Record<string, unknown>,
|
||||
_signal: AbortSignal | undefined,
|
||||
_onUpdate: unknown | undefined,
|
||||
_ctx: unknown,
|
||||
) => {
|
||||
if (!plugin) {
|
||||
return {
|
||||
content: [{ type: "text" as const, text: "Plugin not available" }],
|
||||
details: {},
|
||||
};
|
||||
}
|
||||
|
||||
// Create context for this specific tool call
|
||||
const context = await this.createToolContext(plugin);
|
||||
|
||||
try {
|
||||
const result = await this.withTimeout(
|
||||
pluginTool.execute(params as Record<string, unknown>, context),
|
||||
timeout,
|
||||
`Tool ${pluginTool.name} execution timed out`,
|
||||
);
|
||||
|
||||
// Convert PluginToolResult to AgentToolResult
|
||||
return {
|
||||
content: result.content,
|
||||
isError: result.isError ?? false,
|
||||
details: result.details ?? {},
|
||||
};
|
||||
} catch (err) {
|
||||
return {
|
||||
content: [{ type: "text" as const, text: `Tool execution failed: ${err instanceof Error ? err.message : String(err)}` }],
|
||||
isError: true,
|
||||
details: {},
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
// Use Type.Any for plugin tool parameters since plugins use JSON Schema
|
||||
// which is compatible with TypeBox's Any type
|
||||
const anySchema = Type.Any();
|
||||
|
||||
return {
|
||||
name: `plugin_${pluginTool.name}`,
|
||||
label: pluginTool.name,
|
||||
description: pluginTool.description,
|
||||
parameters: anySchema,
|
||||
execute: wrappedExecute,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the plugin ID that owns a tool.
|
||||
* We infer it from the loader's perspective - tools are stored per plugin.
|
||||
*/
|
||||
private getPluginIdForTool(tool: PluginToolDefinition): string | undefined {
|
||||
const loadedPlugins = this.options.pluginLoader.getLoadedPlugins();
|
||||
for (const plugin of loadedPlugins) {
|
||||
if (plugin.tools?.some((t) => t.name === tool.name)) {
|
||||
return plugin.manifest.id;
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a plugin context for tool execution.
|
||||
*/
|
||||
private async createToolContext(plugin: FusionPlugin): Promise<PluginContext> {
|
||||
const settings = await this.getPluginSettings(plugin.manifest.id);
|
||||
return {
|
||||
pluginId: plugin.manifest.id,
|
||||
taskStore: this.options.taskStore,
|
||||
settings,
|
||||
logger: this.createPluginLogger(plugin.manifest.id),
|
||||
emitEvent: (event: string, data: unknown) => {
|
||||
this.log.log(`[plugin:${plugin.manifest.id}] Event: ${event}`, data);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get settings for a plugin from the store.
|
||||
*/
|
||||
private async getPluginSettings(pluginId: string): Promise<Record<string, unknown>> {
|
||||
try {
|
||||
const plugin = await this.options.pluginStore.getPlugin(pluginId);
|
||||
return plugin.settings;
|
||||
} catch {
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a logger for a plugin.
|
||||
*/
|
||||
private createPluginLogger(pluginId: string): import("@fusion/core").PluginLogger {
|
||||
const prefix = `[plugin:${pluginId}]`;
|
||||
return {
|
||||
info: (...args: unknown[]) => this.log.log(prefix, ...args),
|
||||
warn: (...args: unknown[]) => this.log.warn(prefix, ...args),
|
||||
error: (...args: unknown[]) => this.log.error(prefix, ...args),
|
||||
debug: (...args: unknown[]) => {
|
||||
if (process.env.DEBUG?.includes("plugins")) {
|
||||
this.log.log(prefix, ...args);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ── Cache Invalidation ───────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Invalidate the tools cache, forcing rebuild on next access.
|
||||
*/
|
||||
private invalidateToolsCache(): void {
|
||||
this.toolsCacheVersion++;
|
||||
this.log.log(`Tools cache invalidated (version: ${this.toolsCacheVersion})`);
|
||||
}
|
||||
|
||||
// ── Store Event Subscriptions ────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Subscribe to TaskStore events for task lifecycle hooks.
|
||||
*/
|
||||
private subscribeToStoreEvents(): void {
|
||||
this.options.taskStore.on("task:created", this.handleTaskCreated);
|
||||
this.options.taskStore.on("task:moved", this.handleTaskMoved);
|
||||
}
|
||||
|
||||
/**
|
||||
* Unsubscribe from TaskStore events.
|
||||
*/
|
||||
private unsubscribeFromStoreEvents(): void {
|
||||
this.options.taskStore.off("task:created", this.handleTaskCreated);
|
||||
this.options.taskStore.off("task:moved", this.handleTaskMoved);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle task created event - invoke onTaskCreated hook.
|
||||
*/
|
||||
private handleTaskCreated = (task: Task): void => {
|
||||
// Fire and forget - don't await
|
||||
void this.invokeHookSafe("onTaskCreated", task);
|
||||
};
|
||||
|
||||
/**
|
||||
* Handle task moved event - invoke onTaskMoved and onTaskCompleted hooks.
|
||||
*/
|
||||
private handleTaskMoved = (event: TaskMovedEvent): void => {
|
||||
const { task, from, to } = event;
|
||||
|
||||
// Fire and forget - don't await
|
||||
void this.invokeHookSafe("onTaskMoved", task, from, to);
|
||||
|
||||
// If task completed, invoke onTaskCompleted hook
|
||||
if (to === "done") {
|
||||
void this.invokeHookSafe("onTaskCompleted", task);
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Invoke a hook with error isolation and logging.
|
||||
*/
|
||||
private async invokeHookSafe(hookName: keyof FusionPlugin["hooks"], ...args: unknown[]): Promise<void> {
|
||||
try {
|
||||
await this.withTimeout(
|
||||
this.invokeHook(hookName, ...args),
|
||||
this.hookTimeoutMs,
|
||||
`Hook ${hookName} timed out`,
|
||||
);
|
||||
} catch (err) {
|
||||
// Error already logged by invokeHook
|
||||
}
|
||||
}
|
||||
|
||||
// ── Event Handlers for Cache ────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Handler for plugin state changes.
|
||||
*/
|
||||
private handlePluginStateChanged = (): void => {
|
||||
this.invalidateToolsCache();
|
||||
};
|
||||
|
||||
/**
|
||||
* Handler for plugin updates.
|
||||
*/
|
||||
private handlePluginUpdated = (): void => {
|
||||
this.invalidateToolsCache();
|
||||
};
|
||||
|
||||
// ── Utilities ────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Execute a promise with a timeout.
|
||||
* Returns the result on success, throws on timeout.
|
||||
*/
|
||||
private withTimeout<T>(promise: Promise<T>, ms: number, timeoutMessage: string): Promise<T> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
reject(new Error(timeoutMessage));
|
||||
}, ms);
|
||||
|
||||
promise
|
||||
.then((result) => {
|
||||
clearTimeout(timer);
|
||||
resolve(result);
|
||||
})
|
||||
.catch((err) => {
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user