diff --git a/src/main/bootstrap-application.ts b/src/main/bootstrap-application.ts index 95028599..c2ceaf03 100644 --- a/src/main/bootstrap-application.ts +++ b/src/main/bootstrap-application.ts @@ -7,6 +7,11 @@ import type { createServiceGraph } from "./service-graph"; import type { createIpcSupport } from "./ipc-support"; import type { createThreadIntegration } from "./thread-integration"; import type { createCheckpointEngine } from "./checkpoint-engine"; +import { + DocumentLifecycleProcessor, + makeLifecycleProcessorOptions, +} from "./document-lifecycle-processor"; +import { registerLifecycleHandlers } from "./document-lifecycle-handlers"; import { prepareLogger } from "./bootstrap/prepare-logger"; import { prepareDatabase } from "./bootstrap/prepare-database"; import { prepareServices } from "./bootstrap/prepare-services"; @@ -39,6 +44,7 @@ interface BootstrapApplicationDependencies { ipcSupport: ReturnType; threadIntegration: ReturnType; checkpointEngine: ReturnType; + lifecycleProcessor: DocumentLifecycleProcessor; }) => void; } @@ -82,11 +88,24 @@ export async function bootstrapApplication(deps: BootstrapApplicationDependencie return orchestration.wakeAgentIfPossible(agentId, companyId, trigger); }; + const lifecycleProcessor = new DocumentLifecycleProcessor( + makeLifecycleProcessorOptions({ db, logger: log }), + ); + registerLifecycleHandlers(lifecycleProcessor, { + dispatchAutomation: (trigger, ctx) => { + const automation = serviceGraph.getAutomationService(); + if (!automation) throw new Error("Automation service is not initialized."); + automation.dispatchAutomation(trigger, ctx); + }, + wakeAgentIfPossible, + }); + const apiServer = wireApiServer({ db, serviceGraph, threadIntegration, checkpointEngine, + lifecycleProcessor, emitEvent, notify, publishDomainChanged, @@ -97,7 +116,18 @@ export async function bootstrapApplication(deps: BootstrapApplicationDependencie apiServerRef.current = apiServer; await apiServer.start(); - deps.onRuntimeReady?.({ db, apiServer, serviceGraph, ipcSupport, threadIntegration, checkpointEngine }); + deps.onRuntimeReady?.({ + db, + apiServer, + serviceGraph, + ipcSupport, + threadIntegration, + checkpointEngine, + lifecycleProcessor, + }); + + lifecycleProcessor.start(); + app.on("before-quit", () => lifecycleProcessor.stop()); if (apiServer && serviceGraph.getBrowserActionService()) { apiServer.setBrowserActionHandler(async (actionId, companyId) => { diff --git a/src/main/bootstrap/wire-api-server.ts b/src/main/bootstrap/wire-api-server.ts index e5a479ea..87401176 100644 --- a/src/main/bootstrap/wire-api-server.ts +++ b/src/main/bootstrap/wire-api-server.ts @@ -12,6 +12,7 @@ import { type DocumentLifecycleInput, type DocumentLifecycleOutcome, } from "../document-lifecycle"; +import type { DocumentLifecycleProcessor } from "../document-lifecycle-processor"; type ServiceGraph = ReturnType; type ThreadIntegration = ReturnType; @@ -40,6 +41,7 @@ export interface WireApiServerDeps { serviceGraph: ServiceGraph; threadIntegration: ThreadIntegration; checkpointEngine: CheckpointEngine; + lifecycleProcessor: DocumentLifecycleProcessor; emitEvent: (event: DesktopEvent) => void; notify: (options: NotificationOptions) => void; publishDomainChanged: () => void; @@ -110,8 +112,7 @@ export function wireApiServer(deps: WireApiServerDeps): AgentApiServer { db: deps.db, threadIntegration: deps.threadIntegration, checkpointEngine: deps.checkpointEngine, - dispatchAutomation, - wakeAgentIfPossible, + lifecycleProcessor: deps.lifecycleProcessor, publishDomainChanged: deps.publishDomainChanged, logger: log, }, diff --git a/src/main/database/document-lifecycle-events.ts b/src/main/database/document-lifecycle-events.ts index 63f20b95..4db89abd 100644 --- a/src/main/database/document-lifecycle-events.ts +++ b/src/main/database/document-lifecycle-events.ts @@ -30,6 +30,15 @@ export interface EnqueueInput { payloadJson?: string; } +export interface ClaimBatchInput { + /** Upper bound on rows claimed in one pass. */ + limit: number; + /** Rows whose `attempts` have already reached this cap stay `pending` and are not claimed here; a dead-letter sweep surfaces them separately. */ + maxAttempts: number; + /** When provided, only claim rows for the given document so the post-commit immediate drain stays scoped. */ + documentId?: string; +} + export function buildDocumentLifecycleEventMethods( getDb: () => DatabaseSync, readAll: ReadAll, @@ -54,11 +63,51 @@ export function buildDocumentLifecycleEventMethods( return id; }, - markRunning(id: string): void { + /** + * Atomically claim up to `limit` pending events, flipping each to + * `running` and incrementing `attempts` in a single statement. + * Returns the claimed rows so the processor can dispatch them + * without racing another worker against the same ids. Rows that + * have already reached `maxAttempts` stay `pending` here — the + * operator sees them via `listExhausted` and either retries + * manually or purges. + */ + claimPendingBatch(input: ClaimBatchInput): DocumentLifecycleEventRow[] { const ts = nowIso(); - getDb().prepare( - "UPDATE document_lifecycle_events SET status = 'running', attempts = attempts + 1, updated_at = ? WHERE id = ?", - ).run(ts, id); + const documentScope = input.documentId ? "AND document_id = ?" : ""; + const selectSql = ` + SELECT id FROM document_lifecycle_events + WHERE status = 'pending' AND attempts < ? ${documentScope} + ORDER BY created_at ASC + LIMIT ? + `; + const updateSql = ` + UPDATE document_lifecycle_events + SET status = 'running', attempts = attempts + 1, updated_at = ? + WHERE id IN (${selectSql}) + RETURNING id, company_id, document_id, hook_type, status, attempts, last_error, payload_json, created_at, updated_at, completed_at + `; + const params: SQLInputValue[] = [ts, input.maxAttempts]; + if (input.documentId) params.push(input.documentId); + params.push(input.limit); + const raw = readAll(updateSql, ...params); + return raw.map(mapLedgerRow); + }, + + /** + * Rescue rows that were claimed by a prior process but never + * reached `succeeded` or `failed` (typically because the process + * crashed mid-hook). Rows whose `updated_at` is older than + * `cutoffIso` move back to `pending` so the next `claimPendingBatch` + * picks them up; `attempts` is NOT rewound because the earlier + * claim already counted against the cap. + */ + reclaimStaleRunning(cutoffIso: string): number { + const ts = nowIso(); + const result = getDb().prepare( + "UPDATE document_lifecycle_events SET status = 'pending', updated_at = ? WHERE status = 'running' AND updated_at < ?", + ).run(ts, cutoffIso); + return Number(result.changes ?? 0); }, markSucceeded(id: string): void { @@ -76,10 +125,17 @@ export function buildDocumentLifecycleEventMethods( }, /** - * 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. + * Return to `pending` after a handler failure that the caller wants + * the processor to retry (e.g. transient network failure). The + * attempt counter is preserved so the cap still bounds retries. */ + markRetryable(id: string, error: string): void { + const ts = nowIso(); + getDb().prepare( + "UPDATE document_lifecycle_events SET status = 'pending', last_error = ?, updated_at = ? WHERE id = ?", + ).run(error.slice(0, 2000), ts, id); + }, + 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", @@ -94,28 +150,38 @@ export function buildDocumentLifecycleEventMethods( limit, ); }, + + /** + * Rows that have reached the attempt cap without succeeding. Surfaced + * to operators so a failing hook is visible instead of disappearing + * into the `pending` queue once the cap locks it out. + */ + listExhausted(maxAttempts: number, 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 status = 'pending' AND attempts >= ? ORDER BY updated_at DESC LIMIT ?", + maxAttempts, + 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) => ({ +interface LedgerRawRow { + 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; +} + +function mapLedgerRow(row: LedgerRawRow): DocumentLifecycleEventRow { + return { id: row.id, companyId: row.company_id, documentId: row.document_id, @@ -127,5 +193,13 @@ function readRows( createdAt: row.created_at, updatedAt: row.updated_at, completedAt: row.completed_at, - })); + }; +} + +function readRows( + readAll: ReadAll, + sql: string, + ...params: SQLInputValue[] +): DocumentLifecycleEventRow[] { + return readAll(sql, ...params).map(mapLedgerRow); } diff --git a/src/main/document-lifecycle-handlers.ts b/src/main/document-lifecycle-handlers.ts new file mode 100644 index 00000000..e9e60607 --- /dev/null +++ b/src/main/document-lifecycle-handlers.ts @@ -0,0 +1,75 @@ +/** + * Registers the concrete lifecycle hook handlers against the processor. + * Handler side is kept thin — read the payload, call into the service + * graph, translate throws into a `retryable` outcome so the processor's + * attempt cap decides when a transient failure becomes permanent. + */ + +import type { AutomationEventContext } from "./automation/types"; +import type { + DocumentLifecycleProcessor, + LifecycleHookContext, + LifecycleHookOutcome, +} from "./document-lifecycle-processor"; + +export interface LifecycleHandlerDeps { + dispatchAutomation: (trigger: string, ctx: AutomationEventContext) => void; + wakeAgentIfPossible: ( + agentId: string | null | undefined, + companyId: string, + trigger: string, + ) => string | null; +} + +export function registerLifecycleHandlers( + processor: DocumentLifecycleProcessor, + deps: LifecycleHandlerDeps, +): void { + processor.register("automation_dispatch", automationDispatchHandler(deps)); + processor.register("reviewer_wake", reviewerWakeHandler(deps)); +} + +function automationDispatchHandler(deps: LifecycleHandlerDeps) { + return (payload: Record, context: LifecycleHookContext): LifecycleHookOutcome => { + const trigger = typeof payload.trigger === "string" ? payload.trigger : "document_created"; + const automationContext: AutomationEventContext = { + companyId: context.companyId, + agentId: stringOrUndefined(payload.agentId), + projectId: stringOrUndefined(payload.projectId), + }; + try { + deps.dispatchAutomation(trigger, automationContext); + return { kind: "succeeded" }; + } catch (error) { + return { kind: "retryable", reason: describeError(error) }; + } + }; +} + +function reviewerWakeHandler(deps: LifecycleHandlerDeps) { + return (payload: Record, context: LifecycleHookContext): LifecycleHookOutcome => { + const reviewerAgentId = stringOrNull(payload.reviewerAgentId); + if (!reviewerAgentId) { + return { kind: "failed", reason: "payload missing reviewerAgentId" }; + } + try { + deps.wakeAgentIfPossible(reviewerAgentId, context.companyId, "assignment"); + return { kind: "succeeded" }; + } catch (error) { + return { kind: "retryable", reason: describeError(error) }; + } + }; +} + +function stringOrUndefined(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function stringOrNull(value: unknown): string | null { + return typeof value === "string" && value.length > 0 ? value : null; +} + +function describeError(error: unknown): string { + if (error instanceof Error) return error.message; + return String(error); +} diff --git a/src/main/document-lifecycle-processor.ts b/src/main/document-lifecycle-processor.ts new file mode 100644 index 00000000..7da5fb1a --- /dev/null +++ b/src/main/document-lifecycle-processor.ts @@ -0,0 +1,227 @@ +/** + * Durable processor for `document_lifecycle_events`. + * + * `createDocumentWithLifecycle` enqueues hook rows inside the same + * transaction as `saveDocument`, so the ledger commit is the new + * durability boundary for cross-process side effects. This processor + * is what moves rows from `pending` to `succeeded` / `failed`: + * + * - `drainForDocument(documentId)` runs synchronously right after + * the save commits so the happy-path latency is unchanged. + * - `processBatch(limit)` runs on a periodic sweep so a crash + * between commit and the immediate drain is self-healing. + * - `reclaimStale()` rescues rows stuck in `running` past the + * configured cutoff — typically because the process died + * mid-handler — and puts them back on the pending queue so the + * next claim picks them up. + * + * Claim semantics are atomic: `claimPendingBatch` flips rows to + * `running` and increments `attempts` in a single UPDATE so two + * processors cannot race onto the same id. `maxAttempts` caps the + * retry budget; rows that exhaust it stay `pending` with their + * attempt count at the cap and surface to the operator through + * `listExhausted`. + */ + +import type { AppDatabase } from "./database"; +import type { DocumentLifecycleEventRow } from "./database/document-lifecycle-events"; + +export interface LifecycleHookContext { + companyId: string; + documentId: string; + eventId: string; +} + +export type LifecycleHookOutcome = + | { kind: "succeeded" } + | { kind: "retryable"; reason: string } + | { kind: "failed"; reason: string }; + +export type LifecycleHookHandler = ( + payload: Record, + context: LifecycleHookContext, +) => LifecycleHookOutcome; + +export interface LifecycleProcessorOptions { + db: AppDatabase; + logger: { + info: (...args: unknown[]) => void; + warn: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; + }; + /** Maximum `attempts` value allowed before a row becomes dead-letter. */ + maxAttempts: number; + /** Cutoff for reclaiming rows stuck in `running`. */ + staleRunningMs: number; + /** Batch size for a single sweep. */ + batchSize: number; + /** Interval for the background sweep. 0 disables periodic work. */ + sweepIntervalMs: number; +} + +const DEFAULT_MAX_ATTEMPTS = 5; +const DEFAULT_STALE_RUNNING_MS = 60_000; +const DEFAULT_BATCH_SIZE = 32; +const DEFAULT_SWEEP_INTERVAL_MS = 15_000; + +export function makeLifecycleProcessorOptions( + overrides: Partial & Pick, +): LifecycleProcessorOptions { + return { + maxAttempts: DEFAULT_MAX_ATTEMPTS, + staleRunningMs: DEFAULT_STALE_RUNNING_MS, + batchSize: DEFAULT_BATCH_SIZE, + sweepIntervalMs: DEFAULT_SWEEP_INTERVAL_MS, + ...overrides, + }; +} + +export class DocumentLifecycleProcessor { + private readonly handlers = new Map(); + private sweepTimer: ReturnType | null = null; + private stopped = false; + + constructor(private readonly options: LifecycleProcessorOptions) {} + + register(hookType: string, handler: LifecycleHookHandler): void { + this.handlers.set(hookType, handler); + } + + /** Immediate drain scoped to one document, used right after commit. */ + drainForDocument(documentId: string): void { + const batch = this.options.db.documentLifecycleEvents.claimPendingBatch({ + limit: this.options.batchSize, + maxAttempts: this.options.maxAttempts, + documentId, + }); + for (const row of batch) this.dispatch(row); + } + + /** Periodic sweep: rescues stragglers across all documents. */ + processBatch(): number { + const batch = this.options.db.documentLifecycleEvents.claimPendingBatch({ + limit: this.options.batchSize, + maxAttempts: this.options.maxAttempts, + }); + for (const row of batch) this.dispatch(row); + return batch.length; + } + + /** Move rows stuck in `running` back to `pending`. */ + reclaimStale(): number { + const cutoff = new Date(Date.now() - this.options.staleRunningMs).toISOString(); + const reclaimed = this.options.db.documentLifecycleEvents.reclaimStaleRunning(cutoff); + if (reclaimed > 0) { + this.options.logger.info( + `[document-lifecycle] reclaimed ${reclaimed} stale running row(s) past ${this.options.staleRunningMs}ms cutoff`, + ); + } + return reclaimed; + } + + /** + * Start the periodic sweeper. Idempotent: calling twice is a no-op. + * `stop()` tears it down so the process can exit cleanly. + */ + start(): void { + if (this.sweepTimer || this.stopped) return; + if (this.options.sweepIntervalMs <= 0) return; + // Run a reclaim + sweep on startup so a crashed process's leftover + // `running` rows are rescued before the first new claim arrives. + this.tick(); + this.sweepTimer = setInterval(() => this.tick(), this.options.sweepIntervalMs); + if (typeof this.sweepTimer.unref === "function") this.sweepTimer.unref(); + } + + stop(): void { + this.stopped = true; + if (this.sweepTimer) { + clearInterval(this.sweepTimer); + this.sweepTimer = null; + } + } + + private tick(): void { + try { + this.reclaimStale(); + } catch (error) { + this.options.logger.warn( + `[document-lifecycle] reclaimStale failed: ${describeError(error)}`, + ); + } + try { + this.processBatch(); + } catch (error) { + this.options.logger.warn( + `[document-lifecycle] processBatch failed: ${describeError(error)}`, + ); + } + } + + private dispatch(row: DocumentLifecycleEventRow): void { + const handler = this.handlers.get(row.hookType); + if (!handler) { + const reason = `No handler registered for hook type "${row.hookType}"`; + this.options.db.documentLifecycleEvents.markFailed(row.id, reason); + this.options.logger.error(`[document-lifecycle] ${reason} (event ${row.id})`); + return; + } + + let payload: Record; + try { + payload = row.payloadJson ? (JSON.parse(row.payloadJson) as Record) : {}; + } catch (error) { + const reason = `payload_json parse failed: ${describeError(error)}`; + this.options.db.documentLifecycleEvents.markFailed(row.id, reason); + this.options.logger.error(`[document-lifecycle] ${reason} (event ${row.id})`); + return; + } + + let outcome: LifecycleHookOutcome; + try { + outcome = handler(payload, { + companyId: row.companyId, + documentId: row.documentId, + eventId: row.id, + }); + } catch (error) { + const message = describeError(error); + this.options.logger.error( + `[document-lifecycle] handler "${row.hookType}" threw for event ${row.id}: ${message}`, + ); + this.recordOutcome(row, { kind: "retryable", reason: message }); + return; + } + this.recordOutcome(row, outcome); + } + + private recordOutcome(row: DocumentLifecycleEventRow, outcome: LifecycleHookOutcome): void { + try { + if (outcome.kind === "succeeded") { + this.options.db.documentLifecycleEvents.markSucceeded(row.id); + return; + } + if (outcome.kind === "failed") { + this.options.db.documentLifecycleEvents.markFailed(row.id, outcome.reason); + return; + } + if (row.attempts >= this.options.maxAttempts) { + this.options.db.documentLifecycleEvents.markFailed( + row.id, + `Retry budget exhausted (${row.attempts}/${this.options.maxAttempts}): ${outcome.reason}`, + ); + return; + } + this.options.db.documentLifecycleEvents.markRetryable(row.id, outcome.reason); + } catch (error) { + this.options.logger.warn( + `[document-lifecycle] recordOutcome failed for ${row.id}: ${describeError(error)}`, + ); + } + } +} + +function describeError(error: unknown): string { + if (error instanceof Error) return error.message; + return String(error); +} diff --git a/src/main/document-lifecycle.ts b/src/main/document-lifecycle.ts index db0d7708..21e65127 100644 --- a/src/main/document-lifecycle.ts +++ b/src/main/document-lifecycle.ts @@ -25,6 +25,7 @@ import { IMPORTANT_DOC_TYPE_SET } from "@shared/document-categories"; import type { AppDatabase } from "./database"; import type { ThreadIntegrationService } from "./thread-integration"; import type { createCheckpointEngine } from "./checkpoint-engine"; +import type { DocumentLifecycleProcessor } from "./document-lifecycle-processor"; export type CheckpointEngineLike = Pick< ReturnType, @@ -35,8 +36,14 @@ export interface DocumentLifecycleDeps { db: AppDatabase; threadIntegration: ThreadIntegrationService; checkpointEngine: CheckpointEngineLike; - dispatchAutomation: (trigger: string, ctx: { companyId: string; [key: string]: unknown }) => void; - wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null; + /** + * Drains the ledger rows this save enqueued. The processor reads from + * `document_lifecycle_events` (not from an in-memory closure) so a + * crash between commit and dispatch is self-healing — the next sweep + * picks up anything left as `pending`. Callers still see the immediate + * happy-path outcome because `drainForDocument` is synchronous. + */ + lifecycleProcessor: Pick; publishDomainChanged: () => void; logger: { warn: (...args: unknown[]) => void; @@ -70,18 +77,10 @@ export interface DocumentLifecycleOutcome { export type SideEffectHook = "automation_dispatch" | "reviewer_wake"; -interface PendingSideEffect { - eventId: string; - hook: SideEffectHook; - run: () => void; -} - export function createDocumentWithLifecycle( deps: DocumentLifecycleDeps, input: DocumentLifecycleInput, ): DocumentLifecycleOutcome { - const pendingSideEffects: PendingSideEffect[] = []; - const outcome = deps.db.withTransaction(() => { const isNew = !input.id; const id = deps.db.saveDocument(input); @@ -109,9 +108,12 @@ export function createDocumentWithLifecycle( 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({ + // Side effects (cross-process / advisory) land in the ledger as + // `pending` rows. The processor drains them post-commit — not a + // memoised closure — so a crash between this commit and the first + // handler attempt is self-healing: the periodic sweeper picks the + // row up on its next tick. + deps.db.documentLifecycleEvents.enqueue({ companyId: input.companyId, documentId: id, hookType: "automation_dispatch", @@ -121,74 +123,38 @@ export function createDocumentWithLifecycle( 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({ + 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); + if (outcome.isNew) { + // Immediate synchronous drain so the happy path behaves identically + // to the pre-processor layout. A failure inside the processor is + // recorded on the ledger row — the ledger, not an in-memory list, + // is what the periodic sweeper retries against. + try { + deps.lifecycleProcessor.drainForDocument(outcome.id); + } catch (error) { + deps.logger.warn( + `[document-lifecycle] drainForDocument(${outcome.id}) threw: ${error instanceof Error ? error.message : String(error)}`, + ); + } } deps.publishDomainChanged(); return outcome; } -function runSideEffect(deps: DocumentLifecycleDeps, pending: PendingSideEffect): void { - try { - deps.db.documentLifecycleEvents.markRunning(pending.eventId); - } catch (error) { - deps.logger.warn( - `[document-lifecycle] markRunning failed for ${pending.eventId}: ${error instanceof Error ? error.message : String(error)}`, - ); - return; - } - - try { - pending.run(); - deps.db.documentLifecycleEvents.markSucceeded(pending.eventId); - } catch (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.db.documentLifecycleEvents.markFailed(pending.eventId, message); - } catch (markError) { - deps.logger.warn( - `[document-lifecycle] markFailed failed for ${pending.eventId}: ${markError instanceof Error ? markError.message : String(markError)}`, - ); - } - } -} - function deriveTaskIdFromRun( db: AppDatabase, runId: string | null, diff --git a/src/main/ipc-content-operations-handlers.ts b/src/main/ipc-content-operations-handlers.ts index fa9e52b5..13e2942c 100644 --- a/src/main/ipc-content-operations-handlers.ts +++ b/src/main/ipc-content-operations-handlers.ts @@ -31,10 +31,10 @@ export function registerContentOperationsHandlers({ ensureCompanyExists, findInvalidReference, publishDomainChanged, - wakeAgentIfPossible, dispatchAutomation, threadIntegration, checkpointEngine, + lifecycleProcessor, logger, }: Pick< RegisterDomainHandlersDependencies, @@ -43,10 +43,10 @@ export function registerContentOperationsHandlers({ | "ensureCompanyExists" | "findInvalidReference" | "publishDomainChanged" - | "wakeAgentIfPossible" | "dispatchAutomation" | "threadIntegration" | "checkpointEngine" + | "lifecycleProcessor" | "logger" >) { registerHandle(IPC_CHANNELS.saveMeeting, async (_event, payload) => { @@ -106,8 +106,7 @@ export function registerContentOperationsHandlers({ db, threadIntegration, checkpointEngine, - dispatchAutomation, - wakeAgentIfPossible, + lifecycleProcessor, publishDomainChanged, logger, }, diff --git a/src/main/ipc-domain-common.ts b/src/main/ipc-domain-common.ts index a4f85a88..45f712bf 100644 --- a/src/main/ipc-domain-common.ts +++ b/src/main/ipc-domain-common.ts @@ -7,6 +7,7 @@ import type { AppDatabase } from "./database"; import type { createSkillService } from "./skill-service"; import type { createBudgetEnforcementService } from "./budget-enforcement-service"; import type { CheckpointEngineLike } from "./document-lifecycle"; +import type { DocumentLifecycleProcessor } from "./document-lifecycle-processor"; import type { ThreadIntegrationService } from "./thread-integration"; interface BrowserActionControl { @@ -60,6 +61,7 @@ export interface RegisterDomainHandlersDependencies { */ threadIntegration: ThreadIntegrationService; checkpointEngine: CheckpointEngineLike; + lifecycleProcessor: Pick; } export { ok, fail } from "./ipc-core-common"; diff --git a/src/main/main-runtime-support.ts b/src/main/main-runtime-support.ts index 7b064f6c..7bdbeee2 100644 --- a/src/main/main-runtime-support.ts +++ b/src/main/main-runtime-support.ts @@ -13,6 +13,7 @@ import { registerOAuthIpcHandlers } from "./ipc-oauth-handlers"; import { fail } from "./ipc-core-common"; import type { ThreadIntegrationService } from "./thread-integration"; import type { createCheckpointEngine } from "./checkpoint-engine"; +import type { DocumentLifecycleProcessor } from "./document-lifecycle-processor"; type BootstrapResult = Awaited>; @@ -23,6 +24,7 @@ interface MainRuntimeState { ipcSupport: BootstrapResult["ipcSupport"] | null; threadIntegration: ThreadIntegrationService | null; checkpointEngine: ReturnType | null; + lifecycleProcessor: DocumentLifecycleProcessor | null; } export function createMainRuntimeSupport(state: MainRuntimeState) { @@ -109,6 +111,13 @@ export function createMainRuntimeSupport(state: MainRuntimeState) { return state.checkpointEngine; } + function requireLifecycleProcessor() { + if (!state.lifecycleProcessor) { + throw new Error("Document lifecycle processor is not initialized."); + } + return state.lifecycleProcessor; + } + function emitEvent(event: Parameters[0]) { requireIpcSupport().emitEvent(event); } @@ -209,6 +218,7 @@ export function createMainRuntimeSupport(state: MainRuntimeState) { }, threadIntegration: requireThreadIntegration(), checkpointEngine: requireCheckpointEngine(), + lifecycleProcessor: requireLifecycleProcessor(), }); const piRuntime = requireServiceGraph().getPiRuntime(); diff --git a/src/main/main-runtime.ts b/src/main/main-runtime.ts index df8af993..02baf005 100644 --- a/src/main/main-runtime.ts +++ b/src/main/main-runtime.ts @@ -6,6 +6,7 @@ import { AppDatabase } from "./database"; import { createMainRuntimeSupport } from "./main-runtime-support"; import type { ThreadIntegrationService } from "./thread-integration"; import type { createCheckpointEngine } from "./checkpoint-engine"; +import type { DocumentLifecycleProcessor } from "./document-lifecycle-processor"; interface CreateMainRuntimeDependencies { hasSingleInstanceLock: boolean; @@ -39,6 +40,7 @@ export function createMainRuntime({ ipcSupport: Awaited>["ipcSupport"] | null; threadIntegration: ThreadIntegrationService | null; checkpointEngine: ReturnType | null; + lifecycleProcessor: DocumentLifecycleProcessor | null; isQuitting: boolean; } = { db: null, @@ -47,6 +49,7 @@ export function createMainRuntime({ ipcSupport: null, threadIntegration: null, checkpointEngine: null, + lifecycleProcessor: null, isQuitting: false, }; const support = createMainRuntimeSupport(state); @@ -83,13 +86,14 @@ export function createMainRuntime({ state.isQuitting = value; }, }, - onRuntimeReady: ({ db, apiServer, serviceGraph, ipcSupport, threadIntegration, checkpointEngine }) => { + onRuntimeReady: ({ db, apiServer, serviceGraph, ipcSupport, threadIntegration, checkpointEngine, lifecycleProcessor }) => { state.db = db; state.apiServer = apiServer; state.serviceGraph = serviceGraph; state.ipcSupport = ipcSupport; state.threadIntegration = threadIntegration; state.checkpointEngine = checkpointEngine; + state.lifecycleProcessor = lifecycleProcessor; // Register IPC handlers immediately so the renderer can call loadSnapshot() // as soon as the window appears — before connector health checks finish. support.registerHandlers(); diff --git a/tests/integration/document-run-provenance.spec.ts b/tests/integration/document-run-provenance.spec.ts index 19fa02f6..cd2b860b 100644 --- a/tests/integration/document-run-provenance.spec.ts +++ b/tests/integration/document-run-provenance.spec.ts @@ -91,6 +91,7 @@ async function createHarness() { }); const token = createAgentToken(agentId, companyId, run.id); const silentLogger = { warn: () => {}, error: () => {} }; + const noopLifecycleProcessor = { drainForDocument: () => {} }; const server = createApiServer(db, () => {}, () => null, { createDocument: (input) => createDocumentWithLifecycle( @@ -100,8 +101,7 @@ async function createHarness() { onDocumentCreated: () => {}, } as never, checkpointEngine: { onDocumentDrafted: () => null } as never, - dispatchAutomation: () => {}, - wakeAgentIfPossible: () => null, + lifecycleProcessor: noopLifecycleProcessor, publishDomainChanged: () => {}, logger: silentLogger, },