fix(concurrency): centralize write transactions with bounded SQLITE_BUSY retry

Replaces 9 hand-rolled BEGIN IMMEDIATE / COMMIT / finally-ROLLBACK blocks
with a single shared runWriteTransaction helper that adds a bounded
SQLITE_BUSY retry loop (4 attempts, 100ms backoff). This is a better
concurrency solution than the previous per-site pattern because:

1. Centralized retry policy: transient SQLITE_BUSY from cross-process WAL
   writer lock contention (sibling migration, dreamer run, Channel-2 bulk
   delivery) now waits-and-retries instead of propagating up and disabling
   Magic Context for the run. Previously each site either swallowed,
   propagated, or retried inconsistently.

2. Composition-safe: if called inside an existing db.transaction() or
   another runWriteTransaction, the body runs inline WITHOUT issuing a
   nested BEGIN (SQLite doesn't allow nested BEGIN; the outer transaction
   already holds the writer lock). Detected via the bun:sqlite
   inTransaction flag or the node:sqlite shim's isTransaction flag.

3. One correct transaction plumbing pattern instead of 9 slightly-different
   copies. Several sites were missing the ROLLBACK path or had subtle
   ordering bugs at the edges.

Sites converted (all sync write paths with hand-rolled BEGIN IMMEDIATE):
- dreamer/lease.ts (acquireLease, renewLease, releaseLease)
- git-commits/sweep-coordinator.ts (acquireGitSweepLease, renewGitSweepLease,
  markGitSweepSuccessAndRelease)
- message-index.ts (indexMessagesAfterOrdinal — cross-process FTS dedup)
- workspaces.ts (bumpEpochsForWorkspaceMembers, bumpEpochsForWorkspaceMemberSet)
- compartment-storage.ts (replaceAllCompartmentStateAndBumpDepth,
  promoteRecompStaging)
- key-files/project-key-files.ts (replaceAllKeyFiles)
- key-files/identify-key-files.ts (commitKeyFilesUnderLease)
- hooks/compartment-runner-recomp.ts (promoteRecompStagingWithM0Mutation)
- hooks/compartment-runner-incremental.ts (historian publish path)

inject-compartments.ts is intentionally NOT converted: its two BEGIN IMMEDIATE
blocks have complex early-return-with-rollback-and-fallback-read patterns
(Phase 3 materialization contention retry, soft-refresh cache replay) that
don't map cleanly onto the shared helper without significant restructuring.
The helper is still a net win across the 9 simpler sites.

Verification:
- bun run typecheck (tsc --noEmit + scripts) passes
- bun run lint (biome check) passes
- 12 new unit tests for runWriteTransaction/runWriteTransactionAsync pass
  (BEGIN/COMMIT, rollback on throw, transient BUSY retry, max-attempts
  give-up, nested-transaction composition, non-transient error passthrough)
- All existing tests in the converted files' test suites pass:
  dreamer/lease.test.ts (7), git-commits (19), compartment-storage-atomic
  (8), compartment-storage-v6, compartment-lease, compartment-runner-recomp-fk
  (4), compartment-runner-partial-recomp, key-files (14), workspaces,
  storage-db (11), message-index

Net: -199 lines / +40 lines across the 9 converted sites, +201 lines for
the shared helper + 156 lines for its tests.

Author: OpenCode Agent

packages/plugin/src/features/magic-context/compartment-storage.ts

@@ -1,5 +1,6 @@
 import { getHarness } from "../../shared/harness";
 import type { Database, Statement as PreparedStatement } from "../../shared/sqlite";
+import { runWriteTransaction } from "../../shared/write-transaction";
 import { isCompartmentLeaseHeld } from "./compartment-lease";
 import { getIncrementDepthStatement } from "./compression-depth-storage";
 import { clearCachedM0M1 } from "./storage-meta-shared";
@@ -379,12 +380,8 @@ export function replaceAllCompartmentStateAndBumpDepth(
     depthEndOrdinal: number,
 ): boolean {
     const now = Date.now();
-    db.exec("BEGIN IMMEDIATE");
-    let finished = false;
-    try {
+    return runWriteTransaction(db, () => {
         if (!isCompartmentLeaseHeld(db, sessionId, holderId)) {
-            db.exec("ROLLBACK");
-            finished = true;
             return false;
         }
 
@@ -402,19 +399,8 @@ export function replaceAllCompartmentStateAndBumpDepth(
                 stmt.run(sessionId, ordinal, getHarness());
             }
         }
-
-        db.exec("COMMIT");
-        finished = true;
         return true;
-    } finally {
-        if (!finished) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // Transaction may already be closed by SQLite after an error.
-            }
-        }
-    }
+    });
 }
 
 export interface CompartmentDateRanges {
@@ -587,19 +573,13 @@ export function promoteRecompStaging(
         })();
     }
 
