Close inbox projection under one transaction

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

@@ -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();

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<typeof buildWakeupMethods>;
   public readonly backupMod: ReturnType<typeof buildBackupMethods>;
   public readonly sessionEntries: ReturnType<typeof buildSessionEntryMethods>;
+  public readonly inboxProjections: ReturnType<typeof buildInboxProjectionMethods>;
   public readonly credentialsTable: ReturnType<typeof buildCredentialMethods>;
   public readonly modelCapabilities: ReturnType<typeof buildModelCapabilityMethods>;
 
@@ -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<T>(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); }

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;
+    },
+  };
+}

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");
+      },
+    },
+  ];
+}

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) {