From 4739dfa585ecdb1bc51070f617e20d470256d5df Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 20:42:13 -0700 Subject: [PATCH 1/7] Price OpenCodex usage once per entry and stop per-entry catalog/overlay reloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OpenCodex spend source (`~/.opencodex/usage.jsonl` → `OpenCodexUsageFanOut` → `OpenCodexUsageAggregator.snapshot`) re-resolved pricing context per entry: `listPriceUSD` called `CostUsagePricing.codexCostUSD` without a pre-resolved models.dev catalog, so every call went through `ModelsDevCache.load` → `FileManager.attributesOfItem` (a stat plus an extended-attribute read), and without a pre-resolved custom-pricing overlay, so every call also re-read the overlay file location. Each windowed entry was priced three times (day, session and hour accumulators), and day keys / hour buckets were recomputed through Calendar per entry. On a 35k-entry log (all inside the 30-day window) that is ~100k stat+xattr syscalls and ~70k Calendar interval computations per refresh — in the running app this was the 25–35 s CPU spike on every adaptive refresh (sampled: `snapshotsBySubscription` → `attributesOfItem` → `getxattr`/`listxattr`). Changes (snapshot output is byte-identical; verified against a reference implementation in tests and by diffing CLI JSON on frozen inputs): - Resolve the models.dev catalog and the custom-pricing overlay once per fan-out / snapshot and pass them down; price each windowed entry once and reuse the value for the day/session/hour/model merges. A missing catalog is substituted with an empty catalog so the degraded path never falls back to per-call loads. - Memoize the local-day key and hour-bucket start per calendar interval using the calendar's own `[start, end)` intervals (DST-correct; no 86400/3600 arithmetic). - `ModelsDevCache.load` reads (mtime, size) via POSIX `stat` instead of `attributesOfItem` (which also reads xattrs); memo/invalidation semantics unchanged. This helps every caller repo-wide. CodexParserHash is regenerated because ModelsDevPricing.swift is in the hashed set; the previous hash (3c984b655688593f) is added to compatiblePredecessorParserHashes since parsing and the persisted row shape are unchanged, so existing cost-usage.sqlite stores are adopted on upgrade instead of rebuilt. Measured (release CodexBarCLI, isolated cache root, real 41.7 MB / ~35k-entry usage.jsonl, same machine, `cost --provider codex --days 30`), OpenCodex path isolated with identical frozen inputs: - OpenCodex path alone (empty codex home, identical frozen inputs, CLI JSON output identical apart from `updatedAt`): cold 14.3 s real / 9.1 s user / 4.9 s sys / 193 G instructions → 2.6 s / 2.4 s / 0.1 s / 40 G warm (store cache hit) 13.8 s / 8.3 s / 5.3 s / 166 G → 1.2 s / 1.1 s / 0.04 s / 13 G - Full `cost --provider codex` CLI run on live data, steady state after the log grew (the app's per-refresh case): ~11 s → ~3.5 s real (7.3–9.2 s → 3.2 s user); cold 26 s → 14 s. Peak footprint unchanged (~430 MB cold/grown, ~120–140 MB warm). Peak memory is unchanged — the remaining transient is the append-only log re-parse (`OpenCodexUsageStore` identity = path|size|mtime), left for a follow-up. Tests: equivalence against an independent reference implementation (mixed providers, estimated/reported/unreported/unsupported, custom overlay, duplicate request IDs, DST transitions in America/Los_Angeles and America/Santiago), metadata-read counting proving one catalog load per snapshot (zero with an injected catalog), day/hour memo boundary cases, and ModelsDevCache memo invalidation on size/mtime change after the stat switch. Implemented by grok-4.6 (xhigh) via implementation-loop; reviewed hunk by hunk plus an independent deep review; one iterate round. Co-Authored-By: Claude Fable 5 --- .../Generated/CodexParserHash.generated.swift | 2 +- .../Vendored/CostUsage/CostUsageStore.swift | 2 + .../Vendored/CostUsage/ModelsDevPricing.swift | 69 +- .../OpenCodexUsageAggregator.swift | 103 ++- .../OpenCodexUsage/OpenCodexUsageFanOut.swift | 7 +- Tests/CodexBarTests/CostUsageStoreTests.swift | 1 + .../CodexBarTests/ModelsDevPricingTests.swift | 54 ++ .../OpenCodexUsageFanOutTests.swift | 678 +++++++++++++++++- .../ProviderArchitectureGatekeeperTests.swift | 2 +- 9 files changed, 892 insertions(+), 26 deletions(-) diff --git a/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift b/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift index db7b5fdd77..cf3cb01d99 100644 --- a/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift +++ b/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift @@ -1,5 +1,5 @@ // Generated by Scripts/regenerate-codex-parser-hash.sh. Do not edit by hand. enum CodexParserHash { - static let value = "3c984b655688593f" + static let value = "fc4cae716e9066c5" } diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift index e64ba4a105..a432ce41ba 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift @@ -82,6 +82,8 @@ actor CostUsageStore { "b975eb705f905b9a", // 0.49.0-0.49.2 SQLite producer with compatible rows. "47144baa8daccf52", // This branch changes only scan scheduling, discovery, and persistence bookkeeping. "2d17f4981b78d07f", // Persisted priority-turn cursor; parser and persisted row shape unchanged. + "3c984b655688593f", // OpenCodex fan-out price-once/bucket memos + cheap models.dev cache stat; + // parser and persisted row shape unchanged. ] /// Test-only crash injection: invoked inside `saveCodexCache`'s transaction after each diff --git a/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift b/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift index 6915086291..6e3946ab21 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift @@ -1,4 +1,9 @@ import Foundation +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#endif #if canImport(FoundationNetworking) import FoundationNetworking #endif @@ -427,14 +432,68 @@ enum ModelsDevCache { static let ttlSeconds: TimeInterval = 24 * 60 * 60 private static let memo = ModelsDevCacheMemo() + @TaskLocal private static var metadataReadRecorder: MetadataReadRecorder? + + final class MetadataReadRecorder: @unchecked Sendable { + private let lock = NSLock() + private var count = 0 + + func record() { + self.lock.lock() + self.count += 1 + self.lock.unlock() + } + + func snapshot() -> Int { + self.lock.lock() + defer { self.lock.unlock() } + return self.count + } + } + static func withMetadataReadRecorderForTesting( + _ recorder: MetadataReadRecorder, + operation: () throws -> T) rethrows -> T + { + try self.$metadataReadRecorder.withValue(recorder) { + try operation() + } + } + + static func withMetadataReadRecorderForTesting( + _ recorder: MetadataReadRecorder, + operation: () async throws -> T) async rethrows -> T + { + try await self.$metadataReadRecorder.withValue(recorder) { + try await operation() + } + } + + /// Cheap POSIX stat for the (mtime, size) memo key. `attributesOfItem` also reads xattrs. + /// `stat(2)` follows a terminal symlink (matching what `Data(contentsOf:)` later reads) whereas + /// `attributesOfItem` did not. private static func fileMetadata(at url: URL) -> (modificationDate: Date?, size: Int?) { - guard let attributes = try? FileManager.default.attributesOfItem(atPath: url.path) else { - return (nil, nil) + self.metadataReadRecorder?.record() + + return url.withUnsafeFileSystemRepresentation { pointer in + guard let pointer else { return (nil, nil) } + var status = stat() + guard stat(pointer, &status) == 0 else { + return (nil, nil) + } + return (Self.modificationDate(from: status), Int(status.st_size)) } - let modificationDate = attributes[.modificationDate] as? Date - let size = (attributes[.size] as? NSNumber)?.intValue - return (modificationDate, size) + } + + private static func modificationDate(from status: stat) -> Date { + #if canImport(Darwin) + let seconds = TimeInterval(status.st_mtimespec.tv_sec) + let nanoseconds = TimeInterval(status.st_mtimespec.tv_nsec) + #else + let seconds = TimeInterval(status.st_mtim.tv_sec) + let nanoseconds = TimeInterval(status.st_mtim.tv_nsec) + #endif + return Date(timeIntervalSince1970: seconds + nanoseconds / 1_000_000_000) } private static func defaultCacheRoot() -> URL { diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift index 05ab5645c0..c6e23951bf 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift @@ -59,7 +59,9 @@ enum OpenCodexUsageAggregator { now: Date, historyDays: Int, calendar: Calendar, - customPricing: CostUsageCustomPricing = .empty) -> CostUsageTokenSnapshot + customPricing: CostUsageCustomPricing = .empty, + modelsDevCatalog: ModelsDevCatalog? = nil, + customPricingOverlay: CostUsageCustomPricing? = nil) -> CostUsageTokenSnapshot { let days = max(1, min(365, historyDays)) let today = calendar.startOfDay(for: now) @@ -76,25 +78,44 @@ enum OpenCodexUsageAggregator { return lhs.requestID < rhs.requestID } + let catalog: ModelsDevCatalog + let overlay: CostUsageCustomPricing + if windowed.isEmpty { + catalog = ModelsDevCatalog(providers: [:]) + overlay = .empty + } else { + catalog = modelsDevCatalog + ?? CostUsagePricing.modelsDevCatalog() + ?? ModelsDevCatalog(providers: [:]) + overlay = customPricingOverlay ?? CostUsagePricing.customPricingOverlay() + } + var daysByKey: [String: DayAccumulator] = [:] var sessions: [String: SessionAccumulator] = [:] var hoursByStart: [Date: HourAccumulator] = [:] + var dayMemo = LocalDayKeyMemo() + var hourMemo = HourStartMemo() for entry in windowed { - let dayKey = CostUsageLocalDay.key(from: entry.timestamp, calendar: calendar) + let cost = Self.listPriceUSD( + entry: entry, + customPricing: customPricing, + modelsDevCatalog: catalog, + customPricingOverlay: overlay) + let dayKey = dayMemo.key(for: entry.timestamp, calendar: calendar) var day = daysByKey[dayKey] ?? DayAccumulator() - Self.merge(entry, into: &day, customPricing: customPricing) + Self.merge(entry, cost: cost, into: &day) daysByKey[dayKey] = day let sessionID = entry.conversationID ?? entry.requestID var session = sessions[sessionID] ?? SessionAccumulator() session.lastActivity = max(session.lastActivity, entry.timestamp) session.requests += 1 - Self.merge(entry, into: &session, customPricing: customPricing) + Self.merge(entry, cost: cost, into: &session) sessions[sessionID] = session - let hour = calendar.dateInterval(of: .hour, for: entry.timestamp)?.start ?? entry.timestamp + let hour = hourMemo.start(for: entry.timestamp, calendar: calendar) var hourBucket = hoursByStart[hour] ?? HourAccumulator() - Self.merge(entry, into: &hourBucket, customPricing: customPricing) + Self.merge(entry, cost: cost, into: &hourBucket) hoursByStart[hour] = hourBucket } @@ -164,8 +185,8 @@ enum OpenCodexUsageAggregator { private static func merge( _ entry: OpenCodexUsageEntry, - into day: inout DayAccumulator, - customPricing: CostUsageCustomPricing) + cost: Double?, + into day: inout DayAccumulator) { let usage = entry.usage if let input = usage?.inputTokens { @@ -197,7 +218,6 @@ enum OpenCodexUsageAggregator { day.unmetered += entry.usageStatus == .unsupported ? 1 : 0 day.unpriced += entry.usageStatus == .unreported ? 1 : 0 - let cost = Self.listPriceUSD(entry: entry, customPricing: customPricing) if let cost { day.cost += cost day.sawCost = true @@ -220,15 +240,14 @@ enum OpenCodexUsageAggregator { private static func merge( _ entry: OpenCodexUsageEntry, - into session: inout SessionAccumulator, - customPricing: CostUsageCustomPricing) + cost: Double?, + into session: inout SessionAccumulator) { session.input = self.add(session.input, entry.usage?.inputTokens) session.output = self.add(session.output, entry.usage?.outputTokens) session.cacheRead = self.add(session.cacheRead, entry.usage?.cacheReadTokens) session.reasoning = self.add(session.reasoning, entry.usage?.reasoningOutputTokens) session.tokens = self.add(session.tokens, entry.resolvedTotalTokens) - let cost = self.listPriceUSD(entry: entry, customPricing: customPricing) session.cost = self.add(session.cost, cost) var model = session.models[entry.model] ?? ModelAccumulator() self.merge(entry, cost: cost, into: &model) @@ -237,14 +256,14 @@ enum OpenCodexUsageAggregator { private static func merge( _ entry: OpenCodexUsageEntry, - into hour: inout HourAccumulator, - customPricing: CostUsageCustomPricing) + cost: Double?, + into hour: inout HourAccumulator) { if let tokens = entry.resolvedTotalTokens { hour.tokens += tokens hour.sawTokens = true } - if let cost = self.listPriceUSD(entry: entry, customPricing: customPricing) { + if let cost { hour.cost += cost hour.sawCost = true } @@ -305,7 +324,9 @@ enum OpenCodexUsageAggregator { private static func listPriceUSD( entry: OpenCodexUsageEntry, - customPricing: CostUsageCustomPricing) -> Double? + customPricing: CostUsageCustomPricing, + modelsDevCatalog: ModelsDevCatalog, + customPricingOverlay: CostUsageCustomPricing) -> Double? { guard entry.usageStatus == .reported || entry.usageStatus == .estimated else { return nil } let usage = entry.usage @@ -335,7 +356,9 @@ enum OpenCodexUsageAggregator { cachedInputTokens: cacheRead, outputTokens: output, cacheWriteInputTokens: cacheWrite, - pricingDate: entry.timestamp) + pricingDate: entry.timestamp, + modelsDevCatalog: modelsDevCatalog, + customPricing: customPricingOverlay) } private static func add(_ lhs: Int?, _ rhs: Int?) -> Int? { @@ -356,3 +379,49 @@ enum OpenCodexUsageAggregator { } } } + +extension OpenCodexUsageAggregator { + /// Reuses the calendar's `[start, next)` day interval while timestamps stay inside it. + /// Day keys still come from `CostUsageLocalDay` so DST and non-Gregorian calendars stay aligned. + private struct LocalDayKeyMemo { + var start = Date.distantPast + var end = Date.distantPast + var key = "" + + mutating func key(for timestamp: Date, calendar: Calendar) -> String { + if timestamp >= self.start, timestamp < self.end { + return self.key + } + let dayCalendar = CostUsageLocalDay.gregorianCalendar(matching: calendar) + guard let interval = dayCalendar.dateInterval(of: .day, for: timestamp) else { + self.start = Date.distantPast + self.end = Date.distantPast + return CostUsageLocalDay.key(from: timestamp, calendar: calendar) + } + self.start = interval.start + self.end = interval.end + self.key = CostUsageLocalDay.key(from: timestamp, calendar: calendar) + return self.key + } + } + + /// Reuses the calendar's hour interval while timestamps stay inside `[start, end)`. + private struct HourStartMemo { + var start = Date.distantPast + var end = Date.distantPast + + mutating func start(for timestamp: Date, calendar: Calendar) -> Date { + if timestamp >= self.start, timestamp < self.end { + return self.start + } + guard let interval = calendar.dateInterval(of: .hour, for: timestamp) else { + self.start = Date.distantPast + self.end = Date.distantPast + return timestamp + } + self.start = interval.start + self.end = interval.end + return self.start + } + } +} diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift index 8bec20f312..5ddc870421 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift @@ -18,13 +18,18 @@ public enum OpenCodexUsageFanOut { } grouped[provider, default: []].append(entry) } + guard !grouped.isEmpty else { return [:] } + let catalog = CostUsagePricing.modelsDevCatalog() ?? ModelsDevCatalog(providers: [:]) + let overlay = CostUsagePricing.customPricingOverlay() return grouped.mapValues { providerEntries in OpenCodexUsageAggregator.snapshot( entries: providerEntries, now: now, historyDays: historyDays, calendar: calendar, - customPricing: customPricing) + customPricing: customPricing, + modelsDevCatalog: catalog, + customPricingOverlay: overlay) } } diff --git a/Tests/CodexBarTests/CostUsageStoreTests.swift b/Tests/CodexBarTests/CostUsageStoreTests.swift index 5636bb88fb..0b64255256 100644 --- a/Tests/CodexBarTests/CostUsageStoreTests.swift +++ b/Tests/CodexBarTests/CostUsageStoreTests.swift @@ -1010,6 +1010,7 @@ extension CostUsageStoreTests { "b975eb705f905b9a", "47144baa8daccf52", "2d17f4981b78d07f", + "3c984b655688593f", ]) let predecessorHash = "43609cc56f76a003" let predecessorVersion = CostUsageStore.combinedSchemaVersion( diff --git a/Tests/CodexBarTests/ModelsDevPricingTests.swift b/Tests/CodexBarTests/ModelsDevPricingTests.swift index a0da2f6370..9d9c156dec 100644 --- a/Tests/CodexBarTests/ModelsDevPricingTests.swift +++ b/Tests/CodexBarTests/ModelsDevPricingTests.swift @@ -1262,6 +1262,60 @@ extension ModelsDevPricingTests { #expect(reloaded.artifact == nil) } + @Test + func `memo invalidates when cache file size changes`() throws { + let root = try Self.cacheRoot() + try ModelsDevCache.save(catalog: Self.fixtureCatalog(), fetchedAt: Date(), cacheRoot: root) + let url = ModelsDevCache.cacheFileURL(cacheRoot: root) + let pinnedDate = Date(timeIntervalSince1970: 1_700_000_000) + try FileManager.default.setAttributes([.modificationDate: pinnedDate], ofItemAtPath: url.path) + let primed = try #require(ModelsDevCache.load(cacheRoot: root).artifact) + let originalSize = try #require( + try (FileManager.default.attributesOfItem(atPath: url.path)[.size]) as? NSNumber).intValue + + // Same pinned mtime, larger invalid payload: a memo keyed only on mtime would still hit. + try Data(repeating: 0x7B, count: originalSize + 16).write(to: url) + try FileManager.default.setAttributes([.modificationDate: pinnedDate], ofItemAtPath: url.path) + + let reloaded = ModelsDevCache.load(cacheRoot: root) + #expect(reloaded.artifact != primed) + #expect(reloaded.error == .invalidJSON) + } + + @Test + func `memo invalidates when cache file mtime changes`() throws { + let root = try Self.cacheRoot() + try ModelsDevCache.save(catalog: Self.fixtureCatalog(), fetchedAt: Date(), cacheRoot: root) + let url = ModelsDevCache.cacheFileURL(cacheRoot: root) + let pinnedDate = Date(timeIntervalSince1970: 1_700_000_000) + try FileManager.default.setAttributes([.modificationDate: pinnedDate], ofItemAtPath: url.path) + let primed = try #require(ModelsDevCache.load(cacheRoot: root).artifact) + + let size = try #require( + try (FileManager.default.attributesOfItem(atPath: url.path)[.size]) as? NSNumber).intValue + try Data(repeating: 0, count: size).write(to: url) + try FileManager.default.setAttributes( + [.modificationDate: Date(timeIntervalSince1970: 1_700_000_100)], + ofItemAtPath: url.path) + + let reloaded = ModelsDevCache.load(cacheRoot: root) + #expect(reloaded.artifact != primed) + #expect(reloaded.error == .invalidJSON) + } + + @Test + func `load metadata check is one stat per load`() throws { + let root = try Self.cacheRoot() + try ModelsDevCache.save(catalog: Self.fixtureCatalog(), fetchedAt: Date(), cacheRoot: root) + let recorder = ModelsDevCache.MetadataReadRecorder() + ModelsDevCache.withMetadataReadRecorderForTesting(recorder) { + _ = ModelsDevCache.load(cacheRoot: root) + #expect(recorder.snapshot() == 1) + _ = ModelsDevCache.load(cacheRoot: root) + #expect(recorder.snapshot() == 2) + } + } + @Test func `client fetches with mock transport`() async throws { let data = try Self.fixtureData() diff --git a/Tests/CodexBarTests/OpenCodexUsageFanOutTests.swift b/Tests/CodexBarTests/OpenCodexUsageFanOutTests.swift index 51f893cf9b..1d6f476610 100644 --- a/Tests/CodexBarTests/OpenCodexUsageFanOutTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageFanOutTests.swift @@ -1,7 +1,7 @@ -import CodexBarCore import Foundation import Testing @testable import CodexBar +@testable import CodexBarCore struct OpenCodexUsageFanOutTests { @Test func `snapshotsBySubscription routes openai spend into codex`() throws { @@ -168,4 +168,680 @@ struct OpenCodexUsageFanOutTests { let result = SpendDashboardSource.mergingOpenCodexInputs([dummy], request: request) #expect(!result.contains(where: { $0.id == SpendDashboardModel.openCodexSourceID })) } + + @Test + func `snapshot matches a naive per-entry reference across DST and overlays`() throws { + let calendar = try Self.losAngelesCalendar() + let now = try Self.date(calendar, DateComponents(year: 2026, month: 11, day: 2, hour: 12, minute: 0)) + let customPricing = CostUsageCustomPricing.parse(Data(""" + { + "openai/gpt-5.4": { "input": 1.5, "output": 6, "cacheRead": 0.15, "cacheWrite": 1.875 } + } + """.utf8)) + let firstFallBack = try Self.date(calendar, DateComponents(year: 2026, month: 11, day: 1, hour: 1, minute: 30)) + let entries = try [ + OpenCodexUsageEntry( + requestID: "dup", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 11, day: 1, hour: 10, minute: 0)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .reported, + conversationID: "chat-dup", + usage: OpenCodexTokenUsage(inputTokens: 1, outputTokens: 1, totalTokens: 2), + totalTokens: 2), + OpenCodexUsageEntry( + requestID: "dup", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 11, day: 1, hour: 10, minute: 5)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .reported, + conversationID: "chat-dup", + usage: OpenCodexTokenUsage( + inputTokens: 80, + outputTokens: 20, + cacheReadInputTokens: 10, + cacheCreationInputTokens: 5, + totalTokens: 100), + totalTokens: 100), + OpenCodexUsageEntry( + requestID: "spring", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 3, day: 8, hour: 1, minute: 30)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .reported, + conversationID: "chat-dst", + usage: OpenCodexTokenUsage(inputTokens: 40, outputTokens: 10, totalTokens: 50), + totalTokens: 50), + OpenCodexUsageEntry( + requestID: "spring-after", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 3, day: 8, hour: 3, minute: 0)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .estimated, + conversationID: "chat-dst", + usage: OpenCodexTokenUsage(inputTokens: 20, outputTokens: 4, totalTokens: 24), + totalTokens: 24), + OpenCodexUsageEntry( + requestID: "fallback-first", + timestamp: firstFallBack, + provider: "opencode-go", + model: "opencode-go/gpt-5.2", + usageStatus: .reported, + conversationID: "chat-fallback", + usage: OpenCodexTokenUsage(inputTokens: 12, outputTokens: 3, totalTokens: 15), + totalTokens: 15), + OpenCodexUsageEntry( + requestID: "fallback-second", + timestamp: firstFallBack.addingTimeInterval(3600), + provider: "opencode-go", + model: "opencode-go/gpt-5.2", + usageStatus: .reported, + conversationID: "chat-fallback", + usage: OpenCodexTokenUsage(inputTokens: 8, outputTokens: 2, totalTokens: 10), + totalTokens: 10), + OpenCodexUsageEntry( + requestID: "unreported", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 11, day: 2, hour: 9, minute: 0)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .unreported, + conversationID: "chat-today"), + OpenCodexUsageEntry( + requestID: "unsupported", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 11, day: 2, hour: 9, minute: 15)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .unsupported, + conversationID: "chat-today", + usage: OpenCodexTokenUsage(inputTokens: 4, outputTokens: 1, totalTokens: 5), + totalTokens: 5), + OpenCodexUsageEntry( + requestID: "unknown-estimated", + timestamp: Self.date(calendar, DateComponents(year: 2026, month: 11, day: 2, hour: 9, minute: 30)), + provider: "openai", + model: "not-a-priced-model-xyz", + usageStatus: .estimated, + conversationID: "chat-today", + usage: OpenCodexTokenUsage(inputTokens: 6, outputTokens: 1, totalTokens: 7), + totalTokens: 7), + OpenCodexUsageEntry( + requestID: "outside-window", + timestamp: Self.date(calendar, DateComponents(year: 2024, month: 1, day: 1, hour: 12, minute: 0)), + provider: "openai", + model: "gpt-5.4", + usageStatus: .reported, + usage: OpenCodexTokenUsage(inputTokens: 999, outputTokens: 1, totalTokens: 1000), + totalTokens: 1000), + OpenCodexUsageEntry( + requestID: "catalog-priced", + timestamp: now, + provider: "openai", + model: "gpt-5.2", + usageStatus: .reported, + conversationID: "chat-catalog", + usage: OpenCodexTokenUsage(inputTokens: 1000, outputTokens: 1000, totalTokens: 2000), + totalTokens: 2000), + ] + + let root = try Self.modelsDevCacheRoot() + defer { try? FileManager.default.removeItem(at: root) } + let fixtureCatalog = Self.fixturePricingCatalog() + #expect(ModelsDevCache.save(catalog: fixtureCatalog, fetchedAt: now, cacheRoot: root)) + let loadedCatalog = try #require(ModelsDevCache.load(cacheRoot: root).artifact?.catalog) + + let snapshot = OpenCodexUsageAggregator.snapshot( + entries: entries, + now: now, + historyDays: 365, + calendar: calendar, + customPricing: customPricing, + modelsDevCatalog: loadedCatalog) + let reference = OpenCodexUsageSnapshotReference.snapshot( + entries: entries, + now: now, + historyDays: 365, + calendar: calendar, + customPricing: customPricing, + modelsDevCacheRoot: root) + + #expect(snapshot == reference) + #expect(snapshot.daily.count >= 3) + #expect(snapshot.hourly.count >= 5) + #expect(snapshot.sessions.contains { $0.sessionID == "chat-dup" && $0.requestCount == 1 }) + + let catalogPriced = try #require( + snapshot.daily.flatMap { $0.modelBreakdowns ?? [] }.first { $0.modelName == "gpt-5.2" }) + let catalogCost = try #require(catalogPriced.costUSD) + let bundledCost = CostUsagePricing.codexCostUSD( + model: "gpt-5.2", + inputTokens: 1000, + cachedInputTokens: 0, + outputTokens: 1000, + cacheWriteInputTokens: 0, + pricingDate: now, + modelsDevCatalog: ModelsDevCatalog(providers: [:]), + customPricing: .empty) + #expect(catalogCost != bundledCost) + } + + @Test + func `snapshot resolves the models.dev catalog once for many entries`() throws { + var calendar = Calendar(identifier: .gregorian) + calendar.timeZone = try #require(TimeZone(secondsFromGMT: 0)) + let now = Date(timeIntervalSince1970: 1_784_179_200) + let entryCount = 60 + let entries = (0.. Calendar { + var calendar = Calendar(identifier: .gregorian) + calendar.timeZone = try #require(TimeZone(identifier: "America/Los_Angeles")) + return calendar + } + + private static func santiagoCalendar() throws -> Calendar { + var calendar = Calendar(identifier: .gregorian) + calendar.timeZone = try #require(TimeZone(identifier: "America/Santiago")) + return calendar + } + + private static func date(_ calendar: Calendar, _ components: DateComponents) throws -> Date { + try #require(calendar.date(from: components)) + } + + private static func assertDayAndHourMemos( + calendar: Calendar, + now: Date, + samples: [(id: String, timestamp: Date, tokens: Int)]) throws + { + let entries = samples.map { sample in + OpenCodexUsageEntry( + requestID: sample.id, + timestamp: sample.timestamp, + provider: "openai", + model: "gpt-5.4", + usageStatus: .reported, + usage: OpenCodexTokenUsage( + inputTokens: sample.tokens, + outputTokens: 0, + totalTokens: sample.tokens), + totalTokens: sample.tokens) + } + let snapshot = OpenCodexUsageAggregator.snapshot( + entries: entries, + now: now, + historyDays: 365, + calendar: calendar) + for sample in samples { + let expectedDay = CostUsageLocalDay.key(from: sample.timestamp, calendar: calendar) + let expectedHour = calendar.dateInterval(of: .hour, for: sample.timestamp)?.start + ?? sample.timestamp + let day = try #require(snapshot.daily.first { $0.date == expectedDay }) + let hour = try #require(snapshot.hourly.first { $0.hour == expectedHour }) + #expect(day.totalTokens ?? 0 >= sample.tokens) + #expect(hour.totalTokens == sample.tokens) + #expect(hour.hour == expectedHour) + } + } + + private static func modelsDevCacheRoot() throws -> URL { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("codexbar-opencodex-modelsdev-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + return root + } + + private static func fixturePricingCatalog() -> ModelsDevCatalog { + ModelsDevCatalog(providers: [ + "openai": ModelsDevProvider( + id: "openai", + name: "OpenAI", + models: [ + "gpt-5.2": ModelsDevModel( + id: "gpt-5.2", + name: nil, + cost: ModelsDevCost(input: 99, output: 199), + limit: nil), + "gpt-5.4": ModelsDevModel( + id: "gpt-5.4", + name: nil, + cost: ModelsDevCost(input: 88, output: 188), + limit: nil), + ]), + "opencode-go": ModelsDevProvider( + id: "opencode-go", + name: nil, + models: [ + "gpt-5.2": ModelsDevModel( + id: "gpt-5.2", + name: nil, + cost: ModelsDevCost(input: 50, output: 80), + limit: nil), + ]), + ]) + } +} + +private enum OpenCodexUsageSnapshotReference { + static func snapshot( + entries: [OpenCodexUsageEntry], + now: Date, + historyDays: Int, + calendar: Calendar, + customPricing: CostUsageCustomPricing, + modelsDevCacheRoot: URL? = nil) -> CostUsageTokenSnapshot + { + let days = max(1, min(365, historyDays)) + let today = calendar.startOfDay(for: now) + let windowStart = calendar.date(byAdding: .day, value: -(days - 1), to: today) ?? today + var unique: [String: OpenCodexUsageEntry] = [:] + for entry in entries { + unique[entry.requestID] = entry + } + let windowed = unique.values.filter { $0.timestamp >= windowStart && $0.timestamp <= now } + .sorted { lhs, rhs in + if lhs.timestamp != rhs.timestamp { + return lhs.timestamp < rhs.timestamp + } + return lhs.requestID < rhs.requestID + } + + var daysByKey: [String: OpenCodexUsageAggregator.DayAccumulator] = [:] + var sessions: [String: OpenCodexUsageAggregator.SessionAccumulator] = [:] + var hoursByStart: [Date: OpenCodexUsageAggregator.HourAccumulator] = [:] + for entry in windowed { + let cost = Self.listPriceUSD( + entry: entry, + customPricing: customPricing, + modelsDevCacheRoot: modelsDevCacheRoot) + let dayKey = CostUsageLocalDay.key(from: entry.timestamp, calendar: calendar) + var day = daysByKey[dayKey] ?? OpenCodexUsageAggregator.DayAccumulator() + Self.merge(entry, cost: cost, into: &day) + daysByKey[dayKey] = day + + let sessionID = entry.conversationID ?? entry.requestID + var session = sessions[sessionID] ?? OpenCodexUsageAggregator.SessionAccumulator() + session.lastActivity = max(session.lastActivity, entry.timestamp) + session.requests += 1 + Self.merge(entry, cost: cost, into: &session) + sessions[sessionID] = session + + let hour = calendar.dateInterval(of: .hour, for: entry.timestamp)?.start ?? entry.timestamp + var hourBucket = hoursByStart[hour] ?? OpenCodexUsageAggregator.HourAccumulator() + Self.merge(entry, cost: cost, into: &hourBucket) + hoursByStart[hour] = hourBucket + } + + let daily = daysByKey.keys.sorted().compactMap { key -> CostUsageDailyReport.Entry? in + guard let day = daysByKey[key] else { return nil } + return Self.entry(dayKey: key, day: day) + } + let sessionRows = sessions.keys.sorted().compactMap { key -> CostUsageSessionBreakdown? in + guard let session = sessions[key] else { return nil } + return CostUsageSessionBreakdown( + sessionID: key, + lastActivity: session.lastActivity, + inputTokens: session.input, + cachedInputTokens: session.cacheRead, + outputTokens: session.output, + reasoningTokens: session.reasoning, + totalTokens: session.tokens, + requestCount: session.requests, + costUSD: session.cost, + modelBreakdowns: Self.modelBreakdowns(session.models)) + } + .sorted { lhs, rhs in + if lhs.lastActivity != rhs.lastActivity { + return lhs.lastActivity > rhs.lastActivity + } + return lhs.sessionID < rhs.sessionID + } + let hourly = hoursByStart.keys.sorted().map { hour in + let bucket = hoursByStart[hour] ?? OpenCodexUsageAggregator.HourAccumulator() + return CostUsageHourlyEntry( + hour: hour, + totalTokens: bucket.sawTokens ? bucket.tokens : nil, + costUSD: bucket.sawCost ? bucket.cost : nil) + } + let todayEntry = CostUsageTokenSnapshot.entry( + in: daily, + forLocalDayContaining: now, + calendar: calendar) + let windowSummary = CostUsageTokenSnapshot( + sessionTokens: nil, + sessionCostUSD: nil, + last30DaysTokens: nil, + last30DaysCostUSD: nil, + historyDays: days, + daily: daily, + sessions: Array(sessionRows.prefix(64)), + updatedAt: now) + .summary(forLastDays: min(30, days), calendar: calendar) + return CostUsageTokenSnapshot( + sessionTokens: todayEntry?.totalTokens ?? (daily.isEmpty ? nil : 0), + sessionCostUSD: todayEntry?.costUSD ?? (daily.isEmpty ? nil : 0), + sessionRequests: todayEntry?.requestCount ?? (daily.isEmpty ? nil : 0), + last30DaysTokens: windowSummary.totalTokens, + last30DaysCostUSD: windowSummary.totalCostUSD, + last30DaysRequests: windowSummary.totalRequests, + historyDays: days, + historyLabel: "OpenCodex usage.jsonl", + costProvenance: .listPriceEstimate, + daily: daily, + sessions: Array(sessionRows.prefix(64)), + hourly: hourly, + updatedAt: now) + } + + private static func merge( + _ entry: OpenCodexUsageEntry, + cost: Double?, + into day: inout OpenCodexUsageAggregator.DayAccumulator) + { + let usage = entry.usage + if let input = usage?.inputTokens { + day.input += input + day.sawInput = true + } + if let output = usage?.outputTokens { + day.output += output + day.sawOutput = true + } + if let cacheRead = usage?.cacheReadTokens { + day.cacheRead += cacheRead + day.sawCacheRead = true + } + if let cacheCreation = usage?.cacheCreationInputTokens { + day.cacheCreation += cacheCreation + day.sawCacheCreation = true + } + if let reasoning = usage?.reasoningOutputTokens { + day.reasoning += reasoning + day.sawReasoning = true + } + if let tokens = entry.resolvedTotalTokens { + day.tokens += tokens + day.sawTokens = true + } + day.priced += entry.usageStatus == .reported ? 1 : 0 + day.estimated += entry.usageStatus == .estimated ? 1 : 0 + day.unmetered += entry.usageStatus == .unsupported ? 1 : 0 + day.unpriced += entry.usageStatus == .unreported ? 1 : 0 + if let cost { + day.cost += cost + day.sawCost = true + } else if entry.usageStatus == .reported { + day.unpriced += 1 + if day.priced > 0 { + day.priced -= 1 + } + } else if entry.usageStatus == .estimated { + day.unpriced += 1 + if day.estimated > 0 { + day.estimated -= 1 + } + } + var model = day.models[entry.model] ?? OpenCodexUsageAggregator.ModelAccumulator() + Self.merge(entry, cost: cost, into: &model) + day.models[entry.model] = model + } + + private static func merge( + _ entry: OpenCodexUsageEntry, + cost: Double?, + into session: inout OpenCodexUsageAggregator.SessionAccumulator) + { + session.input = self.add(session.input, entry.usage?.inputTokens) + session.output = self.add(session.output, entry.usage?.outputTokens) + session.cacheRead = self.add(session.cacheRead, entry.usage?.cacheReadTokens) + session.reasoning = self.add(session.reasoning, entry.usage?.reasoningOutputTokens) + session.tokens = self.add(session.tokens, entry.resolvedTotalTokens) + session.cost = self.add(session.cost, cost) + var model = session.models[entry.model] ?? OpenCodexUsageAggregator.ModelAccumulator() + Self.merge(entry, cost: cost, into: &model) + session.models[entry.model] = model + } + + private static func merge( + _ entry: OpenCodexUsageEntry, + cost: Double?, + into hour: inout OpenCodexUsageAggregator.HourAccumulator) + { + if let tokens = entry.resolvedTotalTokens { + hour.tokens += tokens + hour.sawTokens = true + } + if let cost { + hour.cost += cost + hour.sawCost = true + } + } + + private static func merge( + _ entry: OpenCodexUsageEntry, + cost: Double?, + into model: inout OpenCodexUsageAggregator.ModelAccumulator) + { + model.input = self.add(model.input, entry.usage?.inputTokens) + model.output = self.add(model.output, entry.usage?.outputTokens) + model.cacheRead = self.add(model.cacheRead, entry.usage?.cacheReadTokens) + model.cacheCreation = self.add(model.cacheCreation, entry.usage?.cacheCreationInputTokens) + model.reasoning = self.add(model.reasoning, entry.usage?.reasoningOutputTokens) + if let tokens = entry.resolvedTotalTokens { + model.tokens += tokens + model.sawTokens = true + } + if let cost { + model.cost += cost + model.sawCost = true + } + } + + private static func entry( + dayKey: String, + day: OpenCodexUsageAggregator.DayAccumulator) -> CostUsageDailyReport.Entry + { + CostUsageDailyReport.Entry( + date: dayKey, + inputTokens: day.sawInput ? day.input : nil, + outputTokens: day.sawOutput ? day.output : nil, + cacheReadTokens: day.sawCacheRead ? day.cacheRead : nil, + cacheCreationTokens: day.sawCacheCreation ? day.cacheCreation : nil, + reasoningTokens: day.sawReasoning ? day.reasoning : nil, + totalTokens: day.sawTokens ? day.tokens : nil, + requestCount: day.priced + day.unpriced + day.unmetered + day.estimated, + costUSD: day.sawCost ? day.cost : nil, + modelsUsed: day.models.keys.sorted(), + modelBreakdowns: self.modelBreakdowns(day.models), + unpricedRequestCount: day.unpriced, + unmeteredRequestCount: day.unmetered, + estimatedRequestCount: day.estimated) + } + + private static func modelBreakdowns( + _ models: [String: OpenCodexUsageAggregator.ModelAccumulator]) -> [CostUsageDailyReport.ModelBreakdown] + { + models.keys.sorted().map { name in + let model = models[name] ?? OpenCodexUsageAggregator.ModelAccumulator() + return CostUsageDailyReport.ModelBreakdown( + modelName: name, + costUSD: model.sawCost ? model.cost : nil, + totalTokens: model.sawTokens ? model.tokens : nil, + inputTokens: model.input, + outputTokens: model.output, + cacheReadTokens: model.cacheRead, + cacheCreationTokens: model.cacheCreation, + reasoningTokens: model.reasoning) + } + } + + private static func listPriceUSD( + entry: OpenCodexUsageEntry, + customPricing: CostUsageCustomPricing, + modelsDevCacheRoot: URL?) -> Double? + { + guard entry.usageStatus == .reported || entry.usageStatus == .estimated else { return nil } + let usage = entry.usage + let hasTokenData = entry.resolvedTotalTokens != nil + || usage?.inputTokens != nil + || usage?.outputTokens != nil + || usage?.cacheReadTokens != nil + || usage?.cacheCreationInputTokens != nil + guard hasTokenData else { return nil } + let input = usage?.inputTokens ?? 0 + let output = usage?.outputTokens ?? 0 + let cacheRead = usage?.cacheReadTokens ?? 0 + let cacheWrite = usage?.cacheCreationInputTokens ?? 0 + if let overlay = customPricing.costUSD( + providerID: entry.provider, + model: entry.model, + inputTokens: input, + outputTokens: output, + cacheReadTokens: cacheRead, + cacheWriteTokens: cacheWrite) + { + return overlay + } + return CostUsagePricing.codexCostUSD( + model: entry.model, + inputTokens: input, + cachedInputTokens: cacheRead, + outputTokens: output, + cacheWriteInputTokens: cacheWrite, + pricingDate: entry.timestamp, + modelsDevCacheRoot: modelsDevCacheRoot) + } + + private static func add(_ lhs: Int?, _ rhs: Int?) -> Int? { + switch (lhs, rhs) { + case let (left?, right?): left + right + case let (left?, nil): left + case let (nil, right?): right + case (nil, nil): nil + } + } + + private static func add(_ lhs: Double?, _ rhs: Double?) -> Double? { + switch (lhs, rhs) { + case let (left?, right?): left + right + case let (left?, nil): left + case let (nil, right?): right + case (nil, nil): nil + } + } } diff --git a/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift b/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift index a189eecbaa..782bb3af10 100644 --- a/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift +++ b/Tests/CodexBarTests/ProviderArchitectureGatekeeperTests.swift @@ -3746,7 +3746,7 @@ struct ProviderArchitectureGatekeeperTests { reason: "This exact cost scanner dispatch selects a provider-owned transcript, cache, or pricing format."), AllowedProviderConstruct( path: "Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift", - line: 75, + line: 80, anchor: "[\"anthropic\", \"openai\"].allSatisfy { providerID in", expectedProviderIDs: ["openai"], expectedReferenceCount: 1, From f1e764639522998c44056b426d94998b33a8355e Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 23:16:33 -0700 Subject: [PATCH 2/7] docs: comment the OpenCodex price-once context and memo semantics Explain why the models.dev catalog and the custom-pricing overlay are resolved once per snapshot / fan-out, why a missing catalog is substituted with an empty one (so the degraded path never falls back to per-call ModelsDevCache.load), the two-level overlay precedence in listPriceUSD, why the day-key memo cannot disagree with CostUsageLocalDay.key, and that the metadata-read recorder is task-local test-only instrumentation. Comments only; CodexParserHash is regenerated because ModelsDevPricing.swift is in the hashed set (no shipped hash is affected; the predecessor list is unchanged). Co-Authored-By: Claude Fable 5 --- .../Generated/CodexParserHash.generated.swift | 2 +- .../Vendored/CostUsage/ModelsDevPricing.swift | 3 +++ .../OpenCodexUsageAggregator.swift | 22 ++++++++++++++++++- .../OpenCodexUsage/OpenCodexUsageFanOut.swift | 3 +++ 4 files changed, 28 insertions(+), 2 deletions(-) diff --git a/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift b/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift index cf3cb01d99..324050b2e7 100644 --- a/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift +++ b/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift @@ -1,5 +1,5 @@ // Generated by Scripts/regenerate-codex-parser-hash.sh. Do not edit by hand. enum CodexParserHash { - static let value = "fc4cae716e9066c5" + static let value = "5f8507161b23757c" } diff --git a/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift b/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift index 6e3946ab21..eb807591bf 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/ModelsDevPricing.swift @@ -432,6 +432,9 @@ enum ModelsDevCache { static let ttlSeconds: TimeInterval = 24 * 60 * 60 private static let memo = ModelsDevCacheMemo() + /// Test-only instrumentation: counts `fileMetadata(at:)` reads (one per `load`) so tests can prove callers + /// resolve the catalog once instead of per pricing call. Task-local, so concurrent tests do not see each other's + /// counts, and unset (zero cost) in production. @TaskLocal private static var metadataReadRecorder: MetadataReadRecorder? final class MetadataReadRecorder: @unchecked Sendable { diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift index c6e23951bf..1e7d6eb60b 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageAggregator.swift @@ -54,6 +54,13 @@ enum OpenCodexUsageAggregator { var sawCost = false } + /// Aggregates OpenCodex usage entries into a per-window token/cost snapshot. + /// + /// Pricing context is resolved once per call and shared by every entry: `modelsDevCatalog` is the models.dev + /// catalog and `customPricingOverlay` the app-level custom-pricing overlay file. Callers that aggregate several + /// providers (see `OpenCodexUsageFanOut`) resolve both once and pass them in; when either is nil it is resolved + /// here once. Each windowed entry is priced exactly once and that price feeds the day, session, hour and model + /// accumulators, so output is identical to pricing inside each merge — without a catalog/overlay lookup per call. static func snapshot( entries: [OpenCodexUsageEntry], now: Date, @@ -78,6 +85,10 @@ enum OpenCodexUsageAggregator { return lhs.requestID < rhs.requestID } + // Resolve the pricing context once for the whole snapshot. A missing models.dev catalog becomes an EMPTY + // catalog on purpose: `codexCostUSD` treats a nil catalog as "resolve it yourself" and would fall back to + // `ModelsDevCache.load` (a stat per pricing target) for every entry, whereas an empty catalog yields the same + // nil lookups without any file access. Nothing is resolved when the window is empty. let catalog: ModelsDevCatalog let overlay: CostUsageCustomPricing if windowed.isEmpty { @@ -93,6 +104,8 @@ enum OpenCodexUsageAggregator { var daysByKey: [String: DayAccumulator] = [:] var sessions: [String: SessionAccumulator] = [:] var hoursByStart: [Date: HourAccumulator] = [:] + // `windowed` is sorted by timestamp, so the day/hour memos hit on almost every entry; a miss only costs one + // Calendar interval lookup. Price once per entry and reuse it for the day, session and hour merges. var dayMemo = LocalDayKeyMemo() var hourMemo = HourStartMemo() for entry in windowed { @@ -322,6 +335,11 @@ enum OpenCodexUsageAggregator { } } + /// List-price estimate for one entry. Precedence is unchanged from the per-merge pricing it replaces: + /// 1. `customPricing` — the snapshot's own overlay (provider-scoped rates passed by the caller); + /// 2. `CostUsagePricing.codexCostUSD` with the pre-resolved `customPricingOverlay` (the app-level overlay file, + /// which `codexCostUSD` would otherwise re-load per call) and the pre-resolved models.dev `modelsDevCatalog` + /// (otherwise `ModelsDevCache.load` per call), then the bundled/historical tables. private static func listPriceUSD( entry: OpenCodexUsageEntry, customPricing: CostUsageCustomPricing, @@ -382,7 +400,9 @@ enum OpenCodexUsageAggregator { extension OpenCodexUsageAggregator { /// Reuses the calendar's `[start, next)` day interval while timestamps stay inside it. - /// Day keys still come from `CostUsageLocalDay` so DST and non-Gregorian calendars stay aligned. + /// Day keys still come from `CostUsageLocalDay` so DST and non-Gregorian calendars stay aligned: the key derives + /// y-m-d from the same Gregorian-in-timezone calendar whose `.day` interval is cached here, so the memo can never + /// disagree with computing the key per entry (DST days are simply 23 h / 25 h intervals). private struct LocalDayKeyMemo { var start = Date.distantPast var end = Date.distantPast diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift index 5ddc870421..a060ff8df6 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageFanOut.swift @@ -19,6 +19,9 @@ public enum OpenCodexUsageFanOut { grouped[provider, default: []].append(entry) } guard !grouped.isEmpty else { return [:] } + // Resolve the models.dev catalog and the custom-pricing overlay once for all providers; each snapshot then + // prices its entries against this shared context instead of re-reading both per pricing call (see + // `OpenCodexUsageAggregator.snapshot`). A missing catalog is passed as an empty one for the same reason. let catalog = CostUsagePricing.modelsDevCatalog() ?? ModelsDevCatalog(providers: [:]) let overlay = CostUsagePricing.customPricingOverlay() return grouped.mapValues { providerEntries in From dfbed3020e2a173ff084576c66f0306a14f16851 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Sat, 22 Aug 2026 01:41:49 -0700 Subject: [PATCH 3/7] Parse the OpenCodex usage log incrementally instead of re-reading it every refresh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `~/.opencodex/usage.jsonl` is append-only, but `OpenCodexUsageStore` keyed its sqlite cache on `path|size|mtime`, so the identity changed on every append and the cache never hit in the steady state. Each refresh therefore re-read the whole log (42 MB / ~35k entries on the reference machine), rebuilt a 42 MB `String`, ran one `JSONSerialization` per line, and then did `DELETE FROM entries` plus 35k inserts. Even a cache hit re-decoded a stored JSON payload for every row. The store now keeps a parse cursor in its metadata — log path, file identity (`st_dev`/`st_ino`, not mtime), the byte offset just past the last consumed record, and a SHA256 of the first 64 KiB — and on load either serves the cached rows unchanged, or reads only `[parsedOffset, size)` and merges those rows with `INSERT OR REPLACE`. A changed file identity, a shrunken file, or a prefix-digest mismatch falls back to the full re-parse, so rotation, truncation and in-place rewrites are still handled. A trailing record with no newline is returned but not committed and does not advance the cursor, so a later append that glues bytes onto it cannot desync the cache from a full parse. Two supporting changes: token fields become typed sqlite columns (schema v2), so reading the cache no longer decodes JSON per row, and the parser slices lines out of the file's bytes (`mappedIfSafe` plus `memchr`) instead of materializing the whole log as a `String`. `parseLines` now uses the same splitter, so there is one line-splitting rule instead of two. Measured (release CodexBarCLI, isolated cache root, real 42 MB / 35k-entry log, `cost --provider codex --format json --days 30`, OpenCodex path isolated with identical frozen inputs; CLI JSON output identical apart from `updatedAt`): - cold, no cache: 2.85 s / 422 MB peak -> 2.74 s / 187 MB - warm cache hit: 1.06 s / 112 MB -> 0.52 s / 75 MB - after the log grew (the app's per-refresh case, full run on live data): 4.58 s / 422 MB -> 2.14 s / 110 MB - cache database: 18 MB -> 5.1 MB Baselines are the parent commit (Codex 0.54.2 branch state); against upstream main before the OpenCodex work the same cold case was 19-21 s / ~450 MB. Tests: incremental result equals a full re-parse (append, partial trailing line, newline-less record later glued to more bytes, duplicate request IDs, truncation, rotation with a matching 64 KiB prefix, in-place rewrite), tail-only reads with a byte/line recorder (zero bytes when nothing changed), schema-v2 round trip for a missing `usage`, a single zero-valued field and mismatched totals, a v1 database rebuild, and a busy-writer case proving a failed `BEGIN IMMEDIATE` leaves both the rows and the cursor untouched. Implemented by grok-4.6 (xhigh) via implementation-loop; reviewed hunk by hunk plus an independent deep review whose blocker (an ignored `BEGIN IMMEDIATE` result that could empty the cache under a concurrent writer) and seven further findings were fixed in one iterate round. Co-Authored-By: Claude Fable 5 --- .../OpenCodexUsage/OpenCodexUsageParser.swift | 178 +++++- .../OpenCodexUsage/OpenCodexUsageStore.swift | 420 +++++++++---- .../OpenCodexUsageParserTests.swift | 28 + .../OpenCodexUsageStoreIncrementalTests.swift | 591 ++++++++++++++++++ 4 files changed, 1104 insertions(+), 113 deletions(-) create mode 100644 Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift index cff9b00742..550786536b 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift @@ -1,6 +1,45 @@ +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#endif import Foundation public enum OpenCodexUsageParser { + private static let newline: UInt8 = 0x0A + + @TaskLocal static var logReadRecorderForTesting: LogReadRecorder? + + final class LogReadRecorder: @unchecked Sendable { + private let lock = NSLock() + private var bytesRead: Int64 = 0 + private var completeLines = 0 + + func record(bytes: Int64, lines: Int) { + self.lock.lock() + self.bytesRead += bytes + self.completeLines += lines + self.lock.unlock() + } + + func snapshot() -> (bytesRead: Int64, completeLines: Int) { + self.lock.lock() + defer { self.lock.unlock() } + return (self.bytesRead, self.completeLines) + } + } + + static func withLogReadRecorderForTesting( + _ recorder: LogReadRecorder, + operation: () throws -> T) rethrows -> T + { + try self.$logReadRecorderForTesting.withValue(recorder) { + try operation() + } + } + public static func parseLine(_ line: String) -> OpenCodexUsageEntry? { let trimmed = line.trimmingCharacters(in: .whitespacesAndNewlines) guard !trimmed.isEmpty, let data = trimmed.data(using: .utf8) else { return nil } @@ -15,14 +54,143 @@ public enum OpenCodexUsageParser { } public static func parseLines(_ text: String) -> [OpenCodexUsageEntry] { - text.split(whereSeparator: \.isNewline).compactMap { self.parseLine(String($0)) } + self.parseJSONL(Data(text.utf8), baseOffset: 0).entries } public static func parse(fileURL: URL, fileManager: FileManager = .default) throws -> [OpenCodexUsageEntry] { - guard fileManager.fileExists(atPath: fileURL.path) else { return [] } - let data = try Data(contentsOf: fileURL) - guard let text = String(data: data, encoding: .utf8) else { return [] } - return self.parseLines(text) + try self.parseLog(fileURL: fileURL, from: 0, fileManager: fileManager).entries + } + + public static func parse( + fileURL: URL, + from offset: Int64, + fileManager: FileManager = .default) throws -> (entries: [OpenCodexUsageEntry], nextOffset: Int64) + { + let parsed = try self.parseLog(fileURL: fileURL, from: offset, fileManager: fileManager) + return (parsed.entries, parsed.nextOffset) + } + + static func parseLog( + fileURL: URL, + from offset: Int64, + fileManager: FileManager) throws -> JSONLParseResult + { + guard fileManager.fileExists(atPath: fileURL.path) else { + return JSONLParseResult( + entries: [], + nextOffset: max(0, offset), + bytesRead: 0, + completeLineCount: 0, + newlineTerminatedEntryCount: 0) + } + + let startOffset = max(0, offset) + let data: Data + if startOffset == 0 { + // `.mappedIfSafe` assumes the file is append-only; a shrink under an active mapping is + // formally undefined. + data = try Data(contentsOf: fileURL, options: .mappedIfSafe) + } else { + let handle = try FileHandle(forReadingFrom: fileURL) + defer { try? handle.close() } + try handle.seek(toOffset: UInt64(startOffset)) + data = try handle.readToEnd() ?? Data() + } + + let parsed = self.parseJSONL(data, baseOffset: startOffset) + self.logReadRecorderForTesting?.record(bytes: parsed.bytesRead, lines: parsed.completeLineCount) + return parsed + } + + struct JSONLParseResult: Equatable, Sendable { + var entries: [OpenCodexUsageEntry] + var nextOffset: Int64 + var bytesRead: Int64 + var completeLineCount: Int + var newlineTerminatedEntryCount: Int + + var newlineTerminatedEntries: [OpenCodexUsageEntry] { + Array(self.entries.prefix(self.newlineTerminatedEntryCount)) + } + + var pendingTrailingEntries: [OpenCodexUsageEntry] { + Array(self.entries.dropFirst(self.newlineTerminatedEntryCount)) + } + } + + private static func parseJSONL(_ data: Data, baseOffset: Int64) -> JSONLParseResult { + var entries: [OpenCodexUsageEntry] = [] + var completeLineCount = 0 + var lineStart = data.startIndex + var lastCompleteEnd = data.startIndex + for newlineOffset in self.newlineOffsets(in: data) { + let newlineIndex = data.index(data.startIndex, offsetBy: newlineOffset) + if let entry = self.parseLineData(data[lineStart.. [Int] { + #if canImport(Darwin) || canImport(Glibc) || canImport(Musl) + let scanned: [Int]? = data.withUnsafeBytes { rawBuffer in + guard let baseAddress = rawBuffer.baseAddress else { return nil } + let count = rawBuffer.count + var offsets: [Int] = [] + offsets.reserveCapacity(max(1, count / 64)) + var searchStart = 0 + while searchStart < count { + guard let found = memchr( + baseAddress.advanced(by: searchStart), + Int32(Self.newline), + count - searchStart) + else { + break + } + let newlineOffset = baseAddress.distance(to: UnsafeRawPointer(found)) + offsets.append(newlineOffset) + searchStart = newlineOffset + 1 + } + return offsets + } + if let scanned { + return scanned + } + #endif + var offsets: [Int] = [] + var index = data.startIndex + while index < data.endIndex { + if data[index] == Self.newline { + offsets.append(data.distance(from: data.startIndex, to: index)) + } + index = data.index(after: index) + } + return offsets + } + + private static func parseLineData(_ line: Data) -> OpenCodexUsageEntry? { + guard let text = String(data: line, encoding: .utf8) else { return nil } + return self.parseLine(text) } private static func parse(_ object: [String: Any]) -> OpenCodexUsageEntry? { diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift index 040b76d5d4..68044f58ab 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift @@ -3,12 +3,24 @@ import SQLite3 #elseif canImport(CSQLite3) import CSQLite3 #endif +#if canImport(CryptoKit) +import CryptoKit +#else +import Crypto +#endif +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#endif import Foundation /// Independent OpenCodex usage cache. Never writes Codex `cost-usage.sqlite`. public struct OpenCodexUsageStore: Sendable { public static let databaseFilename = "opencodex-usage.sqlite" - private static let schemaVersion = 1 + private static let schemaVersion = 2 + private static let cursorMetaKey = "parseCursor" + private static let prefixDigestByteLimit = 64 * 1024 private let databaseURL: URL @@ -16,6 +28,13 @@ public struct OpenCodexUsageStore: Sendable { self.databaseURL = cacheRoot.appendingPathComponent(Self.databaseFilename, isDirectory: false) } + static func withLogReadRecorderForTesting( + _ recorder: OpenCodexUsageParser.LogReadRecorder, + operation: () throws -> T) rethrows -> T + { + try OpenCodexUsageParser.withLogReadRecorderForTesting(recorder, operation: operation) + } + public func loadSnapshot( logURL: URL, now: Date, @@ -35,89 +54,144 @@ public struct OpenCodexUsageStore: Sendable { public func loadEntries(logURL: URL, fileManager: FileManager = .default) throws -> [OpenCodexUsageEntry] { guard fileManager.fileExists(atPath: logURL.path) else { return [] } - let attributes = try fileManager.attributesOfItem(atPath: logURL.path) - let size = (attributes[.size] as? NSNumber)?.int64Value ?? 0 - let mtime = (attributes[.modificationDate] as? Date)?.timeIntervalSince1970 ?? 0 - let identity = "\(logURL.path)|\(size)|\(mtime)" - - if let cached = self.readCachedEntries(identity: identity), !cached.isEmpty { - return cached + guard let identity = Self.statLog(at: logURL) else { return [] } + let cursor = self.readCursor() + if let cursor, Self.canReuseCursor(cursor, identity: identity) { + if identity.size == cursor.parsedOffset { + if let cached = self.readCachedEntries() { + return cached + } + } else { + return try self.incrementalReload( + logURL: logURL, + identity: identity, + cursor: cursor, + fileManager: fileManager) + } } + return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) + } - let parsed = try OpenCodexUsageParser.parse(fileURL: logURL, fileManager: fileManager) - var unique: [String: OpenCodexUsageEntry] = [:] - for entry in parsed { - unique[entry.requestID] = entry + func parseCursorForTesting() -> (parsedOffset: Int64, prefixDigest: String, fileIdentity: String)? { + guard let cursor = self.readCursor() else { return nil } + return (cursor.parsedOffset, cursor.prefixDigest, cursor.fileIdentity) + } + + private func fullReload( + logURL: URL, + identity: LogIdentity, + fileManager: FileManager) throws -> [OpenCodexUsageEntry] + { + let parsed = try OpenCodexUsageParser.parseLog(fileURL: logURL, from: 0, fileManager: fileManager) + let entries = Self.dedupedAndSorted(parsed.entries) + let cursor = ParseCursor( + path: identity.path, + fileIdentity: identity.fileIdentity, + parsedOffset: parsed.nextOffset, + prefixDigest: Self.prefixDigest(fileURL: logURL, parsedOffset: parsed.nextOffset) ?? "") + self.replaceCachedEntries(Self.dedupedAndSorted(parsed.newlineTerminatedEntries), cursor: cursor) + return entries + } + + private func incrementalReload( + logURL: URL, + identity: LogIdentity, + cursor: ParseCursor, + fileManager: FileManager) throws -> [OpenCodexUsageEntry] + { + guard let existing = self.readCachedEntries() else { + return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) } - let deduped = unique.values.sorted { - if $0.timestamp != $1.timestamp { - return $0.timestamp < $1.timestamp - } - return $0.requestID < $1.requestID + let parsed = try OpenCodexUsageParser.parseLog( + fileURL: logURL, + from: cursor.parsedOffset, + fileManager: fileManager) + let nextOffset = parsed.nextOffset + let committed = parsed.newlineTerminatedEntries + let pending = parsed.pendingTrailingEntries + if committed.isEmpty, nextOffset == cursor.parsedOffset { + return Self.dedupedAndSorted(existing + pending) } - self.replaceCachedEntries(deduped, identity: identity) - return deduped + let digest = nextOffset == cursor.parsedOffset + ? cursor.prefixDigest + : (Self.prefixDigest(fileURL: logURL, parsedOffset: nextOffset) ?? "") + self.insertCachedEntries( + committed, + cursor: ParseCursor( + path: identity.path, + fileIdentity: identity.fileIdentity, + parsedOffset: nextOffset, + prefixDigest: digest)) + return Self.dedupedAndSorted(existing + committed + pending) } - private func readCachedEntries(identity: String) -> [OpenCodexUsageEntry]? { + private func readCachedEntries() -> [OpenCodexUsageEntry]? { guard let db = self.open(readOnly: true) else { return nil } defer { sqlite3_close(db) } - guard Self.userVersion(db) == Self.schemaVersion, - Self.meta(db, key: "identity") == identity - else { return nil } + guard Self.userVersion(db) == Self.schemaVersion else { return nil } var statement: OpaquePointer? let sql = """ - SELECT request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, payload + SELECT request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, \ + input_tokens, output_tokens, cached_input_tokens, cache_read_input_tokens, \ + cache_creation_input_tokens, reasoning_output_tokens, usage_total_tokens, total_tokens FROM entries """ guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return nil } defer { sqlite3_finalize(statement) } var entries: [OpenCodexUsageEntry] = [] - while sqlite3_step(statement) == SQLITE_ROW { - guard let payload = Self.text(statement, 8), - let data = payload.data(using: .utf8), - let entry = OpenCodexUsageParser.parse(data) - else { continue } - entries.append(entry) + var step = sqlite3_step(statement) + while step == SQLITE_ROW { + if let requestID = Self.text(statement, 0), + let provider = Self.text(statement, 2), + let model = Self.text(statement, 3), + let statusRaw = Self.text(statement, 4) + { + let usage = Self.tokenUsage( + inputTokens: Self.int(statement, 8), + outputTokens: Self.int(statement, 9), + cachedInputTokens: Self.int(statement, 10), + cacheReadInputTokens: Self.int(statement, 11), + cacheCreationInputTokens: Self.int(statement, 12), + reasoningOutputTokens: Self.int(statement, 13), + totalTokens: Self.int(statement, 14)) + entries.append(OpenCodexUsageEntry( + requestID: requestID, + timestamp: Date(timeIntervalSince1970: sqlite3_column_double(statement, 1)), + provider: provider, + model: model, + usageStatus: OpenCodexUsageStatus(rawValue: statusRaw) ?? .unreported, + accountLogLabel: Self.text(statement, 5), + surface: Self.text(statement, 6), + conversationID: Self.text(statement, 7), + usage: usage, + totalTokens: Self.int(statement, 15))) + } + step = sqlite3_step(statement) } - return entries + guard step == SQLITE_DONE else { return nil } + return Self.dedupedAndSorted(entries) + } + + private func replaceCachedEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor) { + self.writeEntries(entries, cursor: cursor, replaceAll: true) } - private func replaceCachedEntries(_ entries: [OpenCodexUsageEntry], identity: String) { + private func insertCachedEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor) { + self.writeEntries(entries, cursor: cursor, replaceAll: false) + } + + private func writeEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor, replaceAll: Bool) { guard let db = self.open(readOnly: false) else { return } defer { sqlite3_close(db) } - _ = sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) - _ = sqlite3_exec(db, "DELETE FROM entries", nil, nil, nil) - Self.setMeta(db, key: "identity", value: identity) - var statement: OpaquePointer? - let sql = """ - INSERT OR REPLACE INTO entries( - request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, payload - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """ - guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { + guard sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) == SQLITE_OK else { return } + if replaceAll { + _ = sqlite3_exec(db, "DELETE FROM entries", nil, nil, nil) + } + Self.setCursor(db, cursor) + guard Self.insertEntries(db, entries) else { _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) return } - defer { sqlite3_finalize(statement) } - for entry in entries { - sqlite3_reset(statement) - sqlite3_clear_bindings(statement) - Self.bind(statement, 1, entry.requestID) - sqlite3_bind_double(statement, 2, entry.timestamp.timeIntervalSince1970) - Self.bind(statement, 3, entry.provider) - Self.bind(statement, 4, entry.model) - Self.bind(statement, 5, entry.usageStatus.rawValue) - Self.bind(statement, 6, entry.accountLogLabel) - Self.bind(statement, 7, entry.surface) - Self.bind(statement, 8, entry.conversationID) - let payload = Self.payloadJSON(entry) - Self.bind(statement, 9, payload) - guard sqlite3_step(statement) == SQLITE_DONE else { - _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) - return - } - } _ = sqlite3_exec(db, "COMMIT", nil, nil, nil) } @@ -144,14 +218,40 @@ public struct OpenCodexUsageStore: Sendable { return db } + private func readCursor() -> ParseCursor? { + guard let db = self.open(readOnly: true) else { return nil } + defer { sqlite3_close(db) } + guard Self.userVersion(db) == Self.schemaVersion, + let raw = Self.meta(db, key: Self.cursorMetaKey), + let data = raw.data(using: .utf8), + let cursor = try? JSONDecoder().decode(ParseCursor.self, from: data) + else { return nil } + return cursor + } + + /// Threat model: the digest covers only the first min(64 KiB, parsedOffset) bytes, so an + /// in-place rewrite past 64 KiB that preserves size and inode is not detected. Rotation + /// changes the inode and truncation trips `size >= parsedOffset`. Acceptable because the log + /// is append-only. + private static func canReuseCursor(_ cursor: ParseCursor, identity: LogIdentity) -> Bool { + cursor.path == identity.path + && cursor.fileIdentity == identity.fileIdentity + && identity.size >= cursor.parsedOffset + && self.prefixDigest(fileURL: identity.url, parsedOffset: cursor.parsedOffset) == cursor.prefixDigest + } + private static func ensureSchema(_ db: OpaquePointer?) { - guard self.userVersion(db) == 0 else { return } + // `!= schemaVersion` deliberately rebuilds a NEWER database too: a downgrade must not + // read a schema it does not understand. + guard self.userVersion(db) != self.schemaVersion else { return } let sql = """ - CREATE TABLE IF NOT EXISTS meta ( + DROP TABLE IF EXISTS entries; + DROP TABLE IF EXISTS meta; + CREATE TABLE meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); - CREATE TABLE IF NOT EXISTS entries ( + CREATE TABLE entries ( request_id TEXT PRIMARY KEY, timestamp REAL NOT NULL, provider TEXT NOT NULL, @@ -160,13 +260,55 @@ public struct OpenCodexUsageStore: Sendable { account_label TEXT, surface TEXT, conversation_id TEXT, - payload TEXT NOT NULL + input_tokens INTEGER, + output_tokens INTEGER, + cached_input_tokens INTEGER, + cache_read_input_tokens INTEGER, + cache_creation_input_tokens INTEGER, + reasoning_output_tokens INTEGER, + usage_total_tokens INTEGER, + total_tokens INTEGER ); """ guard sqlite3_exec(db, sql, nil, nil, nil) == SQLITE_OK else { return } Self.setUserVersion(db, Self.schemaVersion) } + private static func insertEntries(_ db: OpaquePointer?, _ entries: [OpenCodexUsageEntry]) -> Bool { + var statement: OpaquePointer? + let sql = """ + INSERT OR REPLACE INTO entries( + request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, + input_tokens, output_tokens, cached_input_tokens, cache_read_input_tokens, + cache_creation_input_tokens, reasoning_output_tokens, usage_total_tokens, total_tokens + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { return false } + defer { sqlite3_finalize(statement) } + for entry in entries { + sqlite3_reset(statement) + sqlite3_clear_bindings(statement) + Self.bind(statement, 1, entry.requestID) + sqlite3_bind_double(statement, 2, entry.timestamp.timeIntervalSince1970) + Self.bind(statement, 3, entry.provider) + Self.bind(statement, 4, entry.model) + Self.bind(statement, 5, entry.usageStatus.rawValue) + Self.bind(statement, 6, entry.accountLogLabel) + Self.bind(statement, 7, entry.surface) + Self.bind(statement, 8, entry.conversationID) + Self.bind(statement, 9, entry.usage?.inputTokens) + Self.bind(statement, 10, entry.usage?.outputTokens) + Self.bind(statement, 11, entry.usage?.cachedInputTokens) + Self.bind(statement, 12, entry.usage?.cacheReadInputTokens) + Self.bind(statement, 13, entry.usage?.cacheCreationInputTokens) + Self.bind(statement, 14, entry.usage?.reasoningOutputTokens) + Self.bind(statement, 15, entry.usage?.totalTokens) + Self.bind(statement, 16, entry.totalTokens) + guard sqlite3_step(statement) == SQLITE_DONE else { return false } + } + return true + } + private static func userVersion(_ db: OpaquePointer?) -> Int { var statement: OpaquePointer? guard sqlite3_prepare_v2(db, "PRAGMA user_version", -1, &statement, nil) == SQLITE_OK else { return 0 } @@ -190,6 +332,13 @@ public struct OpenCodexUsageStore: Sendable { return Self.text(statement, 0) } + private static func setCursor(_ db: OpaquePointer?, _ cursor: ParseCursor) { + guard let data = try? JSONEncoder().encode(cursor), + let value = String(data: data, encoding: .utf8) + else { return } + Self.setMeta(db, key: Self.cursorMetaKey, value: value) + } + private static func setMeta(_ db: OpaquePointer?, key: String, value: String) { var statement: OpaquePointer? guard sqlite3_prepare_v2( @@ -213,59 +362,114 @@ public struct OpenCodexUsageStore: Sendable { sqlite3_bind_text(statement, index, value, -1, unsafeBitCast(-1, to: sqlite3_destructor_type.self)) } + private static func bind(_ statement: OpaquePointer?, _ index: Int32, _ value: Int?) { + guard let value else { + sqlite3_bind_null(statement, index) + return + } + sqlite3_bind_int64(statement, index, Int64(value)) + } + private static func text(_ statement: OpaquePointer?, _ index: Int32) -> String? { guard let pointer = sqlite3_column_text(statement, index) else { return nil } return String(cString: pointer) } - private static func payloadJSON(_ entry: OpenCodexUsageEntry) -> String { - var object: [String: Any] = [ - "requestId": entry.requestID, - "timestamp": entry.timestamp.timeIntervalSince1970 * 1000, - "provider": entry.provider, - "model": entry.model, - "usageStatus": entry.usageStatus.rawValue, - ] - if let accountLogLabel = entry.accountLogLabel { - object["accountLogLabel"] = accountLogLabel + private static func int(_ statement: OpaquePointer?, _ index: Int32) -> Int? { + guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil } + return Int(sqlite3_column_int64(statement, index)) + } + + // Token fields are stored as independent nullable columns; keep the mapping explicit. + // swiftlint:disable:next function_parameter_count + private static func tokenUsage( + inputTokens: Int?, + outputTokens: Int?, + cachedInputTokens: Int?, + cacheReadInputTokens: Int?, + cacheCreationInputTokens: Int?, + reasoningOutputTokens: Int?, + totalTokens: Int?) -> OpenCodexTokenUsage? + { + if inputTokens == nil, + outputTokens == nil, + cachedInputTokens == nil, + cacheReadInputTokens == nil, + cacheCreationInputTokens == nil, + reasoningOutputTokens == nil, + totalTokens == nil + { + return nil } - if let surface = entry.surface { - object["surface"] = surface + return OpenCodexTokenUsage( + inputTokens: inputTokens, + outputTokens: outputTokens, + cachedInputTokens: cachedInputTokens, + cacheReadInputTokens: cacheReadInputTokens, + cacheCreationInputTokens: cacheCreationInputTokens, + reasoningOutputTokens: reasoningOutputTokens, + totalTokens: totalTokens) + } + + private static func dedupedAndSorted(_ entries: [OpenCodexUsageEntry]) -> [OpenCodexUsageEntry] { + var unique: [String: OpenCodexUsageEntry] = [:] + unique.reserveCapacity(entries.count) + for entry in entries { + unique[entry.requestID] = entry } - if let conversationID = entry.conversationID { - object["conversationId"] = conversationID + return self.sortedEntries(Array(unique.values)) + } + + private static func sortedEntries(_ entries: [OpenCodexUsageEntry]) -> [OpenCodexUsageEntry] { + entries.sorted { + if $0.timestamp != $1.timestamp { + return $0.timestamp < $1.timestamp + } + return $0.requestID < $1.requestID } - if let totalTokens = entry.totalTokens { - object["totalTokens"] = totalTokens + } + + private static func statLog(at url: URL) -> LogIdentity? { + url.withUnsafeFileSystemRepresentation { pointer in + guard let pointer else { return nil } + var status = stat() + guard stat(pointer, &status) == 0 else { return nil } + return LogIdentity( + url: url, + path: url.path, + fileIdentity: "\(status.st_dev):\(status.st_ino)", + size: Int64(status.st_size)) } - if let usage = entry.usage { - var usageObject: [String: Any] = [:] - if let inputTokens = usage.inputTokens { - usageObject["inputTokens"] = inputTokens - } - if let outputTokens = usage.outputTokens { - usageObject["outputTokens"] = outputTokens - } - if let cachedInputTokens = usage.cachedInputTokens { - usageObject["cachedInputTokens"] = cachedInputTokens - } - if let cacheReadInputTokens = usage.cacheReadInputTokens { - usageObject["cacheReadInputTokens"] = cacheReadInputTokens - } - if let cacheCreationInputTokens = usage.cacheCreationInputTokens { - usageObject["cacheCreationInputTokens"] = cacheCreationInputTokens - } - if let reasoningOutputTokens = usage.reasoningOutputTokens { - usageObject["reasoningOutputTokens"] = reasoningOutputTokens - } - if let totalTokens = usage.totalTokens { - usageObject["totalTokens"] = totalTokens + } + + /// Covers only the first min(64 KiB, parsedOffset) bytes; see `canReuseCursor` for the threat model. + private static func prefixDigest(fileURL: URL, parsedOffset: Int64) -> String? { + let length = min(Int64(Self.prefixDigestByteLimit), max(0, parsedOffset)) + let prefix: Data + if length == 0 { + prefix = Data() + } else { + guard let handle = try? FileHandle(forReadingFrom: fileURL) else { return nil } + defer { try? handle.close() } + guard let data = try? handle.read(upToCount: Int(length)), data.count == Int(length) else { + return nil } - object["usage"] = usageObject + prefix = data } - guard let data = try? JSONSerialization.data(withJSONObject: object, options: [.sortedKeys]), - let text = String(data: data, encoding: .utf8) - else { return "{}" } - return text + return SHA256.hash(data: prefix).map { String(format: "%02x", $0) }.joined() + } + + private struct ParseCursor: Equatable, Sendable, Codable { + var path: String + var fileIdentity: String + var parsedOffset: Int64 + var prefixDigest: String + } + + private struct LogIdentity: Equatable, Sendable { + var url: URL + var path: String + var fileIdentity: String + var size: Int64 } } diff --git a/Tests/CodexBarTests/OpenCodexUsageParserTests.swift b/Tests/CodexBarTests/OpenCodexUsageParserTests.swift index 2f405ab586..f58b6b4806 100644 --- a/Tests/CodexBarTests/OpenCodexUsageParserTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageParserTests.swift @@ -38,6 +38,34 @@ struct OpenCodexUsageParserTests { #expect(entries[0].accountLogLabel == nil) } + @Test + func `parseLines splits only on LF so CR and form feed are not record separators`() { + func record(_ id: String) -> String { + """ + {"requestId":"\(id)","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"unreported"} + """ + } + + let crOnly = [record("a"), record("b"), record("c")].joined(separator: "\r") + #expect(OpenCodexUsageParser.parseLines(crOnly).isEmpty) + + let formFeed = [record("f"), record("g")].joined(separator: "\u{000C}") + #expect(OpenCodexUsageParser.parseLines(formFeed).isEmpty) + + let withLineSeparator = record("x") + "\n" + """ + {"requestId":"y","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"unreported","note":"keep\u{2028}together"} + """ + #expect(OpenCodexUsageParser.parseLines(withLineSeparator).map(\.requestID) == ["x", "y"]) + + let withNextLine = record("x") + "\n" + """ + {"requestId":"y","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"unreported","note":"keep\u{0085}together"} + """ + #expect(OpenCodexUsageParser.parseLines(withNextLine).map(\.requestID) == ["x", "y"]) + } + @Test func `does not resolve a default home while tests are running`() { #expect(OpenCodexUsageLog.usageLogURL(environment: ["TESTING_LIBRARY_VERSION": "1"]) == nil) diff --git a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift new file mode 100644 index 0000000000..cbd86fcd5e --- /dev/null +++ b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift @@ -0,0 +1,591 @@ +import Foundation +import SQLite3 +import Testing +@testable import CodexBarCore + +struct OpenCodexUsageStoreIncrementalTests { + @Test + func `incremental load matches a full parse after appended lines`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines( + Harness.line(id: "req-1", input: 10), + Harness.line(id: "req-10", input: 4), + Harness.line(id: "req-ä", input: 7)) + let first = try harness.store.loadEntries(logURL: harness.log) + let firstExpected = try harness.referenceEntries() + #expect(first == firstExpected) + + try harness.appendLines( + Harness.line(id: "req-2", input: 3), + Harness.line(id: "req-b", input: 8), + Harness.line(id: "req-c", input: 1)) + let second = try harness.store.loadEntries(logURL: harness.log) + let secondExpected = try harness.referenceEntries() + #expect(second == secondExpected) + } + + @Test + func `appended lines parse only the tail and an unchanged file reads zero bytes`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines( + Harness.line(id: "seed-1", input: 1), + Harness.line(id: "seed-2", input: 2)) + _ = try harness.store.loadEntries(logURL: harness.log) + + let appended = [ + Harness.line(id: "tail-1", input: 3), + Harness.line(id: "tail-2", input: 4), + Harness.line(id: "tail-3", input: 5), + ] + try harness.appendLines(appended) + + let tailRecorder = OpenCodexUsageParser.LogReadRecorder() + let afterAppend = try OpenCodexUsageStore.withLogReadRecorderForTesting(tailRecorder) { + try harness.store.loadEntries(logURL: harness.log) + } + let afterAppendExpected = try harness.referenceEntries() + #expect(afterAppend == afterAppendExpected) + #expect(tailRecorder.snapshot().completeLines == appended.count) + + let unchangedRecorder = OpenCodexUsageParser.LogReadRecorder() + let unchanged = try OpenCodexUsageStore.withLogReadRecorderForTesting(unchangedRecorder) { + try harness.store.loadEntries(logURL: harness.log) + } + #expect(unchanged == afterAppend) + #expect(unchangedRecorder.snapshot().bytesRead == 0) + #expect(unchangedRecorder.snapshot().completeLines == 0) + } + + @Test + func `partial trailing line is ignored until the newline arrives`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let complete = Harness.line(id: "complete", input: 4) + "\n" + try Data(complete.utf8).write(to: harness.log) + _ = try harness.store.loadEntries(logURL: harness.log) + let completeOffset = try #require(harness.store.parseCursorForTesting()?.parsedOffset) + #expect(completeOffset == Int64(complete.utf8.count)) + + let pending = Harness.line(id: "pending", input: 9) + let pendingBytes = Data(pending.utf8) + let splitIndex = pendingBytes.count / 2 + let head = String(data: pendingBytes.prefix(splitIndex), encoding: .utf8) ?? "" + let tail = String(data: pendingBytes.dropFirst(splitIndex), encoding: .utf8) ?? "" + try harness.append(head) + + let partial = try harness.store.loadEntries(logURL: harness.log) + #expect(partial.map(\.requestID) == ["complete"]) + #expect(harness.store.parseCursorForTesting()?.parsedOffset == completeOffset) + + try harness.append(tail + "\n") + let completed = try harness.store.loadEntries(logURL: harness.log) + #expect(completed.map(\.requestID) == ["complete", "pending"]) + let completedExpected = try harness.referenceEntries() + #expect(completed == completedExpected) + #expect(completed.count(where: { $0.requestID == "pending" }) == 1) + } + + @Test + func `later duplicate request ids win like a full parse`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines(Harness.line(id: "dup", input: 1), Harness.line(id: "other", input: 2)) + _ = try harness.store.loadEntries(logURL: harness.log) + try harness.appendLines(Harness.line(id: "dup", input: 9)) + + let incremental = try harness.store.loadEntries(logURL: harness.log) + let incrementalExpected = try harness.referenceEntries() + #expect(incremental == incrementalExpected) + let duplicate = try #require(incremental.first { $0.requestID == "dup" }) + #expect(duplicate.usage?.inputTokens == 9) + #expect(incremental.count == 2) + } + + @Test + func `truncation and in-place rewrite fall back to a full parse`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let first = Harness.line(id: "id-A", input: 1) + let rewritten = Harness.line(id: "id-B", input: 9) + let width = max(first.utf8.count, rewritten.utf8.count) + let firstPadded = first.padding(toLength: width, withPad: " ", startingAt: 0) + let rewrittenPadded = rewritten.padding(toLength: width, withPad: " ", startingAt: 0) + #expect(Array(firstPadded.utf8) != Array(rewrittenPadded.utf8)) + #expect(firstPadded.utf8.count == rewrittenPadded.utf8.count) + let extra = Harness.line(id: "keep", input: 3) + try harness.writeLines(firstPadded, extra) + _ = try harness.store.loadEntries(logURL: harness.log) + + try harness.rewriteFirstLine(rewrittenPadded) + let afterRewrite = try harness.store.loadEntries(logURL: harness.log) + let rewriteExpected = try harness.referenceEntries() + #expect(afterRewrite == rewriteExpected) + #expect(afterRewrite.map(\.requestID) == ["id-B", "keep"]) + + let prefix = rewrittenPadded + "\n" + try harness.truncate(to: prefix.utf8.count) + let afterTruncate = try harness.store.loadEntries(logURL: harness.log) + let truncateExpected = try harness.referenceEntries() + #expect(afterTruncate == truncateExpected) + #expect(afterTruncate.map(\.requestID) == ["id-B"]) + } + + @Test + func `rotation with a shared 64 kib prefix falls back to a full parse`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let prefixLine = Harness.paddedLine(id: "pad", input: 1, padByteCount: Harness.prefixDigestByteLimit) + let firstTail = Harness.line(id: "id-A", input: 1) + let rotatedTail = Harness.line(id: "rotated", input: 6) + let width = max(firstTail.utf8.count, rotatedTail.utf8.count) + let firstPadded = firstTail.padding(toLength: width, withPad: " ", startingAt: 0) + let rotatedPadded = rotatedTail.padding(toLength: width, withPad: " ", startingAt: 0) + try harness.writeLines(prefixLine, firstPadded) + _ = try harness.store.loadEntries(logURL: harness.log) + let oldOffset = try #require(harness.store.parseCursorForTesting()?.parsedOffset) + + try FileManager.default.removeItem(at: harness.log) + try harness.writeLines(prefixLine, rotatedPadded) + let newSize = try FileManager.default.attributesOfItem(atPath: harness.log.path)[.size] as? Int64 + #expect(newSize ?? 0 >= oldOffset) + + let afterRotation = try harness.store.loadEntries(logURL: harness.log) + let rotationExpected = try harness.referenceEntries() + #expect(afterRotation == rotationExpected) + #expect(afterRotation.map(\.requestID) == ["pad", "rotated"]) + } + + @Test + func `in-place rewrite past the 64 kib digest window is not detected`() throws { + // The prefix digest only covers min(64 KiB, parsedOffset). An in-place rewrite past that + // window that preserves size and inode is invisible; the log is append-only. + let harness = try Harness.make() + defer { harness.tearDown() } + + let prefixLine = Harness.paddedLine(id: "pad", input: 1, padByteCount: Harness.prefixDigestByteLimit) + let firstTail = Harness.line(id: "id-A", input: 1) + let rewrittenTail = Harness.line(id: "id-B", input: 9) + let width = max(firstTail.utf8.count, rewrittenTail.utf8.count) + let firstPadded = firstTail.padding(toLength: width, withPad: " ", startingAt: 0) + let rewrittenPadded = rewrittenTail.padding(toLength: width, withPad: " ", startingAt: 0) + try harness.writeLines(prefixLine, firstPadded) + _ = try harness.store.loadEntries(logURL: harness.log) + + let prefixByteCount = prefixLine.utf8.count + 1 + try harness.rewriteLine(atByteOffset: prefixByteCount, rewrittenPadded) + let afterRewrite = try harness.store.loadEntries(logURL: harness.log) + #expect(afterRewrite.map(\.requestID) == ["id-A", "pad"]) + let full = try OpenCodexUsageParser.parse(fileURL: harness.log) + #expect(full.map(\.requestID) == ["pad", "id-B"]) + } + + @Test + func `schema v1 payload caches rebuild from the log`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines(Harness.line(id: "live", input: 9, output: 2, total: 11)) + try Harness.writeV1Database( + at: harness.cacheRoot.appendingPathComponent(OpenCodexUsageStore.databaseFilename), + staleRequestID: "stale-v1") + + let entries = try harness.store.loadEntries(logURL: harness.log) + let schemaExpected = try harness.referenceEntries() + #expect(entries == schemaExpected) + #expect(entries.map(\.requestID) == ["live"]) + #expect(entries.contains { $0.requestID == "stale-v1" } == false) + #expect(entries[0].usage?.inputTokens == 9) + #expect(entries[0].usage?.outputTokens == 2) + #expect(entries[0].totalTokens == 11) + } + + @Test + func `CRLF file parse matches parseLines and offset parse skips a trailing partial line`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let text = """ + \(Harness.line(id: "one", input: 1))\r + \(Harness.line(id: "two", input: 2)) + + not-json + \(Harness.line(id: "three", input: 3)) + """ + try Data(text.utf8).write(to: harness.log) + + let fromFile = try OpenCodexUsageParser.parse(fileURL: harness.log) + #expect(fromFile.map(\.requestID) == OpenCodexUsageParser.parseLines(text).map(\.requestID)) + #expect(fromFile.map(\.requestID) == ["one", "two", "three"]) + + let complete = Harness.line(id: "full", input: 4) + "\n" + let partial = String(data: Data(Harness.line(id: "half", input: 5).utf8).prefix(12), encoding: .utf8) ?? "" + try Data((complete + partial).utf8).write(to: harness.log) + let sliced = try OpenCodexUsageParser.parse(fileURL: harness.log, from: 0) + #expect(sliced.entries.map(\.requestID) == ["full"]) + #expect(sliced.nextOffset == Int64(complete.utf8.count)) + } + + @Test + func `complete newline-less trailing record does not advance the cursor`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let complete = Harness.line(id: "head", input: 1) + "\n" + let trailing = Harness.line(id: "tail", input: 2) + try Data((complete + trailing).utf8).write(to: harness.log) + + let sliced = try OpenCodexUsageParser.parse(fileURL: harness.log, from: 0) + #expect(sliced.entries.map(\.requestID) == ["head", "tail"]) + #expect(sliced.nextOffset == Int64(complete.utf8.count)) + + let loaded = try harness.store.loadEntries(logURL: harness.log) + #expect(loaded.map(\.requestID) == ["head", "tail"]) + #expect(harness.store.parseCursorForTesting()?.parsedOffset == Int64(complete.utf8.count)) + } + + @Test + func `gluing bytes onto a newline-less trailing record matches a full parse`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + let recordA = Harness.line(id: "a", input: 1) + let recordB = Harness.line(id: "b", input: 2) + try Data(recordA.utf8).write(to: harness.log) + let first = try harness.store.loadEntries(logURL: harness.log) + #expect(first.map(\.requestID) == ["a"]) + #expect(harness.store.parseCursorForTesting()?.parsedOffset == 0) + + try harness.append(recordB + "\n") + let incremental = try harness.store.loadEntries(logURL: harness.log) + let full = try harness.referenceEntries() + #expect(incremental == full) + #expect(full.isEmpty) + } + + @Test + func `schema v2 round trip preserves nil usage zero fields and mismatched totals`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines( + Harness.lineWithoutUsage(id: "no-usage"), + Harness.lineWithZeroInputUsage(id: "zero-field"), + Harness.line(id: "mismatch", input: 10, output: 2, total: 99, usageTotal: 12)) + _ = try harness.store.loadEntries(logURL: harness.log) + let second = try harness.store.loadEntries(logURL: harness.log) + let expected = try harness.referenceEntries() + #expect(second == expected) + + let noUsage = try #require(second.first { $0.requestID == "no-usage" }) + #expect(noUsage.usage == nil) + let zeroField = try #require(second.first { $0.requestID == "zero-field" }) + #expect(zeroField.usage?.inputTokens == 0) + #expect(zeroField.usage?.outputTokens == nil) + #expect(zeroField.usage?.totalTokens == nil) + let mismatch = try #require(second.first { $0.requestID == "mismatch" }) + #expect(mismatch.totalTokens == 99) + #expect(mismatch.usage?.totalTokens == 12) + #expect(mismatch.usage?.inputTokens == 10) + } + + @Test + func `failed begin immediate leaves rows and cursor untouched`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines(Harness.line(id: "keep", input: 1)) + _ = try harness.store.loadEntries(logURL: harness.log) + let cursorBefore = try #require(harness.store.parseCursorForTesting()) + let rowsBefore = try harness.sqliteRequestIDs() + #expect(rowsBefore == ["keep"]) + + try harness.appendLines(Harness.line(id: "new", input: 2)) + try harness.withExclusiveSQLiteWriteLock { + _ = try harness.store.loadEntries(logURL: harness.log) + let cursorAfter = try #require(harness.store.parseCursorForTesting()) + #expect(cursorAfter.parsedOffset == cursorBefore.parsedOffset) + #expect(cursorAfter.prefixDigest == cursorBefore.prefixDigest) + #expect(cursorAfter.fileIdentity == cursorBefore.fileIdentity) + #expect(try harness.sqliteRequestIDs() == rowsBefore) + } + } + + @Test + func `incremental reload falls back to a full parse when the cached rows cannot be read`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines(Harness.line(id: "seed", input: 1)) + _ = try harness.store.loadEntries(logURL: harness.log) + try harness.dropEntriesTable() + try harness.appendLines(Harness.line(id: "tail", input: 2)) + + let recovered = try harness.store.loadEntries(logURL: harness.log) + let expected = try harness.referenceEntries() + #expect(recovered == expected) + #expect(recovered.map(\.requestID) == ["seed", "tail"]) + } +} + +private struct Harness { + let root: URL + let cacheRoot: URL + let log: URL + let store: OpenCodexUsageStore + + static func make() throws -> Harness { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("OpenCodexUsageStoreIncremental-\(UUID().uuidString)", isDirectory: true) + let cacheRoot = root.appendingPathComponent("cache", isDirectory: true) + try FileManager.default.createDirectory(at: cacheRoot, withIntermediateDirectories: true) + return Harness( + root: root, + cacheRoot: cacheRoot, + log: root.appendingPathComponent("usage.jsonl"), + store: OpenCodexUsageStore(cacheRoot: cacheRoot)) + } + + func tearDown() { + try? FileManager.default.removeItem(at: self.root) + } + + static let prefixDigestByteLimit = 64 * 1024 + + static func line( + id: String, + input: Int, + output: Int = 1, + total: Int? = nil, + usageTotal: Int? = nil) -> String + { + let resolvedTotal = total ?? (input + output) + let resolvedUsageTotal = usageTotal ?? resolvedTotal + return """ + {"requestId":"\(id)","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"reported","usage":{"inputTokens":\(input),"outputTokens":\(output),\ + "totalTokens":\(resolvedUsageTotal)},"totalTokens":\(resolvedTotal)} + """ + } + + static func lineWithoutUsage(id: String) -> String { + """ + {"requestId":"\(id)","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"unreported"} + """ + } + + static func lineWithZeroInputUsage(id: String) -> String { + """ + {"requestId":"\(id)","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"reported","usage":{"inputTokens":0}} + """ + } + + static func paddedLine(id: String, input: Int, padByteCount: Int) -> String { + let pad = String(repeating: "x", count: padByteCount) + return String(self.line(id: id, input: input).dropLast()) + ",\"pad\":\"\(pad)\"}" + } + + func writeLines(_ lines: String...) throws { + try self.writeLines(Array(lines)) + } + + func writeLines(_ lines: [String]) throws { + let body = lines.map { $0.hasSuffix("\n") ? $0 : $0 + "\n" }.joined() + try Data(body.utf8).write(to: self.log) + } + + func appendLines(_ lines: [String]) throws { + try self.append(lines.map { $0.hasSuffix("\n") ? $0 : $0 + "\n" }.joined()) + } + + func appendLines(_ lines: String...) throws { + try self.appendLines(Array(lines)) + } + + func append(_ text: String) throws { + let handle = try FileHandle(forUpdating: self.log) + defer { try? handle.close() } + _ = try handle.seekToEnd() + try handle.write(contentsOf: Data(text.utf8)) + } + + func rewriteFirstLine(_ line: String) throws { + try self.rewriteLine(atByteOffset: 0, line) + } + + func rewriteLine(atByteOffset offset: Int, _ line: String) throws { + let replacement = Data((line.hasSuffix("\n") ? line : line + "\n").utf8) + let handle = try FileHandle(forUpdating: self.log) + defer { try? handle.close() } + try handle.seek(toOffset: UInt64(offset)) + try handle.write(contentsOf: replacement) + } + + var databaseURL: URL { + self.cacheRoot.appendingPathComponent(OpenCodexUsageStore.databaseFilename) + } + + func sqliteRequestIDs() throws -> [String] { + var db: OpaquePointer? + guard sqlite3_open_v2(self.databaseURL.path, &db, SQLITE_OPEN_READONLY, nil) == SQLITE_OK else { + sqlite3_close(db) + throw FixtureError.sqlite + } + defer { sqlite3_close(db) } + var statement: OpaquePointer? + guard sqlite3_prepare_v2( + db, + "SELECT request_id FROM entries ORDER BY request_id", + -1, + &statement, + nil) == SQLITE_OK + else { + throw FixtureError.sqlite + } + defer { sqlite3_finalize(statement) } + var ids: [String] = [] + while sqlite3_step(statement) == SQLITE_ROW { + if let pointer = sqlite3_column_text(statement, 0) { + ids.append(String(cString: pointer)) + } + } + return ids + } + + func dropEntriesTable() throws { + var db: OpaquePointer? + guard sqlite3_open(self.databaseURL.path, &db) == SQLITE_OK else { + sqlite3_close(db) + throw FixtureError.sqlite + } + defer { sqlite3_close(db) } + guard sqlite3_exec(db, "DROP TABLE entries", nil, nil, nil) == SQLITE_OK else { + throw FixtureError.sqlite + } + } + + func withExclusiveSQLiteWriteLock(_ body: () throws -> Void) throws { + var db: OpaquePointer? + guard sqlite3_open_v2(self.databaseURL.path, &db, SQLITE_OPEN_READWRITE, nil) == SQLITE_OK else { + sqlite3_close(db) + throw FixtureError.sqlite + } + defer { + _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) + sqlite3_close(db) + } + sqlite3_busy_timeout(db, 0) + guard sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) == SQLITE_OK else { + throw FixtureError.sqlite + } + try body() + } + + func truncate(to byteCount: Int) throws { + let handle = try FileHandle(forUpdating: self.log) + defer { try? handle.close() } + try handle.truncate(atOffset: UInt64(byteCount)) + } + + func referenceEntries() throws -> [OpenCodexUsageEntry] { + let parsed = try OpenCodexUsageParser.parse(fileURL: self.log) + var unique: [String: OpenCodexUsageEntry] = [:] + for entry in parsed { + unique[entry.requestID] = entry + } + return unique.values.sorted { + if $0.timestamp != $1.timestamp { + return $0.timestamp < $1.timestamp + } + return $0.requestID < $1.requestID + } + } + + static func writeV1Database(at databaseURL: URL, staleRequestID: String) throws { + try FileManager.default.createDirectory( + at: databaseURL.deletingLastPathComponent(), + withIntermediateDirectories: true) + var db: OpaquePointer? + guard sqlite3_open(databaseURL.path, &db) == SQLITE_OK else { + sqlite3_close(db) + throw FixtureError.sqlite + } + defer { sqlite3_close(db) } + let schema = """ + CREATE TABLE meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE TABLE entries ( + request_id TEXT PRIMARY KEY, + timestamp REAL NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL, + usage_status TEXT NOT NULL, + account_label TEXT, + surface TEXT, + conversation_id TEXT, + payload TEXT NOT NULL + ); + PRAGMA user_version = 1; + """ + guard sqlite3_exec(db, schema, nil, nil, nil) == SQLITE_OK else { throw FixtureError.sqlite } + + try self.exec( + db, + sql: "INSERT INTO meta(key, value) VALUES(?, ?)", + bind: { statement in + self.bind(statement, 1, "identity") + self.bind(statement, 2, "/tmp/stale|1|1") + }) + + let payload = """ + {"requestId":"\(staleRequestID)","timestamp":1784179200000,"provider":"openai","model":"gpt-5.4",\ + "usageStatus":"reported","usage":{"inputTokens":1,"outputTokens":1,"totalTokens":2},"totalTokens":2} + """ + try self.exec( + db, + sql: """ + INSERT INTO entries( + request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, payload + ) VALUES (?, ?, ?, ?, ?, NULL, NULL, NULL, ?) + """, + bind: { statement in + self.bind(statement, 1, staleRequestID) + sqlite3_bind_double(statement, 2, 1_784_179_200) + self.bind(statement, 3, "openai") + self.bind(statement, 4, "gpt-5.4") + self.bind(statement, 5, "reported") + self.bind(statement, 6, payload) + }) + } + + private static func exec( + _ db: OpaquePointer?, + sql: String, + bind: (OpaquePointer?) -> Void) throws + { + var statement: OpaquePointer? + guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else { throw FixtureError.sqlite } + defer { sqlite3_finalize(statement) } + bind(statement) + guard sqlite3_step(statement) == SQLITE_DONE else { throw FixtureError.sqlite } + } + + private static func bind(_ statement: OpaquePointer?, _ index: Int32, _ value: String) { + sqlite3_bind_text(statement, index, value, -1, unsafeBitCast(-1, to: sqlite3_destructor_type.self)) + } + + private enum FixtureError: Error { + case sqlite + } +} From 7b57ab10b8f9ea2dad8a093195c41163ad503518 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Sat, 22 Aug 2026 10:51:37 -0700 Subject: [PATCH 4/7] Reject stale incremental cursor commits under the write lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The app and the CLI open the same OpenCodex usage cache, so two loaders can race: A reads cursor C1 and parses from it, B reads C1, parses further, takes the write lock and commits C2 > C1, and A then takes the lock and commits its snapshot — moving the durable cursor backwards and re-inserting rows B already stored. The same bytes are then parsed again on the next refresh, and after a later truncation the `size == parsedOffset` cache-hit path can serve rows for content no longer in the log. `writeEntries` now re-reads the durable cursor after `BEGIN IMMEDIATE` succeeds and before any mutation. For an incremental append (`replaceAll == false`), a durable cursor with the same path and file identity whose `parsedOffset` is at least the proposed one means this work is stale: the transaction rolls back without touching the cursor or the rows. Full reloads re-derived the whole file and still replace even a newer cursor, which is what truncation, rotation and a schema rebuild need. `loadEntries` re-runs its cursor path once against the freshly committed durable state and falls back to a full reload if that write is stale too, so a caller never sees a partial result and never loops. Test: `stale incremental write does not regress a newer durable cursor` is ordered rather than racy — load once to capture a cursor, append and load again to commit a newer one, then drive the write path with the first snapshot through a test-only seam and assert the newer cursor survives, no duplicate rows exist, and a subsequent load still equals a full re-parse. Co-Authored-By: Claude Fable 5 --- .../OpenCodexUsage/OpenCodexUsageStore.swift | 122 ++++++++++++++---- .../OpenCodexUsageStoreIncrementalTests.swift | 53 ++++++++ 2 files changed, 149 insertions(+), 26 deletions(-) diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift index 68044f58ab..24878a9918 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift @@ -53,28 +53,64 @@ public struct OpenCodexUsageStore: Sendable { } public func loadEntries(logURL: URL, fileManager: FileManager = .default) throws -> [OpenCodexUsageEntry] { - guard fileManager.fileExists(atPath: logURL.path) else { return [] } - guard let identity = Self.statLog(at: logURL) else { return [] } - let cursor = self.readCursor() - if let cursor, Self.canReuseCursor(cursor, identity: identity) { - if identity.size == cursor.parsedOffset { - if let cached = self.readCachedEntries() { - return cached - } - } else { - return try self.incrementalReload( + var shouldRetryStaleWrite = true + while true { + guard fileManager.fileExists(atPath: logURL.path) else { return [] } + guard let identity = Self.statLog(at: logURL) else { return [] } + let cursor = self.readCursor() + if let cursor, Self.canReuseCursor(cursor, identity: identity) { + if identity.size == cursor.parsedOffset { + if let cached = self.readCachedEntries() { + return cached + } + } else if let entries = try self.incrementalReload( logURL: logURL, identity: identity, cursor: cursor, fileManager: fileManager) + { + return entries + } else if shouldRetryStaleWrite { + // Another loader committed a later same-file cursor. Re-run the cursor + // path once against that durable state, then fall back to a full reload. + shouldRetryStaleWrite = false + continue + } else { + return try self.fullReload( + logURL: logURL, + identity: identity, + fileManager: fileManager) + } } + return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) } - return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) } - func parseCursorForTesting() -> (parsedOffset: Int64, prefixDigest: String, fileIdentity: String)? { + func parseCursorForTesting() -> ( + parsedOffset: Int64, + prefixDigest: String, + fileIdentity: String, + path: String)? + { guard let cursor = self.readCursor() else { return nil } - return (cursor.parsedOffset, cursor.prefixDigest, cursor.fileIdentity) + return (cursor.parsedOffset, cursor.prefixDigest, cursor.fileIdentity, cursor.path) + } + + func writeIncrementalEntriesForTesting( + _ entries: [OpenCodexUsageEntry], + path: String, + fileIdentity: String, + parsedOffset: Int64, + prefixDigest: String) + { + _ = self.writeEntries( + entries, + cursor: ParseCursor( + path: path, + fileIdentity: fileIdentity, + parsedOffset: parsedOffset, + prefixDigest: prefixDigest), + replaceAll: false) } private func fullReload( @@ -93,11 +129,12 @@ public struct OpenCodexUsageStore: Sendable { return entries } + /// Returns `nil` when the incremental write observed a newer same-file durable cursor. private func incrementalReload( logURL: URL, identity: LogIdentity, cursor: ParseCursor, - fileManager: FileManager) throws -> [OpenCodexUsageEntry] + fileManager: FileManager) throws -> [OpenCodexUsageEntry]? { guard let existing = self.readCachedEntries() else { return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) @@ -115,13 +152,16 @@ public struct OpenCodexUsageStore: Sendable { let digest = nextOffset == cursor.parsedOffset ? cursor.prefixDigest : (Self.prefixDigest(fileURL: logURL, parsedOffset: nextOffset) ?? "") - self.insertCachedEntries( + if self.insertCachedEntries( committed, cursor: ParseCursor( path: identity.path, fileIdentity: identity.fileIdentity, parsedOffset: nextOffset, - prefixDigest: digest)) + prefixDigest: digest)) == .stale + { + return nil + } return Self.dedupedAndSorted(existing + committed + pending) } @@ -176,23 +216,38 @@ public struct OpenCodexUsageStore: Sendable { self.writeEntries(entries, cursor: cursor, replaceAll: true) } - private func insertCachedEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor) { + private func insertCachedEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor) -> CachedWriteResult { self.writeEntries(entries, cursor: cursor, replaceAll: false) } - private func writeEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor, replaceAll: Bool) { - guard let db = self.open(readOnly: false) else { return } + @discardableResult + private func writeEntries( + _ entries: [OpenCodexUsageEntry], + cursor: ParseCursor, + replaceAll: Bool) -> CachedWriteResult + { + guard let db = self.open(readOnly: false) else { return .applied } defer { sqlite3_close(db) } - guard sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) == SQLITE_OK else { return } + guard sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) == SQLITE_OK else { return .applied } + // Incremental appends only (`replaceAll == false`): after BEGIN IMMEDIATE, re-read the + // durable cursor. A concurrent loader may already have committed a later parsedOffset + // for this path + fileIdentity; writing this snapshot would move the cursor backwards + // and re-insert rows it already stored. Full reloads re-derived the whole file and must + // still replace even a newer cursor (truncation, rotation, schema rebuild). + if !replaceAll, Self.isStaleIncrementalCursor(proposed: cursor, durable: Self.parseCursor(from: db)) { + _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) + return .stale + } if replaceAll { _ = sqlite3_exec(db, "DELETE FROM entries", nil, nil, nil) } Self.setCursor(db, cursor) guard Self.insertEntries(db, entries) else { _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) - return + return .applied } _ = sqlite3_exec(db, "COMMIT", nil, nil, nil) + return .applied } private func open(readOnly: Bool) -> OpaquePointer? { @@ -221,12 +276,22 @@ public struct OpenCodexUsageStore: Sendable { private func readCursor() -> ParseCursor? { guard let db = self.open(readOnly: true) else { return nil } defer { sqlite3_close(db) } - guard Self.userVersion(db) == Self.schemaVersion, - let raw = Self.meta(db, key: Self.cursorMetaKey), - let data = raw.data(using: .utf8), - let cursor = try? JSONDecoder().decode(ParseCursor.self, from: data) + guard Self.userVersion(db) == Self.schemaVersion else { return nil } + return Self.parseCursor(from: db) + } + + private static func parseCursor(from db: OpaquePointer?) -> ParseCursor? { + guard let raw = self.meta(db, key: cursorMetaKey), + let data = raw.data(using: .utf8) else { return nil } - return cursor + return try? JSONDecoder().decode(ParseCursor.self, from: data) + } + + private static func isStaleIncrementalCursor(proposed: ParseCursor, durable: ParseCursor?) -> Bool { + guard let durable else { return false } + return durable.path == proposed.path + && durable.fileIdentity == proposed.fileIdentity + && durable.parsedOffset >= proposed.parsedOffset } /// Threat model: the digest covers only the first min(64 KiB, parsedOffset) bytes, so an @@ -459,6 +524,11 @@ public struct OpenCodexUsageStore: Sendable { return SHA256.hash(data: prefix).map { String(format: "%02x", $0) }.joined() } + private enum CachedWriteResult: Equatable { + case applied + case stale + } + private struct ParseCursor: Equatable, Sendable, Codable { var path: String var fileIdentity: String diff --git a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift index cbd86fcd5e..f9c552d1a1 100644 --- a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift @@ -318,6 +318,43 @@ struct OpenCodexUsageStoreIncrementalTests { } } + @Test + func `stale incremental write does not regress a newer durable cursor`() throws { + // Deterministic: commit a newer cursor, then force the write path with the older + // snapshot. No thread interleaving. + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines(Harness.line(id: "req-1", input: 1), Harness.line(id: "req-2", input: 2)) + let first = try harness.store.loadEntries(logURL: harness.log) + let firstCursor = try #require(harness.store.parseCursorForTesting()) + + try harness.appendLines(Harness.line(id: "req-3", input: 3), Harness.line(id: "req-4", input: 4)) + _ = try harness.store.loadEntries(logURL: harness.log) + let newerCursor = try #require(harness.store.parseCursorForTesting()) + #expect(newerCursor.parsedOffset > firstCursor.parsedOffset) + + harness.store.writeIncrementalEntriesForTesting( + first, + path: firstCursor.path, + fileIdentity: firstCursor.fileIdentity, + parsedOffset: firstCursor.parsedOffset, + prefixDigest: firstCursor.prefixDigest) + + let afterStale = try #require(harness.store.parseCursorForTesting()) + #expect(afterStale.parsedOffset == newerCursor.parsedOffset) + #expect(afterStale.prefixDigest == newerCursor.prefixDigest) + #expect(afterStale.fileIdentity == newerCursor.fileIdentity) + #expect(afterStale.path == newerCursor.path) + + let expected = try harness.referenceEntries() + #expect(try harness.sqliteEntryCount() == expected.count) + #expect(try harness.sqliteRequestIDs() == expected.map(\.requestID).sorted()) + + let loaded = try harness.store.loadEntries(logURL: harness.log) + #expect(loaded == expected) + } + @Test func `incremental reload falls back to a full parse when the cached rows cannot be read`() throws { let harness = try Harness.make() @@ -434,6 +471,22 @@ private struct Harness { self.cacheRoot.appendingPathComponent(OpenCodexUsageStore.databaseFilename) } + func sqliteEntryCount() throws -> Int { + var db: OpaquePointer? + guard sqlite3_open_v2(self.databaseURL.path, &db, SQLITE_OPEN_READONLY, nil) == SQLITE_OK else { + sqlite3_close(db) + throw FixtureError.sqlite + } + defer { sqlite3_close(db) } + var statement: OpaquePointer? + guard sqlite3_prepare_v2(db, "SELECT COUNT(*) FROM entries", -1, &statement, nil) == SQLITE_OK else { + throw FixtureError.sqlite + } + defer { sqlite3_finalize(statement) } + guard sqlite3_step(statement) == SQLITE_ROW else { throw FixtureError.sqlite } + return Int(sqlite3_column_int64(statement, 0)) + } + func sqliteRequestIDs() throws -> [String] { var db: OpaquePointer? guard sqlite3_open_v2(self.databaseURL.path, &db, SQLITE_OPEN_READONLY, nil) == SQLITE_OK else { From 79baa7694ab97ce384289a5ab61e7faff8fabb8a Mon Sep 17 00:00:00 2001 From: olddonkey Date: Sat, 22 Aug 2026 14:15:53 -0700 Subject: [PATCH 5/7] Revalidate the log after the tail read before committing an incremental result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `loadEntries` validates the file identity and prefix digest before parsing, but the tail read happens after that check. A rotation or replacement in that window made the parse start at the old `parsedOffset` inside the new file, and those bytes were then merged with cached rows belonging to the old file and persisted under the old identity — a refresh could publish a mixed old/new snapshot and store a cursor for a file that no longer exists at that path. `incrementalReload` now re-stats the log after the parse and before anything is inserted or returned, and requires the same path and `st_dev`/`st_ino` plus a cursor that still validates against the post-read stat (size and prefix digest included). On mismatch it discards the parsed tail, writes nothing, and runs a single `fullReload` against the file as it now exists — the same bounded fallback the stale-cursor path uses. A replacement that preserves path, device, inode, size and the first 64 KiB is still undetectable, which is the digest threat model already documented at `canReuseCursor`. Tests drive the window deterministically through a task-local post-parse hook (unset and free in production, with a case asserting it is unset by default): the log is replaced with a different inode while the parse result is in hand, once with a replacement longer than the old offset and once shorter. Both assert the returned entries equal a full re-parse of the replacement, that no request ID from the old file survives, that the persisted cursor describes the replacement, that no duplicate rows exist, and that the next load is a cache hit reading zero bytes. Co-Authored-By: Claude Fable 5 --- .../OpenCodexUsage/OpenCodexUsageStore.swift | 36 +++++++++ .../OpenCodexUsageStoreIncrementalTests.swift | 75 +++++++++++++++++++ 2 files changed, 111 insertions(+) diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift index 24878a9918..2a50599fbc 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift @@ -28,6 +28,9 @@ public struct OpenCodexUsageStore: Sendable { self.databaseURL = cacheRoot.appendingPathComponent(Self.databaseFilename, isDirectory: false) } + /// Test-only. Unset in production; the optional call in `incrementalReload` is a no-op. + @TaskLocal private static var incrementalPostParseHookForTesting: (@Sendable () -> Void)? + static func withLogReadRecorderForTesting( _ recorder: OpenCodexUsageParser.LogReadRecorder, operation: () throws -> T) rethrows -> T @@ -35,6 +38,19 @@ public struct OpenCodexUsageStore: Sendable { try OpenCodexUsageParser.withLogReadRecorderForTesting(recorder, operation: operation) } + static func withIncrementalPostParseHookForTesting( + _ hook: @escaping @Sendable () -> Void, + operation: () throws -> T) rethrows -> T + { + try self.$incrementalPostParseHookForTesting.withValue(hook) { + try operation() + } + } + + static func incrementalPostParseHookInstalledForTesting() -> Bool { + self.incrementalPostParseHookForTesting != nil + } + public func loadSnapshot( logURL: URL, now: Date, @@ -143,6 +159,16 @@ public struct OpenCodexUsageStore: Sendable { fileURL: logURL, from: cursor.parsedOffset, fileManager: fileManager) + Self.incrementalPostParseHookForTesting?() + // Closes the TOCTOU window between the pre-read `canReuseCursor` check and this tail + // parse: a rotation or replacement in that window would otherwise merge cached rows from + // the old file with bytes from the new one and persist a cursor for a path that no longer + // names that file. A replacement that preserves path, st_dev, st_ino, size, AND the first + // min(64 KiB, parsedOffset) bytes remains undetectable. + guard let postIdentity = Self.statLog(at: logURL) else { return [] } + if !Self.isSameLogAfterTailRead(preRead: identity, postRead: postIdentity, cursor: cursor) { + return try self.fullReload(logURL: logURL, identity: postIdentity, fileManager: fileManager) + } let nextOffset = parsed.nextOffset let committed = parsed.newlineTerminatedEntries let pending = parsed.pendingTrailingEntries @@ -305,6 +331,16 @@ public struct OpenCodexUsageStore: Sendable { && self.prefixDigest(fileURL: identity.url, parsedOffset: cursor.parsedOffset) == cursor.prefixDigest } + private static func isSameLogAfterTailRead( + preRead: LogIdentity, + postRead: LogIdentity, + cursor: ParseCursor) -> Bool + { + postRead.path == preRead.path + && postRead.fileIdentity == preRead.fileIdentity + && self.canReuseCursor(cursor, identity: postRead) + } + private static func ensureSchema(_ db: OpaquePointer?) { // `!= schemaVersion` deliberately rebuilds a NEWER database too: a downgrade must not // read a schema it does not understand. diff --git a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift index f9c552d1a1..f2be75c6c9 100644 --- a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift @@ -370,6 +370,81 @@ struct OpenCodexUsageStoreIncrementalTests { #expect(recovered == expected) #expect(recovered.map(\.requestID) == ["seed", "tail"]) } + + @Test + func `incremental post-parse hook is unset by default`() { + #expect(OpenCodexUsageStore.incrementalPostParseHookInstalledForTesting() == false) + } + + @Test + func `post-parse replacement longer than the cursor falls back to a full parse`() throws { + try self.assertPostParseReplacementFallsBack(longerThanCursor: true) + } + + @Test + func `post-parse replacement shorter than the cursor falls back to a full parse`() throws { + try self.assertPostParseReplacementFallsBack(longerThanCursor: false) + } + + private func assertPostParseReplacementFallsBack(longerThanCursor: Bool) throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines( + Harness.line(id: "seed-1", input: 1), + Harness.line(id: "seed-2", input: 2)) + _ = try harness.store.loadEntries(logURL: harness.log) + let firstCursor = try #require(harness.store.parseCursorForTesting()) + try harness.appendLines(Harness.line(id: "tail-1", input: 3)) + + let replacementLines: [String] = if longerThanCursor { + [ + Harness.line(id: "new-1", input: 11), + Harness.line(id: "new-2", input: 12), + Harness.paddedLine( + id: "new-pad", + input: 13, + padByteCount: max(64, Int(firstCursor.parsedOffset))), + ] + } else { + [Harness.line(id: "new-only", input: 5)] + } + let body = replacementLines.map { $0.hasSuffix("\n") ? $0 : $0 + "\n" }.joined() + let bodyCount = Int64(body.utf8.count) + if longerThanCursor { + #expect(bodyCount > firstCursor.parsedOffset) + } else { + #expect(bodyCount < firstCursor.parsedOffset) + } + #expect(OpenCodexUsageStore.incrementalPostParseHookInstalledForTesting() == false) + + let log = harness.log + let loaded = try OpenCodexUsageStore.withIncrementalPostParseHookForTesting { + try? FileManager.default.removeItem(at: log) + try? Data(body.utf8).write(to: log) + } operation: { + try harness.store.loadEntries(logURL: log) + } + + let expected = try harness.referenceEntries() + #expect(loaded == expected) + let loadedIDs = loaded.map(\.requestID) + #expect(Set(loadedIDs).count == loadedIDs.count) + #expect(Set(loadedIDs).isDisjoint(with: ["seed-1", "seed-2", "tail-1"])) + #expect(try harness.sqliteRequestIDs() == loadedIDs.sorted()) + + let cursor = try #require(harness.store.parseCursorForTesting()) + #expect(cursor.path == harness.log.path) + #expect(cursor.fileIdentity != firstCursor.fileIdentity) + #expect(cursor.parsedOffset == bodyCount) + + let recorder = OpenCodexUsageParser.LogReadRecorder() + let cached = try OpenCodexUsageStore.withLogReadRecorderForTesting(recorder) { + try harness.store.loadEntries(logURL: harness.log) + } + #expect(cached == loaded) + #expect(recorder.snapshot().bytesRead == 0) + } } private struct Harness { From e331eab0d714910ff16b10e64f588a7797dac394 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Sat, 22 Aug 2026 15:36:03 -0700 Subject: [PATCH 6/7] Harden the incremental cache against concurrent writers, truncation and I/O errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An independent side-effect review found three ways this cache could misbehave at runtime. None are theoretical; two are regressions this branch introduced. Mixed ledgers in one cache. Every OpenCodex home shares one cache database while the log path differs per home, and this branch had replaced the per-read identity check with a cursor check made from a separate connection. A loader could read its own cursor, have another home's CLI replace the cache, then read that home's rows as its baseline — and the write-lock check only rejected same-file older offsets, so it kept those rows, reset the cursor and appended its own tail. The cursor and the cached rows are now read in one transaction, the write carries the exact base cursor the parse was derived from, and an incremental write is rejected whenever the durable cursor differs from that base in any field. A crash class. The full parse mapped the whole log and then walked it; a truncation between mapping and touching those pages raises SIGBUS, which Swift cannot catch — the app dies. The parser now opens one descriptor, snapshots its size with fstat, and reads exactly the range it intends to parse, treating a short read as "changed under us". That also bounds a read that would otherwise chase a growing file, and gives the full-reload path the same post-read identity check the incremental path already had. The cost is that a full parse now holds the file's bytes: on a 42 MB log the cold rebuild peaks at ~290 MB instead of ~187 MB, still well under the ~485 MB of the current implementation, and it only runs on a cold cache or a schema rebuild — the per-refresh path reads only the appended tail and is unchanged at ~118 MB. Transient failures shown as "no spend". Any stat failure returned an empty array, and the dashboard turns that into a confirmed-empty state that removes the OpenCodex source. Only ENOENT/ENOTDIR now mean "absent"; every other errno throws so the source is marked unavailable instead. Schema v2 also moves to its own database filename, so downgrading to an older build leaves it a usable v1 cache rather than one it can read but never write. Tests: an incremental load whose cache was replaced by another home keeps only its own entries; a circular symlink and an unreadable directory throw instead of returning empty; a missing log still returns empty without throwing; the legacy v1 database survives a v2 rebuild. The cached-state accessor is `parseCursor`, not `cursor`: the provider-architecture gate reads a bare `.cursor` as a reference to the Cursor provider, and this one is a parse position, matching the persisted metadata key. Co-Authored-By: Claude Fable 5 --- .../OpenCodexUsage/OpenCodexUsageParser.swift | 149 +++++++++++-- .../OpenCodexUsage/OpenCodexUsageStore.swift | 197 ++++++++++++------ .../OpenCodexUsageParserTests.swift | 36 +++- .../OpenCodexUsageStoreIncrementalTests.swift | 85 ++++++++ 4 files changed, 384 insertions(+), 83 deletions(-) diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift index 550786536b..607c8605ea 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageParser.swift @@ -5,10 +5,21 @@ import Glibc #elseif canImport(Musl) import Musl #endif +#if canImport(CryptoKit) +import CryptoKit +#else +import Crypto +#endif import Foundation public enum OpenCodexUsageParser { private static let newline: UInt8 = 0x0A + /// Must match `OpenCodexUsageStore`'s prefix-digest window. + private static let prefixDigestByteLimit = 64 * 1024 + + struct ChangedUnderReadError: Error, Equatable { + let path: String + } @TaskLocal static var logReadRecorderForTesting: LogReadRecorder? @@ -73,31 +84,56 @@ public enum OpenCodexUsageParser { static func parseLog( fileURL: URL, from offset: Int64, - fileManager: FileManager) throws -> JSONLParseResult + fileManager _: FileManager) throws -> JSONLParseResult { - guard fileManager.fileExists(atPath: fileURL.path) else { - return JSONLParseResult( - entries: [], - nextOffset: max(0, offset), - bytesRead: 0, - completeLineCount: 0, - newlineTerminatedEntryCount: 0) + let handle: FileHandle + do { + handle = try FileHandle(forReadingFrom: fileURL) + } catch { + if Self.isConfirmedAbsence(error) { + return JSONLParseResult( + entries: [], + nextOffset: max(0, offset), + bytesRead: 0, + completeLineCount: 0, + newlineTerminatedEntryCount: 0) + } + throw error } + defer { try? handle.close() } + var status = stat() + guard fstat(handle.fileDescriptor, &status) == 0 else { + throw Self.posixError(errno, path: fileURL.path) + } + let fileIdentity = "\(status.st_dev):\(status.st_ino)" + let size = Int64(status.st_size) let startOffset = max(0, offset) + if startOffset > size { + throw ChangedUnderReadError(path: fileURL.path) + } + + // Hold the snapshotted bytes instead of `.mappedIfSafe`. A mapping can SIGBUS if the + // file is truncated before those pages are touched; Swift cannot catch that. Full parse + // holds the file (~42 MB on a heavy machine) and only runs on a cold cache or rebuild. + // The steady-state path reads only the appended tail (`parsedOffset.. String + { + let length = min(Int64(Self.prefixDigestByteLimit), max(0, nextOffset)) + let prefix: Data + if length == 0 { + prefix = Data() + } else if let fileData, Int64(fileData.count) >= length { + prefix = Data(fileData.prefix(Int(length))) + } else { + try handle.seek(toOffset: 0) + prefix = try Self.readExact(handle, byteCount: length, path: path) + } + return SHA256.hash(data: prefix).map { String(format: "%02x", $0) }.joined() + } + + private static func readExact(_ handle: FileHandle, byteCount: Int64, path: String) throws -> Data { + let count = Int(byteCount) + var data = Data() + data.reserveCapacity(count) + while data.count < count { + let chunk = try handle.read(upToCount: count - data.count) ?? Data() + if chunk.isEmpty { + throw ChangedUnderReadError(path: path) + } + data.append(chunk) + } + return data + } + + private static func isConfirmedAbsence(_ error: Error) -> Bool { + var current: Error? = error + while let err = current { + let nsError = err as NSError + if nsError.domain == NSPOSIXErrorDomain { + return nsError.code == Int(ENOENT) || nsError.code == Int(ENOTDIR) + } + if nsError.domain == NSCocoaErrorDomain, + nsError.code == CocoaError.fileReadNoSuchFile.rawValue + || nsError.code == CocoaError.fileNoSuchFile.rawValue + { + return true + } + current = nsError.userInfo[NSUnderlyingErrorKey] as? Error + } + return false + } + + private static func posixError(_ code: Int32, path: String) -> NSError { + NSError( + domain: NSPOSIXErrorDomain, + code: Int(code), + userInfo: [NSFilePathErrorKey: path]) + } } diff --git a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift index 2a50599fbc..9fae10b9b2 100644 --- a/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/OpenCodexUsage/OpenCodexUsageStore.swift @@ -12,12 +12,16 @@ import Crypto import Darwin #elseif canImport(Glibc) import Glibc +#elseif canImport(Musl) +import Musl #endif import Foundation /// Independent OpenCodex usage cache. Never writes Codex `cost-usage.sqlite`. public struct OpenCodexUsageStore: Sendable { - public static let databaseFilename = "opencodex-usage.sqlite" + /// Schema v2 lives in a versioned filename so a v1 build keeps using `opencodex-usage.sqlite`. + /// Leave that older file alone; do not delete it. + public static let databaseFilename = "opencodex-usage-v2.sqlite" private static let schemaVersion = 2 private static let cursorMetaKey = "parseCursor" private static let prefixDigestByteLimit = 64 * 1024 @@ -71,32 +75,27 @@ public struct OpenCodexUsageStore: Sendable { public func loadEntries(logURL: URL, fileManager: FileManager = .default) throws -> [OpenCodexUsageEntry] { var shouldRetryStaleWrite = true while true { - guard fileManager.fileExists(atPath: logURL.path) else { return [] } - guard let identity = Self.statLog(at: logURL) else { return [] } - let cursor = self.readCursor() - if let cursor, Self.canReuseCursor(cursor, identity: identity) { - if identity.size == cursor.parsedOffset { - if let cached = self.readCachedEntries() { - return cached - } - } else if let entries = try self.incrementalReload( + guard let identity = try Self.statLog(at: logURL) else { return [] } + if let cached = self.readCachedState(), Self.canReuseCursor(cached.parseCursor, identity: identity) { + if identity.size == cached.parseCursor.parsedOffset { + return cached.entries + } + if let entries = try self.incrementalReload( logURL: logURL, identity: identity, - cursor: cursor, + cursor: cached.parseCursor, + existing: cached.entries, fileManager: fileManager) { return entries - } else if shouldRetryStaleWrite { - // Another loader committed a later same-file cursor. Re-run the cursor - // path once against that durable state, then fall back to a full reload. + } + if shouldRetryStaleWrite { + // Another loader changed the durable cursor. Re-run the cursor path once + // against that durable state, then fall back to a full reload. shouldRetryStaleWrite = false continue - } else { - return try self.fullReload( - logURL: logURL, - identity: identity, - fileManager: fileManager) } + return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) } return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) } @@ -119,54 +118,102 @@ public struct OpenCodexUsageStore: Sendable { parsedOffset: Int64, prefixDigest: String) { - _ = self.writeEntries( - entries, - cursor: ParseCursor( - path: path, - fileIdentity: fileIdentity, - parsedOffset: parsedOffset, - prefixDigest: prefixDigest), - replaceAll: false) + let cursor = ParseCursor( + path: path, + fileIdentity: fileIdentity, + parsedOffset: parsedOffset, + prefixDigest: prefixDigest) + _ = self.writeEntries(entries, cursor: cursor, replaceAll: false, baseCursor: cursor) } private func fullReload( logURL: URL, identity: LogIdentity, - fileManager: FileManager) throws -> [OpenCodexUsageEntry] + fileManager: FileManager, + allowRetry: Bool = true) throws -> [OpenCodexUsageEntry] { - let parsed = try OpenCodexUsageParser.parseLog(fileURL: logURL, from: 0, fileManager: fileManager) + let parsed: OpenCodexUsageParser.JSONLParseResult + do { + parsed = try OpenCodexUsageParser.parseLog(fileURL: logURL, from: 0, fileManager: fileManager) + } catch { + if error is OpenCodexUsageParser.ChangedUnderReadError, allowRetry { + guard let current = try Self.statLog(at: logURL) else { return [] } + return try self.fullReload( + logURL: logURL, + identity: current, + fileManager: fileManager, + allowRetry: false) + } + throw error + } + guard let parsedIdentity = parsed.fileIdentity else { return [] } + let pathIdentity: LogIdentity? + do { + pathIdentity = try Self.statLog(at: logURL) + } catch { + if allowRetry { + guard let current = try Self.statLog(at: logURL) else { return [] } + return try self.fullReload( + logURL: logURL, + identity: current, + fileManager: fileManager, + allowRetry: false) + } + throw error + } + guard let pathIdentity else { return [] } + if pathIdentity.fileIdentity != parsedIdentity { + if allowRetry { + return try self.fullReload( + logURL: logURL, + identity: pathIdentity, + fileManager: fileManager, + allowRetry: false) + } + return Self.dedupedAndSorted(parsed.entries) + } let entries = Self.dedupedAndSorted(parsed.entries) let cursor = ParseCursor( path: identity.path, - fileIdentity: identity.fileIdentity, + fileIdentity: parsedIdentity, parsedOffset: parsed.nextOffset, - prefixDigest: Self.prefixDigest(fileURL: logURL, parsedOffset: parsed.nextOffset) ?? "") + prefixDigest: parsed.prefixDigest) self.replaceCachedEntries(Self.dedupedAndSorted(parsed.newlineTerminatedEntries), cursor: cursor) return entries } - /// Returns `nil` when the incremental write observed a newer same-file durable cursor. + /// Returns `nil` when the incremental write observed a durable cursor that is not `cursor`. private func incrementalReload( logURL: URL, identity: LogIdentity, cursor: ParseCursor, + existing: [OpenCodexUsageEntry], fileManager: FileManager) throws -> [OpenCodexUsageEntry]? { - guard let existing = self.readCachedEntries() else { - return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) + let parsed: OpenCodexUsageParser.JSONLParseResult + do { + parsed = try OpenCodexUsageParser.parseLog( + fileURL: logURL, + from: cursor.parsedOffset, + fileManager: fileManager) + } catch { + if error is OpenCodexUsageParser.ChangedUnderReadError { + return try self.fullReload(logURL: logURL, identity: identity, fileManager: fileManager) + } + throw error } - let parsed = try OpenCodexUsageParser.parseLog( - fileURL: logURL, - from: cursor.parsedOffset, - fileManager: fileManager) Self.incrementalPostParseHookForTesting?() // Closes the TOCTOU window between the pre-read `canReuseCursor` check and this tail // parse: a rotation or replacement in that window would otherwise merge cached rows from // the old file with bytes from the new one and persist a cursor for a path that no longer // names that file. A replacement that preserves path, st_dev, st_ino, size, AND the first - // min(64 KiB, parsedOffset) bytes remains undetectable. - guard let postIdentity = Self.statLog(at: logURL) else { return [] } - if !Self.isSameLogAfterTailRead(preRead: identity, postRead: postIdentity, cursor: cursor) { + // min(64 KiB, parsedOffset) bytes remains undetectable. `parsed.fileIdentity` is the + // descriptor we actually read. + guard let parsedIdentity = parsed.fileIdentity else { return [] } + guard let postIdentity = try Self.statLog(at: logURL) else { return [] } + if parsedIdentity != identity.fileIdentity + || !Self.isSameLogAfterTailRead(preRead: identity, postRead: postIdentity, cursor: cursor) + { return try self.fullReload(logURL: logURL, identity: postIdentity, fileManager: fileManager) } let nextOffset = parsed.nextOffset @@ -175,14 +222,13 @@ public struct OpenCodexUsageStore: Sendable { if committed.isEmpty, nextOffset == cursor.parsedOffset { return Self.dedupedAndSorted(existing + pending) } - let digest = nextOffset == cursor.parsedOffset - ? cursor.prefixDigest - : (Self.prefixDigest(fileURL: logURL, parsedOffset: nextOffset) ?? "") + let digest = nextOffset == cursor.parsedOffset ? cursor.prefixDigest : parsed.prefixDigest if self.insertCachedEntries( committed, + baseCursor: cursor, cursor: ParseCursor( path: identity.path, - fileIdentity: identity.fileIdentity, + fileIdentity: parsedIdentity, parsedOffset: nextOffset, prefixDigest: digest)) == .stale { @@ -191,10 +237,19 @@ public struct OpenCodexUsageStore: Sendable { return Self.dedupedAndSorted(existing + committed + pending) } - private func readCachedEntries() -> [OpenCodexUsageEntry]? { + private func readCachedState() -> (parseCursor: ParseCursor, entries: [OpenCodexUsageEntry])? { guard let db = self.open(readOnly: true) else { return nil } defer { sqlite3_close(db) } guard Self.userVersion(db) == Self.schemaVersion else { return nil } + guard sqlite3_exec(db, "BEGIN", nil, nil, nil) == SQLITE_OK else { return nil } + defer { _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) } + guard let cursor = Self.parseCursor(from: db), + let entries = Self.readEntries(from: db) + else { return nil } + return (parseCursor: cursor, entries: entries) + } + + private static func readEntries(from db: OpaquePointer?) -> [OpenCodexUsageEntry]? { var statement: OpaquePointer? let sql = """ SELECT request_id, timestamp, provider, model, usage_status, account_label, surface, conversation_id, \ @@ -242,25 +297,30 @@ public struct OpenCodexUsageStore: Sendable { self.writeEntries(entries, cursor: cursor, replaceAll: true) } - private func insertCachedEntries(_ entries: [OpenCodexUsageEntry], cursor: ParseCursor) -> CachedWriteResult { - self.writeEntries(entries, cursor: cursor, replaceAll: false) + private func insertCachedEntries( + _ entries: [OpenCodexUsageEntry], + baseCursor: ParseCursor, + cursor: ParseCursor) -> CachedWriteResult + { + self.writeEntries(entries, cursor: cursor, replaceAll: false, baseCursor: baseCursor) } @discardableResult private func writeEntries( _ entries: [OpenCodexUsageEntry], cursor: ParseCursor, - replaceAll: Bool) -> CachedWriteResult + replaceAll: Bool, + baseCursor: ParseCursor? = nil) -> CachedWriteResult { guard let db = self.open(readOnly: false) else { return .applied } defer { sqlite3_close(db) } guard sqlite3_exec(db, "BEGIN IMMEDIATE", nil, nil, nil) == SQLITE_OK else { return .applied } // Incremental appends only (`replaceAll == false`): after BEGIN IMMEDIATE, re-read the - // durable cursor. A concurrent loader may already have committed a later parsedOffset - // for this path + fileIdentity; writing this snapshot would move the cursor backwards - // and re-insert rows it already stored. Full reloads re-derived the whole file and must - // still replace even a newer cursor (truncation, rotation, schema rebuild). - if !replaceAll, Self.isStaleIncrementalCursor(proposed: cursor, durable: Self.parseCursor(from: db)) { + // durable cursor. Reject the write whenever it differs from the exact base cursor this + // parse was derived from — another writer replaced the cache, including a different + // home/path. Full reloads re-derived the whole file and must still replace even a newer + // cursor (truncation, rotation, schema rebuild). + if !replaceAll, Self.isStaleIncrementalCursor(base: baseCursor, durable: Self.parseCursor(from: db)) { _ = sqlite3_exec(db, "ROLLBACK", nil, nil, nil) return .stale } @@ -313,11 +373,8 @@ public struct OpenCodexUsageStore: Sendable { return try? JSONDecoder().decode(ParseCursor.self, from: data) } - private static func isStaleIncrementalCursor(proposed: ParseCursor, durable: ParseCursor?) -> Bool { - guard let durable else { return false } - return durable.path == proposed.path - && durable.fileIdentity == proposed.fileIdentity - && durable.parsedOffset >= proposed.parsedOffset + private static func isStaleIncrementalCursor(base: ParseCursor?, durable: ParseCursor?) -> Bool { + durable != base } /// Threat model: the digest covers only the first min(64 KiB, parsedOffset) bytes, so an @@ -342,8 +399,10 @@ public struct OpenCodexUsageStore: Sendable { } private static func ensureSchema(_ db: OpaquePointer?) { - // `!= schemaVersion` deliberately rebuilds a NEWER database too: a downgrade must not - // read a schema it does not understand. + // Schema v2 lives in `opencodex-usage-v2.sqlite` so a v1 build keeps using + // `opencodex-usage.sqlite`. Never delete that older file. + // `!= schemaVersion` still rebuilds THIS file if the version is wrong (corrupt header, + // leftover user_version 0, or a copied v1 payload). guard self.userVersion(db) != self.schemaVersion else { return } let sql = """ DROP TABLE IF EXISTS entries; @@ -530,11 +589,19 @@ public struct OpenCodexUsageStore: Sendable { } } - private static func statLog(at url: URL) -> LogIdentity? { - url.withUnsafeFileSystemRepresentation { pointer in - guard let pointer else { return nil } + private static func statLog(at url: URL) throws -> LogIdentity? { + try url.withUnsafeFileSystemRepresentation { pointer in + guard let pointer else { + throw POSIXError(.EINVAL) + } var status = stat() - guard stat(pointer, &status) == 0 else { return nil } + guard stat(pointer, &status) == 0 else { + let err = errno + if err == ENOENT || err == ENOTDIR { + return nil + } + throw POSIXError(POSIXErrorCode(rawValue: err) ?? .EIO) + } return LogIdentity( url: url, path: url.path, diff --git a/Tests/CodexBarTests/OpenCodexUsageParserTests.swift b/Tests/CodexBarTests/OpenCodexUsageParserTests.swift index f58b6b4806..a264d4c136 100644 --- a/Tests/CodexBarTests/OpenCodexUsageParserTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageParserTests.swift @@ -1,3 +1,10 @@ +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#endif import Foundation import Testing @testable import CodexBarCore @@ -125,8 +132,33 @@ struct OpenCodexUsageParserTests { #expect(snapshot.sessions[0].sessionID == "chat-1") #expect(snapshot.sessions[0].reasoningTokens == 3) #expect(snapshot.costProvenance == .listPriceEstimate) - #expect(OpenCodexUsageStore.databaseFilename == "opencodex-usage.sqlite") - #expect(FileManager.default.fileExists(atPath: root.appendingPathComponent("opencodex-usage.sqlite").path)) + #expect(OpenCodexUsageStore.databaseFilename == "opencodex-usage-v2.sqlite") + #expect(FileManager.default.fileExists( + atPath: root.appendingPathComponent("opencodex-usage-v2.sqlite").path)) + } + + @Test + func `missing usage log parses as empty`() throws { + let url = FileManager.default.temporaryDirectory + .appendingPathComponent("OpenCodexUsageParserMissing-\(UUID().uuidString).jsonl") + #expect(try OpenCodexUsageParser.parse(fileURL: url).isEmpty) + } + + @Test(.enabled(if: geteuid() != 0)) + func `unreadable usage log throws instead of returning empty`() throws { + let root = FileManager.default.temporaryDirectory + .appendingPathComponent("OpenCodexUsageParserUnreadable-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + let url = root.appendingPathComponent("usage.jsonl") + try Data("{\"requestId\":\"x\"}\n".utf8).write(to: url) + try FileManager.default.setAttributes([.posixPermissions: 0o000], ofItemAtPath: url.path) + defer { + try? FileManager.default.setAttributes([.posixPermissions: 0o600], ofItemAtPath: url.path) + try? FileManager.default.removeItem(at: root) + } + #expect(throws: (any Error).self) { + _ = try OpenCodexUsageParser.parse(fileURL: url) + } } @Test diff --git a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift index f2be75c6c9..a8dc8fe202 100644 --- a/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift +++ b/Tests/CodexBarTests/OpenCodexUsageStoreIncrementalTests.swift @@ -1,3 +1,10 @@ +#if canImport(Darwin) +import Darwin +#elseif canImport(Glibc) +import Glibc +#elseif canImport(Musl) +import Musl +#endif import Foundation import SQLite3 import Testing @@ -196,15 +203,19 @@ struct OpenCodexUsageStoreIncrementalTests { try Harness.writeV1Database( at: harness.cacheRoot.appendingPathComponent(OpenCodexUsageStore.databaseFilename), staleRequestID: "stale-v1") + let legacyURL = harness.cacheRoot.appendingPathComponent("opencodex-usage.sqlite") + try Harness.writeV1Database(at: legacyURL, staleRequestID: "legacy-v1") let entries = try harness.store.loadEntries(logURL: harness.log) let schemaExpected = try harness.referenceEntries() #expect(entries == schemaExpected) #expect(entries.map(\.requestID) == ["live"]) #expect(entries.contains { $0.requestID == "stale-v1" } == false) + #expect(entries.contains { $0.requestID == "legacy-v1" } == false) #expect(entries[0].usage?.inputTokens == 9) #expect(entries[0].usage?.outputTokens == 2) #expect(entries[0].totalTokens == 11) + #expect(FileManager.default.fileExists(atPath: legacyURL.path)) } @Test @@ -445,6 +456,80 @@ struct OpenCodexUsageStoreIncrementalTests { #expect(cached == loaded) #expect(recorder.snapshot().bytesRead == 0) } + + @Test + func `incremental load of log A does not keep another home's cached rows`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + + try harness.writeLines( + Harness.line(id: "a-seed-1", input: 1), + Harness.line(id: "a-seed-2", input: 2)) + _ = try harness.store.loadEntries(logURL: harness.log) + try harness.appendLines(Harness.line(id: "a-tail", input: 3)) + + let logB = harness.root.appendingPathComponent("usage-b.jsonl") + try Data([ + Harness.line(id: "b-1", input: 11) + "\n", + Harness.line(id: "b-2", input: 12) + "\n", + ].joined().utf8).write(to: logB) + + let store = harness.store + let loaded = try OpenCodexUsageStore.withIncrementalPostParseHookForTesting { + do { + _ = try store.loadEntries(logURL: logB) + } catch { + Issue.record(error) + } + } operation: { + try store.loadEntries(logURL: harness.log) + } + + let expected = try harness.referenceEntries() + #expect(loaded == expected) + #expect(loaded.map(\.requestID) == ["a-seed-1", "a-seed-2", "a-tail"]) + #expect(Set(loaded.map(\.requestID)).isDisjoint(with: ["b-1", "b-2"])) + #expect(try harness.sqliteRequestIDs() == ["a-seed-1", "a-seed-2", "a-tail"].sorted()) + #expect(try harness.sqliteRequestIDs().contains("b-1") == false) + #expect(try harness.sqliteRequestIDs().contains("b-2") == false) + } + + @Test + func `missing log returns an empty snapshot without throwing`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + #expect(try harness.store.loadEntries(logURL: harness.log) == []) + } + + @Test + func `stat failure other than absence throws instead of returning empty`() throws { + let harness = try Harness.make() + defer { harness.tearDown() } + let loop = harness.root.appendingPathComponent("loop.jsonl") + try FileManager.default.createSymbolicLink( + atPath: loop.path, + withDestinationPath: loop.lastPathComponent) + #expect(throws: (any Error).self) { + _ = try harness.store.loadEntries(logURL: loop) + } + } + + @Test(.enabled(if: geteuid() != 0)) + func `permission failure on the log directory throws instead of returning empty`() throws { + let harness = try Harness.make() + let logDir = harness.root.appendingPathComponent("logdir", isDirectory: true) + try FileManager.default.createDirectory(at: logDir, withIntermediateDirectories: true) + let log = logDir.appendingPathComponent("usage.jsonl") + try Data((Harness.line(id: "hidden", input: 1) + "\n").utf8).write(to: log) + try FileManager.default.setAttributes([.posixPermissions: 0o000], ofItemAtPath: logDir.path) + defer { + try? FileManager.default.setAttributes([.posixPermissions: 0o700], ofItemAtPath: logDir.path) + harness.tearDown() + } + #expect(throws: (any Error).self) { + _ = try harness.store.loadEntries(logURL: log) + } + } } private struct Harness { From c94cace09d64343c8df29d32fbb90ce2a1e782c4 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 25 Aug 2026 00:42:01 -0700 Subject: [PATCH 7/7] test: isolate concurrent SQLite checkpoint fixtures Co-authored-by: olddonkey --- .../Vendored/CostUsage/CostUsageStore+CodexCache.swift | 6 +++++- .../CodexBarCore/Vendored/CostUsage/CostUsageStore.swift | 6 ++++-- Tests/CodexBarTests/CostUsageStoreTests.swift | 4 ++-- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+CodexCache.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+CodexCache.swift index 2957fd1bcb..074a0c2b79 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+CodexCache.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+CodexCache.swift @@ -86,7 +86,11 @@ extension CostUsageStore { requestedUntilDay: budgetProtectionWindow.untilKey, calendar: calendar) guard !result.catchUpRequired else { return result } - Self.identicalContentPreLockCheckpointForTesting?() + if let checkpoint = Self.identicalContentPreLockCheckpointForTesting, + checkpoint.databaseURL == self.databaseURL + { + checkpoint.checkpoint() + } guard self.beginSaveTransaction() else { var retry = result retry.catchUpRequired = true diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift index f639df7e72..7258adcaca 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift @@ -90,8 +90,10 @@ actor CostUsageStore { /// 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)? - /// Test-only interleaving point after optimistic identity succeeds and before its writer lock. - nonisolated(unsafe) static var identicalContentPreLockCheckpointForTesting: (() -> Void)? + /// Test-only interleaving point scoped to one database so parallel store fixtures stay isolated. + nonisolated(unsafe) static var identicalContentPreLockCheckpointForTesting: ( + databaseURL: URL, + checkpoint: () -> Void)? /// Test-only traversal proof for persisted Codex catch-up reconciliation. Never set in production. nonisolated(unsafe) static var codexCatchUpReconciliationVisitForTesting: (() -> Void)? diff --git a/Tests/CodexBarTests/CostUsageStoreTests.swift b/Tests/CodexBarTests/CostUsageStoreTests.swift index 103f094112..e4f7e93682 100644 --- a/Tests/CodexBarTests/CostUsageStoreTests.swift +++ b/Tests/CodexBarTests/CostUsageStoreTests.swift @@ -594,13 +594,13 @@ extension CostUsageStoreTests { reread.lastScanUnixMs = 2000 let interloper = try SQLiteTestConnection(url: store.databaseURL) var checkpointError: Error? - CostUsageStore.identicalContentPreLockCheckpointForTesting = { + CostUsageStore.identicalContentPreLockCheckpointForTesting = (store.databaseURL, { do { try interloper.execute("UPDATE files SET parsed_bytes = 999 WHERE path = '\(path)'") } catch { checkpointError = error } - } + }) defer { CostUsageStore.identicalContentPreLockCheckpointForTesting = nil } let result = save(reread)