refactor(services): combine runtime orchestration ownership

This commit is contained in:
ZacharyZhang-NY committed 2026-04-24 01:10:54 -04:00
1 parent fd86fc6668
commit 3609456ca5
9 files changed
+286 -380

No files matched your search

+22 -45
View File
@@ -12,15 +12,12 @@ import type { NotificationOptions } from "./notification-service";
import type { createAutomationService } from "./automation-service";
import type { createBrowserActionService } from "./browser-action-service";
import type { createOnboardingService } from "./onboarding-service";
import type { createOrchestrationService } from "./orchestration-service";
import { createUpdaterService } from "./updater-service";
import { createLeafServices } from "./service-graph/create-leaf-services";
import { createDesktopShell } from "./service-graph/create-desktop-shell";
import { buildWorkerService } from "./service-graph/create-worker-service";
import { buildOrchestrationService } from "./service-graph/create-orchestration-service";
import { createRuntimeOrchestrationService } from "./service-graph/create-runtime-orchestration-service";
import { createBackgroundServices } from "./service-graph/create-background-services";
import { createLifecycle } from "./service-graph/create-lifecycle";
import { createServiceEventHub } from "./service-graph/service-event-hub";
interface ServiceGraphDependencies {
db: AppDatabase;
@@ -58,8 +55,9 @@ interface ServiceGraphDependencies {
/**
* Thin orchestrator: assembles services in a deterministic topological order
* and exposes them via read-only getters. Cross-service callbacks are routed
* through a typed event hub so construction order stays explicit.
* and exposes them via read-only getters. The hot run path is owned by
* runtimeOrchestrationService, which holds worker/orchestration callbacks
* together so graph construction stays explicit.
*/
export function createServiceGraph(deps: ServiceGraphDependencies) {
const {
@@ -90,8 +88,7 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
// the lifecycle drain has torn services down.
let desktopShellService: ReturnType<typeof createDesktopShell> | null = null;
let connectorHealthService: ReturnType<typeof createConnectorHealthService> | null = null;
let workerService: ReturnType<typeof buildWorkerService> | null = null;
let orchestrationService: ReturnType<typeof createOrchestrationService> | null = null;
let runtimeOrchestrationService: ReturnType<typeof createRuntimeOrchestrationService> | null = null;
let automationService: ReturnType<typeof createAutomationService> | null = null;
let browserActionService: ReturnType<typeof createBrowserActionService> | null = null;
let onboardingService: ReturnType<typeof createOnboardingService> | null = null;
@@ -100,7 +97,6 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
let leaf: ReturnType<typeof createLeafServices> | null = null;
let runtimeMessageInjector: RuntimeMessageInjector | null = null;
const piRuntimeRef: { current: PiRuntimeServices | null } = { current: null };
const serviceEvents = createServiceEventHub();
function requireLeaf() {
if (!leaf) throw new Error("Leaf services are not initialized.");
@@ -152,16 +148,13 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
credentialStore: piRuntimeRef.current.credentialStore,
});
workerService = buildWorkerService({
runtimeOrchestrationService = createRuntimeOrchestrationService({
db,
logger,
workerPath,
isQuittingRef,
publishDomainChanged,
emitEvent,
wakeAgentIfPossible: serviceEvents.wakeAgentIfPossible,
dispatchRunToWorker: serviceEvents.dispatchRunToWorker,
processRunCompletion: serviceEvents.processRunCompletion,
onRunStarted,
onRunCompleted,
notify,
@@ -169,41 +162,28 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
sessionService: leaf.sessionManagementService,
piRuntimeRef,
getRuntimeMessageInjector: () => runtimeMessageInjector,
createRunToken,
getApiPort: () => apiServerRef.current?.port ?? 0,
skillService: leaf.skillService,
checkpointEngine,
});
runtimeMessageInjector = new RuntimeMessageInjector({
db,
runtimeBridge: piRuntimeRef.current.runtimeBridge,
postToWorker: (message) => {
const worker = workerService?.getWorker();
const worker = runtimeOrchestrationService?.getWorker();
if (!worker) throw new Error("Worker is not initialised — cannot deliver steer message.");
worker.postMessage(message);
},
getWakeAgent: () => {
if (!serviceEvents.hasOrchestrationHandlers()) return null;
return (agentId, companyId, trigger) => serviceEvents.wakeAgentIfPossible(agentId, companyId, trigger);
const runtime = runtimeOrchestrationService;
if (!runtime) return null;
return (agentId, companyId, trigger) => runtime.wakeAgentIfPossible(agentId, companyId, trigger);
},
logger,
});
orchestrationService = buildOrchestrationService({
db,
logger,
publishDomainChanged,
emitEvent,
notify,
createRunToken,
getApiPort: () => apiServerRef.current?.port ?? 0,
isQuittingRef,
skillService: leaf.skillService,
budgetService: leaf.budgetEnforcementService,
sessionService: leaf.sessionManagementService,
workerService,
dispatchAutomation: serviceEvents.dispatchAutomation,
checkpointEngine,
});
serviceEvents.registerOrchestration(orchestrationService);
const background = createBackgroundServices({
db,
logger,
@@ -213,15 +193,15 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
isQuittingRef,
browserManagerRef,
skillService: leaf.skillService,
orchestrationService,
orchestrationService: runtimeOrchestrationService,
seedDefaultAutomationRules,
});
automationService = background.automationService;
browserActionService = background.browserActionService;
onboardingService = background.onboardingService;
serviceEvents.registerAutomation(automationService);
serviceEvents.assertReady();
workerService.createWorker();
runtimeOrchestrationService.bindAutomation(automationService);
runtimeOrchestrationService.assertReady();
runtimeOrchestrationService.createWorker();
appLifecycleService = createLifecycle({
db,
@@ -230,23 +210,20 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
apiServerRef,
browserManagerRef,
piRuntimeRef,
orchestrationService,
runtimeOrchestrationService,
automationService,
workerService,
desktopShellService,
clearServiceRefs: () => {
desktopShellService = null;
workerService = null;
runtimeOrchestrationService = null;
browserActionService = null;
automationService = null;
connectorHealthService = null;
onboardingService = null;
orchestrationService = null;
updaterService = null;
appLifecycleService = null;
runtimeMessageInjector = null;
leaf = null;
serviceEvents.clear();
},
});
}
@@ -255,9 +232,9 @@ export function createServiceGraph(deps: ServiceGraphDependencies) {
init,
getDesktopShellService: () => desktopShellService,
getConnectorHealthService: () => connectorHealthService,
getWorkerService: () => workerService,
getWorkerService: () => runtimeOrchestrationService,
getBrowserActionService: () => browserActionService,
getOrchestrationService: () => orchestrationService,
getOrchestrationService: () => runtimeOrchestrationService,
getAutomationService: () => automationService,
getOnboardingService: () => onboardingService,
getUpdaterService: () => updaterService,
@@ -4,7 +4,7 @@ import type { DesktopEvent } from "@shared/types";
import { createAutomationService } from "../automation-service";
import { createBrowserActionService } from "../browser-action-service";
import { createOnboardingService } from "../onboarding-service";
import type { createOrchestrationService } from "../orchestration-service";
import type { createRuntimeOrchestrationService } from "./create-runtime-orchestration-service";
import type { createSkillService } from "../skill-service";
import type { NotificationOptions } from "../notification-service";
@@ -22,7 +22,7 @@ export interface BackgroundServicesDeps {
isQuittingRef: { current: boolean };
browserManagerRef: { current: BrowserManager | null };
skillService: ReturnType<typeof createSkillService>;
orchestrationService: ReturnType<typeof createOrchestrationService>;
orchestrationService: ReturnType<typeof createRuntimeOrchestrationService>;
seedDefaultAutomationRules: (companyId: string) => void;
}
+5 -8
View File
@@ -5,9 +5,8 @@ import { BrowserManager } from "../browser-manager";
import { createAppLifecycleService } from "../app-lifecycle-service";
import type { createAutomationService } from "../automation-service";
import type { createDesktopShellService } from "../desktop-shell-service";
import type { createOrchestrationService } from "../orchestration-service";
import type { createWorkerService } from "../worker-service";
import type { PiRuntimeServices } from "../service-graph-pi-runtime";
import type { createRuntimeOrchestrationService } from "./create-runtime-orchestration-service";
/** Overall budget for awaited shutdown drain. */
const SHUTDOWN_BUDGET_MS = 5_000;
@@ -24,9 +23,8 @@ export interface LifecycleDeps {
apiServerRef: { current: AgentApiServer | null };
browserManagerRef: { current: BrowserManager | null };
piRuntimeRef: { current: PiRuntimeServices | null };
orchestrationService: ReturnType<typeof createOrchestrationService>;
runtimeOrchestrationService: ReturnType<typeof createRuntimeOrchestrationService>;
automationService: ReturnType<typeof createAutomationService>;
workerService: ReturnType<typeof createWorkerService>;
desktopShellService: ReturnType<typeof createDesktopShellService>;
/**
* Clears graph-owned service refs (so getters return null) after the drain
@@ -50,9 +48,8 @@ export function createLifecycle(deps: LifecycleDeps) {
apiServerRef,
browserManagerRef,
piRuntimeRef,
orchestrationService,
runtimeOrchestrationService,
automationService,
workerService,
desktopShellService,
clearServiceRefs,
} = deps;
@@ -71,7 +68,7 @@ export function createLifecycle(deps: LifecycleDeps) {
}
async function drain(): Promise<number> {
try { orchestrationService.stopAllSchedulers(); } catch (err) { logger.warn("[lifecycle] stopAllSchedulers failed:", err); }
try { runtimeOrchestrationService.stopAllSchedulers(); } catch (err) { logger.warn("[lifecycle] stopAllSchedulers failed:", err); }
try { automationService.destroy(); } catch (err) { logger.warn("[lifecycle] automationService.destroy failed:", err); }
try { apiServerRef.current?.stop(); } catch (err) { logger.warn("[lifecycle] apiServer.stop failed:", err); }
apiServerRef.current = null;
@@ -83,7 +80,7 @@ export function createLifecycle(deps: LifecycleDeps) {
if (typeof handle.unref === "function") handle.unref();
});
const outcome = await Promise.race([
workerService.shutdown().then(() => "drained" as const),
runtimeOrchestrationService.shutdown().then(() => "drained" as const),
timeout,
]);
@@ -1,83 +0,0 @@
import type { AppDatabase } from "../database";
import type { DesktopEvent } from "@shared/types";
import { createOrchestrationService } from "../orchestration-service";
import type { createBudgetEnforcementService } from "../budget-enforcement-service";
import type { createSessionManagementService } from "../session-management-service";
import type { createSkillService } from "../skill-service";
import type { createWorkerService } from "../worker-service";
import type { NotificationOptions } from "../notification-service";
import type { OrchestrationDependencies } from "../orchestration/types";
export interface OrchestrationServiceDeps {
db: AppDatabase;
logger: {
debug: (...args: unknown[]) => void;
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
publishDomainChanged: () => void;
emitEvent: (event: DesktopEvent) => void;
notify: (options: NotificationOptions) => void;
createRunToken: (agentId: string, companyId: string, runId: string) => string;
getApiPort: () => number;
isQuittingRef: { current: boolean };
skillService: ReturnType<typeof createSkillService>;
budgetService: ReturnType<typeof createBudgetEnforcementService>;
sessionService: ReturnType<typeof createSessionManagementService>;
workerService: ReturnType<typeof createWorkerService>;
dispatchAutomation: OrchestrationDependencies["dispatchAutomation"];
checkpointEngine?: { shouldPauseHeartbeat: (companyId: string, agentId: string) => boolean };
}
/** Build the orchestration service with concrete worker wiring and event dispatch. */
export function buildOrchestrationService({
db,
logger,
publishDomainChanged,
emitEvent,
notify,
createRunToken,
getApiPort,
isQuittingRef,
skillService,
budgetService,
sessionService,
workerService,
dispatchAutomation,
checkpointEngine,
}: OrchestrationServiceDeps) {
return createOrchestrationService({
db,
logger,
publishDomainChanged,
emitEvent,
notify,
createRunToken,
getApiPort,
dispatchAutomation,
hasWorkerReady: () => Boolean(workerService.getWorker()),
enqueueWorkerRun: (payload) => {
workerService.getWorker()?.postMessage({ type: "enqueue-run", payload });
},
isQuitting: () => isQuittingRef.current,
listRuntimeSkillEntries: (companyId, agentId, connectorId) =>
skillService.listRuntimeSkillEntries(companyId, agentId, connectorId),
budgetService,
checkpointEngine,
getSessionHandoff: (agentId) => sessionService.getHandoffForAgent(agentId),
evaluatePreRunSessionRotation: (agentId, _connectorId) => {
const agent = db.core.entity.getAgent(agentId);
if (!agent.sessionId) return { rotated: false, handoff: null, sessionId: null };
const session = db.core.tasks.getSessionById(agent.sessionId);
if (!session) return { rotated: false, handoff: null, sessionId: agent.sessionId };
const result = sessionService.evaluateRotation(session);
if (result.shouldRotate) {
const newSessionId = sessionService.rotateSession(session);
const handoff = sessionService.getHandoffForAgent(agentId);
return { rotated: true, handoff, sessionId: newSessionId };
}
return { rotated: false, handoff: null, sessionId: agent.sessionId };
},
});
}
@@ -0,0 +1,192 @@
import type { AppDatabase } from "../database";
import type { DesktopEvent } from "@shared/types";
import { createWorkerService } from "../worker-service";
import { createOrchestrationService } from "../orchestration-service";
import type { createBudgetEnforcementService } from "../budget-enforcement-service";
import type { createSessionManagementService } from "../session-management-service";
import type { createSkillService } from "../skill-service";
import type { PiRuntimeServices } from "../service-graph-pi-runtime";
import type { NotificationOptions } from "../notification-service";
import type { RuntimeMessageInjector } from "../agent-runtime";
import type { WorkerServiceDependencies } from "../worker/entry/types";
import type { OrchestrationDependencies } from "../orchestration/types";
export interface RuntimeOrchestrationServiceDeps {
db: AppDatabase;
logger: {
debug: (...args: unknown[]) => void;
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
workerPath: string;
isQuittingRef: { current: boolean };
publishDomainChanged: () => void;
emitEvent: (event: DesktopEvent) => void;
notify: (options: NotificationOptions) => void;
createRunToken: (agentId: string, companyId: string, runId: string) => string;
getApiPort: () => number;
skillService: ReturnType<typeof createSkillService>;
budgetService: ReturnType<typeof createBudgetEnforcementService>;
sessionService: ReturnType<typeof createSessionManagementService>;
piRuntimeRef: { current: PiRuntimeServices | null };
getRuntimeMessageInjector: () => RuntimeMessageInjector | null;
onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void;
onRunCompleted?: (
runId: string,
taskId: string | null,
agentId: string,
companyId: string,
status: string,
summary: string | null,
errorMessage: string | null,
) => void;
checkpointEngine?: { shouldPauseHeartbeat: (companyId: string, agentId: string) => boolean };
}
type AutomationHandlers = {
dispatchAutomation: OrchestrationDependencies["dispatchAutomation"];
processRunCompletion: WorkerServiceDependencies["processRunCompletion"];
};
/**
* Owns the two halves of the run execution loop as one object:
* orchestration decides which run should start, and the worker service
* owns the utility process that executes it. Keeping both in one factory
* removes the service-graph event hub from the hot run path.
*/
export function createRuntimeOrchestrationService(deps: RuntimeOrchestrationServiceDeps) {
const {
db,
logger,
workerPath,
isQuittingRef,
publishDomainChanged,
emitEvent,
notify,
createRunToken,
getApiPort,
skillService,
budgetService,
sessionService,
piRuntimeRef,
getRuntimeMessageInjector,
onRunStarted,
onRunCompleted,
checkpointEngine,
} = deps;
let automationHandlers: AutomationHandlers | null = null;
let orchestrationService: ReturnType<typeof createOrchestrationService> | null = null;
function requireAutomation(): AutomationHandlers {
if (!automationHandlers) {
throw new Error("Runtime orchestration automation handlers are not bound.");
}
return automationHandlers;
}
function requireOrchestration(): ReturnType<typeof createOrchestrationService> {
if (!orchestrationService) {
throw new Error("Runtime orchestration service is not initialized.");
}
return orchestrationService;
}
const workerService = createWorkerService({
db,
logger,
workerPath,
isQuitting: () => isQuittingRef.current,
publishDomainChanged,
emitEvent,
wakeAgentIfPossible: (agentId, companyId, trigger) =>
requireOrchestration().wakeAgentIfPossible(agentId, companyId, trigger),
dispatchRunToWorker: (params) => {
requireOrchestration().dispatchRunToWorker(params);
},
processRunCompletion: (run, event) => {
requireAutomation().processRunCompletion(run, event);
},
onRunStarted,
onRunCompleted,
notify,
budgetService,
sessionService,
getRuntimeBridge: () => piRuntimeRef.current?.runtimeBridge ?? null,
getRuntimeMessageInjector,
});
orchestrationService = createOrchestrationService({
db,
logger,
publishDomainChanged,
emitEvent,
notify,
createRunToken,
getApiPort,
dispatchAutomation: (trigger, ctx) => {
requireAutomation().dispatchAutomation(trigger, ctx);
},
hasWorkerReady: () => Boolean(workerService.getWorker()),
enqueueWorkerRun: (payload) => {
workerService.getWorker()?.postMessage({ type: "enqueue-run", payload });
},
isQuitting: () => isQuittingRef.current,
listRuntimeSkillEntries: (companyId, agentId, connectorId) =>
skillService.listRuntimeSkillEntries(companyId, agentId, connectorId),
budgetService,
checkpointEngine,
getSessionHandoff: (agentId) => sessionService.getHandoffForAgent(agentId),
evaluatePreRunSessionRotation: (agentId, _connectorId) => {
const agent = db.core.entity.getAgent(agentId);
if (!agent.sessionId) return { rotated: false, handoff: null, sessionId: null };
const session = db.core.tasks.getSessionById(agent.sessionId);
if (!session) return { rotated: false, handoff: null, sessionId: agent.sessionId };
const result = sessionService.evaluateRotation(session);
if (result.shouldRotate) {
const newSessionId = sessionService.rotateSession(session);
const handoff = sessionService.getHandoffForAgent(agentId);
return { rotated: true, handoff, sessionId: newSessionId };
}
return { rotated: false, handoff: null, sessionId: agent.sessionId };
},
});
return {
bindAutomation(service: AutomationHandlers): void {
if (automationHandlers) {
throw new Error("Runtime orchestration automation handlers are already bound.");
}
automationHandlers = service;
},
assertReady(): void {
requireAutomation();
requireOrchestration();
},
createWorker: workerService.createWorker,
getWorker: workerService.getWorker,
shutdown: workerService.shutdown,
destroy: workerService.destroy,
resolveWorkspace: (...args: Parameters<ReturnType<typeof createOrchestrationService>["resolveWorkspace"]>) =>
requireOrchestration().resolveWorkspace(...args),
wakeAgentIfPossible: (...args: Parameters<ReturnType<typeof createOrchestrationService>["wakeAgentIfPossible"]>) =>
requireOrchestration().wakeAgentIfPossible(...args),
dispatchRunToWorker: (...args: Parameters<ReturnType<typeof createOrchestrationService>["dispatchRunToWorker"]>) =>
requireOrchestration().dispatchRunToWorker(...args),
queueTaskRun: (...args: Parameters<ReturnType<typeof createOrchestrationService>["queueTaskRun"]>) =>
requireOrchestration().queueTaskRun(...args),
createTaskRunForDeferredDispatch: (
...args: Parameters<ReturnType<typeof createOrchestrationService>["createTaskRunForDeferredDispatch"]>
) => requireOrchestration().createTaskRunForDeferredDispatch(...args),
resumeQueuedRuns: (...args: Parameters<ReturnType<typeof createOrchestrationService>["resumeQueuedRuns"]>) =>
requireOrchestration().resumeQueuedRuns(...args),
startAgentHeartbeatRun: (
...args: Parameters<ReturnType<typeof createOrchestrationService>["startAgentHeartbeatRun"]>
) => requireOrchestration().startAgentHeartbeatRun(...args),
startAllSchedulers: (...args: Parameters<ReturnType<typeof createOrchestrationService>["startAllSchedulers"]>) =>
requireOrchestration().startAllSchedulers(...args),
stopAllSchedulers: (...args: Parameters<ReturnType<typeof createOrchestrationService>["stopAllSchedulers"]>) =>
requireOrchestration().stopAllSchedulers(...args),
};
}
@@ -1,79 +0,0 @@
import type { AppDatabase } from "../database";
import type { DesktopEvent } from "@shared/types";
import { createWorkerService } from "../worker-service";
import type { createBudgetEnforcementService } from "../budget-enforcement-service";
import type { createSessionManagementService } from "../session-management-service";
import type { PiRuntimeServices } from "../service-graph-pi-runtime";
import type { NotificationOptions } from "../notification-service";
import type { RuntimeMessageInjector } from "../agent-runtime";
import type { WorkerServiceDependencies } from "../worker/entry/types";
export interface WorkerServiceDeps {
db: AppDatabase;
logger: {
debug: (...args: unknown[]) => void;
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
workerPath: string;
isQuittingRef: { current: boolean };
publishDomainChanged: () => void;
emitEvent: (event: DesktopEvent) => void;
wakeAgentIfPossible: WorkerServiceDependencies["wakeAgentIfPossible"];
dispatchRunToWorker: WorkerServiceDependencies["dispatchRunToWorker"];
processRunCompletion: WorkerServiceDependencies["processRunCompletion"];
onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void;
onRunCompleted?: (runId: string, taskId: string | null, agentId: string, companyId: string, status: string, summary: string | null, errorMessage: string | null) => void;
notify: (options: NotificationOptions) => void;
budgetService: ReturnType<typeof createBudgetEnforcementService>;
sessionService: ReturnType<typeof createSessionManagementService>;
piRuntimeRef: { current: PiRuntimeServices | null };
/**
* Resolver for the late-constructed runtime message injector. The worker
* message handler needs it to route `steer-ack` child messages, but the
* injector is assembled after the worker service itself.
*/
getRuntimeMessageInjector: () => RuntimeMessageInjector | null;
}
/** Build the worker service with dispatch callbacks supplied by the service event hub. */
export function buildWorkerService(deps: WorkerServiceDeps) {
const {
db,
logger,
workerPath,
isQuittingRef,
publishDomainChanged,
emitEvent,
wakeAgentIfPossible,
dispatchRunToWorker,
processRunCompletion,
onRunStarted,
onRunCompleted,
notify,
budgetService,
sessionService,
piRuntimeRef,
getRuntimeMessageInjector,
} = deps;
return createWorkerService({
db,
logger,
workerPath,
isQuitting: () => isQuittingRef.current,
publishDomainChanged,
emitEvent,
wakeAgentIfPossible,
dispatchRunToWorker,
processRunCompletion,
onRunStarted,
onRunCompleted,
notify,
budgetService,
sessionService,
getRuntimeBridge: () => piRuntimeRef.current?.runtimeBridge ?? null,
getRuntimeMessageInjector,
});
}
@@ -1,75 +0,0 @@
import type { RunRecord } from "@shared/types";
import type { RunFinishedEvent } from "../automation/types";
import type { createOrchestrationService } from "../orchestration-service";
import type { AutomationDispatch } from "../orchestration/types";
type OrchestrationServiceEvents = Pick<
ReturnType<typeof createOrchestrationService>,
"wakeAgentIfPossible" | "dispatchRunToWorker"
>;
type AutomationServiceEvents = {
dispatchAutomation: (trigger: string, ctx: AutomationDispatch) => void;
processRunCompletion: (run: RunRecord, event: RunFinishedEvent) => void;
};
export function createServiceEventHub() {
let orchestration: OrchestrationServiceEvents | null = null;
let automation: AutomationServiceEvents | null = null;
function requireOrchestration(): OrchestrationServiceEvents {
if (!orchestration) {
throw new Error("Service event hub orchestration handlers are not registered.");
}
return orchestration;
}
function requireAutomation(): AutomationServiceEvents {
if (!automation) {
throw new Error("Service event hub automation handlers are not registered.");
}
return automation;
}
return {
registerOrchestration(service: OrchestrationServiceEvents): void {
if (orchestration) {
throw new Error("Service event hub orchestration handlers are already registered.");
}
orchestration = service;
},
registerAutomation(service: AutomationServiceEvents): void {
if (automation) {
throw new Error("Service event hub automation handlers are already registered.");
}
automation = service;
},
hasOrchestrationHandlers(): boolean {
return Boolean(orchestration);
},
wakeAgentIfPossible: (
agentId: string | null | undefined,
companyId: string,
trigger: string,
) => requireOrchestration().wakeAgentIfPossible(agentId, companyId, trigger),
dispatchRunToWorker: (
params: Parameters<OrchestrationServiceEvents["dispatchRunToWorker"]>[0],
) => {
requireOrchestration().dispatchRunToWorker(params);
},
dispatchAutomation: (trigger: string, ctx: AutomationDispatch) => {
requireAutomation().dispatchAutomation(trigger, ctx);
},
processRunCompletion: (run: RunRecord, event: RunFinishedEvent) => {
requireAutomation().processRunCompletion(run, event);
},
assertReady(): void {
requireOrchestration();
requireAutomation();
},
clear(): void {
orchestration = null;
automation = null;
},
};
}
@@ -0,0 +1,65 @@
import { describe, expect, it, vi } from "vitest";
vi.mock("electron", () => ({
app: { isPackaged: false },
utilityProcess: { fork: vi.fn() },
powerMonitor: {
on: vi.fn(),
removeListener: vi.fn(),
},
}));
import { createRuntimeOrchestrationService } from "../../src/main/service-graph/create-runtime-orchestration-service";
function createService() {
return createRuntimeOrchestrationService({
db: {} as never,
logger: {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
},
workerPath: "/tmp/worker.js",
isQuittingRef: { current: false },
publishDomainChanged: vi.fn(),
emitEvent: vi.fn(),
notify: vi.fn(),
createRunToken: vi.fn(() => "token"),
getApiPort: vi.fn(() => 0),
skillService: {
listRuntimeSkillEntries: vi.fn(() => []),
} as never,
budgetService: {
evaluateBudgetGate: vi.fn(() => ({ allowed: true, blockedBy: [] })),
} as never,
sessionService: {
getHandoffForAgent: vi.fn(() => null),
evaluateRotation: vi.fn(() => ({ shouldRotate: false, reason: null })),
rotateSession: vi.fn(),
} as never,
piRuntimeRef: { current: null },
getRuntimeMessageInjector: vi.fn(() => null),
});
}
describe("runtime orchestration service", () => {
it("requires automation handlers to be bound exactly once", () => {
const service = createService();
const handlers = {
dispatchAutomation: vi.fn(),
processRunCompletion: vi.fn(),
};
expect(() => service.assertReady()).toThrow(
"Runtime orchestration automation handlers are not bound.",
);
service.bindAutomation(handlers);
expect(() => service.assertReady()).not.toThrow();
expect(() => service.bindAutomation(handlers)).toThrow(
"Runtime orchestration automation handlers are already bound.",
);
});
});
-88
View File
@@ -1,88 +0,0 @@
import { describe, expect, it } from "vitest";
import type { RunRecord } from "@shared/types";
import { createServiceEventHub } from "../../src/main/service-graph/service-event-hub";
type Hub = ReturnType<typeof createServiceEventHub>;
type DispatchRunPayload = Parameters<Hub["dispatchRunToWorker"]>[0];
describe("service event hub", () => {
it("fails fast before required service handlers are registered", () => {
const hub = createServiceEventHub();
expect(() => hub.wakeAgentIfPossible("agent-1", "company-1", "heartbeat")).toThrow(
"Service event hub orchestration handlers are not registered.",
);
expect(() => hub.dispatchAutomation("task.created", { companyId: "company-1" })).toThrow(
"Service event hub automation handlers are not registered.",
);
expect(() => hub.assertReady()).toThrow(
"Service event hub orchestration handlers are not registered.",
);
});
it("routes orchestration and automation events through registered handlers", () => {
const hub = createServiceEventHub();
const calls: string[] = [];
const runPayload = { runId: "run-1" } as DispatchRunPayload;
const run = { id: "run-1" } as RunRecord;
const runFinished = {
status: "succeeded",
summary: "done",
errorMessage: null,
} as const;
hub.registerOrchestration({
wakeAgentIfPossible: (agentId, companyId, trigger) => {
calls.push(`wake:${agentId}:${companyId}:${trigger}`);
return "run-1";
},
dispatchRunToWorker: (params) => {
expect(params).toBe(runPayload);
calls.push("dispatch-run");
},
});
hub.registerAutomation({
dispatchAutomation: (trigger, ctx) => {
calls.push(`automation:${trigger}:${ctx.companyId}`);
},
processRunCompletion: (completedRun, event) => {
expect(completedRun).toBe(run);
expect(event).toBe(runFinished);
calls.push("run-completion");
},
});
expect(hub.hasOrchestrationHandlers()).toBe(true);
expect(hub.wakeAgentIfPossible("agent-1", "company-1", "heartbeat")).toBe("run-1");
hub.dispatchRunToWorker(runPayload);
hub.dispatchAutomation("task.created", { companyId: "company-1" });
hub.processRunCompletion(run, runFinished);
hub.assertReady();
expect(calls).toEqual([
"wake:agent-1:company-1:heartbeat",
"dispatch-run",
"automation:task.created:company-1",
"run-completion",
]);
});
it("rejects duplicate registrations and clears handlers during shutdown", () => {
const hub = createServiceEventHub();
const orchestrationHandlers = {
wakeAgentIfPossible: () => null,
dispatchRunToWorker: () => undefined,
};
hub.registerOrchestration(orchestrationHandlers);
expect(() => hub.registerOrchestration(orchestrationHandlers)).toThrow(
"Service event hub orchestration handlers are already registered.",
);
hub.clear();
expect(hub.hasOrchestrationHandlers()).toBe(false);
expect(() => hub.wakeAgentIfPossible(null, "company-1", "heartbeat")).toThrow(
"Service event hub orchestration handlers are not registered.",
);
});
});