refactor(service-graph): replace late bindings with event hub

This commit is contained in:
ZacharyZhang-NY committed 2026-04-22 16:55:20 -04:00
1 parent 9e712f2746
commit e540cbb2a6
9 files changed
+205 -140

No files matched your search

+23 -36
View File
@@ -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<typeof createLeafServices> | 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<ReturnType<typeof createOrchestrationService>>("orchestrationService");
const automationRef = new LateBinding<ReturnType<typeof createAutomationService>>("automationService");
const browserActionRef = new LateBinding<ReturnType<typeof createBrowserActionService>>("browserActionService");
const updaterRef = new LateBinding<ReturnType<typeof createUpdaterService>>("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<unknown>,
automationRef as LateBinding<unknown>,
browserActionRef as LateBinding<unknown>,
updaterRef as LateBinding<unknown>,
],
});
}
@@ -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<typeof createAutomationService>;
browserActionService: ReturnType<typeof createBrowserActionService>;
onboardingService: ReturnType<typeof createOnboardingService>;
updaterService: ReturnType<typeof createUpdaterService>;
}
/**
* 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 };
}
+4 -10
View File
@@ -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<unknown>> | DesktopResult<unknown>,
) => void;
updaterRef: LateBinding<ReturnType<typeof createUpdaterService>>;
checkForUpdates: () => Promise<void>;
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({
@@ -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<LateBinding<unknown>>;
}
/**
@@ -62,7 +56,6 @@ export function createLifecycle(deps: LifecycleDeps) {
workerService,
desktopShellService,
clearServiceRefs,
lateBindings,
} = deps;
let drainPromise: Promise<number> | 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();
@@ -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<typeof createBudgetEnforcementService>;
sessionService: ReturnType<typeof createSessionManagementService>;
workerService: ReturnType<typeof createWorkerService>;
automationRef: LateBinding<ReturnType<typeof createAutomationService>>;
browserActionRef: LateBinding<ReturnType<typeof createBrowserActionService>>;
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,
@@ -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<ReturnType<typeof createOrchestrationService>>;
automationRef: LateBinding<ReturnType<typeof createAutomationService>>;
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<typeof createWorkerService>[0]["dispatchRunToWorker"] = (params) =>
orchestrationRef.get().dispatchRunToWorker(params);
const processRunCompletion = (
run: RunRecord,
event: Parameters<ReturnType<typeof createAutomationService>["processRunCompletion"]>[1],
) => {
automationRef.get().processRunCompletion(run, event);
};
return createWorkerService({
db,
logger,
-37
View File
@@ -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<T> {
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;
}
}
@@ -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<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;
},
};
}
+88
View File
@@ -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<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.",
);
});
});