fix: wrap the cost-store save cycle in one transaction so a crash mid-save is all-or-nothing (refs #2760)

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Fixed
 - OpenRouter: keep optional key-quota enrichment on its one-second production fast join while making degraded results explicit and preventing loaded CI parity runs from mistaking the fallback snapshot for a golden mismatch (fixes #2778).
+- Codex: the SQLite cost store now writes each save cycle inside one transaction, so a crash or kill mid-save can never leave session rows updated against stale day aggregates — the previous state survives intact, matching the old JSON path's atomic file replace (refs #2760).
 - Codex: SQLite cost saves no longer rescan every stored row and snapshot per file — baseline counts and file lookups are precomputed once, cutting a large-corpus (1,700+ sessions) save pass from minutes of CPU to seconds (refs #2760).
 - Codex: restore JSON-cache retention semantics lost in the SQLite cutover — discovery pruning now reaches the scanner's round-tripped payload so deleted files stop resurfacing, the row budget never sacrifices in-window or recently active sessions, and fork-parent protection again drops stale lineage-only parents (refs #2760).
 - Codex: the SQLite cost store no longer deletes the whole database on transient failures — lock contention from a concurrent CLI/app writer, disk-full, or a constraint violation now preserve history and only genuine corruption or schema drift triggers a rebuild, which is now logged (refs #2760).

Package.swift

@@ -102,6 +102,16 @@ let package = Package(
                     .enableUpcomingFeature("StrictConcurrency"),
                 ],
                 linkerSettings: sqlite3LinkerSettings),
+            // Crash-test subprocess: tests SIGKILL it mid-save to prove the cost store's
+            // save cycle is atomic. Not shipped; built only as a test dependency.
+            .executableTarget(
+                name: "CodexBarCostStoreCrashProbe",
+                dependencies: ["CodexBarCore"],
+                path: "Sources/CodexBarCostStoreCrashProbe",
+                swiftSettings: [
+                    .enableUpcomingFeature("StrictConcurrency"),
+                ],
+                linkerSettings: sqlite3LinkerSettings),
             // Sole owner of the adaptive refresh decision table. Package-internal so the app and
             // offline replay tool share behavior without publishing another library product.
             .target(
@@ -211,7 +221,7 @@ let package = Package(
 
         targets.append(.testTarget(
             name: "CodexBarTests",
-            dependencies: ["CodexBar", "CodexBarCore", "CodexBarCLI", "CodexBarWidget"],
+            dependencies: ["CodexBar", "CodexBarCore", "CodexBarCLI", "CodexBarCostStoreCrashProbe", "CodexBarWidget"],
             path: "Tests",
             exclude: [
                 "AdaptiveReplayCLITests",

Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+CodexCache.swift

@@ -24,11 +24,17 @@ extension CostUsageStore {
     {
         let previous = self.readSnapshot()
         let canReuseStoredRows = previous.metadata.timeZoneIdentifier == calendar.timeZone.identifier
-        self.deleteRemovedFiles(previous: previous, cache: cache)
         let previousFilesByPath = Dictionary(uniqueKeysWithValues: previous.files.map { ($0.path, $0) })
         let snapshotCountsByPath = previous.tokenSnapshots
             .reduce(into: [String: Int]()) { $0[$1.path, default: 0] += 1 }
         let rowCountsByPath = previous.usageRows.reduce(into: [String: Int]()) { $0[$1.path, default: 0] += 1 }
+        // One transaction spans every table the save cycle touches, so a crash or failure
+        // midway can never leave e.g. files upserted while day_aggregates stay stale.
+        // Budget enforcement below runs outside: it checkpoints the WAL and vacuums, which
+        // SQLite forbids inside an open transaction.
+        self.beginSaveTransaction()
+        self.deleteRemovedFiles(previous: previous, cache: cache)
+        var persistedFiles = 0
         for (path, usage) in cache.files.sorted(by: { $0.key < $1.key }) {
             self.persistFile(
                 path: path,
@@ -39,11 +45,14 @@ extension CostUsageStore {
                     rowCount: rowCountsByPath[path] ?? 0,
                     canReuseRows: canReuseStoredRows),
                 calendar: calendar)
+            persistedFiles += 1
+            Self.saveCycleCheckpointForTesting?(persistedFiles)
         }
         _ = self.replaceDayAggregates(Self.globalAggregates(cache: cache))
         _ = self.setMetadata(Self.metadata(cache: cache, calendar: calendar))
         _ = self.setDiscoveryState(Self.discoveryState(cache.codexSessionDiscovery))
         _ = self.setLookbackState(Self.lookbackState(cache.codexActiveLookbackState))
+        self.endSaveTransaction()
         let result = self.enforceBudgets(
             maxRows: rowBudget,
             maxFileBytes: fileBudgetBytes,

Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Writes.swift

@@ -472,6 +472,12 @@ extension CostUsageStore {
     }
 
     static func inTransaction<T>(_ database: OpaquePointer, _ operation: () throws -> T) throws -> T {
+        // Flatten when a transaction is already open (the save cycle's outer BEGIN
+        // IMMEDIATE): SQLite has no nested transactions, and the outer scope owns
+        // commit/rollback for every write made inside it.
+        guard sqlite3_get_autocommit(database) != 0 else {
+            return try operation()
+        }
         try self.execute(database, "BEGIN IMMEDIATE")
         do {
             let value = try operation()

Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift

@@ -61,6 +61,11 @@ actor CostUsageStore {
         parserHash: CodexParserHash.value)
     static let cacheGeneration = "sqlite:\(CostUsageStore.schemaVersion)"
 
+    /// Test-only crash injection: invoked inside `saveCodexCache`'s transaction after each
+    /// persisted file with the running count, so a crash-safety harness can SIGKILL the
+    /// process at a deterministic mid-save point. Never set in production.
+    nonisolated(unsafe) static var saveCycleCheckpointForTesting: ((Int) -> Void)?
+
     /// Process-wide serialization keeps every writable store connection on the same queue.
     /// This matches the scan pipeline's single-writer contract without multiplying executor
     /// threads when tests or short-lived readers create several store actors.
@@ -75,6 +80,11 @@ actor CostUsageStore {
     private let expectedParserHash: String
     private var connection: SQLiteConnection?
     private(set) var rebuildCount = 0
+    /// While a save cycle's enclosing transaction is open, nested `withDatabase` calls join
+    /// it instead of opening their own connection scope, and the first failure aborts the
+    /// remainder of the cycle so the outer transaction rolls back as a unit.
+    private var activeTransactionDatabase: OpaquePointer?
+    private var activeTransactionError: Error?
 
     init(
         cacheRoot: URL? = nil,
@@ -142,6 +152,18 @@ extension CostUsageStore {
     }
 
     func withDatabase<T>(default fallback: T, _ operation: (OpaquePointer) throws -> T) -> T {
+        if let database = self.activeTransactionDatabase {
+            // A failed statement may have aborted the enclosing transaction; running the
+            // remaining writes would commit them individually in autocommit mode, which is
+            // exactly the partial-save state the transaction exists to prevent. Skip them.
+            guard self.activeTransactionError == nil else { return fallback }
+            do {
+                return try operation(database)
+            } catch {
+                self.activeTransactionError = error
+                return fallback
+            }
+        }
         do {
             let database = try self.ensureDatabase()
             return try operation(database)
@@ -167,6 +189,39 @@ extension CostUsageStore {
         }
     }
 
+    /// Opens the save cycle's all-or-nothing transaction: a single BEGIN IMMEDIATE spanning
+    /// every store call made until `endSaveTransaction()`. Nested `withDatabase` calls join
+    /// the open transaction and the first inner failure aborts the cycle, so a crash or
+    /// error midway leaves the previous on-disk state fully intact — matching the old JSON
+    /// path's atomic single-file replace. If the transaction cannot open, subsequent writes
+    /// proceed unprotected, exactly like the pre-transaction behavior.
+    func beginSaveTransaction() {
+        _ = self.withDatabase(default: false) { database in
+            try Self.execute(database, "BEGIN IMMEDIATE")
+            self.activeTransactionDatabase = database
+            return true
+        }
+    }
+
+    /// Commits the open save transaction, or rolls it back when a nested write failed. The
+    /// stored failure is rethrown through `withDatabase` so the usual rebuild-vs-preserve
+    /// classification still applies to it.
+    @discardableResult
+    func endSaveTransaction() -> Bool {
+        guard self.activeTransactionDatabase != nil else { return false }
+        let failure = self.activeTransactionError
+        self.activeTransactionDatabase = nil
+        self.activeTransactionError = nil
+        return self.withDatabase(default: false) { database in
+            if let failure {
+                try? Self.execute(database, "ROLLBACK")
+                throw failure
+            }
+            try Self.execute(database, "COMMIT")
+            return true
+        }
+    }
+
     /// Destroying the database is only the right recovery for corruption or schema drift.
     /// Transient and data-shape failures (lock contention from a second process, disk full,
     /// out of memory, a constraint violation from bad input) must not delete user history:

Sources/CodexBarCore/Vendored/CostUsage/CostUsageStoreCrashHarness.swift

@@ -0,0 +1,86 @@
+import Foundation
+
+#if canImport(Darwin)
+import Darwin
+#elseif canImport(Glibc)
+import Glibc
+#endif
+
+/// Crash-safety harness for the SQLite cost store. The `CodexBarCostStoreCrashProbe`
+/// executable drives these entry points so tests can SIGKILL a real process at a
+/// deterministic point inside `saveCodexCache` and then prove, from a fresh process, that
+/// the interrupted save left no partial state behind. Fixtures live here (not in the test
+/// target) so the probe subprocess and the asserting test share one source of truth.
+package enum CostUsageStoreCrashHarness {
+    static let scanWindow = (sinceKey: "0000-01-01", untilKey: "9999-12-31")
+
+    /// Both processes must agree on the calendar: `loadCodexCache` discards the cache when
+    /// the stored time zone differs from the caller's.
+    static var fixtureCalendar: Calendar {
+        var calendar = Calendar(identifier: .gregorian)
+        calendar.timeZone = TimeZone(identifier: "UTC")!
+        return calendar
+    }
+
+    static func seededCache() -> CostUsageCache {
+        var cache = CostUsageCache()
+        cache.files = [
+            "/sessions/a.jsonl": self.usage(session: "session-a", input: 1),
+            "/sessions/b.jsonl": self.usage(session: "session-b", input: 2),
+            "/sessions/c.jsonl": self.usage(session: "session-c", input: 3),
+        ]
+        cache.days = ["2026-08-01": ["gpt-5.6-sol": [6, 0, 0]]]
+        return cache
+    }
+
+    /// Differs from the seed in every way a torn save could mix: changed tallies for kept
+    /// files, one file removed, one added, and new global day totals.
+    static func updatedCache() -> CostUsageCache {
+        var cache = CostUsageCache()
+        cache.files = [
+            "/sessions/a.jsonl": self.usage(session: "session-a", input: 10),
+            "/sessions/b.jsonl": self.usage(session: "session-b", input: 20),
+            "/sessions/d.jsonl": self.usage(session: "session-d", input: 40),
+        ]
+        cache.days = ["2026-08-01": ["gpt-5.6-sol": [70, 0, 0]]]
+        return cache
+    }
+
+    @discardableResult
+    package static func seed(cacheRoot: URL) -> Bool {
+        self.save(self.seededCache(), cacheRoot: cacheRoot, killAfterFiles: nil)
+    }
+
+    /// Saves the updated fixture. With `killAfterFiles` set, the process raises SIGKILL
+    /// inside the save transaction once that many files have been persisted — after real
+    /// table writes have been issued, before the cycle's aggregates and metadata.
+    @discardableResult
+    package static func saveUpdate(cacheRoot: URL, killAfterFiles: Int?) -> Bool {
+        self.save(self.updatedCache(), cacheRoot: cacheRoot, killAfterFiles: killAfterFiles)
+    }
+
+    private static func save(_ cache: CostUsageCache, cacheRoot: URL, killAfterFiles: Int?) -> Bool {
+        if let killAfterFiles {
+            CostUsageStore.saveCycleCheckpointForTesting = { persistedFiles in
+                if persistedFiles >= killAfterFiles {
+                    kill(getpid(), SIGKILL)
+                }
+            }
+        }
+        let store = CostUsageStore(cacheRoot: cacheRoot)
+        _ = store.syncSaveCodexCache(
+            cache,
+            calendar: self.fixtureCalendar,
+            requestedScanWindow: self.scanWindow)
+        return true
+    }
+
+    private static func usage(session: String, input: Int) -> CostUsageFileUsage {
+        var usage = CostUsageFileUsage(
+            mtimeUnixMs: 1000,
+            size: 100,
+            days: ["2026-08-01": ["gpt-5.6-sol": [input, 0, 0]]])
+        usage.sessionId = session
+        return usage
+    }
+}

Sources/CodexBarCostStoreCrashProbe/main.swift

@@ -0,0 +1,31 @@
+import CodexBarCore
+import Foundation
+
+// Subprocess driver for CostUsageStoreCrashSafetyTests: the test SIGKILLs this process
+// mid-save to prove the store's save cycle is all-or-nothing.
+//
+// Usage: CodexBarCostStoreCrashProbe <seed|save|crash-save> <cacheRoot> [killAfterFiles]
+
+let arguments = CommandLine.arguments
+guard arguments.count >= 3 else {
+    FileHandle.standardError.write(Data("usage: <seed|save|crash-save> <cacheRoot> [killAfterFiles]\n".utf8))
+    exit(64)
+}
+
+let cacheRoot = URL(fileURLWithPath: arguments[2], isDirectory: true)
+switch arguments[1] {
+case "seed":
+    CostUsageStoreCrashHarness.seed(cacheRoot: cacheRoot)
+    exit(0)
+case "save":
+    CostUsageStoreCrashHarness.saveUpdate(cacheRoot: cacheRoot, killAfterFiles: nil)
+    exit(0)
+case "crash-save":
+    let killAfterFiles = arguments.count > 3 ? Int(arguments[3]) : 1
+    CostUsageStoreCrashHarness.saveUpdate(cacheRoot: cacheRoot, killAfterFiles: killAfterFiles)
+    // The checkpoint hook must have killed the process during the save.
+    FileHandle.standardError.write(Data("crash-save survived the save cycle\n".utf8))
+    exit(70)
+default:
+    exit(64)
+}

Tests/CodexBarTests/CLIServeRouterTests.swift

@@ -570,7 +570,7 @@ struct CLIServeRouterTests {
         {
             _ = await counter.increment()
             do {
-                try await Task.sleep(nanoseconds: 200_000_000)
+                try await Task.sleep(nanoseconds: 5_000_000_000)
                 return Self.response("[{\"provider\":\"codex\",\"call\":1}]")
             } catch {
                 return CodexBarCLI.serveTimeoutResponse()
@@ -633,7 +633,7 @@ struct CLIServeRouterTests {
                         requestTimeout: 0.01)
                     {
                         _ = await counter.increment()
-                        try? await Task.sleep(nanoseconds: 200_000_000)
+                        try? await Task.sleep(nanoseconds: 5_000_000_000)
                         return Self.response("[{\"provider\":\"codex\"}]")
                     }
                 }
@@ -727,7 +727,7 @@ struct CLIServeRouterTests {
             requestTimeout: 0.01)
         {
             _ = await counter.increment()
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response("[{\"provider\":\"codex\",\"call\":2}]")
         }
 
@@ -785,7 +785,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.01,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response(#"[{"provider":"codex","call":3}]"#)
         }
         let timeoutRows = try Self.jsonRows(timedOut)
@@ -929,7 +929,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.05,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response("[]")
         }
         #expect(timedOut.status == .gatewayTimeout)
@@ -972,7 +972,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.05,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response("[]")
         }
         #expect(timedOut.status == .gatewayTimeout)
@@ -1258,7 +1258,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.05,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response(
                 #"[{"provider":"codex","account":"shared","call":3}]"#,
                 usageCacheKeys: ["account-2"])
@@ -1388,7 +1388,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.05,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response(
                 #"[{"provider":"antigravity","account":"second@example.com","call":2}]"#,
                 usageCacheKeys: [nil])
@@ -1426,7 +1426,7 @@ struct CLIServeRouterTests {
             refreshInterval: 0.05,
             requestTimeout: 0.01)
         {
-            try? await Task.sleep(nanoseconds: 200_000_000)
+            try? await Task.sleep(nanoseconds: 5_000_000_000)
             return Self.response("[]", usageCacheKeys: [])
         }
         #expect(timedOut.status == .gatewayTimeout)