From e540cbb2a6b58f4aaa5879cc9e29683ec720b505 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Wed, 22 Apr 2026 16:55:20 -0400 Subject: [PATCH] refactor(service-graph): replace late bindings with event hub --- src/main/service-graph.ts | 59 +++++-------- .../create-background-services.ts | 9 +- .../service-graph/create-desktop-shell.ts | 14 +-- src/main/service-graph/create-lifecycle.ts | 8 -- .../create-orchestration-service.ts | 20 +---- .../service-graph/create-worker-service.ts | 35 ++------ src/main/service-graph/late-binding.ts | 37 -------- src/main/service-graph/service-event-hub.ts | 75 ++++++++++++++++ tests/unit/service-event-hub.test.ts | 88 +++++++++++++++++++ 9 files changed, 205 insertions(+), 140 deletions(-) delete mode 100644 src/main/service-graph/late-binding.ts create mode 100644 src/main/service-graph/service-event-hub.ts create mode 100644 tests/unit/service-event-hub.test.ts diff --git a/src/main/service-graph.ts b/src/main/service-graph.ts index 8fc70429..d169b8c0 100644 --- a/src/main/service-graph.ts +++ b/src/main/service-graph.ts @@ -13,14 +13,14 @@ 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 type { createUpdaterService } from "./updater-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 { createBackgroundServices } from "./service-graph/create-background-services"; import { createLifecycle } from "./service-graph/create-lifecycle"; -import { LateBinding } from "./service-graph/late-binding"; +import { createServiceEventHub } from "./service-graph/service-event-hub"; interface ServiceGraphDependencies { db: AppDatabase; @@ -58,9 +58,8 @@ interface ServiceGraphDependencies { /** * Thin orchestrator: assembles services in a deterministic topological order - * and exposes them via read-only getters. All wiring is performed via - * explicit dependency passing; late-bound cycles are handled by - * `LateBinding` slots that throw if accessed before assignment. + * and exposes them via read-only getters. Cross-service callbacks are routed + * through a typed event hub so construction order stays explicit. */ export function createServiceGraph(deps: ServiceGraphDependencies) { const { @@ -101,16 +100,7 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { let leaf: ReturnType | null = null; let runtimeMessageInjector: RuntimeMessageInjector | null = null; const piRuntimeRef: { current: PiRuntimeServices | null } = { current: null }; - - // Late-bound slots resolve circular construction: worker needs orchestration - // (for dispatchRunToWorker), orchestration needs worker (for enqueueWorkerRun), - // orchestration needs automation (for dispatchAutomation), automation/orchestration - // need browser-action (for dispatchBrowserAction). Each is filled inside - // init() before the corresponding getter closure is ever invoked. - const orchestrationRef = new LateBinding>("orchestrationService"); - const automationRef = new LateBinding>("automationService"); - const browserActionRef = new LateBinding>("browserActionService"); - const updaterRef = new LateBinding>("updaterService"); + const serviceEvents = createServiceEventHub(); function requireLeaf() { if (!leaf) throw new Error("Leaf services are not initialized."); @@ -121,6 +111,8 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { nativeTheme.themeSource = db.snap.listSnapshot().theme; leaf = createLeafServices({ db, logger, notify, publishDomainChanged }); + updaterService = createUpdaterService({ db, logger, notify }); + updaterService.attachListeners(); desktopShellService = createDesktopShell({ db, @@ -134,7 +126,11 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { emitEvent, notify, registerHandle, - updaterRef, + checkForUpdates: async () => { + const updater = updaterService; + if (!updater) throw new Error("Updater service is not initialized."); + await updater.checkForUpdates({ manual: true }); + }, logger, }); @@ -163,8 +159,9 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { isQuittingRef, publishDomainChanged, emitEvent, - orchestrationRef, - automationRef, + wakeAgentIfPossible: serviceEvents.wakeAgentIfPossible, + dispatchRunToWorker: serviceEvents.dispatchRunToWorker, + processRunCompletion: serviceEvents.processRunCompletion, onRunStarted, onRunCompleted, notify, @@ -183,9 +180,8 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { worker.postMessage(message); }, getWakeAgent: () => { - if (!orchestrationRef.isAssigned()) return null; - const orch = orchestrationRef.get(); - return (agentId, companyId, trigger) => orch.wakeAgentIfPossible(agentId, companyId, trigger); + if (!serviceEvents.hasOrchestrationHandlers()) return null; + return (agentId, companyId, trigger) => serviceEvents.wakeAgentIfPossible(agentId, companyId, trigger); }, logger, }); @@ -203,13 +199,10 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { budgetService: leaf.budgetEnforcementService, sessionService: leaf.sessionManagementService, workerService, - automationRef, - browserActionRef, + dispatchAutomation: serviceEvents.dispatchAutomation, checkpointEngine, }); - orchestrationRef.set(orchestrationService); - - workerService.createWorker(); + serviceEvents.registerOrchestration(orchestrationService); const background = createBackgroundServices({ db, @@ -226,10 +219,9 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { automationService = background.automationService; browserActionService = background.browserActionService; onboardingService = background.onboardingService; - updaterService = background.updaterService; - automationRef.set(automationService); - browserActionRef.set(browserActionService); - updaterRef.set(updaterService); + serviceEvents.registerAutomation(automationService); + serviceEvents.assertReady(); + workerService.createWorker(); appLifecycleService = createLifecycle({ db, @@ -254,13 +246,8 @@ export function createServiceGraph(deps: ServiceGraphDependencies) { appLifecycleService = null; runtimeMessageInjector = null; leaf = null; + serviceEvents.clear(); }, - lateBindings: [ - orchestrationRef as LateBinding, - automationRef as LateBinding, - browserActionRef as LateBinding, - updaterRef as LateBinding, - ], }); } diff --git a/src/main/service-graph/create-background-services.ts b/src/main/service-graph/create-background-services.ts index 773ef603..53303471 100644 --- a/src/main/service-graph/create-background-services.ts +++ b/src/main/service-graph/create-background-services.ts @@ -4,7 +4,6 @@ import type { DesktopEvent } from "@shared/types"; import { createAutomationService } from "../automation-service"; import { createBrowserActionService } from "../browser-action-service"; import { createOnboardingService } from "../onboarding-service"; -import { createUpdaterService } from "../updater-service"; import type { createOrchestrationService } from "../orchestration-service"; import type { createSkillService } from "../skill-service"; import type { NotificationOptions } from "../notification-service"; @@ -31,12 +30,11 @@ export interface BackgroundServices { automationService: ReturnType; browserActionService: ReturnType; onboardingService: ReturnType; - updaterService: ReturnType; } /** * Build background services that run off the main dispatch loop: automation - * triggers, browser-action dispatch, onboarding hire flow, and the updater. + * triggers, browser-action dispatch, and onboarding hire flow. * Orchestration is fully assembled at this point, so all dependencies are * concrete references rather than lazy slots. */ @@ -88,8 +86,5 @@ export function createBackgroundServices({ }, }); - const updaterService = createUpdaterService({ db, logger, notify }); - updaterService.attachListeners(); - - return { automationService, browserActionService, onboardingService, updaterService }; + return { automationService, browserActionService, onboardingService }; } diff --git a/src/main/service-graph/create-desktop-shell.ts b/src/main/service-graph/create-desktop-shell.ts index 699a96ee..bf60b046 100644 --- a/src/main/service-graph/create-desktop-shell.ts +++ b/src/main/service-graph/create-desktop-shell.ts @@ -4,8 +4,6 @@ import type { DesktopEvent } from "@shared/types"; import type { DesktopResult } from "@shared/contracts"; import { createDesktopShellService } from "../desktop-shell-service"; import type { NotificationOptions } from "../notification-service"; -import type { LateBinding } from "./late-binding"; -import type { createUpdaterService } from "../updater-service"; export interface DesktopShellDeps { db: AppDatabase; @@ -22,15 +20,11 @@ export interface DesktopShellDeps { channel: string, handler: (event: IpcMainInvokeEvent, ...args: unknown[]) => Promise> | DesktopResult, ) => void; - updaterRef: LateBinding>; + checkForUpdates: () => Promise; logger: { warn: (...args: unknown[]) => void }; } -/** - * Build the desktop shell service. `updaterRef` is late-bound because the - * updater is constructed after the shell; the menu's "Check for updates" - * handler resolves it at call time. - */ +/** Build the desktop shell service with concrete updater and window callbacks. */ export function createDesktopShell({ db, preloadPath, @@ -43,7 +37,7 @@ export function createDesktopShell({ emitEvent, notify, registerHandle, - updaterRef, + checkForUpdates, logger, }: DesktopShellDeps) { const desktopShellService = createDesktopShellService({ @@ -59,7 +53,7 @@ export function createDesktopShell({ isQuittingRef.current = true; }, onCheckForUpdates: async () => { - await updaterRef.get().checkForUpdates(); + await checkForUpdates(); }, onBackupProfile: async () => { const target = await import("electron").then(({ dialog }) => dialog.showOpenDialog({ diff --git a/src/main/service-graph/create-lifecycle.ts b/src/main/service-graph/create-lifecycle.ts index 930936df..c44681d5 100644 --- a/src/main/service-graph/create-lifecycle.ts +++ b/src/main/service-graph/create-lifecycle.ts @@ -8,7 +8,6 @@ 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 { LateBinding } from "./late-binding"; /** Overall budget for awaited shutdown drain (worker exit + API close). */ const SHUTDOWN_BUDGET_MS = 5_000; @@ -35,11 +34,6 @@ export interface LifecycleDeps { * old onBeforeQuit. */ clearServiceRefs: () => void; - /** - * Late bindings owned by the graph; cleared after drain so any stray - * post-quit access fails loudly rather than returning stale data. - */ - lateBindings: Array>; } /** @@ -62,7 +56,6 @@ export function createLifecycle(deps: LifecycleDeps) { workerService, desktopShellService, clearServiceRefs, - lateBindings, } = deps; let drainPromise: Promise | null = null; @@ -113,7 +106,6 @@ export function createLifecycle(deps: LifecycleDeps) { try { db.close(); } catch (err) { logger.warn("[lifecycle] db.close failed:", err); } clearServiceRefs(); - for (const binding of lateBindings) binding.clear(); if (outcome === "timeout") { const stuck = listActiveRunIds(); diff --git a/src/main/service-graph/create-orchestration-service.ts b/src/main/service-graph/create-orchestration-service.ts index 7438a980..d64526f7 100644 --- a/src/main/service-graph/create-orchestration-service.ts +++ b/src/main/service-graph/create-orchestration-service.ts @@ -5,10 +5,8 @@ import type { createBudgetEnforcementService } from "../budget-enforcement-servi import type { createSessionManagementService } from "../session-management-service"; import type { createSkillService } from "../skill-service"; import type { createWorkerService } from "../worker-service"; -import type { createAutomationService, AutomationEventContext } from "../automation-service"; -import type { createBrowserActionService } from "../browser-action-service"; import type { NotificationOptions } from "../notification-service"; -import type { LateBinding } from "./late-binding"; +import type { OrchestrationDependencies } from "../orchestration/types"; export interface OrchestrationServiceDeps { db: AppDatabase; @@ -28,17 +26,11 @@ export interface OrchestrationServiceDeps { budgetService: ReturnType; sessionService: ReturnType; workerService: ReturnType; - automationRef: LateBinding>; - browserActionRef: LateBinding>; + dispatchAutomation: OrchestrationDependencies["dispatchAutomation"]; checkpointEngine?: { shouldPauseHeartbeat: (companyId: string, agentId: string) => boolean }; } -/** - * Build the orchestration service with resolved dependencies. Late bindings - * for automation and browser-action are required because orchestration's - * dispatchAutomation closure is invoked long after assembly — the slots are - * filled in the same init() pass, before any scheduler tick. - */ +/** Build the orchestration service with concrete worker wiring and event dispatch. */ export function buildOrchestrationService({ db, logger, @@ -52,13 +44,9 @@ export function buildOrchestrationService({ budgetService, sessionService, workerService, - automationRef, + dispatchAutomation, checkpointEngine, }: OrchestrationServiceDeps) { - const dispatchAutomation = (trigger: string, ctx: AutomationEventContext) => { - automationRef.get().dispatchAutomation(trigger, ctx); - }; - return createOrchestrationService({ db, logger, diff --git a/src/main/service-graph/create-worker-service.ts b/src/main/service-graph/create-worker-service.ts index 8ff9d48b..c58c5648 100644 --- a/src/main/service-graph/create-worker-service.ts +++ b/src/main/service-graph/create-worker-service.ts @@ -1,14 +1,12 @@ import type { AppDatabase } from "../database"; -import type { DesktopEvent, RunRecord } from "@shared/types"; +import type { DesktopEvent } from "@shared/types"; import { createWorkerService } from "../worker-service"; -import type { createOrchestrationService } from "../orchestration-service"; -import type { createAutomationService } from "../automation-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 { LateBinding } from "./late-binding"; +import type { WorkerServiceDependencies } from "../worker/entry/types"; export interface WorkerServiceDeps { db: AppDatabase; @@ -22,8 +20,9 @@ export interface WorkerServiceDeps { isQuittingRef: { current: boolean }; publishDomainChanged: () => void; emitEvent: (event: DesktopEvent) => void; - orchestrationRef: LateBinding>; - automationRef: LateBinding>; + 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; @@ -38,11 +37,7 @@ export interface WorkerServiceDeps { getRuntimeMessageInjector: () => RuntimeMessageInjector | null; } -/** - * Build the worker service. Orchestration and automation are late-bound via - * LateBinding slots because they are constructed after the worker but close - * over the worker's dispatch functions. - */ +/** Build the worker service with dispatch callbacks supplied by the service event hub. */ export function buildWorkerService(deps: WorkerServiceDeps) { const { db, @@ -51,8 +46,9 @@ export function buildWorkerService(deps: WorkerServiceDeps) { isQuittingRef, publishDomainChanged, emitEvent, - orchestrationRef, - automationRef, + wakeAgentIfPossible, + dispatchRunToWorker, + processRunCompletion, onRunStarted, onRunCompleted, notify, @@ -62,19 +58,6 @@ export function buildWorkerService(deps: WorkerServiceDeps) { getRuntimeMessageInjector, } = deps; - const wakeAgentIfPossible = (agentId: string | null | undefined, companyId: string, trigger: string) => - orchestrationRef.get().wakeAgentIfPossible(agentId, companyId, trigger); - - const dispatchRunToWorker: Parameters[0]["dispatchRunToWorker"] = (params) => - orchestrationRef.get().dispatchRunToWorker(params); - - const processRunCompletion = ( - run: RunRecord, - event: Parameters["processRunCompletion"]>[1], - ) => { - automationRef.get().processRunCompletion(run, event); - }; - return createWorkerService({ db, logger, diff --git a/src/main/service-graph/late-binding.ts b/src/main/service-graph/late-binding.ts deleted file mode 100644 index 4151ed36..00000000 --- a/src/main/service-graph/late-binding.ts +++ /dev/null @@ -1,37 +0,0 @@ -/** - * One-time settable slot for services that participate in a cycle during - * assembly. `set()` may be called exactly once; `get()` throws until the - * slot has been set. Eliminates the lazy-null `let` pattern that service-graph - * previously used to wire `workerService.dispatchRunToWorker` against - * `orchestrationService` before the latter exists. - */ -export class LateBinding { - private value: T | null = null; - private assigned = false; - - constructor(private readonly label: string) {} - - set(value: T): void { - if (this.assigned) { - throw new Error(`LateBinding<${this.label}> is already assigned.`); - } - this.value = value; - this.assigned = true; - } - - get(): T { - if (!this.assigned || this.value === null) { - throw new Error(`LateBinding<${this.label}> accessed before assignment.`); - } - return this.value; - } - - isAssigned(): boolean { - return this.assigned; - } - - clear(): void { - this.value = null; - this.assigned = false; - } -} diff --git a/src/main/service-graph/service-event-hub.ts b/src/main/service-graph/service-event-hub.ts new file mode 100644 index 00000000..35587181 --- /dev/null +++ b/src/main/service-graph/service-event-hub.ts @@ -0,0 +1,75 @@ +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/service-event-hub.test.ts b/tests/unit/service-event-hub.test.ts new file mode 100644 index 00000000..0f2f3e6b --- /dev/null +++ b/tests/unit/service-event-hub.test.ts @@ -0,0 +1,88 @@ +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.", + ); + }); +});