From cb0ac485da56434f1eab18e1785656a5fe0abab7 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:46:12 -0400 Subject: [PATCH] Close inbox projection under one transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The old `prepareRun` projection drifted: append-to-shadow, thread-event, and `markAgentInboxDelivered` ran as three independent writes, so a crash between the append and the delivered mark replayed the same inbox row on the next run and double-populated the transcript. v46 installs `agent_message_projections` keyed by (message_id, session_id) as the idempotent gate. The projection closure now wraps the ledger claim, the shadow append, the thread-event write, and the delivered-mark inside one `BEGIN IMMEDIATE` transaction (exposed via `AppDatabase.withTransaction`). `claimProjection` returns false on a replay, so the whole closure short-circuits — no double transcript, no double thread event. --- src/main/agent-runtime/runtime-bridge.ts | 66 ++++++++++++++---------- src/main/database-base.ts | 15 ++++++ src/main/database/inbox-projections.ts | 32 ++++++++++++ src/main/database/migrations/v46-v46.ts | 44 ++++++++++++++++ src/main/database/schema.ts | 2 + 5 files changed, 133 insertions(+), 26 deletions(-) create mode 100644 src/main/database/inbox-projections.ts create mode 100644 src/main/database/migrations/v46-v46.ts diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index 456df068..10fe1322 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -481,35 +481,49 @@ export class RuntimeBridge { // Project any pending inbox messages into the session transcript // before loading priorMessages — the recipient must observe them - // as plain user turns in its context. Append fails closed: if - // persistence throws, the inbox row stays unread and we surface - // the error so the worker aborts. + // as plain user turns in its context. + // + // The projection closure is now atomic: the ledger claim, the + // session-shadow append, the thread-event write, and the + // delivered-mark all commit inside one `BEGIN IMMEDIATE` + // transaction. Pre-v46 layout let those four writes drift apart, + // so a crash before the delivered-mark replayed the same inbox + // row on the next run and double-populated the transcript. The + // `agent_message_projections` ledger is the idempotent gate — + // `claimProjection` returns false when the `(message_id, + // session_id)` pair is already recorded, so a replay skips the + // whole closure. const store = this.getOrCreateStore(payload.taskSessionId); const inboxRecords = this.deps.db.listAgentInboxUnread(payload.companyId, payload.agentId); - for (const record of inboxRecords) { - const projected = projectInboxMessageToPi(record, { - getAgentName: (id) => this.resolveAgentName(id), - }); - store.appendMessage(projected); - this.registerInboundInjection(runId, record); - // priorMessages do not fire `message_end`, so write the thread - // event directly. Steer-injected messages (mid-run) go through - // `recordThreadEvent` via the pi event loop instead. - this.recordInboundMessageAsThreadEvent( - { - runId, - agentId: payload.agentId, - companyId: payload.companyId, - taskId: payload.taskId, - }, - record, - ); - } + const acceptedRecords: AgentMessageRecord[] = []; if (inboxRecords.length > 0) { - this.deps.db.markAgentInboxDelivered( - payload.agentId, - inboxRecords.map((r) => r.id), - ); + this.deps.db.withTransaction(() => { + for (const record of inboxRecords) { + const claimed = this.deps.db.inboxProjections.claimProjection(record.id, payload.taskSessionId); + if (!claimed) continue; + const projected = projectInboxMessageToPi(record, { + getAgentName: (id) => this.resolveAgentName(id), + }); + store.appendMessage(projected); + this.recordInboundMessageAsThreadEvent( + { + runId, + agentId: payload.agentId, + companyId: payload.companyId, + taskId: payload.taskId, + }, + record, + ); + acceptedRecords.push(record); + } + this.deps.db.markAgentInboxDelivered( + payload.agentId, + inboxRecords.map((r) => r.id), + ); + }); + } + for (const record of acceptedRecords) { + this.registerInboundInjection(runId, record); } const priorMessages = store.loadMessages(); diff --git a/src/main/database-base.ts b/src/main/database-base.ts index d8a38392..a4197082 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -23,6 +23,7 @@ import { buildKnowledgeMethods } from "./database/knowledge"; import { buildWakeupMethods } from "./database/wakeup"; import { buildBackupMethods } from "./database/backup"; import { buildSessionEntryMethods } from "./database/session-entries"; +import { buildInboxProjectionMethods } from "./database/inbox-projections"; import { buildCredentialMethods } from "./database/credentials"; import { buildModelCapabilityMethods } from "./database/model-capabilities-crud"; import { MODEL_CAPABILITY_SEED } from "./model-capability-seed"; @@ -69,6 +70,7 @@ export class AppDatabaseBase { public readonly wakeup: ReturnType; public readonly backupMod: ReturnType; public readonly sessionEntries: ReturnType; + public readonly inboxProjections: ReturnType; public readonly credentialsTable: ReturnType; public readonly modelCapabilities: ReturnType; @@ -143,6 +145,7 @@ export class AppDatabaseBase { this.knowledgeMod = buildKnowledgeMethods(getDb, ra); this.wakeup = buildWakeupMethods(getDb, ra, tx); this.sessionEntries = buildSessionEntryMethods(getDb, ra, tx); + this.inboxProjections = buildInboxProjectionMethods(getDb); this.credentialsTable = buildCredentialMethods(getDb, ra, tx); this.modelCapabilities = buildModelCapabilityMethods(getDb, ra, tx); this.backupMod = buildBackupMethods({ @@ -443,6 +446,18 @@ export class AppDatabaseBase { } } + /** + * Run `operation` inside a `BEGIN IMMEDIATE` transaction. Public + * surface for callers outside the database module that need to bind + * multiple writes into one atomic unit — the inbox projection closure + * is the canonical example, where the append to the session shadow, + * the thread event, and the delivered-mark all have to commit + * together or roll back together. + */ + withTransaction(operation: () => T): T { + return this.transaction(operation); + } + // ─── Methods used by subclass constructor callbacks ──────────────────────── addActivity(input: { companyId: string; actor: string; action: string; entityType: string; entityId: string; detail: string }) { this.activityMod.addActivity(input); } diff --git a/src/main/database/inbox-projections.ts b/src/main/database/inbox-projections.ts new file mode 100644 index 00000000..8fde4b46 --- /dev/null +++ b/src/main/database/inbox-projections.ts @@ -0,0 +1,32 @@ +import type { DatabaseSync } from "node:sqlite"; +import { nowIso } from "./helpers"; + +/** + * Ledger accessor for `agent_message_projections`. + * + * The projection path (inbox → session shadow) used to assume + * `markAgentInboxDelivered` acted as its completion mark, but that + * write lived outside the transaction that appended to the shadow + * and wrote the thread event. A crash in that window produced a + * replay that double-inserted both. The ledger is the idempotent + * gate: a single `INSERT OR IGNORE` per `(message_id, session_id)` + * short-circuits the rest of the projection closure when the pair + * has already committed. + */ +export function buildInboxProjectionMethods(getDb: () => DatabaseSync) { + return { + /** + * Attempt to claim a projection slot. Returns true when this call + * wrote a fresh row, false when the pair was already recorded. + * Callers run this inside the same transaction that performs the + * shadow append, the thread-event write, and the delivered mark, + * so either all four writes commit or none do. + */ + claimProjection(messageId: string, sessionId: string): boolean { + const result = getDb().prepare( + "INSERT OR IGNORE INTO agent_message_projections (message_id, session_id, projected_at) VALUES (?, ?, ?)", + ).run(messageId, sessionId, nowIso()); + return Number(result.changes ?? 0) > 0; + }, + }; +} diff --git a/src/main/database/migrations/v46-v46.ts b/src/main/database/migrations/v46-v46.ts new file mode 100644 index 00000000..26908f51 --- /dev/null +++ b/src/main/database/migrations/v46-v46.ts @@ -0,0 +1,44 @@ +import log from "electron-log/main.js"; +import type { Migration, MigrationDeps } from "./types"; + +export function createMigrationsV46V46(deps: MigrationDeps): Migration[] { + const { db } = deps; + return [ + { + // Ledger for `prepareRun` inbox → session-shadow projection. + // + // Before v46 the projection sequence — append to the session + // shadow, write the thread event, mark the inbox row delivered — + // ran as three independent SQL writes. A crash or worker restart + // between the append and the delivered mark caused the next run + // to replay the same inbox row, duplicating transcript turns and + // thread events. + // + // The ledger is keyed by the natural identity of a projection + // attempt: the combination of the inbox message and the session + // shadow it lands on. `INSERT OR IGNORE` against that key is the + // idempotent gate — the first insert wins, a replay silently + // skips. Callers wrap the ledger insert, the shadow append, the + // thread-event write, and the delivered-mark in a single + // `BEGIN IMMEDIATE` transaction so the four writes commit + // together or roll back together. + version: 46, + name: "agent_message_projections_ledger", + fn: () => { + db.exec(` + CREATE TABLE IF NOT EXISTS agent_message_projections ( + message_id TEXT NOT NULL, + session_id TEXT NOT NULL, + projected_at TEXT NOT NULL, + PRIMARY KEY (message_id, session_id) + ) + `); + db.exec(` + CREATE INDEX IF NOT EXISTS idx_agent_message_projections_session + ON agent_message_projections(session_id, projected_at) + `); + log.info("[migration v46] agent_message_projections ledger installed"); + }, + }, + ]; +} diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 88765d94..481b6538 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -10,6 +10,7 @@ import { createMigrationsV42V42 } from "./migrations/v42-v42"; 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 { columnExists, nowIso, @@ -195,6 +196,7 @@ export function runMigrations( ...createMigrationsV43V43(deps), ...createMigrationsV44V44(deps), ...createMigrationsV45V45(deps), + ...createMigrationsV46V46(deps), ]; for (const migration of migrations) {