diff --git a/src/main/service-graph.ts b/src/main/service-graph.ts index d591e4d2..4b74d6e2 100644 --- a/src/main/service-graph.ts +++ b/src/main/service-graph.ts @@ -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 | null = null; let connectorHealthService: ReturnType | null = null; - let workerService: ReturnType | null = null; - let orchestrationService: ReturnType | null = null; + let runtimeOrchestrationService: ReturnType | null = null; let automationService: ReturnType | null = null; let browserActionService: ReturnType | null = null; let onboardingService: ReturnType | null = null; @@ -100,7 +97,6 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { let leaf: ReturnType | 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, diff --git a/src/main/service-graph/create-background-services.ts b/src/main/service-graph/create-background-services.ts index 86d23682..a0e866b8 100644 --- a/src/main/service-graph/create-background-services.ts +++ b/src/main/service-graph/create-background-services.ts @@ -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; - orchestrationService: ReturnType; + orchestrationService: ReturnType; seedDefaultAutomationRules: (companyId: string) => void; } diff --git a/src/main/service-graph/create-lifecycle.ts b/src/main/service-graph/create-lifecycle.ts index 703d9b43..7a9c6545 100644 --- a/src/main/service-graph/create-lifecycle.ts +++ b/src/main/service-graph/create-lifecycle.ts @@ -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; + runtimeOrchestrationService: ReturnType; automationService: ReturnType; - workerService: ReturnType; desktopShellService: ReturnType; /** * 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 { - 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, ]); diff --git a/src/main/service-graph/create-orchestration-service.ts b/src/main/service-graph/create-orchestration-service.ts deleted file mode 100644 index 21083026..00000000 --- a/src/main/service-graph/create-orchestration-service.ts +++ /dev/null @@ -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; - budgetService: ReturnType; - sessionService: ReturnType; - workerService: ReturnType; - 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 }; - }, - }); -} diff --git a/src/main/service-graph/create-runtime-orchestration-service.ts b/src/main/service-graph/create-runtime-orchestration-service.ts new file mode 100644 index 00000000..2ae26bfd --- /dev/null +++ b/src/main/service-graph/create-runtime-orchestration-service.ts @@ -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; + budgetService: ReturnType; + sessionService: ReturnType; + 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 | null = null; + + function requireAutomation(): AutomationHandlers { + if (!automationHandlers) { + throw new Error("Runtime orchestration automation handlers are not bound."); + } + return automationHandlers; + } + + function requireOrchestration(): ReturnType { + 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["resolveWorkspace"]>) => + requireOrchestration().resolveWorkspace(...args), + wakeAgentIfPossible: (...args: Parameters["wakeAgentIfPossible"]>) => + requireOrchestration().wakeAgentIfPossible(...args), + dispatchRunToWorker: (...args: Parameters["dispatchRunToWorker"]>) => + requireOrchestration().dispatchRunToWorker(...args), + queueTaskRun: (...args: Parameters["queueTaskRun"]>) => + requireOrchestration().queueTaskRun(...args), + createTaskRunForDeferredDispatch: ( + ...args: Parameters["createTaskRunForDeferredDispatch"]> + ) => requireOrchestration().createTaskRunForDeferredDispatch(...args), + resumeQueuedRuns: (...args: Parameters["resumeQueuedRuns"]>) => + requireOrchestration().resumeQueuedRuns(...args), + startAgentHeartbeatRun: ( + ...args: Parameters["startAgentHeartbeatRun"]> + ) => requireOrchestration().startAgentHeartbeatRun(...args), + startAllSchedulers: (...args: Parameters["startAllSchedulers"]>) => + requireOrchestration().startAllSchedulers(...args), + stopAllSchedulers: (...args: Parameters["stopAllSchedulers"]>) => + requireOrchestration().stopAllSchedulers(...args), + }; +} diff --git a/src/main/service-graph/create-worker-service.ts b/src/main/service-graph/create-worker-service.ts deleted file mode 100644 index c58c5648..00000000 --- a/src/main/service-graph/create-worker-service.ts +++ /dev/null @@ -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; - sessionService: ReturnType; - 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, - }); -} diff --git a/src/main/service-graph/service-event-hub.ts b/src/main/service-graph/service-event-hub.ts deleted file mode 100644 index 35587181..00000000 --- a/src/main/service-graph/service-event-hub.ts +++ /dev/null @@ -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, - "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[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; - }, - }; -} diff --git a/tests/unit/runtime-orchestration-service.test.ts b/tests/unit/runtime-orchestration-service.test.ts new file mode 100644 index 00000000..17c6624d --- /dev/null +++ b/tests/unit/runtime-orchestration-service.test.ts @@ -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.", + ); + }); +}); diff --git a/tests/unit/service-event-hub.test.ts b/tests/unit/service-event-hub.test.ts deleted file mode 100644 index 0f2f3e6b..00000000 --- a/tests/unit/service-event-hub.test.ts +++ /dev/null @@ -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; -type DispatchRunPayload = Parameters[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.", - ); - }); -});