From a78e5ebbce886a8c1fa5eba5bc0c4671aba98b28 Mon Sep 17 00:00:00 2001 From: pavbar Date: Mon, 10 Aug 2026 10:10:39 -0500 Subject: [PATCH 1/2] fix: detect semantic Codex cost catch-up progress --- Sources/CodexBarCore/CostUsageFetcher.swift | 85 ++++++- .../CostUsageCatchUpProgressTests.swift | 233 ++++++++++++++++++ .../UsageStoreCodexCostCatchUpTests.swift | 62 +++++ 3 files changed, 371 insertions(+), 9 deletions(-) create mode 100644 Tests/CodexBarTests/CostUsageCatchUpProgressTests.swift diff --git a/Sources/CodexBarCore/CostUsageFetcher.swift b/Sources/CodexBarCore/CostUsageFetcher.swift index f459c144ff..bc107b4456 100644 --- a/Sources/CodexBarCore/CostUsageFetcher.swift +++ b/Sources/CodexBarCore/CostUsageFetcher.swift @@ -312,19 +312,19 @@ public struct CostUsageFetcher: Sendable { } let scoped = CostUsageScanner.codexCache(cache, scopedTo: roots) - var progressHasher = Hasher() - for (path, usage) in scoped.files.sorted(by: { $0.key < $1.key }) { - progressHasher.combine(path) - progressHasher.combine(usage.codexScanFileId) - progressHasher.combine(usage.parsedBytes) - progressHasher.combine(usage.size) - progressHasher.combine(usage.codexScanComplete) - } + // Bounded passes can advance work without changing parsed file bytes. + // Directory validation can move between partitions, + // active lookback can discover older sessions, + // and buffered fork retries can resolve a parent dependency. + // Include each semantic cursor so the stall detector sees real progress. + let progressKey = self.codexScanProgressKey( + cache: cache, + scopedFiles: scoped.files) let hasIncompleteFile = scoped.files.values.contains { $0.codexScanComplete == false } let pending = cache.codexScanCatchUpPending == true || hasIncompleteFile return CodexScanCatchUpStatus( pending: pending, - progressKey: "\(scoped.files.count):\(progressHasher.finalize())", + progressKey: progressKey, processedBytes: cache.codexScanProcessedBytes ?? 0, totalBytes: cache.codexScanTotalBytes ?? 0, completedFiles: cache.codexScanCompletedFiles ?? 0, @@ -1372,4 +1372,71 @@ extension CostUsageFetcher { #endif return nil } + + static func codexScanProgressKey( + cache: CostUsageCache, + scopedFiles: [String: CostUsageFileUsage]) -> String + { + var progressHasher = Hasher() + // The aggregate closes the remaining bounded-work gap for already indexed files that grew: + // each completed stale append advances the count, while a fully parsed active append leaves + // it unchanged. + progressHasher.combine(cache.codexScanCompletedFiles) + // Stable file membership tracks newly completed bounded work. Mutable byte counts belong + // only to unfinished files: a completed live session can append between every pass without + // advancing the finite backlog. Buffered retries track dependency state rather than counts, + // so appends cannot mask a parent dependency that remains unresolved. + for (path, usage) in scopedFiles.sorted(by: { $0.key < $1.key }) { + progressHasher.combine(path) + progressHasher.combine(usage.codexScanFileId) + progressHasher.combine(usage.codexScanComplete) + if usage.codexScanComplete == false { + progressHasher.combine(usage.parsedBytes) + progressHasher.combine(usage.size) + progressHasher.combine(usage.codexJSONLResumeState?.offset) + } + let hasBufferedRetry = usage.hasBufferedCodexForkRetryLines + progressHasher.combine(hasBufferedRetry) + if hasBufferedRetry { + progressHasher.combine(usage.forkedFromId) + progressHasher.combine(usage.forkBaselineDependencyKey) + progressHasher.combine(usage.codexBufferedSubagentLines?.isEmpty == false) + progressHasher.combine(usage.codexBufferedUnresolvedForkLines?.isEmpty == false) + } + } + + if let discovery = cache.codexSessionDiscovery { + progressHasher.combine(discovery.generation) + progressHasher.combine(discovery.directoryPaths.count) + progressHasher.combine(discovery.nextDirectoryIndex) + progressHasher.combine(discovery.filePaths.count) + progressHasher.combine(discovery.nextFileIndex) + progressHasher.combine(discovery.headScan?.path) + progressHasher.combine(discovery.headScan?.offset) + progressHasher.combine(discovery.headScan?.resumeState?.offset) + progressHasher.combine(discovery.filePathBySessionId.count) + progressHasher.combine(discovery.missingSessionIds.sorted()) + progressHasher.combine(discovery.pendingSessionIds.sorted()) + progressHasher.combine(discovery.validationDirectoryIndex) + progressHasher.combine(discovery.isComplete) + } else { + progressHasher.combine("no-discovery") + } + + if let lookback = cache.codexActiveLookbackState { + progressHasher.combine(lookback.scanSinceKey) + progressHasher.combine(lookback.rootPaths.sorted()) + for (root, dayKey) in lookback.nextDayKeyByRoot.sorted(by: { $0.key < $1.key }) { + progressHasher.combine(root) + progressHasher.combine(dayKey) + } + progressHasher.combine(lookback.completedRootPaths.sorted()) + progressHasher.combine(lookback.pendingFilePaths.sorted()) + progressHasher.combine(lookback.legacyRecursivePendingRootPaths.sorted()) + } else { + progressHasher.combine("no-lookback") + } + + return "v2:\(scopedFiles.count):\(progressHasher.finalize())" + } } diff --git a/Tests/CodexBarTests/CostUsageCatchUpProgressTests.swift b/Tests/CodexBarTests/CostUsageCatchUpProgressTests.swift new file mode 100644 index 0000000000..be74d2aa4e --- /dev/null +++ b/Tests/CodexBarTests/CostUsageCatchUpProgressTests.swift @@ -0,0 +1,233 @@ +import Foundation +import Testing +@testable import CodexBarCore + +struct CostUsageCatchUpProgressTests { + @Test + func `progress key includes semantic discovery cursor progress`() { + var cache = CostUsageCache() + cache.codexSessionDiscovery = CostUsageCodexSessionDiscovery( + roots: ["/sessions"], + generation: "generation-1", + directoryStamps: [:], + directoryPaths: ["/sessions/2026", "/sessions/2026/08"], + nextDirectoryIndex: 2, + filePaths: [], + nextFileIndex: 0, + fileStamps: [:], + headScan: nil, + filePathBySessionId: [:], + missingSessionIds: ["missing-parent"], + pendingSessionIds: [], + validationDirectoryIndex: 0, + isComplete: true) + + let initial = CostUsageFetcher.codexScanProgressKey(cache: cache, scopedFiles: [:]) + cache.codexSessionDiscovery?.validationDirectoryIndex = 1 + let advanced = CostUsageFetcher.codexScanProgressKey(cache: cache, scopedFiles: [:]) + + #expect(advanced != initial) + } + + @Test + func `progress tracks stable membership but ignores complete live appends`() { + let path = "/sessions/live.jsonl" + let complete = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 100, + days: [:], + parsedBytes: 100, + codexScanFileId: "1:1", + codexScanComplete: true) + let empty = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [:]) + let initial = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: complete]) + var appended = complete + appended.size = 125 + appended.parsedBytes = 125 + let afterLiveAppend = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: appended]) + + #expect(initial != empty) + #expect(afterLiveAppend == initial) + } + + @Test + func `progress tracks unfinished file bytes`() { + let path = "/sessions/unfinished.jsonl" + var unfinished = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 125, + days: [:], + parsedBytes: 110, + codexScanFileId: "1:1", + codexScanComplete: false) + let unfinishedInitial = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: unfinished]) + unfinished.parsedBytes = 120 + let unfinishedAdvanced = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: unfinished]) + + #expect(unfinishedAdvanced != unfinishedInitial) + } + + @Test + func `progress tracks completed aggregate for existing file backlog`() { + let path = "/sessions/existing.jsonl" + let usage = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 125, + days: [:], + parsedBytes: 125, + codexScanFileId: "1:1", + codexScanComplete: true) + var before = CostUsageCache() + before.codexScanCompletedFiles = 0 + var after = before + after.codexScanCompletedFiles = 1 + + #expect(CostUsageFetcher.codexScanProgressKey(cache: after, scopedFiles: [path: usage]) + != CostUsageFetcher.codexScanProgressKey(cache: before, scopedFiles: [path: usage])) + } + + @Test + func `completed buffered appends do not hide a stalled dependency`() { + let path = "/sessions/fork.jsonl" + let line = CostUsageScanner.CodexBufferedFastLine( + lineIndex: 1, + ordinal: nil, + line: .interAgentCommunication(triggerTurn: false)) + let buffered = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 100, + days: [:], + parsedBytes: 100, + forkedFromId: "missing-parent", + codexScanFileId: "1:1", + codexScanComplete: true, + codexBufferedUnresolvedForkLines: [line]) + let initial = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: buffered]) + + var appended = buffered + appended.size = 125 + appended.parsedBytes = 125 + appended.codexBufferedUnresolvedForkLines = [line, line] + let afterAppend = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: appended]) + + var dependencyResolved = appended + dependencyResolved.forkBaselineDependencyKey = "parent:resolved" + let afterDependencyChange = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: dependencyResolved]) + + var replayed = dependencyResolved + replayed.codexBufferedUnresolvedForkLines = nil + let afterReplay = CostUsageFetcher.codexScanProgressKey( + cache: CostUsageCache(), + scopedFiles: [path: replayed]) + + #expect(afterAppend == initial) + #expect(afterDependencyChange != afterAppend) + #expect(afterReplay != afterDependencyChange) + } + + @Test + func `progress key includes resumable discovery head offset`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + let day = try env.makeLocalNoon(year: 2026, month: 5, day: 10) + let iso = env.isoString(for: day) + let body = #"{"type":"session_meta","timestamp":"\#(iso)","payload":{"session_id":"known-session","cwd":""# + + String(repeating: "x", count: 512) + + #""}}"# + + "\n" + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "budgeted-head.jsonl", + contents: body) + + let firstBudget = CostUsageScanner.CodexScanBudget(maxFileBytes: 32, maxBytesPerRefresh: 32) + let firstIndex = CostUsageScanner.CodexSessionFileIndex( + files: [fileURL], + roots: [env.codexSessionsRoot], + cachedDiscovery: nil, + scanBudget: firstBudget) + _ = try firstIndex.lookup(sessionId: "absent-session") + let firstDiscovery = firstIndex.persistedState + + let secondBudget = CostUsageScanner.CodexScanBudget(maxFileBytes: 32, maxBytesPerRefresh: 32) + let secondIndex = CostUsageScanner.CodexSessionFileIndex( + files: [fileURL], + roots: [env.codexSessionsRoot], + cachedDiscovery: firstDiscovery, + scanBudget: secondBudget) + _ = try secondIndex.lookup(sessionId: "absent-session") + let secondDiscovery = secondIndex.persistedState + + let firstCommittedOffset = try #require(firstDiscovery.headScan?.offset) + let secondCommittedOffset = try #require(secondDiscovery.headScan?.offset) + let firstResumeOffset = try #require(firstDiscovery.headScan?.resumeState?.offset) + let secondResumeOffset = try #require(secondDiscovery.headScan?.resumeState?.offset) + var firstCache = CostUsageCache() + firstCache.codexSessionDiscovery = firstDiscovery + var secondCache = CostUsageCache() + secondCache.codexSessionDiscovery = secondDiscovery + + #expect(secondCommittedOffset == firstCommittedOffset) + #expect(secondResumeOffset > firstResumeOffset) + #expect(CostUsageFetcher.codexScanProgressKey(cache: secondCache, scopedFiles: [:]) + != CostUsageFetcher.codexScanProgressKey(cache: firstCache, scopedFiles: [:])) + } + + @Test + func `progress key includes active lookback cursor and ignores dictionary insertion order`() { + var initialCache = CostUsageCache() + initialCache.codexActiveLookbackState = CostUsageCodexActiveLookbackState( + scanSinceKey: "2026-07-01", + rootPaths: ["/sessions", "/archived_sessions"], + nextDayKeyByRoot: [ + "/sessions": "2026-07-02", + "/archived_sessions": "2026-07-03", + ]) + var advancedCache = initialCache + advancedCache.codexActiveLookbackState?.nextDayKeyByRoot["/sessions"] = "2026-07-01" + + let first = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 10, + days: [:], + parsedBytes: 10, + codexScanFileId: "1:1", + codexScanComplete: true) + let second = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 20, + days: [:], + parsedBytes: 10, + codexScanFileId: "2:2", + codexScanComplete: false) + var forward: [String: CostUsageFileUsage] = [:] + forward["/sessions/a.jsonl"] = first + forward["/sessions/b.jsonl"] = second + var reverse: [String: CostUsageFileUsage] = [:] + reverse["/sessions/b.jsonl"] = second + reverse["/sessions/a.jsonl"] = first + + let initial = CostUsageFetcher.codexScanProgressKey(cache: initialCache, scopedFiles: forward) + let advanced = CostUsageFetcher.codexScanProgressKey(cache: advancedCache, scopedFiles: forward) + let reordered = CostUsageFetcher.codexScanProgressKey(cache: initialCache, scopedFiles: reverse) + + #expect(advanced != initial) + #expect(reordered == initial) + } +} diff --git a/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift b/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift index 9ccd317d33..c0343cd156 100644 --- a/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift +++ b/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift @@ -87,6 +87,68 @@ struct UsageStoreCodexCostCatchUpTests { #expect(store.codexCostCatchUpActivity?.pauseReason == .noProgress) } + @Test + func `catch-up continues when existing complete file backlog advances`() async throws { + let store = try Self.makeStore(suite: "existing-complete-backlog") + let first = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 125, + days: [:], + parsedBytes: 125, + codexScanFileId: "1:1", + codexScanComplete: true) + let second = CostUsageScanner.makeFileUsage( + mtimeUnixMs: 1, + size: 125, + days: [:], + parsedBytes: 125, + codexScanFileId: "2:2", + codexScanComplete: true) + let files = [ + "/sessions/first.jsonl": first, + "/sessions/second.jsonl": second, + ] + var caches = [CostUsageCache(), CostUsageCache(), CostUsageCache()] + caches[0].codexScanCompletedFiles = 0 + caches[1].codexScanCompletedFiles = 1 + caches[2].codexScanCompletedFiles = 2 + let keys = caches.map { + CostUsageFetcher.codexScanProgressKey(cache: $0, scopedFiles: files) + } + var statusLoadCount = 0 + var advanceCount = 0 + store._test_tokenUsageSnapshotLoaderOverride = { _, _, now, _, _ in + Self.tokenSnapshot(cost: 1, now: now) + } + store._test_codexCostCatchUpStatusOverride = { _ in + statusLoadCount += 1 + return CostUsageFetcher.CodexScanCatchUpStatus( + pending: statusLoadCount == 1, + progressKey: statusLoadCount == 1 ? keys[0] : keys[2]) + } + store._test_codexCostCatchUpAdvanceOverride = { _, _, _ in + advanceCount += 1 + return CostUsageFetcher.CodexScanCatchUpStatus( + pending: advanceCount < 2, + progressKey: keys[advanceCount]) + } + store._test_codexCostCatchUpSleepOverride = { _ in + await Task.yield() + } + store._test_codexCostCatchUpResourceStateOverride = { + (.ac, false, .nominal) + } + + store.startCodexCostCatchUpIfNeeded(mode: .accelerated) + await Self.waitUntil { + store.codexCostCatchUpTask == nil + } + + #expect(Set(keys).count == 3) + #expect(advanceCount == 2) + #expect(store.codexCostCatchUpActivity?.phase == .complete) + } + @Test func `accelerated catch-up runs without an inter-pass delay and publishes progress`() async throws { let store = try Self.makeStore(suite: "accelerated") From b1d5c1109938442b3e4aab84d9d16df5d0089697 Mon Sep 17 00:00:00 2001 From: pavbar Date: Mon, 10 Aug 2026 12:10:35 -0500 Subject: [PATCH 2/2] fix: stop cyclic Codex cost catch-up --- .../UsageStore+CodexCostCatchUp.swift | 4 +-- ...Store+SpendDashboardCodexCostCatchUp.swift | 4 +-- .../UsageStoreCodexCostCatchUpTests.swift | 33 +++++++++++++++++++ ...eSpendDashboardCodexCostCatchUpTests.swift | 33 +++++++++++++++++++ 4 files changed, 70 insertions(+), 4 deletions(-) diff --git a/Sources/CodexBar/UsageStore+CodexCostCatchUp.swift b/Sources/CodexBar/UsageStore+CodexCostCatchUp.swift index 39de988e04..94e290145b 100644 --- a/Sources/CodexBar/UsageStore+CodexCostCatchUp.swift +++ b/Sources/CodexBar/UsageStore+CodexCostCatchUp.swift @@ -106,7 +106,7 @@ extension UsageStore { context: context, phase: status.pending ? .indexing : .complete) var didAdvance = false - var previousActiveDuration: TimeInterval? + var (previousActiveDuration, seenProgressKeys): (TimeInterval?, Set) = (nil, [status.progressKey]) while status.pending { do { guard self.codexCostCatchUpContextIsCurrent(context) else { return } @@ -183,7 +183,7 @@ extension UsageStore { pauseReason: .user) return } - if nextStatus.pending, nextStatus.progressKey == status.progressKey { + if nextStatus.pending, !seenProgressKeys.insert(nextStatus.progressKey).inserted { self.publishCodexCostCatchUpActivity( status: nextStatus, context: context, diff --git a/Sources/CodexBar/UsageStore+SpendDashboardCodexCostCatchUp.swift b/Sources/CodexBar/UsageStore+SpendDashboardCodexCostCatchUp.swift index 0ee8c911ec..7f312b7576 100644 --- a/Sources/CodexBar/UsageStore+SpendDashboardCodexCostCatchUp.swift +++ b/Sources/CodexBar/UsageStore+SpendDashboardCodexCostCatchUp.swift @@ -117,7 +117,7 @@ extension UsageStore { var didChangeCache = false var previousActiveDuration: TimeInterval? - var stalledCacheIdentities: Set = [] + var (stalledCacheIdentities, seenKeysByCache) = (Set(), statuses.mapValues { Set([$0.progressKey]) }) while Self.spendDashboardCodexCatchUpIsPending(statuses) { do { guard self.spendDashboardCodexCostCatchUpContextIsCurrent(context) else { return } @@ -197,7 +197,7 @@ extension UsageStore { didChangeCache = didChangeCache || nextStatus.progressKey != previousStatus?.progressKey statuses[account.cacheIdentity] = nextStatus if nextStatus.pending, - nextStatus.progressKey == previousStatus?.progressKey + !seenKeysByCache[account.cacheIdentity, default: []].insert(nextStatus.progressKey).inserted { stalledCacheIdentities.insert(account.cacheIdentity) } else { diff --git a/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift b/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift index c0343cd156..7dbdea1fb8 100644 --- a/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift +++ b/Tests/CodexBarTests/UsageStoreCodexCostCatchUpTests.swift @@ -87,6 +87,39 @@ struct UsageStoreCodexCostCatchUpTests { #expect(store.codexCostCatchUpActivity?.pauseReason == .noProgress) } + @Test + func `catch-up stops when bounded progress revisits an earlier semantic state`() async throws { + let store = try Self.makeStore(suite: "cyclic-progress") + let progressKeys = ["validation-1", "validation-2", "validation-0"] + var advanceCount = 0 + store._test_codexCostCatchUpStatusOverride = { _ in + CostUsageFetcher.CodexScanCatchUpStatus( + pending: true, + progressKey: "validation-0") + } + store._test_codexCostCatchUpAdvanceOverride = { _, _, _ in + advanceCount += 1 + return CostUsageFetcher.CodexScanCatchUpStatus( + pending: true, + progressKey: progressKeys[min(advanceCount - 1, progressKeys.count - 1)]) + } + store._test_codexCostCatchUpSleepOverride = { _ in + await Task.yield() + } + store._test_codexCostCatchUpResourceStateOverride = { + (.ac, false, .nominal) + } + + store.startCodexCostCatchUpIfNeeded(mode: .accelerated) + await Self.waitUntil { + store.codexCostCatchUpTask == nil + } + + #expect(advanceCount == 3) + #expect(store.codexCostCatchUpActivity?.phase == .paused) + #expect(store.codexCostCatchUpActivity?.pauseReason == .noProgress) + } + @Test func `catch-up continues when existing complete file backlog advances`() async throws { let store = try Self.makeStore(suite: "existing-complete-backlog") diff --git a/Tests/CodexBarTests/UsageStoreSpendDashboardCodexCostCatchUpTests.swift b/Tests/CodexBarTests/UsageStoreSpendDashboardCodexCostCatchUpTests.swift index 93a5419a75..f523344cb9 100644 --- a/Tests/CodexBarTests/UsageStoreSpendDashboardCodexCostCatchUpTests.swift +++ b/Tests/CodexBarTests/UsageStoreSpendDashboardCodexCostCatchUpTests.swift @@ -130,6 +130,39 @@ struct UsageStoreSpendDashboardCodexCostCatchUpTests { #expect(store.spendDashboardCodexCostCatchUpActivity?.pauseReason == .noProgress) } + @Test + func `dashboard catch-up stalls a cache that revisits an earlier semantic state`() async throws { + let store = try Self.makeStore(suite: "cyclic-progress") + let accounts = [Self.account(id: "cyclic", cacheIdentity: "cache-cyclic")] + let progressKeys = ["validation-1", "validation-2", "validation-0"] + var advanceCount = 0 + store._test_spendDashboardCodexCostCatchUpStatusOverride = { _ in + Self.status(pending: true, key: "validation-0", processedBytes: 25) + } + store._test_spendDashboardCodexCostCatchUpAdvanceOverride = { _, _, _ in + advanceCount += 1 + return Self.status( + pending: true, + key: progressKeys[min(advanceCount - 1, progressKeys.count - 1)], + processedBytes: 25) + } + store._test_spendDashboardCodexCostCatchUpSleepOverride = { _ in + await Task.yield() + } + store._test_spendDashboardCodexCostCatchUpResourceStateOverride = { + (.ac, false, .nominal) + } + + store.startSpendDashboardCodexCostCatchUpIfNeeded(accounts: accounts, mode: .accelerated) + await Self.waitUntil { + store.spendDashboardCodexCostCatchUpTask == nil + } + + #expect(advanceCount == 3) + #expect(store.spendDashboardCodexCostCatchUpActivity?.phase == .paused) + #expect(store.spendDashboardCodexCostCatchUpActivity?.pauseReason == .noProgress) + } + @Test func `dashboard synchronization keeps an accelerated account queue accelerated`() throws { let store = try Self.makeStore(suite: "preserve-mode")