feat(KB-501): implement multi-project runtime architecture with HybridExecutor
- Add ProjectRuntime interface with InProcessRuntime and ChildProcessRuntime implementations - Implement HybridExecutor for multi-project task orchestration - Add IPC protocol for host-worker communication in child-process mode - Implement health monitoring with automatic restart for child processes - Add comprehensive tests for hybrid-executor and project-runtime - Update AGENTS.md with multi-project runtime documentation - Add changeset for hybrid-executor-runtime release
This commit is contained in:
gsxdsm
344
packages/engine/src/__tests__/hybrid-executor.test.ts
Normal file
344
packages/engine/src/__tests__/hybrid-executor.test.ts
Normal file
@@ -0,0 +1,344 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import type { CentralCore, RegisteredProject } from "@fusion/core";
|
||||
import { HybridExecutor } from "../hybrid-executor.js";
|
||||
import type { ProjectRuntimeConfig } from "../project-runtime.js";
|
||||
|
||||
// Mock the ProjectManager
|
||||
const mockRuntimes = new Map();
|
||||
const mockProjectIds: string[] = [];
|
||||
|
||||
vi.mock("../project-manager.js", () => ({
|
||||
ProjectManager: vi.fn().mockImplementation(() => ({
|
||||
addProject: vi.fn().mockImplementation((config: ProjectRuntimeConfig) => {
|
||||
const runtime = {
|
||||
start: vi.fn().mockResolvedValue(undefined),
|
||||
stop: vi.fn().mockResolvedValue(undefined),
|
||||
getStatus: vi.fn().mockReturnValue("active"),
|
||||
getTaskStore: vi.fn(),
|
||||
getScheduler: vi.fn(),
|
||||
getMetrics: vi.fn().mockReturnValue({
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
}),
|
||||
on: vi.fn().mockReturnThis(),
|
||||
};
|
||||
mockRuntimes.set(config.projectId, runtime);
|
||||
mockProjectIds.push(config.projectId);
|
||||
return Promise.resolve(runtime);
|
||||
}),
|
||||
removeProject: vi.fn().mockImplementation((projectId: string) => {
|
||||
mockRuntimes.delete(projectId);
|
||||
const index = mockProjectIds.indexOf(projectId);
|
||||
if (index > -1) mockProjectIds.splice(index, 1);
|
||||
return Promise.resolve(undefined);
|
||||
}),
|
||||
getRuntime: vi.fn().mockImplementation((projectId: string) => {
|
||||
return mockRuntimes.get(projectId);
|
||||
}),
|
||||
listRuntimes: vi.fn().mockImplementation(() => {
|
||||
return Array.from(mockRuntimes.values());
|
||||
}),
|
||||
getProjectIds: vi.fn().mockImplementation(() => {
|
||||
return [...mockProjectIds];
|
||||
}),
|
||||
getGlobalMetrics: vi.fn().mockResolvedValue({
|
||||
totalInFlightTasks: 0,
|
||||
totalActiveAgents: 0,
|
||||
runtimeCountByStatus: {
|
||||
active: 0,
|
||||
paused: 0,
|
||||
errored: 0,
|
||||
stopped: 0,
|
||||
starting: 0,
|
||||
stopping: 0,
|
||||
},
|
||||
totalRuntimes: 0,
|
||||
}),
|
||||
acquireGlobalSlot: vi.fn().mockResolvedValue(true),
|
||||
releaseGlobalSlot: vi.fn().mockResolvedValue(undefined),
|
||||
stopAll: vi.fn().mockImplementation(() => {
|
||||
mockRuntimes.clear();
|
||||
mockProjectIds.length = 0;
|
||||
return Promise.resolve(undefined);
|
||||
}),
|
||||
on: vi.fn().mockReturnThis(),
|
||||
})),
|
||||
}));
|
||||
|
||||
describe("HybridExecutor", () => {
|
||||
let executor: HybridExecutor;
|
||||
let mockCentralCore: CentralCore;
|
||||
const mockProject: RegisteredProject = {
|
||||
id: "proj_test123",
|
||||
name: "Test Project",
|
||||
path: "/tmp/test-project",
|
||||
status: "initializing",
|
||||
isolationMode: "in-process",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
mockCentralCore = {
|
||||
listProjects: vi.fn().mockResolvedValue([mockProject]),
|
||||
getProject: vi.fn().mockResolvedValue(mockProject),
|
||||
registerProject: vi.fn().mockResolvedValue(mockProject),
|
||||
unregisterProject: vi.fn().mockResolvedValue(undefined),
|
||||
updateProject: vi.fn().mockResolvedValue(mockProject),
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
updateProjectHealth: vi.fn().mockResolvedValue(undefined),
|
||||
logActivity: vi.fn().mockResolvedValue(undefined),
|
||||
acquireGlobalSlot: vi.fn().mockResolvedValue(true),
|
||||
releaseGlobalSlot: vi.fn().mockResolvedValue(undefined),
|
||||
removeAllListeners: vi.fn(),
|
||||
on: vi.fn().mockReturnThis(),
|
||||
} as unknown as CentralCore;
|
||||
|
||||
executor = new HybridExecutor(mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await executor.shutdown();
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
vi.clearAllMocks();
|
||||
mockRuntimes.clear();
|
||||
mockProjectIds.length = 0;
|
||||
});
|
||||
|
||||
describe("initialization", () => {
|
||||
it("should start with initialized = false", () => {
|
||||
expect(executor.isInitialized()).toBe(false);
|
||||
});
|
||||
|
||||
it("should set initialized = true after initialize()", async () => {
|
||||
await executor.initialize();
|
||||
expect(executor.isInitialized()).toBe(true);
|
||||
});
|
||||
|
||||
it("should load existing projects on initialize", async () => {
|
||||
await executor.initialize();
|
||||
expect(mockCentralCore.listProjects).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should setup CentralCore listeners on initialize", async () => {
|
||||
await executor.initialize();
|
||||
expect(mockCentralCore.on).toHaveBeenCalledWith(
|
||||
"project:registered",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockCentralCore.on).toHaveBeenCalledWith(
|
||||
"project:unregistered",
|
||||
expect.any(Function)
|
||||
);
|
||||
expect(mockCentralCore.on).toHaveBeenCalledWith(
|
||||
"project:updated",
|
||||
expect.any(Function)
|
||||
);
|
||||
});
|
||||
|
||||
it("should be idempotent (initialize multiple times)", async () => {
|
||||
await executor.initialize();
|
||||
const callCount = (mockCentralCore.listProjects as ReturnType<typeof vi.fn>).mock.calls.length;
|
||||
|
||||
await executor.initialize();
|
||||
expect(mockCentralCore.listProjects).toHaveBeenCalledTimes(callCount);
|
||||
});
|
||||
});
|
||||
|
||||
describe("project lifecycle", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
});
|
||||
|
||||
it("should add a project runtime", async () => {
|
||||
const runtime = await executor.addProject(testConfig);
|
||||
expect(runtime).toBeDefined();
|
||||
});
|
||||
|
||||
it("should emit project:added when adding project", async () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("project:added", handler);
|
||||
|
||||
await executor.addProject(testConfig);
|
||||
|
||||
expect(handler).toHaveBeenCalledWith({
|
||||
projectId: "proj_test123",
|
||||
projectName: "Test Project",
|
||||
});
|
||||
});
|
||||
|
||||
it("should remove a project runtime", async () => {
|
||||
await executor.addProject(testConfig);
|
||||
await executor.removeProject("proj_test123");
|
||||
// Should not throw
|
||||
});
|
||||
|
||||
it("should emit project:removed when removing project", async () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("project:removed", handler);
|
||||
|
||||
await executor.addProject(testConfig);
|
||||
await executor.removeProject("proj_test123");
|
||||
|
||||
expect(handler).toHaveBeenCalledWith({
|
||||
projectId: "proj_test123",
|
||||
projectName: "Test Project",
|
||||
});
|
||||
});
|
||||
|
||||
it("should get a runtime by project ID", async () => {
|
||||
await executor.addProject(testConfig);
|
||||
const runtime = executor.getRuntime("proj_test123");
|
||||
expect(runtime).toBeDefined();
|
||||
});
|
||||
|
||||
it("should list all runtimes", async () => {
|
||||
await executor.addProject(testConfig);
|
||||
const runtimes = executor.listRuntimes();
|
||||
expect(Array.isArray(runtimes)).toBe(true);
|
||||
});
|
||||
|
||||
it("should get all project IDs", async () => {
|
||||
await executor.addProject(testConfig);
|
||||
const ids = executor.getProjectIds();
|
||||
expect(ids).toContain("proj_test123");
|
||||
});
|
||||
});
|
||||
|
||||
describe("global metrics", () => {
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
});
|
||||
|
||||
it("should get global metrics", async () => {
|
||||
const metrics = await executor.getGlobalMetrics();
|
||||
|
||||
expect(metrics).toHaveProperty("totalInFlightTasks");
|
||||
expect(metrics).toHaveProperty("totalActiveAgents");
|
||||
expect(metrics).toHaveProperty("runtimeCountByStatus");
|
||||
expect(metrics).toHaveProperty("totalRuntimes");
|
||||
});
|
||||
});
|
||||
|
||||
describe("concurrency slots", () => {
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
});
|
||||
|
||||
it("should acquire global slot", async () => {
|
||||
const acquired = await executor.acquireGlobalSlot("proj_test123");
|
||||
expect(acquired).toBe(true);
|
||||
});
|
||||
|
||||
it("should release global slot", async () => {
|
||||
await executor.releaseGlobalSlot("proj_test123");
|
||||
// Should not throw
|
||||
});
|
||||
});
|
||||
|
||||
describe("event forwarding", () => {
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
});
|
||||
|
||||
it("should support task:created event", () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("task:created", handler);
|
||||
|
||||
// The event should be listenable
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should support task:moved event", () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("task:moved", handler);
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should support task:updated event", () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("task:updated", handler);
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should support error event", () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("error", handler);
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should support health:changed event", () => {
|
||||
const handler = vi.fn();
|
||||
executor.on("health:changed", handler);
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("shutdown", () => {
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
});
|
||||
|
||||
it("should set initialized = false after shutdown", async () => {
|
||||
await executor.shutdown();
|
||||
expect(executor.isInitialized()).toBe(false);
|
||||
});
|
||||
|
||||
it("should be safe to call shutdown multiple times", async () => {
|
||||
await executor.shutdown();
|
||||
await executor.shutdown(); // Should not throw
|
||||
expect(executor.isInitialized()).toBe(false);
|
||||
});
|
||||
|
||||
it("should remove CentralCore listeners on shutdown", async () => {
|
||||
await executor.shutdown();
|
||||
expect(mockCentralCore.removeAllListeners).toHaveBeenCalledWith(
|
||||
"project:registered"
|
||||
);
|
||||
expect(mockCentralCore.removeAllListeners).toHaveBeenCalledWith(
|
||||
"project:unregistered"
|
||||
);
|
||||
expect(mockCentralCore.removeAllListeners).toHaveBeenCalledWith(
|
||||
"project:updated"
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("updateProject", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
beforeEach(async () => {
|
||||
await executor.initialize();
|
||||
await executor.addProject(testConfig);
|
||||
});
|
||||
|
||||
it("should throw if runtime not found", async () => {
|
||||
await expect(
|
||||
executor.updateProject("non-existent", { maxConcurrent: 4 })
|
||||
).rejects.toThrow("Runtime not found");
|
||||
});
|
||||
});
|
||||
});
|
||||
362
packages/engine/src/__tests__/project-runtime.test.ts
Normal file
362
packages/engine/src/__tests__/project-runtime.test.ts
Normal file
@@ -0,0 +1,362 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { Task, TaskStore, CentralCore } from "@fusion/core";
|
||||
import type { Scheduler } from "../scheduler.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeStatus,
|
||||
RuntimeMetrics,
|
||||
ProjectRuntimeEvents,
|
||||
} from "../project-runtime.js";
|
||||
import { InProcessRuntime } from "../runtimes/in-process-runtime.js";
|
||||
import { ChildProcessRuntime } from "../runtimes/child-process-runtime.js";
|
||||
|
||||
/**
|
||||
* Mock implementation of ProjectRuntime for interface compliance testing.
|
||||
* This verifies the interface contract without relying on the full implementation.
|
||||
*/
|
||||
class MockProjectRuntime
|
||||
extends EventEmitter<ProjectRuntimeEvents>
|
||||
implements ProjectRuntime
|
||||
{
|
||||
private _status: RuntimeStatus = "stopped";
|
||||
private _metrics: RuntimeMetrics = {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
constructor(private config: ProjectRuntimeConfig) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
this._status = "starting";
|
||||
this.emit("health-changed", { status: "starting", previous: "stopped" });
|
||||
this._status = "active";
|
||||
this.emit("health-changed", { status: "active", previous: "starting" });
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this._status = "stopping";
|
||||
this.emit("health-changed", { status: "stopping", previous: "active" });
|
||||
this._status = "stopped";
|
||||
this.emit("health-changed", { status: "stopped", previous: "stopping" });
|
||||
}
|
||||
|
||||
getStatus(): RuntimeStatus {
|
||||
return this._status;
|
||||
}
|
||||
|
||||
getTaskStore(): TaskStore {
|
||||
throw new Error("Mock: getTaskStore not implemented");
|
||||
}
|
||||
|
||||
getScheduler(): Scheduler {
|
||||
throw new Error("Mock: getScheduler not implemented");
|
||||
}
|
||||
|
||||
getMetrics(): RuntimeMetrics {
|
||||
return { ...this._metrics };
|
||||
}
|
||||
|
||||
simulateTaskCreated(task: Task): void {
|
||||
this.emit("task:created", task);
|
||||
}
|
||||
|
||||
simulateTaskMoved(
|
||||
task: Task,
|
||||
from: string,
|
||||
to: string
|
||||
): void {
|
||||
this.emit("task:moved", { task, from, to });
|
||||
}
|
||||
|
||||
simulateTaskUpdated(task: Task): void {
|
||||
this.emit("task:updated", task);
|
||||
}
|
||||
|
||||
simulateError(error: Error): void {
|
||||
this.emit("error", error);
|
||||
}
|
||||
}
|
||||
|
||||
describe("ProjectRuntime Interface", () => {
|
||||
const testConfig: ProjectRuntimeConfig = {
|
||||
projectId: "proj_test123",
|
||||
workingDirectory: "/tmp/test-project",
|
||||
isolationMode: "in-process",
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
};
|
||||
|
||||
describe("interface contract", () => {
|
||||
let runtime: MockProjectRuntime;
|
||||
|
||||
beforeEach(() => {
|
||||
runtime = new MockProjectRuntime(testConfig);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
});
|
||||
|
||||
it("should extend EventEmitter", () => {
|
||||
expect(runtime).toBeInstanceOf(EventEmitter);
|
||||
});
|
||||
|
||||
it("should have required methods", () => {
|
||||
expect(typeof runtime.start).toBe("function");
|
||||
expect(typeof runtime.stop).toBe("function");
|
||||
expect(typeof runtime.getStatus).toBe("function");
|
||||
expect(typeof runtime.getTaskStore).toBe("function");
|
||||
expect(typeof runtime.getScheduler).toBe("function");
|
||||
expect(typeof runtime.getMetrics).toBe("function");
|
||||
});
|
||||
|
||||
it("should return RuntimeStatus from getStatus()", () => {
|
||||
const status = runtime.getStatus();
|
||||
expect(["active", "paused", "errored", "stopped", "starting", "stopping"]).toContain(status);
|
||||
});
|
||||
|
||||
it("should return RuntimeMetrics from getMetrics()", () => {
|
||||
const metrics = runtime.getMetrics();
|
||||
expect(metrics).toHaveProperty("inFlightTasks");
|
||||
expect(metrics).toHaveProperty("activeAgents");
|
||||
expect(metrics).toHaveProperty("lastActivityAt");
|
||||
expect(typeof metrics.inFlightTasks).toBe("number");
|
||||
expect(typeof metrics.activeAgents).toBe("number");
|
||||
expect(typeof metrics.lastActivityAt).toBe("string");
|
||||
});
|
||||
});
|
||||
|
||||
describe("status lifecycle", () => {
|
||||
let runtime: MockProjectRuntime;
|
||||
|
||||
beforeEach(() => {
|
||||
runtime = new MockProjectRuntime(testConfig);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
});
|
||||
|
||||
it("should start with status 'stopped'", () => {
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
});
|
||||
|
||||
it("should transition through starting to active", async () => {
|
||||
const transitions: Array<{ status: RuntimeStatus; previous: RuntimeStatus }> = [];
|
||||
runtime.on("health-changed", (data) => {
|
||||
transitions.push(data);
|
||||
});
|
||||
|
||||
await runtime.start();
|
||||
|
||||
expect(runtime.getStatus()).toBe("active");
|
||||
expect(transitions).toContainEqual({ status: "starting", previous: "stopped" });
|
||||
expect(transitions).toContainEqual({ status: "active", previous: "starting" });
|
||||
});
|
||||
|
||||
it("should transition through stopping to stopped", async () => {
|
||||
await runtime.start();
|
||||
|
||||
const transitions: Array<{ status: RuntimeStatus; previous: RuntimeStatus }> = [];
|
||||
runtime.on("health-changed", (data) => {
|
||||
transitions.push(data);
|
||||
});
|
||||
|
||||
await runtime.stop();
|
||||
|
||||
expect(runtime.getStatus()).toBe("stopped");
|
||||
expect(transitions).toContainEqual({ status: "stopping", previous: "active" });
|
||||
expect(transitions).toContainEqual({ status: "stopped", previous: "stopping" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("event emission", () => {
|
||||
let runtime: MockProjectRuntime;
|
||||
|
||||
beforeEach(() => {
|
||||
runtime = new MockProjectRuntime(testConfig);
|
||||
});
|
||||
|
||||
it("should emit task:created events", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("task:created", handler);
|
||||
|
||||
const mockTask = { id: "KB-001", title: "Test Task" } as Task;
|
||||
runtime.simulateTaskCreated(mockTask);
|
||||
|
||||
expect(handler).toHaveBeenCalledWith(mockTask);
|
||||
});
|
||||
|
||||
it("should emit task:moved events", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("task:moved", handler);
|
||||
|
||||
const mockTask = { id: "KB-001", title: "Test Task" } as Task;
|
||||
runtime.simulateTaskMoved(mockTask, "todo", "in-progress");
|
||||
|
||||
expect(handler).toHaveBeenCalledWith({
|
||||
task: mockTask,
|
||||
from: "todo",
|
||||
to: "in-progress",
|
||||
});
|
||||
});
|
||||
|
||||
it("should emit task:updated events", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("task:updated", handler);
|
||||
|
||||
const mockTask = { id: "KB-001", title: "Updated Task" } as Task;
|
||||
runtime.simulateTaskUpdated(mockTask);
|
||||
|
||||
expect(handler).toHaveBeenCalledWith(mockTask);
|
||||
});
|
||||
|
||||
it("should emit error events", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("error", handler);
|
||||
|
||||
const error = new Error("Test error");
|
||||
runtime.simulateError(error);
|
||||
|
||||
expect(handler).toHaveBeenCalledWith(error);
|
||||
});
|
||||
|
||||
it("should support removing event listeners with off()", () => {
|
||||
const handler = vi.fn();
|
||||
runtime.on("task:created", handler);
|
||||
runtime.off("task:created", handler);
|
||||
|
||||
const mockTask = { id: "KB-001", title: "Test Task" } as Task;
|
||||
runtime.simulateTaskCreated(mockTask);
|
||||
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("real implementations", () => {
|
||||
let mockCentralCore: CentralCore;
|
||||
|
||||
beforeEach(() => {
|
||||
mockCentralCore = {
|
||||
getGlobalConcurrencyState: vi.fn().mockResolvedValue({
|
||||
globalMaxConcurrent: 4,
|
||||
currentlyActive: 0,
|
||||
queuedCount: 0,
|
||||
projectsActive: {},
|
||||
}),
|
||||
recordTaskCompletion: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as CentralCore;
|
||||
});
|
||||
|
||||
describe("InProcessRuntime", () => {
|
||||
let runtime: InProcessRuntime;
|
||||
|
||||
beforeEach(() => {
|
||||
runtime = new InProcessRuntime(testConfig, mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
});
|
||||
|
||||
it("should implement ProjectRuntime interface", () => {
|
||||
expect(typeof runtime.start).toBe("function");
|
||||
expect(typeof runtime.stop).toBe("function");
|
||||
expect(typeof runtime.getStatus).toBe("function");
|
||||
expect(typeof runtime.getTaskStore).toBe("function");
|
||||
expect(typeof runtime.getScheduler).toBe("function");
|
||||
expect(typeof runtime.getMetrics).toBe("function");
|
||||
});
|
||||
|
||||
it("should be an EventEmitter", () => {
|
||||
expect(runtime).toBeInstanceOf(EventEmitter);
|
||||
});
|
||||
});
|
||||
|
||||
describe("ChildProcessRuntime", () => {
|
||||
let runtime: ChildProcessRuntime;
|
||||
|
||||
beforeEach(() => {
|
||||
runtime = new ChildProcessRuntime(testConfig, mockCentralCore);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await runtime.stop();
|
||||
} catch {
|
||||
// Ignore cleanup errors
|
||||
}
|
||||
});
|
||||
|
||||
it("should implement ProjectRuntime interface", () => {
|
||||
expect(typeof runtime.start).toBe("function");
|
||||
expect(typeof runtime.stop).toBe("function");
|
||||
expect(typeof runtime.getStatus).toBe("function");
|
||||
expect(typeof runtime.getTaskStore).toBe("function");
|
||||
expect(typeof runtime.getScheduler).toBe("function");
|
||||
expect(typeof runtime.getMetrics).toBe("function");
|
||||
});
|
||||
|
||||
it("should be an EventEmitter", () => {
|
||||
expect(runtime).toBeInstanceOf(EventEmitter);
|
||||
});
|
||||
|
||||
it("should throw for getTaskStore() (not accessible in child mode)", () => {
|
||||
expect(() => runtime.getTaskStore()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
|
||||
it("should throw for getScheduler() (not accessible in child mode)", () => {
|
||||
expect(() => runtime.getScheduler()).toThrow("not accessible in ChildProcessRuntime");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("type guards", () => {
|
||||
it("should validate RuntimeStatus values", () => {
|
||||
const validStatuses: RuntimeStatus[] = [
|
||||
"active",
|
||||
"paused",
|
||||
"errored",
|
||||
"stopped",
|
||||
"starting",
|
||||
"stopping",
|
||||
];
|
||||
|
||||
for (const status of validStatuses) {
|
||||
expect(status).toBeDefined();
|
||||
}
|
||||
});
|
||||
|
||||
it("should validate RuntimeMetrics structure", () => {
|
||||
const metrics: RuntimeMetrics = {
|
||||
inFlightTasks: 0,
|
||||
activeAgents: 0,
|
||||
lastActivityAt: new Date().toISOString(),
|
||||
memoryBytes: 1024 * 1024,
|
||||
};
|
||||
|
||||
expect(metrics.inFlightTasks).toBe(0);
|
||||
expect(metrics.activeAgents).toBe(0);
|
||||
expect(typeof metrics.lastActivityAt).toBe("string");
|
||||
expect(typeof metrics.memoryBytes).toBe("number");
|
||||
});
|
||||
});
|
||||
});
|
||||
441
packages/engine/src/hybrid-executor.ts
Normal file
441
packages/engine/src/hybrid-executor.ts
Normal file
@@ -0,0 +1,441 @@
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { Task, CentralCore, RegisteredProject } from "@fusion/core";
|
||||
import { ProjectManager } from "./project-manager.js";
|
||||
import type {
|
||||
ProjectRuntime,
|
||||
ProjectRuntimeConfig,
|
||||
RuntimeStatus,
|
||||
GlobalMetrics,
|
||||
} from "./project-runtime.js";
|
||||
import type { ProjectManagerEvents } from "./project-manager.js";
|
||||
import { hybridExecutorLog } from "./logger.js";
|
||||
|
||||
/**
|
||||
* Events emitted by HybridExecutor.
|
||||
*/
|
||||
export interface HybridExecutorEvents {
|
||||
/** Emitted when a task is created in any project */
|
||||
"task:created": [data: { projectId: string; projectName: string; task: Task }];
|
||||
/** Emitted when a task is moved in any project */
|
||||
"task:moved": [
|
||||
data: {
|
||||
projectId: string;
|
||||
projectName: string;
|
||||
task: Task;
|
||||
from: string;
|
||||
to: string;
|
||||
}
|
||||
];
|
||||
/** Emitted when a task is updated in any project */
|
||||
"task:updated": [data: { projectId: string; projectName: string; task: Task }];
|
||||
/** Emitted when a task execution completes */
|
||||
"task:completed": [data: { projectId: string; taskId: string; success: boolean }];
|
||||
/** Emitted when a task execution fails */
|
||||
"task:failed": [data: { projectId: string; taskId: string; error: string }];
|
||||
/** Emitted when an error occurs in any project */
|
||||
"error": [data: { projectId: string; projectName: string; error: Error }];
|
||||
/** Emitted when project health status changes */
|
||||
"health:changed": [
|
||||
data: {
|
||||
projectId: string;
|
||||
projectName: string;
|
||||
status: RuntimeStatus;
|
||||
previous: RuntimeStatus;
|
||||
}
|
||||
];
|
||||
/** Emitted when a project runtime is added */
|
||||
"project:added": [data: { projectId: string; projectName: string }];
|
||||
/** Emitted when a project runtime is removed */
|
||||
"project:removed": [data: { projectId: string; projectName: string }];
|
||||
}
|
||||
|
||||
/**
|
||||
* Options for creating a HybridExecutor.
|
||||
*/
|
||||
export interface HybridExecutorOptions {
|
||||
/** Called when a task is scheduled */
|
||||
onTaskScheduled?: (projectId: string, task: Task) => void;
|
||||
/** Called when a task is blocked by dependencies */
|
||||
onTaskBlocked?: (projectId: string, task: Task, blockedBy: string[]) => void;
|
||||
/** Called when a task completes */
|
||||
onTaskCompleted?: (projectId: string, taskId: string, success: boolean) => void;
|
||||
/** Called when a task fails */
|
||||
onTaskFailed?: (projectId: string, taskId: string, error: Error) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* HybridExecutor — Multi-project task execution orchestrator.
|
||||
*
|
||||
* Manages the lifecycle of project runtimes (both in-process and child-process),
|
||||
* coordinates task execution across all registered projects, and enforces
|
||||
* global concurrency limits from CentralCore.
|
||||
*
|
||||
* This is the main entry point for multi-project task execution in kb. It sits
|
||||
* between CentralCore (project registry) and the individual ProjectRuntimes,
|
||||
* routing tasks to the appropriate runtime based on project configuration.
|
||||
*
|
||||
* ## Architecture
|
||||
*
|
||||
* ```
|
||||
* ┌─────────────────────────────────────────────────────────────┐
|
||||
* │ HybridExecutor │
|
||||
* │ ┌─────────────────────────────────────────────────────┐ │
|
||||
* │ │ ProjectManager (internal) │ │
|
||||
* │ │ ┌──────────────┐ ┌──────────────┐ ┌─────────────┐ │ │
|
||||
* │ │ │ Project A │ │ Project B │ │ Project C │ │ │
|
||||
* │ │ │ (in-process) │ │(child-process│ │(in-process) │ │ │
|
||||
* │ │ └──────────────┘ └──────────────┘ └─────────────┘ │ │
|
||||
* │ └─────────────────────────────────────────────────────┘ │
|
||||
* └─────────────────────────────────────────────────────────────┘
|
||||
* │
|
||||
* ┌─────────┴──────────┐
|
||||
* ▼ ▼
|
||||
* ┌──────────┐ ┌──────────┐
|
||||
* │CentralCore│ │ Scheduler │
|
||||
* │ (registry)│ │ (per proj)│
|
||||
* └──────────┘ └──────────┘
|
||||
* ```
|
||||
*
|
||||
* ## Example
|
||||
*
|
||||
* ```typescript
|
||||
* const central = new CentralCore();
|
||||
* await central.init();
|
||||
*
|
||||
* const executor = new HybridExecutor(central);
|
||||
* await executor.initialize();
|
||||
*
|
||||
* // Add a project (must be registered in CentralCore first)
|
||||
* const project = await central.registerProject({
|
||||
* name: "My Project",
|
||||
* path: "/path/to/project"
|
||||
* });
|
||||
*
|
||||
* await executor.addProject({
|
||||
* projectId: project.id,
|
||||
* workingDirectory: project.path,
|
||||
* isolationMode: "in-process",
|
||||
* maxConcurrent: 2,
|
||||
* maxWorktrees: 4,
|
||||
* });
|
||||
*
|
||||
* // Listen for events
|
||||
* executor.on("task:completed", ({ projectId, taskId }) => {
|
||||
* console.log(`Task ${taskId} completed in ${projectId}`);
|
||||
* });
|
||||
*
|
||||
* // Graceful shutdown
|
||||
* await executor.shutdown();
|
||||
* ```
|
||||
*
|
||||
* @see ProjectManager - The underlying project orchestration class
|
||||
* @see ProjectRuntime - The runtime interface for individual projects
|
||||
*/
|
||||
export class HybridExecutor extends EventEmitter<HybridExecutorEvents> {
|
||||
private projectManager: ProjectManager;
|
||||
private initialized = false;
|
||||
|
||||
/**
|
||||
* @param centralCore - CentralCore reference for global coordination
|
||||
* @param options - Optional configuration callbacks
|
||||
*/
|
||||
constructor(
|
||||
private centralCore: CentralCore,
|
||||
private options: HybridExecutorOptions = {}
|
||||
) {
|
||||
super();
|
||||
this.setMaxListeners(100);
|
||||
|
||||
// Create internal ProjectManager
|
||||
this.projectManager = new ProjectManager(centralCore);
|
||||
|
||||
// Set up event forwarding from ProjectManager
|
||||
this.setupEventForwarding();
|
||||
|
||||
hybridExecutorLog.log("HybridExecutor created");
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the HybridExecutor and load existing projects.
|
||||
*
|
||||
* Loads all registered projects from CentralCore and creates appropriate
|
||||
* runtimes based on their isolation mode configuration.
|
||||
*/
|
||||
async initialize(): Promise<void> {
|
||||
if (this.initialized) {
|
||||
hybridExecutorLog.warn("HybridExecutor already initialized");
|
||||
return;
|
||||
}
|
||||
|
||||
hybridExecutorLog.log("Initializing HybridExecutor...");
|
||||
|
||||
// Load all registered projects from CentralCore
|
||||
const projects = await this.centralCore.listProjects();
|
||||
|
||||
// Start runtimes for all active projects
|
||||
for (const project of projects) {
|
||||
if (project.status === "active" || project.status === "initializing") {
|
||||
try {
|
||||
await this.addProject({
|
||||
projectId: project.id,
|
||||
workingDirectory: project.path,
|
||||
isolationMode: project.isolationMode,
|
||||
maxConcurrent: project.settings?.maxConcurrent ?? 2,
|
||||
maxWorktrees: project.settings?.maxWorktrees ?? 4,
|
||||
settings: project.settings,
|
||||
});
|
||||
hybridExecutorLog.log(`Loaded project runtime for ${project.name}`);
|
||||
} catch (error) {
|
||||
hybridExecutorLog.error(
|
||||
`Failed to load project ${project.id}:`,
|
||||
error instanceof Error ? error.message : String(error)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Listen for CentralCore project events
|
||||
this.setupCentralCoreListeners();
|
||||
|
||||
this.initialized = true;
|
||||
hybridExecutorLog.log("HybridExecutor initialized");
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a project runtime and start it.
|
||||
*
|
||||
* @param config - Runtime configuration (must match a registered project)
|
||||
* @returns The created and started ProjectRuntime
|
||||
* @throws Error if project not found in CentralCore or runtime already exists
|
||||
*/
|
||||
async addProject(config: ProjectRuntimeConfig): Promise<ProjectRuntime> {
|
||||
const runtime = await this.projectManager.addProject(config);
|
||||
|
||||
const project = await this.centralCore.getProject(config.projectId);
|
||||
this.emit("project:added", {
|
||||
projectId: config.projectId,
|
||||
projectName: project?.name ?? config.projectId,
|
||||
});
|
||||
|
||||
return runtime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a project runtime and stop it.
|
||||
*
|
||||
* @param projectId - Project ID to remove
|
||||
* @throws Error if runtime not found
|
||||
*/
|
||||
async removeProject(projectId: string): Promise<void> {
|
||||
const project = await this.centralCore.getProject(projectId);
|
||||
|
||||
await this.projectManager.removeProject(projectId);
|
||||
|
||||
this.emit("project:removed", {
|
||||
projectId,
|
||||
projectName: project?.name ?? projectId,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Update a project's runtime configuration.
|
||||
*
|
||||
* If the isolation mode changes, the old runtime will be stopped and a new
|
||||
* one started with the new configuration.
|
||||
*
|
||||
* @param projectId - Project ID to update
|
||||
* @param config - New runtime configuration
|
||||
*/
|
||||
async updateProject(
|
||||
projectId: string,
|
||||
config: Partial<ProjectRuntimeConfig>
|
||||
): Promise<ProjectRuntime> {
|
||||
const existingRuntime = this.projectManager.getRuntime(projectId);
|
||||
if (!existingRuntime) {
|
||||
throw new Error(`Runtime not found for project ${projectId}`);
|
||||
}
|
||||
|
||||
const currentStatus = existingRuntime.getStatus();
|
||||
const currentMode =
|
||||
existingRuntime instanceof
|
||||
(await import("./runtimes/child-process-runtime.js")).ChildProcessRuntime
|
||||
? "child-process"
|
||||
: "in-process";
|
||||
|
||||
// If isolation mode changed, need to recreate the runtime
|
||||
if (config.isolationMode && config.isolationMode !== currentMode) {
|
||||
hybridExecutorLog.log(
|
||||
`Isolation mode changed for ${projectId}: ${currentMode} → ${config.isolationMode}`
|
||||
);
|
||||
|
||||
// Stop old runtime
|
||||
await this.projectManager.removeProject(projectId);
|
||||
|
||||
// Get the full current config
|
||||
const fullConfig: ProjectRuntimeConfig = {
|
||||
projectId,
|
||||
workingDirectory: config.workingDirectory ?? "/tmp",
|
||||
isolationMode: config.isolationMode,
|
||||
maxConcurrent: config.maxConcurrent ?? 2,
|
||||
maxWorktrees: config.maxWorktrees ?? 4,
|
||||
settings: config.settings,
|
||||
};
|
||||
|
||||
// Start new runtime with new mode
|
||||
return await this.addProject(fullConfig);
|
||||
}
|
||||
|
||||
// For other config changes, just return the existing runtime
|
||||
// (specific config updates can be added here as needed)
|
||||
return existingRuntime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a runtime by project ID.
|
||||
*/
|
||||
getRuntime(projectId: string): ProjectRuntime | undefined {
|
||||
return this.projectManager.getRuntime(projectId);
|
||||
}
|
||||
|
||||
/**
|
||||
* List all managed runtimes.
|
||||
*/
|
||||
listRuntimes(): ProjectRuntime[] {
|
||||
return this.projectManager.listRuntimes();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all project IDs.
|
||||
*/
|
||||
getProjectIds(): string[] {
|
||||
return this.projectManager.getProjectIds();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get global metrics aggregated across all runtimes.
|
||||
*/
|
||||
async getGlobalMetrics(): Promise<GlobalMetrics> {
|
||||
return this.projectManager.getGlobalMetrics();
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire a global concurrency slot.
|
||||
*
|
||||
* @param projectId - Project requesting the slot
|
||||
* @returns true if slot acquired, false if at limit
|
||||
*/
|
||||
async acquireGlobalSlot(projectId: string): Promise<boolean> {
|
||||
return this.projectManager.acquireGlobalSlot(projectId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a global concurrency slot.
|
||||
*
|
||||
* @param projectId - Project releasing the slot
|
||||
*/
|
||||
async releaseGlobalSlot(projectId: string): Promise<void> {
|
||||
return this.projectManager.releaseGlobalSlot(projectId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Graceful shutdown of all runtimes.
|
||||
*
|
||||
* Stops accepting new tasks, waits for active tasks to complete (with timeout),
|
||||
* and shuts down all runtimes in parallel.
|
||||
*/
|
||||
async shutdown(): Promise<void> {
|
||||
if (!this.initialized) {
|
||||
return;
|
||||
}
|
||||
|
||||
hybridExecutorLog.log("Shutting down HybridExecutor...");
|
||||
|
||||
// Stop listening to CentralCore events
|
||||
this.centralCore.removeAllListeners("project:registered");
|
||||
this.centralCore.removeAllListeners("project:unregistered");
|
||||
this.centralCore.removeAllListeners("project:updated");
|
||||
|
||||
// Stop all runtimes
|
||||
await this.projectManager.stopAll();
|
||||
|
||||
this.initialized = false;
|
||||
hybridExecutorLog.log("HybridExecutor shutdown complete");
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the HybridExecutor is initialized.
|
||||
*/
|
||||
isInitialized(): boolean {
|
||||
return this.initialized;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up event forwarding from ProjectManager to HybridExecutor listeners.
|
||||
*/
|
||||
private setupEventForwarding(): void {
|
||||
// Forward task:created
|
||||
this.projectManager.on("task:created", (data) => {
|
||||
this.emit("task:created", data);
|
||||
});
|
||||
|
||||
// Forward task:moved
|
||||
this.projectManager.on("task:moved", (data) => {
|
||||
this.emit("task:moved", data);
|
||||
});
|
||||
|
||||
// Forward task:updated
|
||||
this.projectManager.on("task:updated", (data) => {
|
||||
this.emit("task:updated", data);
|
||||
});
|
||||
|
||||
// Forward errors
|
||||
this.projectManager.on("error", (data) => {
|
||||
this.emit("error", data);
|
||||
});
|
||||
|
||||
// Forward health changes
|
||||
this.projectManager.on("health:changed", (data) => {
|
||||
this.emit("health:changed", data);
|
||||
});
|
||||
|
||||
// Forward runtime added/removed as project added/removed
|
||||
this.projectManager.on("runtime:added", (data) => {
|
||||
this.emit("project:added", data);
|
||||
});
|
||||
|
||||
this.projectManager.on("runtime:removed", (data) => {
|
||||
this.emit("project:removed", data);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up listeners for CentralCore project events.
|
||||
*/
|
||||
private setupCentralCoreListeners(): void {
|
||||
// When a new project is registered, we don't auto-add it
|
||||
// The user must explicitly call addProject()
|
||||
this.centralCore.on("project:registered", (project: RegisteredProject) => {
|
||||
hybridExecutorLog.log(`New project registered: ${project.name} (${project.id})`);
|
||||
});
|
||||
|
||||
// When a project is unregistered, remove its runtime
|
||||
this.centralCore.on("project:unregistered", (projectId: string) => {
|
||||
hybridExecutorLog.log(`Project unregistered: ${projectId}`);
|
||||
const runtime = this.projectManager.getRuntime(projectId);
|
||||
if (runtime) {
|
||||
this.removeProject(projectId).catch((error: unknown) => {
|
||||
hybridExecutorLog.error(
|
||||
`Failed to remove runtime for ${projectId}:`,
|
||||
error instanceof Error ? error.message : String(error)
|
||||
);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
// When a project is updated, check if we need to update the runtime
|
||||
this.centralCore.on("project:updated", (project: RegisteredProject) => {
|
||||
hybridExecutorLog.log(`Project updated: ${project.name} (${project.id})`);
|
||||
// Could trigger runtime reconfiguration here if needed
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -24,11 +24,15 @@ export {
|
||||
type RuntimeMetrics,
|
||||
type ProjectRuntimeEvents,
|
||||
type GlobalMetrics,
|
||||
type TaskExecutionResult,
|
||||
type RuntimeHealth,
|
||||
type RuntimeEventType,
|
||||
} from "./project-runtime.js";
|
||||
|
||||
export { InProcessRuntime } from "./runtimes/in-process-runtime.js";
|
||||
export { ChildProcessRuntime } from "./runtimes/child-process-runtime.js";
|
||||
export { ProjectManager, type ProjectManagerEvents } from "./project-manager.js";
|
||||
export { HybridExecutor, type HybridExecutorEvents, type HybridExecutorOptions } from "./hybrid-executor.js";
|
||||
|
||||
// ── IPC Protocol ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -73,3 +73,6 @@ export const ipcLog = createLogger("ipc");
|
||||
|
||||
/** Logger for the project manager subsystem. */
|
||||
export const projectManagerLog = createLogger("project-manager");
|
||||
|
||||
/** Logger for the hybrid executor subsystem. */
|
||||
export const hybridExecutorLog = createLogger("hybrid-executor");
|
||||
|
||||
@@ -128,6 +128,53 @@ export interface ProjectRuntime extends EventEmitter<ProjectRuntimeEvents> {
|
||||
getMetrics(): RuntimeMetrics;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of a task execution operation.
|
||||
*/
|
||||
export interface TaskExecutionResult {
|
||||
/** Whether the task execution was successful */
|
||||
success: boolean;
|
||||
/** The task ID that was executed */
|
||||
taskId: string;
|
||||
/** Error message if execution failed */
|
||||
error?: string;
|
||||
/** Number of steps completed during execution */
|
||||
stepsCompleted?: number;
|
||||
/** ISO-8601 timestamp when execution started */
|
||||
startedAt?: string;
|
||||
/** ISO-8601 timestamp when execution completed */
|
||||
completedAt?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Health metrics for a ProjectRuntime instance.
|
||||
* Used for detailed health monitoring and diagnostics.
|
||||
*/
|
||||
export interface RuntimeHealth {
|
||||
/** Current status of the runtime */
|
||||
status: RuntimeStatus;
|
||||
/** Number of tasks currently in-progress */
|
||||
activeTasks: number;
|
||||
/** Memory usage in bytes */
|
||||
memoryUsage: number;
|
||||
/** ISO-8601 timestamp of the last activity */
|
||||
lastActivityAt: string;
|
||||
/** Number of errors encountered */
|
||||
errorCount: number;
|
||||
/** Optional last error message */
|
||||
lastError?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runtime event types for event handlers.
|
||||
*/
|
||||
export type RuntimeEventType =
|
||||
| "task:created"
|
||||
| "task:completed"
|
||||
| "task:failed"
|
||||
| "health:changed"
|
||||
| "error";
|
||||
|
||||
/**
|
||||
* Global metrics aggregated across all project runtimes.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user