From c779fc28586b9c1e91e83a680cf88f7f484bf995 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 23 Apr 2026 20:43:44 -0400 Subject: [PATCH] fix: make onboarding bootstrap atomic --- src/main/onboarding-service.ts | 170 ++++++++++-------- src/main/orchestration/index.ts | 1 + src/main/orchestration/task-runner.ts | 89 ++++++--- .../create-background-services.ts | 4 +- .../integration/onboarding-model-pool.spec.ts | 5 +- .../onboarding-transaction.spec.ts | 121 +++++++++++++ 6 files changed, 286 insertions(+), 104 deletions(-) create mode 100644 tests/integration/onboarding-transaction.spec.ts diff --git a/src/main/onboarding-service.ts b/src/main/onboarding-service.ts index c8ed97c7..781c2dc4 100644 --- a/src/main/onboarding-service.ts +++ b/src/main/onboarding-service.ts @@ -13,6 +13,10 @@ interface OnboardingBootstrapSummary { runId: string | null; } +interface InitialExecutionLoopResult extends OnboardingBootstrapSummary { + dispatchAfterCommit: () => void; +} + interface BootstrapInput { companyName: string; companyDescription: string; @@ -33,9 +37,9 @@ interface BootstrapInput { interface OnboardingDependencies { db: AppDatabase; seedDefaultAutomationRules: (companyId: string) => void; - queueTaskRun: (taskId: string, trigger: string, actor: string) => { + createTaskRunForDeferredDispatch: (taskId: string, trigger: string, actor: string) => { ok: boolean; - data?: string; + data?: { runId: string; dispatchAfterCommit: () => void }; error?: { code: string; message: string }; }; publishDomainChanged: () => void; @@ -78,7 +82,7 @@ function resolveOnboardingConnector( export function createOnboardingService({ db, seedDefaultAutomationRules, - queueTaskRun, + createTaskRunForDeferredDispatch, publishDomainChanged, scanAndAssignSkills, logger, @@ -123,7 +127,7 @@ export function createOnboardingService({ }); } - function createInitialExecutionLoop(companyId: string, workspaceId: string | null): OnboardingBootstrapSummary { + function createInitialExecutionLoop(companyId: string, workspaceId: string | null): InitialExecutionLoopResult { const companyRows = db.queryAll<{ id: string; name: string; description: string }>( "select id, name, description from companies where id = ? limit 1", companyId, @@ -205,7 +209,7 @@ export function createOnboardingService({ requiresFilesystem: true, }); - const startRunResult = queueTaskRun(taskId, "assignment", "system"); + const startRunResult = createTaskRunForDeferredDispatch(taskId, "assignment", "system"); if (!startRunResult.ok && !DEFERRABLE_RUN_ERRORS.has(startRunResult.error?.code ?? "")) { throw new Error(startRunResult.error?.message ?? "Failed to queue the initial execution loop."); } @@ -217,96 +221,104 @@ export function createOnboardingService({ leadAgentId: leadAgent.id, projectId, taskId, - runId: startRunResult.ok ? startRunResult.data ?? null : null, + runId: startRunResult.ok ? startRunResult.data?.runId ?? null : null, + dispatchAfterCommit: startRunResult.ok ? startRunResult.data?.dispatchAfterCommit ?? (() => {}) : () => {}, }; } function bootstrapOnboarding(input: BootstrapInput): OnboardingBootstrapSummary { - const connectors = db.projection.snapshot.listSnapshot({ companyId: null }).connectors; - const selectedConnectorId = resolveOnboardingConnector(connectors, input.connectorId); + const result = db.withTransaction(() => { + const connectors = db.projection.snapshot.listSnapshot({ companyId: null }).connectors; + const selectedConnectorId = resolveOnboardingConnector(connectors, input.connectorId); - const companyId = db.core.company.saveCompany({ - name: input.companyName.trim(), - description: input.companyDescription.trim(), - status: "active", - autoApproveHires: input.autoApproveHires, - reviewDeliverables: input.reviewDeliverables, - executionMode: input.executionMode ?? "ai_ceo", - }); - - // Persist the user's per-connector disabled-model selections - // against the freshly-created company id. Onboarding is entirely - // pre-commit; the pool settings arrive in the input payload and - // land in the DB only once a company exists to scope them. - for (const pool of input.connectorPoolSelections ?? []) { - db.core.settings.setDisabledModels(companyId, pool.connectorId, pool.disabledModels); - } - - // Validate the CEO model selection against the now-persisted pool - // scoped to this new company. Must run AFTER the pool is stored, - // otherwise the validator sees an empty pool and accepts anything. - const modelSelection = validateConnectorModelSelection(connectorPolicyDepsFromDb(db), companyId, selectedConnectorId, input.connectorModel ?? null); - if (!modelSelection.ok) { - throw new Error(modelSelection.error); - } - - const workspaceId = createDefaultWorkspace(companyId, input.workspacePath); - - const goalTitle = input.goalTitle.trim(); - if (goalTitle) { - db.core.entity.saveGoal({ - companyId, - title: goalTitle, - description: input.goalDescription.trim() || `Achieve: ${goalTitle}`, + const companyId = db.core.company.saveCompany({ + name: input.companyName.trim(), + description: input.companyDescription.trim(), status: "active", - parentId: null, - ownerAgentId: null, + autoApproveHires: input.autoApproveHires, + reviewDeliverables: input.reviewDeliverables, + executionMode: input.executionMode ?? "ai_ceo", }); - } - const ceoLocale = getLocale(); - const ceoId = db.core.agents.saveAgent({ - companyId, - name: "CEO", - role: "CEO", - title: ceoLocale === "zh" ? "首席执行官" : "Chief Executive Officer", - reportsTo: null, - connectorId: selectedConnectorId, - workspaceId, - model: input.connectorModel?.trim() || null, - capabilities: ceoLocale === "zh" - ? "高管领导力、战略规划、团队建设、目标分解、授权管理和组织运营。负责招募领导团队、制定战略，并协调公司所有运营活动。" - : "Executive leadership, strategic planning, team building, goal decomposition, delegation, and organizational management. Responsible for hiring the leadership team, setting strategy, and coordinating all company operations.", - budgetMonthlyUsd: 250, - status: "idle", - // CEO is pinned to the user's pick at onboarding. Never let the - // dispatcher re-route it. - modelStrategy: "explicit", - roleLevel: "c_suite", + // Persist the user's per-connector disabled-model selections + // against the freshly-created company id. Onboarding is entirely + // pre-commit; the pool settings arrive in the input payload and + // land in the DB only once a company exists to scope them. + for (const pool of input.connectorPoolSelections ?? []) { + db.core.settings.setDisabledModels(companyId, pool.connectorId, pool.disabledModels); + } + + // Validate the CEO model selection against the now-persisted pool + // scoped to this new company. Must run AFTER the pool is stored, + // otherwise the validator sees an empty pool and accepts anything. + const modelSelection = validateConnectorModelSelection(connectorPolicyDepsFromDb(db), companyId, selectedConnectorId, input.connectorModel ?? null); + if (!modelSelection.ok) { + throw new Error(modelSelection.error); + } + + const workspaceId = createDefaultWorkspace(companyId, input.workspacePath); + + const goalTitle = input.goalTitle.trim(); + if (goalTitle) { + db.core.entity.saveGoal({ + companyId, + title: goalTitle, + description: input.goalDescription.trim() || `Achieve: ${goalTitle}`, + status: "active", + parentId: null, + ownerAgentId: null, + }); + } + + const ceoLocale = getLocale(); + const ceoId = db.core.agents.saveAgent({ + companyId, + name: "CEO", + role: "CEO", + title: ceoLocale === "zh" ? "首席执行官" : "Chief Executive Officer", + reportsTo: null, + connectorId: selectedConnectorId, + workspaceId, + model: input.connectorModel?.trim() || null, + capabilities: ceoLocale === "zh" + ? "高管领导力、战略规划、团队建设、目标分解、授权管理和组织运营。负责招募领导团队、制定战略，并协调公司所有运营活动。" + : "Executive leadership, strategic planning, team building, goal decomposition, delegation, and organizational management. Responsible for hiring the leadership team, setting strategy, and coordinating all company operations.", + budgetMonthlyUsd: 250, + status: "idle", + // CEO is pinned to the user's pick at onboarding. Never let the + // dispatcher re-route it. + modelStrategy: "explicit", + roleLevel: "c_suite", + }); + db.core.agents.setHeartbeat(ceoId, true, 120); + db.core.settings.setCurrentCompany(companyId); + seedDefaultAutomationRules(companyId); + + const kickoff = createInitialExecutionLoop(companyId, workspaceId); + + db.core.activity.addActivity({ + companyId, + actor: "board", + action: "company.bootstrapped", + entityType: "company", + entityId: companyId, + detail: `Bootstrapped from onboarding with connector ${selectedConnectorId}.`, + }); + + return { kickoff, ceoId }; }); - db.core.agents.setHeartbeat(ceoId, true, 120); - db.core.settings.setCurrentCompany(companyId); - seedDefaultAutomationRules(companyId); + result.kickoff.dispatchAfterCommit(); if (scanAndAssignSkills) { - void scanAndAssignSkills(companyId, ceoId).catch((error) => { + void scanAndAssignSkills(result.kickoff.companyId, result.ceoId).catch((error) => { const message = error instanceof Error ? error.message : String(error); - logger?.warn(`[onboarding] skill scan failed for agent ${ceoId}: ${message}`); + logger?.warn(`[onboarding] skill scan failed for agent ${result.ceoId}: ${message}`); }); } - const kickoff = createInitialExecutionLoop(companyId, workspaceId); - - db.core.activity.addActivity({ - companyId, - actor: "board", - action: "company.bootstrapped", - entityType: "company", - entityId: companyId, - detail: `Bootstrapped from onboarding with connector ${selectedConnectorId}.`, - }); publishDomainChanged(); - return kickoff; + const { dispatchAfterCommit: _dispatchAfterCommit, ...summary } = result.kickoff; + return summary; } return { diff --git a/src/main/orchestration/index.ts b/src/main/orchestration/index.ts index aae01e4b..20055681 100644 --- a/src/main/orchestration/index.ts +++ b/src/main/orchestration/index.ts @@ -38,6 +38,7 @@ export function createOrchestrationService(deps: OrchestrationDependencies) { wakeAgentIfPossible: heartbeat.wakeAgentIfPossible, dispatchRunToWorker: runDispatch.dispatchRunToWorker, queueTaskRun: taskRunner.queueTaskRun, + createTaskRunForDeferredDispatch: taskRunner.createTaskRunForDeferredDispatch, resumeQueuedRuns: taskRunner.resumeQueuedRuns, startAgentHeartbeatRun: heartbeat.startAgentHeartbeatRun, startAllSchedulers: scheduler.startAllSchedulers, diff --git a/src/main/orchestration/task-runner.ts b/src/main/orchestration/task-runner.ts index 767be72a..8601ff75 100644 --- a/src/main/orchestration/task-runner.ts +++ b/src/main/orchestration/task-runner.ts @@ -6,6 +6,12 @@ import type { ResolveWorkspaceFn } from "./run-dispatch"; import { WorkspaceConfigurationError } from "./resolve-workspace"; import { pickModelForAgent } from "../model-dispatch"; +interface PreparedTaskRun { + runId: string; + taskTitle: string; + dispatchParams: RunDispatchParams; +} + export function createTaskRunner( deps: OrchestrationDependencies, resolveWorkspace: ResolveWorkspaceFn, @@ -29,11 +35,15 @@ export function createTaskRunner( return { ok: false, error: { code, message } }; } - function queueTaskRun(taskId: string, trigger: string, actor: string): DesktopResult { + function prepareTaskRun( + taskId: string, + trigger: string, + options: { requireWorkerReady: boolean }, + ): DesktopResult { if (isQuitting()) { return fail("APP_SHUTTING_DOWN", "Application is shutting down."); } - if (!hasWorkerReady()) { + if (options.requireWorkerReady && !hasWorkerReady()) { logger.warn(`[run] Worker not ready — cannot queue run for task ${taskId} (trigger: ${trigger})`); return fail("WORKER_UNAVAILABLE", "Orchestration worker is not ready."); } @@ -112,34 +122,69 @@ export function createTaskRunner( } const companyBasic = db.core.company.getCompanyBasic(task.companyId); - dispatchRunToWorker({ + return ok({ runId: run.id, - companyId: task.companyId, - agent, - task, - connector, - workspace, - workspaceMode: workspaceResolution.lockMode, - observationPath: workspaceResolution.observationPath, - companyName: companyBasic?.name ?? "Unknown", - companyDescription: companyBasic?.description ?? "", - snapshot: db.projection.snapshot.listSnapshot(), - trigger, - resolvedModel: assignment.modelId, + taskTitle: task.title, + dispatchParams: { + runId: run.id, + companyId: task.companyId, + agent, + task, + connector, + workspace, + workspaceMode: workspaceResolution.lockMode, + observationPath: workspaceResolution.observationPath, + companyName: companyBasic?.name ?? "Unknown", + companyDescription: companyBasic?.description ?? "", + snapshot: db.projection.snapshot.listSnapshot(), + trigger, + resolvedModel: assignment.modelId, + }, }); - const postDispatchRun = db.core.runs.getRun(run.id); + } + + function recordQueuedRun(prepared: PreparedTaskRun, actor: string) { + const postDispatchRun = db.core.runs.getRun(prepared.runId); if (postDispatchRun.status === "queued" || postDispatchRun.status === "running") { db.core.activity.addActivity({ - companyId: task.companyId, + companyId: prepared.dispatchParams.companyId, actor, action: "run.queued", entityType: "run", - entityId: run.id, - detail: task.title, + entityId: prepared.runId, + detail: prepared.taskTitle, }); - emitEvent({ type: "run-updated", runId: run.id, action: "created" }); + emitEvent({ type: "run-updated", runId: prepared.runId, action: "created" }); } - return ok(run.id); + } + + function dispatchPreparedTaskRun(prepared: PreparedTaskRun, actor: string) { + dispatchRunToWorker(prepared.dispatchParams); + recordQueuedRun(prepared, actor); + } + + function queueTaskRun(taskId: string, trigger: string, actor: string): DesktopResult { + const prepared = prepareTaskRun(taskId, trigger, { requireWorkerReady: true }); + if (!prepared.ok) return prepared; + dispatchPreparedTaskRun(prepared.data, actor); + return ok(prepared.data.runId); + } + + function createTaskRunForDeferredDispatch(taskId: string, trigger: string, actor: string): DesktopResult<{ runId: string; dispatchAfterCommit: () => void }> { + const prepared = prepareTaskRun(taskId, trigger, { requireWorkerReady: false }); + if (!prepared.ok) return prepared; + return ok({ + runId: prepared.data.runId, + dispatchAfterCommit: () => { + if (isQuitting()) return; + if (!hasWorkerReady()) { + logger.warn(`[run] Worker not ready — leaving queued run ${prepared.data.runId} for recovery (trigger: ${trigger})`); + recordQueuedRun(prepared.data, actor); + return; + } + dispatchPreparedTaskRun(prepared.data, actor); + }, + }); } function resumeQueuedRuns() { @@ -236,5 +281,5 @@ export function createTaskRunner( } } - return { queueTaskRun, resumeQueuedRuns }; + return { queueTaskRun, createTaskRunForDeferredDispatch, resumeQueuedRuns }; } diff --git a/src/main/service-graph/create-background-services.ts b/src/main/service-graph/create-background-services.ts index 5eb6d40d..86d23682 100644 --- a/src/main/service-graph/create-background-services.ts +++ b/src/main/service-graph/create-background-services.ts @@ -75,8 +75,8 @@ export function createBackgroundServices({ const onboardingService = createOnboardingService({ db, seedDefaultAutomationRules, - queueTaskRun: (taskId, trigger, actor) => - orchestrationService.queueTaskRun(taskId, trigger, actor), + createTaskRunForDeferredDispatch: (taskId, trigger, actor) => + orchestrationService.createTaskRunForDeferredDispatch(taskId, trigger, actor), publishDomainChanged, scanAndAssignSkills: async (companyId: string, _agentId: string) => { // All company skills are visible to every agent via progressive disclosure. diff --git a/tests/integration/onboarding-model-pool.spec.ts b/tests/integration/onboarding-model-pool.spec.ts index 7aed9d3b..b5adce18 100644 --- a/tests/integration/onboarding-model-pool.spec.ts +++ b/tests/integration/onboarding-model-pool.spec.ts @@ -43,7 +43,10 @@ function createBootstrapHarness(db: AppDatabase) { db, seedDefaultAutomationRules: noop, publishDomainChanged: noop, - queueTaskRun: () => ({ ok: true, data: "run-1" }), + createTaskRunForDeferredDispatch: () => ({ + ok: true, + data: { runId: "run-1", dispatchAfterCommit: noop }, + }), }); } diff --git a/tests/integration/onboarding-transaction.spec.ts b/tests/integration/onboarding-transaction.spec.ts new file mode 100644 index 00000000..268c09f2 --- /dev/null +++ b/tests/integration/onboarding-transaction.spec.ts @@ -0,0 +1,121 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, describe, expect, it } from "vitest"; +import { getModels } from "@mariozechner/pi-ai"; +import { AppDatabase } from "../../src/main/database"; +import { getConnectorDefinition } from "../../src/main/connectors"; +import { createOnboardingService } from "../../src/main/onboarding-service"; + +const tempDirs: string[] = []; + +async function createTempDir(prefix: string) { + const dir = await mkdtemp(join(tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +function markConnectorReady(db: AppDatabase, connectorId: "openai-codex") { + db.core.connectors.updateConnectorHealth({ + id: connectorId, + status: "ready", + authState: "authenticated", + version: "test", + lastError: null, + }); +} + +function buildBootstrapInput(workspacePath: string, overrides: Partial["bootstrapOnboarding"]>[0]> = {}) { + return { + companyName: "Atomic Onboarding Co", + companyDescription: "Verifies onboarding transaction boundaries", + workspacePath, + autoApproveHires: true, + reviewDeliverables: false, + executionMode: "ai_ceo", + goalTitle: "Create one atomic company", + goalDescription: "Every onboarding write commits together.", + connectorId: "openai-codex" as const, + connectorModel: "", + ...overrides, + }; +} + +function expectNoOnboardingState(db: AppDatabase) { + expect(db.projection.snapshot.listSnapshot().companies).toHaveLength(0); + expect(db.core.settings.getCurrentCompanyId()).toBeNull(); + expect(db.queryAll("select id from agents")).toHaveLength(0); + expect(db.queryAll("select id from goals")).toHaveLength(0); + expect(db.queryAll("select id from projects")).toHaveLength(0); + expect(db.queryAll("select id from tasks")).toHaveLength(0); + expect(db.queryAll("select id from runs")).toHaveLength(0); +} + +describe("onboarding transaction", () => { + it("rolls back every persisted onboarding row when model validation rejects the selected model", async () => { + const profileDir = await createTempDir("agentcompany-onboarding-tx-model-"); + const workspaceDir = await createTempDir("agentcompany-onboarding-tx-workspace-"); + const db = new AppDatabase(profileDir); + db.init(); + markConnectorReady(db, "openai-codex"); + + const definition = getConnectorDefinition("openai-codex"); + const disabledModel = getModels(definition.provider as never)[0]?.id; + if (!disabledModel) { + throw new Error("openai-codex requires a model for onboarding transaction coverage."); + } + + const service = createOnboardingService({ + db, + seedDefaultAutomationRules: () => {}, + publishDomainChanged: () => {}, + createTaskRunForDeferredDispatch: () => ({ + ok: true, + data: { runId: "run-1", dispatchAfterCommit: () => {} }, + }), + }); + + expect(() => service.bootstrapOnboarding(buildBootstrapInput(workspaceDir, { + connectorModel: disabledModel, + connectorPoolSelections: [ + { connectorId: "openai-codex", disabledModels: [{ id: disabledModel, label: disabledModel }] }, + ], + }))).toThrow(`${definition.label} does not allow model "${disabledModel}".`); + + expectNoOnboardingState(db); + db.close(); + }); + + it("rolls back every persisted onboarding row and skips external side effects when initial run creation fails", async () => { + const profileDir = await createTempDir("agentcompany-onboarding-tx-run-"); + const workspaceDir = await createTempDir("agentcompany-onboarding-tx-run-workspace-"); + const db = new AppDatabase(profileDir); + db.init(); + markConnectorReady(db, "openai-codex"); + + let publishCount = 0; + let scanCount = 0; + const service = createOnboardingService({ + db, + seedDefaultAutomationRules: () => {}, + publishDomainChanged: () => { + publishCount += 1; + }, + scanAndAssignSkills: async () => { + scanCount += 1; + }, + createTaskRunForDeferredDispatch: () => ({ ok: false, error: { code: "RUN_CREATE_FAILED", message: "run insert failed" } }), + }); + + expect(() => service.bootstrapOnboarding(buildBootstrapInput(workspaceDir))).toThrow("run insert failed"); + + expectNoOnboardingState(db); + expect(publishCount).toBe(0); + expect(scanCount).toBe(0); + db.close(); + }); +});