-    db.exec("BEGIN IMMEDIATE");
-    let finished = false;
-    try {
+    return runWriteTransaction(db, () => {
         if (!isCompartmentLeaseHeld(db, sessionId, holderId)) {
-            db.exec("ROLLBACK");
-            finished = true;
             return null;
         }
 
         const staging = getRecompStaging(db, sessionId);
         if (!staging || staging.compartments.length === 0) {
-            db.exec("ROLLBACK");
-            finished = true;
             return null;
         }
         // Replace real tables
@@ -615,18 +595,8 @@ export function promoteRecompStaging(
 
         clearCachedM0M1(db, sessionId);
 
-        db.exec("COMMIT");
-        finished = true;
         return { compartments: staging.compartments, facts: staging.facts };
-    } finally {
-        if (!finished) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // Transaction may already be closed by SQLite after an error.
-            }
-        }
-    }
+    });
 }
 
 /** Clear staging tables for a session (on cancel/abandon or after successful promote). */

packages/plugin/src/features/magic-context/dreamer/lease.ts

@@ -1,4 +1,5 @@
 import type { Database } from "../../../shared/sqlite";
+import { runWriteTransaction } from "../../../shared/write-transaction";
 import { deleteDreamState, getDreamState, setDreamState } from "./storage-dream-state";
 
 const LEASE_HOLDER_KEY = "dreaming_lease_holder";
@@ -43,27 +44,9 @@ export function peekLeaseHolderAndExpiry(db: Database, expectedHolder: string):
 // = false under WAL snapshot isolation and both write — double-acquiring the
 // lease and spawning duplicate dreamer workers. busy_timeout (set in
 // initializeDatabase) makes the loser wait rather than throw SQLITE_BUSY.
