refactor(bootstrap): extract bootstrap-application into orchestrator + modules

Splits the 391-line bootstrapApplication god function into a thin
orchestrator plus 7 focused modules under src/main/bootstrap/:
prepare-logger, prepare-database, prepare-services (service graph +
ipc wiring), wire-api-server (14-callback dictionary), recover-
scheduled-posts, run-startup-health-checks, recover-interrupted-tasks
(crash notification + clean-shutdown task resumption). Original
startup order, before-quit cleanup, and error handling are preserved;
the entry file drops to 153 lines.
This commit is contained in:
ZacharyZhang-NY committed 2026-04-16 23:34:13 -04:00
1 parent dd007760b8
commit a2a90f00e6
8 files changed
+487 -308

No files matched your search

+14
View File
@@ -0,0 +1,14 @@
import type { AppDatabase } from "../database";
export interface PrepareDatabaseDeps {
profileDir: string;
appDatabaseFactory: (profileDir: string) => AppDatabase;
initSigningKey: (profileDir: string) => void;
}
export function prepareDatabase(deps: PrepareDatabaseDeps): AppDatabase {
const db = deps.appDatabaseFactory(deps.profileDir);
db.init();
deps.initSigningKey(db.profileDir);
return db;
}
+16
View File
@@ -0,0 +1,16 @@
import { app } from "electron";
import log from "electron-log/main.js";
export function prepareLogger() {
log.initialize();
log.transports.console.format = "[{h}:{i}:{s}] [{level}] {text}";
log.transports.file.format = "[{y}-{m}-{d} {h}:{i}:{s}.{ms}] [{level}] {text}";
if (app.isPackaged) {
log.transports.console.level = "warn";
log.transports.file.level = "info";
} else {
log.transports.console.level = "info";
log.transports.file.level = "debug";
}
return log;
}
+134
View File
@@ -0,0 +1,134 @@
import type { AgentApiServer } from "../api-server";
import type { AppDatabase } from "../database";
import type { BrowserManager } from "../browser-manager";
import { createServiceGraph } from "../service-graph";
import { createIpcSupport } from "../ipc-support";
import { createNotificationService } from "../notification-service";
import { createThreadIntegration } from "../thread-integration";
import { createCheckpointEngine } from "../checkpoint-engine";
import { createRunHousekeeping } from "../run-housekeeping";
import type { DesktopEvent } from "@shared/types";
import type { NotificationOptions } from "../notification-service";
export interface PrepareServicesDeps {
db: AppDatabase;
apiServerRef: { current: AgentApiServer | null };
browserManagerRef: { current: BrowserManager | null };
isQuittingRef: { current: boolean };
preloadPath: string;
rendererDist: string;
iconPath: string;
isDev: boolean;
devServerUrl?: string;
devServerOrigin: string | null;
windowStatePath: string;
workerPath: string;
logger: {
debug: (...args: unknown[]) => void;
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
createRunToken: (agentId: string, companyId: string, runId: string) => string;
seedDefaultAutomationRules: (companyId: string) => void;
}
export interface ServiceGraphHandles {
serviceGraph: ReturnType<typeof createServiceGraph>;
ipcSupport: ReturnType<typeof createIpcSupport>;
threadIntegration: ReturnType<typeof createThreadIntegration>;
notificationService: ReturnType<typeof createNotificationService>;
checkpointEngine: ReturnType<typeof createCheckpointEngine>;
notify: (options: NotificationOptions) => void;
emitEvent: (event: DesktopEvent) => void;
publishDomainChanged: () => void;
}
export function prepareServices(deps: PrepareServicesDeps): ServiceGraphHandles {
let ipcSupport: ReturnType<typeof createIpcSupport> | null = null;
let serviceGraph: ReturnType<typeof createServiceGraph> | null = null;
const notificationService = createNotificationService({
getMainWindow: () => serviceGraph?.getDesktopShellService()?.getMainWindow() ?? null,
emitEvent: (event: DesktopEvent) => ipcSupport?.emitEvent(event),
logger: deps.logger,
});
const notify = (options: NotificationOptions) => {
if (!ipcSupport) {
throw new Error("IPC support is not initialized.");
}
ipcSupport.notify(options);
};
const emitEvent = (event: DesktopEvent) => ipcSupport?.emitEvent(event);
const publishDomainChanged = () => ipcSupport?.publishDomainChanged();
const threadIntegration = createThreadIntegration({
db: deps.db,
logger: deps.logger,
emitEvent,
publishDomainChanged,
notify: (options) => notify({ ...options, urgency: "informational" }),
});
const checkpointEngine = createCheckpointEngine({
db: deps.db,
logger: deps.logger,
threadIntegration,
publishDomainChanged,
notify,
});
const runHousekeeping = createRunHousekeeping({
db: deps.db,
logger: deps.logger,
threadIntegration,
});
serviceGraph = createServiceGraph({
db: deps.db,
apiServerRef: deps.apiServerRef,
browserManagerRef: deps.browserManagerRef,
isQuittingRef: deps.isQuittingRef,
preloadPath: deps.preloadPath,
rendererDist: deps.rendererDist,
iconPath: deps.iconPath,
isDev: deps.isDev,
devServerUrl: deps.devServerUrl,
windowStatePath: deps.windowStatePath,
workerPath: deps.workerPath,
logger: deps.logger,
emitEvent,
publishDomainChanged,
notify,
createRunToken: deps.createRunToken,
seedDefaultAutomationRules: deps.seedDefaultAutomationRules,
onRunStarted: (runId, taskId, agentId, companyId) =>
threadIntegration.onRunStarted(runId, taskId, agentId, companyId),
onRunCompleted: (runId, taskId, agentId, companyId, status, summary, errorMessage) =>
runHousekeeping.onRunCompleted(runId, taskId, agentId, companyId, status, summary, errorMessage),
checkpointEngine,
});
ipcSupport = createIpcSupport({
isDev: deps.isDev,
devServerOrigin: deps.devServerOrigin,
logger: deps.logger,
getMainWindow: () => serviceGraph?.getDesktopShellService()?.getMainWindow() ?? null,
getDb: () => ({
hasCompany: (companyId: string) => deps.db.hasCompany(companyId),
belongsToCompany: (table: string, id: string, companyId: string) =>
deps.db.belongsToCompany(table as Parameters<typeof deps.db.belongsToCompany>[0], id, companyId),
}),
notificationService,
});
serviceGraph.init();
return {
serviceGraph,
ipcSupport,
threadIntegration,
notificationService,
checkpointEngine,
notify,
emitEvent,
publishDomainChanged,
};
}
@@ -0,0 +1,71 @@
import type { AppDatabase } from "../database";
import type { DesktopEvent } from "@shared/types";
import type { NotificationOptions } from "../notification-service";
interface Logger {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
}
export interface RecoverInterruptedTasksDeps {
db: AppDatabase;
logger: Logger;
emitEvent: (event: DesktopEvent) => void;
notify: (options: NotificationOptions) => void;
publishDomainChanged: () => void;
wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
}
/** Emit the session-recovery notification for runs interrupted by crash. */
export function emitCrashRecoveryNotification(deps: RecoverInterruptedTasksDeps): void {
const interruptedCount = deps.db.lastInterruptedCount;
if (interruptedCount <= 0) return;
deps.logger.warn(`[bootstrap] Detected ${interruptedCount} interrupted runs from previous session. Sending recovery notification.`);
const interruptedRuns = deps.db.getInterruptedRunsForRecovery();
if (interruptedRuns.length === 0) return;
deps.emitEvent({ type: "recovery-needed", interruptedRuns });
deps.notify({
title: "Session Recovered",
body: `${interruptedRuns.length} interrupted task(s) detected from previous session`,
urgency: "informational",
navigation: { section: "overview" },
});
}
/**
* Resume tasks interrupted by a clean shutdown. worker-service marks running/
* queued runs as `interrupted` on destroy but does not reset the tasks; without
* this recovery the assigned agents never re-wake to continue them.
*/
export function recoverShutdownInterruptedTasks(deps: RecoverInterruptedTasksDeps): void {
try {
const snapshot = deps.db.listSnapshot();
const shutdownRuns = snapshot.runs.filter(
(r) => r.status === "interrupted" &&
(r.summary === "Application shut down." || r.summary === "Worker process crashed."),
);
if (shutdownRuns.length === 0) return;
const resumedAgents = new Set<string>();
for (const run of shutdownRuns) {
try {
const task = deps.db.getTask(run.taskId);
if (task.status === "in_progress" || task.status === "blocked") {
deps.db.advanceTaskStatus(run.taskId, "todo", "Auto-resumed after app restart.");
deps.db.releaseTaskLock(run.taskId, run.id);
}
if (!resumedAgents.has(run.agentId)) {
resumedAgents.add(run.agentId);
deps.wakeAgentIfPossible(run.agentId, run.companyId, "app_restart_resume");
}
} catch (error) {
deps.logger.warn(`[bootstrap] Skipped shutdown recovery for run ${run.id} (task/agent may have been deleted):`, error);
}
}
if (resumedAgents.size > 0) {
deps.logger.info(`[bootstrap] Resumed ${shutdownRuns.length} interrupted task(s) across ${resumedAgents.size} agent(s) after clean shutdown recovery.`);
deps.publishDomainChanged();
}
} catch (err) {
deps.logger.warn(`[bootstrap] Shutdown recovery failed: ${err instanceof Error ? err.message : String(err)}`);
}
}
@@ -0,0 +1,45 @@
import type { AppDatabase } from "../database";
interface Logger {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
}
export interface RecoverScheduledPostsDeps {
db: AppDatabase;
logger: Logger;
isQuittingRef: { current: boolean };
getBrowserActionService: () => { dispatchBrowserAction: (actionId: string, companyId: string) => Promise<void> } | null;
}
export function recoverScheduledPosts(deps: RecoverScheduledPostsDeps): Set<ReturnType<typeof setTimeout>> {
const timers = new Set<ReturnType<typeof setTimeout>>();
try {
const snapshot = deps.db.listSnapshot();
for (const company of snapshot.companies) {
const queuedActions = deps.db.listBrowserActions(company.id, { status: "queued" });
for (const action of queuedActions) {
if (!action.payloadJson) continue;
try {
const payload = JSON.parse(action.payloadJson) as { scheduledAt?: string };
if (!payload.scheduledAt) continue;
const delay = new Date(payload.scheduledAt).getTime() - Date.now();
if (delay <= 0) continue;
const timer = setTimeout(() => {
timers.delete(timer);
if (deps.isQuittingRef.current) return;
const browserActions = deps.getBrowserActionService();
if (browserActions) {
void browserActions.dispatchBrowserAction(action.id, company.id);
}
}, delay);
timers.add(timer);
deps.logger.info(`[bootstrap] Re-registered scheduled post timer: action ${action.id} in ${Math.round(delay / 1000)}s`);
} catch { /* ignore malformed payload */ }
}
}
} catch (error) {
deps.logger.warn(`[bootstrap] Failed to re-register scheduled post timers: ${error instanceof Error ? error.message : String(error)}`);
}
return timers;
}
@@ -0,0 +1,28 @@
import type { AppDatabase } from "../database";
import type { createConnectorHealthService } from "../connector-health-service";
interface Logger {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
}
type ConnectorHealthService = ReturnType<typeof createConnectorHealthService>;
export async function runStartupHealthChecks(
db: AppDatabase,
connectorHealth: ConnectorHealthService | null | undefined,
logger: Logger,
): Promise<void> {
if (!connectorHealth) return;
const connectors = db.listSnapshot().connectors;
const results = await Promise.allSettled(
connectors.map((connector) => connectorHealth.checkConnector(connector.id)),
);
for (let i = 0; i < results.length; i++) {
const result = results[i]!;
if (result.status === "rejected") {
logger.warn(`[bootstrap] Connector health check failed for ${connectors[i]?.id}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`);
}
}
logger.info(`[bootstrap] Connector health checks completed (${results.filter((r) => r.status === "fulfilled").length}/${connectors.length} succeeded)`);
}
+109
View File
@@ -0,0 +1,109 @@
import type { AgentApiServer } from "../api-server";
import type { AppDatabase } from "../database";
import type { DesktopEvent, SectionId } from "@shared/types";
import type { NotificationOptions } from "../notification-service";
import type { createServiceGraph } from "../service-graph";
import type { createThreadIntegration } from "../thread-integration";
import type { createCheckpointEngine } from "../checkpoint-engine";
import type { AutomationEventContext } from "../automation/types";
type ServiceGraph = ReturnType<typeof createServiceGraph>;
type ThreadIntegration = ReturnType<typeof createThreadIntegration>;
type CheckpointEngine = ReturnType<typeof createCheckpointEngine>;
export interface CreateApiServerFn {
(db: AppDatabase, callbacks: {
onDomainChanged: () => void;
onWakeAgent: (agentId: string, companyId: string, trigger: string) => string | null;
onTaskCreated: (taskId: string, companyId: string) => void;
onTaskStatusChanged: (taskId: string, companyId: string, prev: string, next: string) => void;
onCommentPosted: (taskId: string, companyId: string, authorAgentId: string | null) => void;
onApprovalCreated: (approvalId: string, companyId: string, taskId: string | null, approvalType: string, summary: string) => void;
onApprovalResolved: (approvalId: string, companyId: string, decision: string) => void;
onDocumentCreated: (companyId: string, authorAgentId: string | null, projectId: string | null, documentId: string, documentTitle: string, documentType: string, taskId: string | null) => void;
onAutoAssignTask: (taskId: string, companyId: string) => string | null;
onHumanMessage: (companyId: string, toAgentId?: string | null, threadId?: string | null, waitStateId?: string | null) => void;
emitEvent: (event: DesktopEvent) => void;
notify: (options: { title: string; body: string; urgency: "critical" | "informational"; navigation?: { section: SectionId; entityId?: string }; batchKey?: string }) => void;
}, allowedOrigins: string[]): AgentApiServer;
}
export interface WireApiServerDeps {
db: AppDatabase;
serviceGraph: ServiceGraph;
threadIntegration: ThreadIntegration;
checkpointEngine: CheckpointEngine;
emitEvent: (event: DesktopEvent) => void;
notify: (options: NotificationOptions) => void;
publishDomainChanged: () => void;
isDev: boolean;
devServerUrl?: string;
createApiServer: CreateApiServerFn;
}
export function wireApiServer(deps: WireApiServerDeps): AgentApiServer {
const wakeAgentIfPossible = (agentId: string | null | undefined, companyId: string, trigger: string) => {
const orchestration = deps.serviceGraph.getOrchestrationService();
if (!orchestration) throw new Error("Orchestration service is not initialized.");
return orchestration.wakeAgentIfPossible(agentId, companyId, trigger);
};
const dispatchAutomation = (trigger: string, ctx: AutomationEventContext) => {
const automation = deps.serviceGraph.getAutomationService();
if (!automation) throw new Error("Automation service is not initialized.");
automation.dispatchAutomation(trigger, ctx);
};
const onTaskCreated = (taskId: string, companyId: string) => {
const automation = deps.serviceGraph.getAutomationService();
if (!automation) throw new Error("Automation service is not initialized.");
automation.handleTaskCreated(taskId, companyId);
deps.threadIntegration.onTaskCreated(taskId, companyId);
};
const onTaskStatusChanged = (taskId: string, companyId: string, prev: string, next: string) => {
const automation = deps.serviceGraph.getAutomationService();
if (!automation) throw new Error("Automation service is not initialized.");
automation.handleTaskStatusChange(taskId, companyId, prev, next);
deps.threadIntegration.onTaskStatusChanged(taskId, companyId, prev, next);
if (prev === "todo" && next === "in_progress") {
try {
const task = deps.db.getTask(taskId);
deps.checkpointEngine.onTaskStarted(taskId, companyId, task.assigneeAgentId);
} catch { /* task may not exist */ }
}
};
const onApprovalResolved = (approvalId: string, companyId: string, decision: string) => {
const automation = deps.serviceGraph.getAutomationService();
if (!automation) throw new Error("Automation service is not initialized.");
automation.handleApprovalResolved(approvalId, companyId, decision);
deps.threadIntegration.onApprovalResolved(approvalId, companyId, decision);
};
const allowedOrigins = [deps.isDev && deps.devServerUrl ? new URL(deps.devServerUrl).origin : null, "app://-"]
.filter((origin): origin is string => Boolean(origin));
return deps.createApiServer(deps.db, {
onDomainChanged: deps.publishDomainChanged,
onWakeAgent: (agentId, companyId, trigger) => wakeAgentIfPossible(agentId, companyId, trigger),
onTaskCreated,
onTaskStatusChanged,
onCommentPosted: (taskId, companyId, authorAgentId) => {
dispatchAutomation("comment_posted", { companyId, taskId, agentId: authorAgentId ?? undefined });
},
onApprovalCreated: (approvalId, companyId, taskId, approvalType, summary) => {
deps.threadIntegration.onApprovalCreated(approvalId, companyId, taskId, approvalType, summary);
},
onApprovalResolved,
onDocumentCreated: (companyId, authorAgentId, projectId, documentId, documentTitle, documentType, taskId) => {
dispatchAutomation("document_created", { companyId, agentId: authorAgentId ?? undefined, projectId: projectId ?? undefined });
deps.threadIntegration.onDocumentCreated(companyId, authorAgentId, taskId, documentTitle);
deps.checkpointEngine.onDocumentDrafted(companyId, documentId, documentTitle, documentType, authorAgentId, taskId);
},
onAutoAssignTask: (taskId, companyId) => deps.db.autoAssignTask(taskId, companyId),
onHumanMessage: (companyId, toAgentId, threadId, waitStateId) => {
deps.threadIntegration.onHumanMessage(companyId, toAgentId, threadId, waitStateId);
deps.publishDomainChanged();
},
emitEvent: deps.emitEvent,
notify: deps.notify,
}, allowedOrigins);
}