From edbebe610900c7e83bb66919cc61a3764a6b09a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Sun, 19 Apr 2026 15:53:55 -0400 Subject: [PATCH] Make document lifecycle fan-out durable and atomic Pre-v47 `createDocumentWithLifecycle` wrapped each hook in its own try/catch and always returned success, so a failure in any of the downstream hooks (thread event, checkpoint approval, deliverable review, automation dispatch, reviewer wake) left a live document row behind with a silently incomplete lifecycle. Two invariants now hold: - DB-writing hooks (thread integration, checkpoint approval, deliverable review) run inside the same `BEGIN IMMEDIATE` transaction as `saveDocument`. A throw rolls back the document row along with every partial write, so an operator never sees a persisted document with a missing thread or skipped approval. - Cross-process side effects (automation dispatch, reviewer wake) are enqueued as `document_lifecycle_events` rows inside the same transaction and processed post-commit. Success stamps `status = 'succeeded'`; failure stamps `status = 'failed'` with `last_error`, so partial lifecycle state is observable via `listFailedForCompany` instead of disappearing into the log tail. Migration v47 installs the ledger; `buildDocumentLifecycleEventMethods` exposes the enqueue/running/succeeded/failed transitions; the lifecycle service drives both. --- src/main/database-base.ts | 3 + .../database/document-lifecycle-events.ts | 131 +++++++++ src/main/database/migrations/v47-v47.ts | 53 ++++ src/main/database/schema.ts | 2 + src/main/document-lifecycle.ts | 254 ++++++++++-------- 5 files changed, 326 insertions(+), 117 deletions(-) create mode 100644 src/main/database/document-lifecycle-events.ts create mode 100644 src/main/database/migrations/v47-v47.ts diff --git a/src/main/database-base.ts b/src/main/database-base.ts index a4197082..b6adad96 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -24,6 +24,7 @@ import { buildWakeupMethods } from "./database/wakeup"; import { buildBackupMethods } from "./database/backup"; import { buildSessionEntryMethods } from "./database/session-entries"; import { buildInboxProjectionMethods } from "./database/inbox-projections"; +import { buildDocumentLifecycleEventMethods } from "./database/document-lifecycle-events"; import { buildCredentialMethods } from "./database/credentials"; import { buildModelCapabilityMethods } from "./database/model-capabilities-crud"; import { MODEL_CAPABILITY_SEED } from "./model-capability-seed"; @@ -71,6 +72,7 @@ export class AppDatabaseBase { public readonly backupMod: ReturnType; public readonly sessionEntries: ReturnType; public readonly inboxProjections: ReturnType; + public readonly documentLifecycleEvents: ReturnType; public readonly credentialsTable: ReturnType; public readonly modelCapabilities: ReturnType; @@ -146,6 +148,7 @@ export class AppDatabaseBase { this.wakeup = buildWakeupMethods(getDb, ra, tx); this.sessionEntries = buildSessionEntryMethods(getDb, ra, tx); this.inboxProjections = buildInboxProjectionMethods(getDb); + this.documentLifecycleEvents = buildDocumentLifecycleEventMethods(getDb, ra); this.credentialsTable = buildCredentialMethods(getDb, ra, tx); this.modelCapabilities = buildModelCapabilityMethods(getDb, ra, tx); this.backupMod = buildBackupMethods({ diff --git a/src/main/database/document-lifecycle-events.ts b/src/main/database/document-lifecycle-events.ts new file mode 100644 index 00000000..63f20b95 --- /dev/null +++ b/src/main/database/document-lifecycle-events.ts @@ -0,0 +1,131 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync, SQLInputValue } from "node:sqlite"; +import type { ReadAll } from "./helpers"; +import { nowIso } from "./helpers"; + +/** + * Document-lifecycle event row shape returned by listers. + * `status` moves `pending -> running -> (succeeded | failed)`. `attempts` + * tracks how many processor passes have tried this hook so a sweeper can + * bound retries without losing the ledger. + */ +export interface DocumentLifecycleEventRow { + id: string; + companyId: string; + documentId: string; + hookType: string; + status: "pending" | "running" | "succeeded" | "failed"; + attempts: number; + lastError: string | null; + payloadJson: string; + createdAt: string; + updatedAt: string; + completedAt: string | null; +} + +export interface EnqueueInput { + companyId: string; + documentId: string; + hookType: string; + payloadJson?: string; +} + +export function buildDocumentLifecycleEventMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, +) { + return { + enqueue(input: EnqueueInput): string { + const id = randomUUID(); + const ts = nowIso(); + getDb().prepare( + `INSERT INTO document_lifecycle_events + (id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at) + VALUES (?, ?, ?, ?, 'pending', 0, NULL, ?, ?, ?, NULL)`, + ).run( + id, + input.companyId, + input.documentId, + input.hookType, + input.payloadJson ?? "{}", + ts, + ts, + ); + return id; + }, + + markRunning(id: string): void { + const ts = nowIso(); + getDb().prepare( + "UPDATE document_lifecycle_events SET status = 'running', attempts = attempts + 1, updated_at = ? WHERE id = ?", + ).run(ts, id); + }, + + markSucceeded(id: string): void { + const ts = nowIso(); + getDb().prepare( + "UPDATE document_lifecycle_events SET status = 'succeeded', last_error = NULL, updated_at = ?, completed_at = ? WHERE id = ?", + ).run(ts, ts, id); + }, + + markFailed(id: string, error: string): void { + const ts = nowIso(); + getDb().prepare( + "UPDATE document_lifecycle_events SET status = 'failed', last_error = ?, updated_at = ? WHERE id = ?", + ).run(error.slice(0, 2000), ts, id); + }, + + /** + * Claim the batch of pending events for a document so the caller can + * drain them post-commit. Ordered by `created_at` so the fan-out + * reproduces the enqueue order. + */ + listPendingForDocument(documentId: string): DocumentLifecycleEventRow[] { + return readRows(readAll, + "SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE document_id = ? AND status = 'pending' ORDER BY created_at ASC", + documentId, + ); + }, + + listFailedForCompany(companyId: string, limit = 200): DocumentLifecycleEventRow[] { + return readRows(readAll, + "SELECT id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at FROM document_lifecycle_events WHERE company_id = ? AND status = 'failed' ORDER BY updated_at DESC LIMIT ?", + companyId, + limit, + ); + }, + }; +} + +function readRows( + readAll: ReadAll, + sql: string, + ...params: SQLInputValue[] +): DocumentLifecycleEventRow[] { + const raw = readAll<{ + id: string; + company_id: string; + document_id: string; + hook_type: string; + status: "pending" | "running" | "succeeded" | "failed"; + attempts: number; + last_error: string | null; + payload_json: string; + created_at: string; + updated_at: string; + completed_at: string | null; + }>(sql, ...params); + return raw.map((row) => ({ + id: row.id, + companyId: row.company_id, + documentId: row.document_id, + hookType: row.hook_type, + status: row.status, + attempts: row.attempts, + lastError: row.last_error, + payloadJson: row.payload_json, + createdAt: row.created_at, + updatedAt: row.updated_at, + completedAt: row.completed_at, + })); +} diff --git a/src/main/database/migrations/v47-v47.ts b/src/main/database/migrations/v47-v47.ts new file mode 100644 index 00000000..b46e2a8d --- /dev/null +++ b/src/main/database/migrations/v47-v47.ts @@ -0,0 +1,53 @@ +import log from "electron-log/main.js"; +import type { Migration, MigrationDeps } from "./types"; + +export function createMigrationsV47V47(deps: MigrationDeps): Migration[] { + const { db } = deps; + return [ + { + // Durable fan-out ledger for document-creation side effects. + // + // Before v47 `createDocumentWithLifecycle` wrapped every downstream + // hook (automation dispatch, thread integration, checkpoint engine, + // reviewer wake) in an individual `try/catch` and always returned + // success. A failure in any single hook left the document row live + // with a silently incomplete lifecycle — thread views, approvals, + // or reviewer wakes disappeared without trace. + // + // The ledger is the tracked-state surface the operator reads to + // reason about lifecycle coverage: one row per hook per document + // is inserted inside the same transaction as the document write. + // The processor advances `status` from `pending` to `succeeded` or + // `failed`, stamps `last_error` on failure, and `attempts` lets the + // sweeper retry within a bounded policy. + version: 47, + name: "document_lifecycle_events_ledger", + fn: () => { + db.exec(` + CREATE TABLE IF NOT EXISTS document_lifecycle_events ( + id TEXT PRIMARY KEY, + company_id TEXT NOT NULL, + document_id TEXT NOT NULL, + hook_type TEXT NOT NULL, + status TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + payload_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT + ) + `); + db.exec(` + CREATE INDEX IF NOT EXISTS idx_document_lifecycle_events_company + ON document_lifecycle_events(company_id, status, updated_at) + `); + db.exec(` + CREATE INDEX IF NOT EXISTS idx_document_lifecycle_events_document + ON document_lifecycle_events(document_id, hook_type) + `); + log.info("[migration v47] document_lifecycle_events ledger installed"); + }, + }, + ]; +} diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 481b6538..d92121b4 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -11,6 +11,7 @@ import { createMigrationsV43V43 } from "./migrations/v43-v43"; import { createMigrationsV44V44 } from "./migrations/v44-v44"; import { createMigrationsV45V45 } from "./migrations/v45-v45"; import { createMigrationsV46V46 } from "./migrations/v46-v46"; +import { createMigrationsV47V47 } from "./migrations/v47-v47"; import { columnExists, nowIso, @@ -197,6 +198,7 @@ export function runMigrations( ...createMigrationsV44V44(deps), ...createMigrationsV45V45(deps), ...createMigrationsV46V46(deps), + ...createMigrationsV47V47(deps), ]; for (const migration of migrations) { diff --git a/src/main/document-lifecycle.ts b/src/main/document-lifecycle.ts index 5c7a0cce..db0d7708 100644 --- a/src/main/document-lifecycle.ts +++ b/src/main/document-lifecycle.ts @@ -1,20 +1,23 @@ /** * Unified document-creation lifecycle. * - * Previous layout had two divergent entry points: the HTTP API - * (`routes-files.ts`) fired thread, checkpoint, automation, reviewer-wake, - * and deliverable-review side effects, while the desktop IPC save path - * (`ipc-content-operations-handlers.ts`) only fired automation and the - * reviewer wake. A document created from the API and one created from the - * desktop reviewer ended up in inconsistent thread-event and checkpoint - * states, so the UI showed different lifecycle progress depending on - * which surface produced the row. + * Two invariants: * - * This module is the single owner of "what happens when a new document - * lands in the DB". Both entry points call `createDocumentWithLifecycle` - * and rely on its deterministic fan-out. Update-only edits (when `id` is - * present) skip the creation-time side effects but still publish the - * domain-change event via `publishDomainChanged`. + * 1. DB-writing hooks (thread events, checkpoint approval, + * deliverable-review approval) commit atomically with the + * document row. A failure in any of them rolls back the save so + * the operator never sees a live document with a missing thread + * or a skipped approval. + * + * 2. Cross-process side effects (automation dispatch, reviewer wake) + * are durable via the `document_lifecycle_events` ledger. Every + * hook attempt lands a row and the processor stamps its outcome; + * a crash between "document saved" and "reviewer woken" is + * observable as a `failed` ledger row instead of a silent gap. + * + * The HTTP API route and the desktop IPC handler both call + * `createDocumentWithLifecycle`, so both surfaces end up in the same + * state after a save — no more divergence between the two entry points. */ import log from "electron-log/main.js"; @@ -55,99 +58,132 @@ export interface DocumentLifecycleInput { version: number; status: string; tagsJson: string; - /** Run that produced the save call, or null for operator-authored saves. */ sourceRunId: string | null; } export interface DocumentLifecycleOutcome { id: string; isNew: boolean; - /** Task derived from the originating run, if any — used for thread/approval binding. */ taskId: string | null; - /** Approval raised by the deliverable-review gate, when applicable. */ deliverableReviewApprovalId: string | null; } -/** - * Save a document and dispatch every downstream side effect associated - * with its creation. Side effects are fired inside individual `try`s - * because this is the authoritative fanout — an early throw should not - * prevent later hooks from running (e.g. an automation failure must not - * suppress the thread event). - */ +export type SideEffectHook = "automation_dispatch" | "reviewer_wake"; + +interface PendingSideEffect { + eventId: string; + hook: SideEffectHook; + run: () => void; +} + export function createDocumentWithLifecycle( deps: DocumentLifecycleDeps, input: DocumentLifecycleInput, ): DocumentLifecycleOutcome { - const isNew = !input.id; - const id = deps.db.saveDocument(input); - const taskId = deriveTaskIdFromRun(deps.db, input.sourceRunId, input.companyId, deps.logger); + const pendingSideEffects: PendingSideEffect[] = []; - if (isNew) { - fireCreationHooks(deps, input, id, taskId); - } + const outcome = deps.db.withTransaction(() => { + const isNew = !input.id; + const id = deps.db.saveDocument(input); + const taskId = deriveTaskIdFromRun(deps.db, input.sourceRunId, input.companyId, deps.logger); - let deliverableReviewApprovalId: string | null = null; - if (isNew && IMPORTANT_DOC_TYPE_SET.has(input.type)) { - deliverableReviewApprovalId = maybeRaiseDeliverableReviewApproval(deps, input, id, taskId); + let deliverableReviewApprovalId: string | null = null; + if (isNew) { + // DB-writing hooks run inside the transaction. Any throw rolls + // back `saveDocument` along with the thread event / approvals. + deps.threadIntegration.onDocumentCreated( + input.companyId, + input.authorAgentId, + taskId, + input.title, + ); + deps.checkpointEngine.onDocumentDrafted( + input.companyId, + id, + input.title, + input.type, + input.authorAgentId, + taskId, + ); + if (IMPORTANT_DOC_TYPE_SET.has(input.type)) { + deliverableReviewApprovalId = maybeRaiseDeliverableReviewApproval(deps, input, id, taskId); + } + + // Side effects (cross-process / advisory) land in the ledger so + // a post-commit failure is observable instead of a silent gap. + const automationEventId = deps.db.documentLifecycleEvents.enqueue({ + companyId: input.companyId, + documentId: id, + hookType: "automation_dispatch", + payloadJson: JSON.stringify({ + trigger: "document_created", + agentId: input.authorAgentId, + projectId: input.projectId, + }), + }); + pendingSideEffects.push({ + eventId: automationEventId, + hook: "automation_dispatch", + run: () => deps.dispatchAutomation("document_created", { + companyId: input.companyId, + agentId: input.authorAgentId ?? undefined, + projectId: input.projectId ?? undefined, + }), + }); + + if (input.reviewerAgentId && input.reviewerAgentId !== input.authorAgentId) { + const wakeEventId = deps.db.documentLifecycleEvents.enqueue({ + companyId: input.companyId, + documentId: id, + hookType: "reviewer_wake", + payloadJson: JSON.stringify({ reviewerAgentId: input.reviewerAgentId }), + }); + const reviewerAgentId = input.reviewerAgentId; + const companyId = input.companyId; + pendingSideEffects.push({ + eventId: wakeEventId, + hook: "reviewer_wake", + run: () => { + deps.wakeAgentIfPossible(reviewerAgentId, companyId, "assignment"); + }, + }); + } + } + + return { id, isNew, taskId, deliverableReviewApprovalId }; + }); + + for (const pending of pendingSideEffects) { + runSideEffect(deps, pending); } deps.publishDomainChanged(); - return { id, isNew, taskId, deliverableReviewApprovalId }; + return outcome; } -function fireCreationHooks( - deps: DocumentLifecycleDeps, - input: DocumentLifecycleInput, - documentId: string, - taskId: string | null, -): void { +function runSideEffect(deps: DocumentLifecycleDeps, pending: PendingSideEffect): void { try { - deps.dispatchAutomation("document_created", { - companyId: input.companyId, - agentId: input.authorAgentId ?? undefined, - projectId: input.projectId ?? undefined, - }); + deps.db.documentLifecycleEvents.markRunning(pending.eventId); } catch (error) { deps.logger.warn( - `[document-lifecycle] dispatchAutomation failed for document ${documentId}: ${error instanceof Error ? error.message : String(error)}`, + `[document-lifecycle] markRunning failed for ${pending.eventId}: ${error instanceof Error ? error.message : String(error)}`, ); + return; } try { - deps.threadIntegration.onDocumentCreated( - input.companyId, - input.authorAgentId, - taskId, - input.title, - ); + pending.run(); + deps.db.documentLifecycleEvents.markSucceeded(pending.eventId); } catch (error) { - deps.logger.warn( - `[document-lifecycle] threadIntegration.onDocumentCreated failed for ${documentId}: ${error instanceof Error ? error.message : String(error)}`, + const message = error instanceof Error ? error.message : String(error); + deps.logger.error( + `[document-lifecycle] hook ${pending.hook} failed for event ${pending.eventId}: ${message}`, ); - } - - try { - deps.checkpointEngine.onDocumentDrafted( - input.companyId, - documentId, - input.title, - input.type, - input.authorAgentId, - taskId, - ); - } catch (error) { - deps.logger.warn( - `[document-lifecycle] checkpointEngine.onDocumentDrafted failed for ${documentId}: ${error instanceof Error ? error.message : String(error)}`, - ); - } - - if (input.reviewerAgentId && input.reviewerAgentId !== input.authorAgentId) { try { - deps.wakeAgentIfPossible(input.reviewerAgentId, input.companyId, "assignment"); - } catch (error) { + deps.db.documentLifecycleEvents.markFailed(pending.eventId, message); + } catch (markError) { deps.logger.warn( - `[document-lifecycle] reviewer wake failed for ${documentId}: ${error instanceof Error ? error.message : String(error)}`, + `[document-lifecycle] markFailed failed for ${pending.eventId}: ${markError instanceof Error ? markError.message : String(markError)}`, ); } } @@ -172,60 +208,44 @@ function deriveTaskIdFromRun( } } -/** - * Raises a `deliverable_review` approval when the company has opted in, - * the document type is important, execution mode is `ai_ceo` (cofounder - * mode handles reviews via the checkpoint engine), and no pending - * deliverable review already exists. Mirrors the gate that used to live - * inline in `routes-files.ts`; consolidating it here ensures operator- - * created documents get the same reviewer treatment. - */ function maybeRaiseDeliverableReviewApproval( deps: DocumentLifecycleDeps, input: DocumentLifecycleInput, documentId: string, taskId: string | null, ): string | null { - try { - const companySettings = deps.db.getCompanyApprovalSettings(input.companyId); - if (!companySettings?.reviewDeliverables) return null; - if (deps.db.getCompanyExecutionMode(input.companyId) === "ai_cofounder") return null; - if (deps.db.hasPendingDeliverableReviews(input.companyId)) return null; + const companySettings = deps.db.getCompanyApprovalSettings(input.companyId); + if (!companySettings?.reviewDeliverables) return null; + if (deps.db.getCompanyExecutionMode(input.companyId) === "ai_cofounder") return null; + if (deps.db.hasPendingDeliverableReviews(input.companyId)) return null; - const approvalId = deps.db.requestApproval({ - companyId: input.companyId, - relatedTaskId: taskId, - requestedByAgentId: input.authorAgentId ?? "", - type: "deliverable_review", - payloadSummary: `Review deliverable: ${input.title}`.slice(0, 280), - impactSummary: `New ${input.type.replaceAll("_", " ")} created. Awaiting your review before workflow continues.`.slice(0, 400), - payloadJson: JSON.stringify({ - documentIds: [documentId], - documents: [{ id: documentId, type: input.type, title: input.title }], - }), - }); + const approvalId = deps.db.requestApproval({ + companyId: input.companyId, + relatedTaskId: taskId, + requestedByAgentId: input.authorAgentId ?? "", + type: "deliverable_review", + payloadSummary: `Review deliverable: ${input.title}`.slice(0, 280), + impactSummary: `New ${input.type.replaceAll("_", " ")} created. Awaiting your review before workflow continues.`.slice(0, 400), + payloadJson: JSON.stringify({ + documentIds: [documentId], + documents: [{ id: documentId, type: input.type, title: input.title }], + }), + }); - if (input.authorAgentId) { - try { - const agent = deps.db.getAgent(input.authorAgentId); - if (agent.heartbeatEnabled) { - deps.db.setHeartbeat(input.authorAgentId, false, agent.heartbeatIntervalSec || 120); - } - } catch (error) { - deps.logger.warn( - `[document-lifecycle] heartbeat pause failed for agent ${input.authorAgentId}: ${error instanceof Error ? error.message : String(error)}`, - ); + if (input.authorAgentId) { + try { + const agent = deps.db.getAgent(input.authorAgentId); + if (agent.heartbeatEnabled) { + deps.db.setHeartbeat(input.authorAgentId, false, agent.heartbeatIntervalSec || 120); } + } catch (error) { + deps.logger.warn( + `[document-lifecycle] heartbeat pause failed for agent ${input.authorAgentId}: ${error instanceof Error ? error.message : String(error)}`, + ); } - - return approvalId; - } catch (error) { - deps.logger.warn( - `[document-lifecycle] deliverable-review gate failed for document ${documentId}: ${error instanceof Error ? error.message : String(error)}`, - ); - return null; } + + return approvalId; } -/** Re-export for backwards-compatible logging in callers that do not have electron-log wired. */ export const defaultLogger: DocumentLifecycleDeps["logger"] = log;