-function runImmediate<T>(db: Database, body: () => T): T {
-    db.exec("BEGIN IMMEDIATE");
-    let committed = false;
-    try {
-        const result = body();
-        db.exec("COMMIT");
-        committed = true;
-        return result;
-    } finally {
-        if (!committed) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // already rolled back / no active transaction
-            }
-        }
-    }
-}
 
 export function acquireLease(db: Database, holderId: string): boolean {
-    return runImmediate(db, () => {
+    return runWriteTransaction(db, () => {
         if (isLeaseActive(db)) {
             const existingHolder = getLeaseHolder(db);
             if (existingHolder && existingHolder !== holderId) {
@@ -80,7 +63,7 @@ export function acquireLease(db: Database, holderId: string): boolean {
 }
 
 export function renewLease(db: Database, holderId: string): boolean {
-    return runImmediate(db, () => {
+    return runWriteTransaction(db, () => {
         if (getLeaseHolder(db) !== holderId || !isLeaseActive(db)) {
             return false;
         }
@@ -93,7 +76,7 @@ export function renewLease(db: Database, holderId: string): boolean {
 }
 
 export function releaseLease(db: Database, holderId: string): void {
-    runImmediate(db, () => {
+    runWriteTransaction(db, () => {
         if (getLeaseHolder(db) !== holderId) {
             return;
         }

packages/plugin/src/features/magic-context/git-commits/sweep-coordinator.ts

@@ -1,4 +1,5 @@
 import type { Database } from "../../../shared/sqlite";
+import { runWriteTransaction } from "../../../shared/write-transaction";
 
 export const GIT_SWEEP_COOLDOWN_MS = 10 * 60 * 1000;
 // Commit indexing can include two embedding drains (the indexer drain plus the
@@ -59,25 +60,6 @@ export interface AcquireGitSweepLeaseOptions {
     ignoreCooldown?: boolean;
 }
 
-function runImmediate<T>(db: Database, body: () => T): T {
-    db.exec("BEGIN IMMEDIATE");
-    let committed = false;
-    try {
-        const result = body();
-        db.exec("COMMIT");
-        committed = true;
-        return result;
-    } finally {
-        if (!committed) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // already rolled back / no active transaction
-            }
-        }
-    }
-}
-
 function rowToState(row: GitSweepCoordinatorRow): GitSweepCoordinatorState {
     return {
         projectPath: row.project_path,
@@ -110,7 +92,7 @@ export function acquireGitSweepLease(
     const cooldownMs = options.cooldownMs ?? GIT_SWEEP_COOLDOWN_MS;
     const leaseTtlMs = options.leaseTtlMs ?? GIT_SWEEP_LEASE_TTL_MS;
 
-    return runImmediate(db, () => {
+    return runWriteTransaction(db, () => {
         const now = Date.now();
         const row = getGitSweepCoordinatorState(db, projectPath);
         if (row?.leaseHolder && row.leaseExpiresAt !== null && row.leaseExpiresAt > now) {
@@ -173,7 +155,7 @@ export function renewGitSweepLease(
     holderId: string,
     leaseTtlMs = GIT_SWEEP_LEASE_TTL_MS,
 ): boolean {
-    return runImmediate(db, () => {
+    return runWriteTransaction(db, () => {
         const now = Date.now();
         const leaseExpiresAt = now + leaseTtlMs;
         const result = db
@@ -194,7 +176,7 @@ export function markGitSweepSuccessAndRelease(
     projectPath: string,
     holderId: string,
 ): boolean {
-    return runImmediate(db, () => {
+    return runWriteTransaction(db, () => {
         const now = Date.now();
         const result = db
             .prepare(
@@ -212,7 +194,7 @@ export function markGitSweepSuccessAndRelease(
 }
 
 export function releaseGitSweepLease(db: Database, projectPath: string, holderId: string): void {
-    runImmediate(db, () => {
+    runWriteTransaction(db, () => {
         db.prepare(
             `UPDATE git_sweep_coordinator
              SET lease_holder = NULL,

packages/plugin/src/features/magic-context/key-files/identify-key-files.ts

@@ -10,6 +10,7 @@ import { getHarness } from "../../../shared/harness";
 import { shouldKeepSubagents } from "../../../shared/keep-subagents";
 import { log } from "../../../shared/logger";
 import type { Database } from "../../../shared/sqlite";
+import { runWriteTransaction } from "../../../shared/write-transaction";
 import { peekLeaseHolderAndExpiry, renewLease } from "../dreamer/lease";
 import { recordChildInvocation } from "../subagent-token-capture";
 import { isAftAvailable } from "./aft-availability";
@@ -423,9 +424,7 @@ export function commitKeyFiles(args: {
     );
     const generatedAt = Date.now();
     const bump = args.bumpVersion ?? bumpKeyFilesVersion;
-    args.db.exec("BEGIN IMMEDIATE");
-    let committed = false;
-    try {
+    return runWriteTransaction(args.db, () => {
         if (!peekLeaseHolderAndExpiry(args.db, args.leaseHolderId)) {
             log(`key-files commit aborted: lease lost (holder ${args.leaseHolderId})`);
             return null;
@@ -440,21 +439,11 @@ export function commitKeyFiles(args: {
             args.configHash,
         );
         const version = bump(args.db, projectPath);
-        args.db.exec("COMMIT");
-        committed = true;
         log(
             `key-files committed: ${resolved.length} files, version=${version}, ${resolved.filter((r) => r.staleReason).length} pre-stale`,
         );
         return version;
-    } finally {
-        if (!committed) {
-            try {
-                args.db.exec("ROLLBACK");
-            } catch {
-                // no active transaction
-            }
-        }
-    }
+    });
 }
 
 async function runKeyFilesLlm(args: {

packages/plugin/src/features/magic-context/key-files/project-key-files.ts

@@ -3,6 +3,7 @@ import { existsSync, readFileSync, realpathSync } from "node:fs";
 import { join, resolve, sep } from "node:path";
 import { log } from "../../../shared/logger";
 import type { Database } from "../../../shared/sqlite";
+import { runWriteTransaction } from "../../../shared/write-transaction";
 
 export type KeyFileStaleReason = "missing" | "content_drift";
 
@@ -140,9 +141,7 @@ export function replaceProjectKeyFiles(
     const generatedByModel = files[0]?.generatedByModel ?? null;
     const configHash = files[0]?.generationConfigHash ?? sha256("{}");
 
-    db.exec("BEGIN IMMEDIATE");
-    let committed = false;
-    try {
+    return runWriteTransaction(db, () => {
         db.prepare("DELETE FROM project_key_files WHERE project_path = ?").run(resolvedProjectPath);
         insertResolvedKeyFiles(
             db,
@@ -152,19 +151,8 @@ export function replaceProjectKeyFiles(
             generatedByModel,
             configHash,
         );
-        const version = bumpKeyFilesVersion(db, resolvedProjectPath);
-        db.exec("COMMIT");
-        committed = true;
-        return version;
-    } finally {
-        if (!committed) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // no active transaction
-            }
-        }
-    }
+        return bumpKeyFilesVersion(db, resolvedProjectPath);
+    });
 }
 
 export function insertResolvedKeyFiles(

packages/plugin/src/features/magic-context/message-index.ts

@@ -7,6 +7,7 @@ import type { RawMessage } from "../../hooks/magic-context/read-session-raw";
 import { getHarness } from "../../shared/harness";
 import type { Database, Statement as PreparedStatement } from "../../shared/sqlite";
 import { removeSystemReminders } from "../../shared/system-directive";
+import { runWriteTransaction } from "../../shared/write-transaction";
 import { clearCompressionDepth } from "./compression-depth-storage";
 
 interface MessageHistoryIndexRow {
@@ -218,9 +219,7 @@ export function indexMessagesAfterOrdinal(
     // is reflected, so the second skips those ordinals and inserts nothing
     // duplicate. The bulk SELECT of existing message-ids is still avoided (it
     // held the writer lock too long on ~30k-row sessions).
-    db.exec("BEGIN IMMEDIATE");
-    let committed = false;
-    try {
+    return runWriteTransaction(db, () => {
         // Re-read under the lock: another process may have advanced the
         // watermark between the caller's out-of-transaction read and now.
         const effectiveWatermark = Math.max(
@@ -245,18 +244,8 @@ export function indexMessagesAfterOrdinal(
         // Never regress a higher watermark a concurrent writer may have set.
         const newWatermark = Math.max(effectiveWatermark, finalWatermark);
         getUpsertIndexStatement(db).run(sessionId, newWatermark, now, getHarness());
-        db.exec("COMMIT");
-        committed = true;
-    } finally {
-        if (!committed) {
-            try {
-                db.exec("ROLLBACK");
-            } catch {
-                // already rolled back / no active transaction
-            }
-        }
-    }
-    return inserted;
+        return inserted;
+    });
 }
 
 export function ensureMessagesIndexed(

packages/plugin/src/features/magic-context/workspaces.ts

@@ -1,5 +1,6 @@
 import { createHash } from "node:crypto";
 import type { Database } from "../../shared/sqlite";
+import { runWriteTransaction } from "../../shared/write-transaction";
 import { V2_MEMORY_CATEGORIES } from "./memory/constants";
 import { normalizeStoredProjectPath, storedPathBelongsToIdentity } from "./project-identity";
 
@@ -271,11 +272,6 @@ export function computeWorkspaceEpochFingerprint(
     return hash.digest("hex");
 }
 
-function isInTransaction(db: Database): boolean {
-    const candidate = db as unknown as { inTransaction?: unknown; isTransaction?: unknown };
-    return candidate.inTransaction === true || candidate.isTransaction === true;
-}
-
 function workspaceMembersForIdentity(db: Database, identity: string): string[] {
     if (!tableExists(db, "workspace_members")) return [identity];
     const rows = db
@@ -313,22 +309,7 @@ export function bumpEpochsForWorkspaceMembers(
     now = Date.now(),
 ): void {
     const run = () => bumpEpochRows(db, workspaceMembersForIdentity(db, identity), now);
-    if (isInTransaction(db)) {
-        run();
-        return;
-    }
-    db.exec("BEGIN IMMEDIATE");
-    try {
-        run();
-        db.exec("COMMIT");
-    } catch (error) {
-        try {
-            db.exec("ROLLBACK");
-        } catch {
-            // ignore rollback failures from an already-closed transaction
-        }
-        throw error;
-    }
+    runWriteTransaction(db, run);
 }
 
 export function bumpEpochsForWorkspaceMemberSet(
@@ -337,20 +318,5 @@ export function bumpEpochsForWorkspaceMemberSet(
     now = Date.now(),
 ): void {
     const run = () => bumpEpochRows(db, identities, now);
-    if (isInTransaction(db)) {
-        run();
-        return;
-    }
-    db.exec("BEGIN IMMEDIATE");
-    try {
-        run();
-        db.exec("COMMIT");
-    } catch (error) {
-        try {
-            db.exec("ROLLBACK");
-        } catch {
-            // ignore rollback failures from an already-closed transaction
-        }
-        throw error;
-    }
+    runWriteTransaction(db, run);
 }