From 024e6d494546e77642d64a75d88ac28da9523de7 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 11 Jul 2026 02:56:47 -0700 Subject: [PATCH] Revert "fix: limit Pi cumulative Ultra token overcount (#2043)" This reverts commit db93fc92abca3c1d7e1bf2dc5c27f932e838910a. --- CHANGELOG.md | 1 - .../CodexBarCore/PiSessionCostScanner.swift | 229 ++++-------------- .../PiSessionCostScannerTests.swift | 145 ----------- 3 files changed, 49 insertions(+), 326 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 707ae54c3e..b8119354d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,6 @@ - Codex cost history: bound malformed session-metadata lines and release read chunks promptly, preventing metadata pre-scans from retaining memory in proportion to oversized JSONL records. Thanks @Yuxin-Qiao! - Widgets: add Cursor to configurable and switcher widgets with accurate legacy Requests and current Total, Auto, and API quota labels (#2040). Thanks @Zihao-Qi! - Menus: return oversized tracked menus to the provider header after a manual refresh without moving background updates, highlighted rows, open submenus, or newer menu/provider sessions (#2046). Thanks @ss251! -- Codex cost history: reconcile large monotonic Pi usage snapshots only within their narrowest request, message, turn, or task lineage, preventing forked sessions from multiplying cumulative token totals (#2037). Thanks @kiranmagic7! ## 0.42.0 — 2026-07-11 diff --git a/Sources/CodexBarCore/PiSessionCostScanner.swift b/Sources/CodexBarCore/PiSessionCostScanner.swift index 15fa584e77..27b356892b 100644 --- a/Sources/CodexBarCore/PiSessionCostScanner.swift +++ b/Sources/CodexBarCore/PiSessionCostScanner.swift @@ -46,33 +46,6 @@ enum PiSessionCostScanner { let modelName: String } - private struct UsageSample { - let sequence: Int - let provider: UsageProvider - let dayKey: String - let modelName: String - let lineageKey: String - let pricingDate: Date - let usage: PiPackedUsage - - func replacingUsage(_ usage: PiPackedUsage) -> UsageSample { - UsageSample( - sequence: self.sequence, - provider: self.provider, - dayKey: self.dayKey, - modelName: self.modelName, - lineageKey: self.lineageKey, - pricingDate: self.pricingDate, - usage: usage) - } - } - - private struct UsageSampleGroupKey: Hashable { - let providerRawValue: String - let modelName: String - let lineageKey: String - } - private struct ModelsDevPricingContext { let catalog: ModelsDevCatalog? let cacheRoot: URL? @@ -88,7 +61,7 @@ enum PiSessionCostScanner { private static let costScale = 1_000_000_000.0 /// Bump for Pi-only cost formula changes not represented by the parser or pricing fingerprints. - private static let costFormulaVersion = 2 + private static let costFormulaVersion = 1 private static let maxLineBytes = 16 * 1024 * 1024 private static let maxSafeRoundedInt = Double(Int.max) - 1 private static let sessionStartFilenameRegex = try? NSRegularExpression( @@ -262,9 +235,7 @@ enum PiSessionCostScanner { } private static func defaultPiSessionsRoot(options: Options) -> URL { - if let override = options.piSessionsRoot { - return override - } + if let override = options.piSessionsRoot { return override } return FileManager.default.homeDirectoryForCurrentUser .appendingPathComponent(".pi", isDirectory: true) .appendingPathComponent("agent", isDirectory: true) @@ -341,6 +312,35 @@ enum PiSessionCostScanner { return } + if !context.forceRescan, + let cached, + size > cached.size, + cached.parsedBytes > 0, + cached.parsedBytes <= size + { + let delta = try self.parsePiSessionFile( + fileURL: fileURL, + range: context.range, + startOffset: cached.parsedBytes, + initialModelContext: cached.lastModelContext, + pricingContext: context.pricingContext, + checkCancellation: context.checkCancellation) + if !delta.contributions.isEmpty { + self.applyContributions( + daysByProvider: &cache.daysByProvider, + contributions: delta.contributions, + sign: 1) + } + let merged = self.mergedContributions(existing: cached.contributions, delta: delta.contributions) + storeFileUsage(PiSessionFileUsage( + mtimeUnixMs: mtimeMs, + size: size, + parsedBytes: delta.parsedBytes, + lastModelContext: delta.lastModelContext, + contributions: merged)) + return + } + if let cached { self.applyContributions( daysByProvider: &cache.daysByProvider, @@ -374,7 +374,6 @@ enum PiSessionCostScanner { checkCancellation: CostUsageScanner.CancellationCheck? = nil) throws -> ParseResult { var currentModelContext = initialModelContext - var samples: [UsageSample] = [] var contributions: [String: [String: [String: PiPackedUsage]]] = [:] func add(provider: UsageProvider, dayKey: String, modelName: String, usage: PiPackedUsage) { @@ -438,16 +437,13 @@ enum PiSessionCostScanner { guard let identity else { return } guard let date = self.timestampDate(entry: object, message: message) else { return } let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: date) - let usage = self.extractRawUsage(message: message) - let sequence = samples.count - samples.append(UsageSample( - sequence: sequence, + let usage = self.extractUsage( provider: identity.provider, - dayKey: dayKey, modelName: identity.modelName, - lineageKey: self.lineageKey(entry: object, message: message, fallbackSequence: sequence), + message: message, pricingDate: date, - usage: usage)) + pricingContext: pricingContext) + add(provider: identity.provider, dayKey: dayKey, modelName: identity.modelName, usage: usage) } }) } catch is CancellationError { @@ -456,16 +452,6 @@ enum PiSessionCostScanner { parsedBytes = startOffset } - for sample in self.adjustedUsageSamples(samples) { - let usage = self.pricedUsage( - provider: sample.provider, - modelName: sample.modelName, - usage: sample.usage, - pricingDate: sample.pricingDate, - pricingContext: pricingContext) - add(provider: sample.provider, dayKey: sample.dayKey, modelName: sample.modelName, usage: usage) - } - return ParseResult( contributions: contributions, parsedBytes: parsedBytes, @@ -569,38 +555,6 @@ enum PiSessionCostScanner { ?? self.parseTimestampValue(entry["timestamp"]) } - private static func lineageKey( - entry: [String: Any], - message: [String: Any], - fallbackSequence: Int) -> String - { - // Broader turn and task IDs can span independent requests, so prefer the narrowest stable identity. - let keys = [ - "requestId", - "request_id", - "messageId", - "message_id", - "turnId", - "turn_id", - "taskId", - "task_id", - "id", - ] - for key in keys { - if let value = (message[key] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines), - !value.isEmpty - { - return "\(key):\(value)" - } - if let value = (entry[key] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines), - !value.isEmpty - { - return "\(key):\(value)" - } - } - return "sample:\(fallbackSequence)" - } - private static func parseTimestampValue(_ value: Any?) -> Date? { if let number = value as? NSNumber { let raw = number.doubleValue @@ -624,7 +578,13 @@ enum PiSessionCostScanner { return nil } - private static func extractRawUsage(message: [String: Any]) -> PiPackedUsage { + private static func extractUsage( + provider: UsageProvider, + modelName: String, + message: [String: Any], + pricingDate: Date? = nil, + pricingContext: ModelsDevPricingContext? = nil) -> PiPackedUsage + { let usage = (message["usage"] as? [String: Any]) ?? [:] let input = self.readNonNegativeInt( usage["input"] @@ -664,123 +624,32 @@ enum PiSessionCostScanner { let derivedTotal = input + cacheRead + cacheWrite + output let totalTokens = max(directTotal, derivedTotal) - return PiPackedUsage( + let rawUsage = PiPackedUsage( inputTokens: input, cacheReadTokens: cacheRead, cacheWriteTokens: cacheWrite, outputTokens: output, totalTokens: totalTokens) - } - - private static func pricedUsage( - provider: UsageProvider, - modelName: String, - usage: PiPackedUsage, - pricingDate: Date? = nil, - pricingContext: ModelsDevPricingContext? = nil) -> PiPackedUsage - { // Pi JSONL does not record Anthropic cache retention, so use Pi's persisted default tariff. let costUSD = self.computedCostUSD( provider: provider, modelName: modelName, - usage: usage, + usage: rawUsage, pricingDate: pricingDate, pricingContext: pricingContext) let costNanos = costUSD.map { Int64(($0 * self.costScale).rounded()) } ?? 0 return PiPackedUsage( - inputTokens: usage.inputTokens, - cacheReadTokens: usage.cacheReadTokens, - cacheWriteTokens: usage.cacheWriteTokens, - outputTokens: usage.outputTokens, - totalTokens: usage.totalTokens, + inputTokens: rawUsage.inputTokens, + cacheReadTokens: rawUsage.cacheReadTokens, + cacheWriteTokens: rawUsage.cacheWriteTokens, + outputTokens: rawUsage.outputTokens, + totalTokens: rawUsage.totalTokens, costNanos: costNanos, costSampleCount: costUSD == nil ? 0 : 1, usageSampleCount: 1) } - private static func adjustedUsageSamples(_ samples: [UsageSample]) -> [UsageSample] { - guard samples.count > 1 else { return samples } - - let grouped = Dictionary(grouping: samples) { sample in - UsageSampleGroupKey( - providerRawValue: sample.provider.rawValue, - modelName: sample.modelName, - lineageKey: sample.lineageKey) - } - - var adjusted: [UsageSample] = [] - adjusted.reserveCapacity(samples.count) - - for group in grouped.values { - let sorted = group.sorted { $0.sequence < $1.sequence } - guard self.shouldTreatAsCumulative(sorted) else { - adjusted.append(contentsOf: sorted) - continue - } - - var previous = PiPackedUsage() - for sample in sorted { - let delta = self.usageDelta(from: previous, to: sample.usage) - previous = sample.usage - guard !delta.isZero else { continue } - adjusted.append(sample.replacingUsage(delta)) - } - } - - return adjusted.sorted { $0.sequence < $1.sequence } - } - - private static func shouldTreatAsCumulative(_ samples: [UsageSample]) -> Bool { - guard samples.count >= 4 else { return false } - let usages = samples.map(\.usage) - guard zip(usages, usages.dropFirst()).allSatisfy({ self.isNonDecreasing(previous: $0, current: $1) }) - else { - return false - } - - let rawTotal = usages.reduce(0) { partial, usage in - self.clampedAdd(partial, self.usageTotal(usage)) - } - guard let final = usages.last.map(self.usageTotal), final > 0 else { return false } - - return Double(rawTotal) >= Double(final) * 3.0 - && rawTotal - final >= 1_000_000 - } - - private static func isNonDecreasing(previous: PiPackedUsage, current: PiPackedUsage) -> Bool { - current.inputTokens >= previous.inputTokens - && current.cacheReadTokens >= previous.cacheReadTokens - && current.cacheWriteTokens >= previous.cacheWriteTokens - && current.outputTokens >= previous.outputTokens - && current.totalTokens >= previous.totalTokens - } - - private static func usageDelta(from previous: PiPackedUsage, to current: PiPackedUsage) -> PiPackedUsage { - let input = max(0, current.inputTokens - previous.inputTokens) - let cacheRead = max(0, current.cacheReadTokens - previous.cacheReadTokens) - let cacheWrite = max(0, current.cacheWriteTokens - previous.cacheWriteTokens) - let output = max(0, current.outputTokens - previous.outputTokens) - let total = max(0, current.totalTokens - previous.totalTokens) - return PiPackedUsage( - inputTokens: input, - cacheReadTokens: cacheRead, - cacheWriteTokens: cacheWrite, - outputTokens: output, - totalTokens: max(total, input + cacheRead + cacheWrite + output)) - } - - private static func usageTotal(_ usage: PiPackedUsage) -> Int { - let derivedTotal = [usage.inputTokens, usage.cacheReadTokens, usage.cacheWriteTokens, usage.outputTokens] - .reduce(0, self.clampedAdd) - return max(usage.totalTokens, derivedTotal) - } - - private static func clampedAdd(_ lhs: Int, _ rhs: Int) -> Int { - let added = lhs.addingReportingOverflow(rhs) - return added.overflow ? Int.max : added.partialValue - } - private static func computedCostUSD( provider: UsageProvider, modelName: String, diff --git a/Tests/CodexBarTests/PiSessionCostScannerTests.swift b/Tests/CodexBarTests/PiSessionCostScannerTests.swift index 8d5a72af41..e451cca1db 100644 --- a/Tests/CodexBarTests/PiSessionCostScannerTests.swift +++ b/Tests/CodexBarTests/PiSessionCostScannerTests.swift @@ -806,151 +806,6 @@ struct PiSessionCostScannerTests { } extension PiSessionCostScannerTests { - @Test - func `pi scanner deltas cumulative ultra usage samples per lineage`() throws { - try self.assertLargeMonotonicPiUsage( - relativePath: "2026-07-11T10-00-00-000Z_ultra-cumulative.jsonl", - expected: (6_000_000, 600_000, 6000, 6_606_000), - messageFields: { _ in ["turnId": "ultra-fork-turn"] }) - } - - @Test - func `pi scanner keeps ordinary increasing rows below cumulative guardrail raw`() throws { - let env = try CostUsageTestEnvironment() - defer { env.cleanup() } - - let day = try env.makeLocalNoon(year: 2026, month: 7, day: 11) - - func assistant(step: Int) -> [String: Any] { - let timestamp = day.addingTimeInterval(TimeInterval(step)) - let input = step * 100 - let output = step * 10 - return [ - "type": "message", - "timestamp": env.isoString(for: timestamp), - "message": [ - "role": "assistant", - "provider": "openai-codex", - "model": "gpt-5.4", - "turnId": "small-increasing-turn", - "timestamp": Int(timestamp.timeIntervalSince1970 * 1000), - "usage": [ - "input": input, - "output": output, - "totalTokens": input + output, - ], - ], - ] - } - - _ = try env.writePiSessionFile( - relativePath: "2026-07-11T10-00-00-000Z_small-increasing.jsonl", - contents: env.jsonl((1...5).map(assistant(step:)))) - - let report = PiSessionCostScanner.loadDailyReport( - provider: .codex, - since: day, - until: day, - now: day, - options: PiSessionCostScanner.Options( - piSessionsRoot: env.piSessionsRoot, - cacheRoot: env.cacheRoot, - refreshMinIntervalSeconds: 0)) - - #expect(report.data.count == 1) - #expect(report.data.first?.inputTokens == 1500) - #expect(report.data.first?.outputTokens == 150) - #expect(report.data.first?.totalTokens == 1650) - } - - @Test - func `pi scanner keeps unique requests additive under shared turn and task ids`() throws { - try self.assertLargeMonotonicPiUsage( - relativePath: "2026-07-11T10-00-00-000Z_shared-task-unique-requests.jsonl", - expected: (21_000_000, 2_100_000, 21000, 23_121_000), - messageFields: { step in - [ - "turnId": "shared-pi-turn", - "taskId": "shared-pi-task", - "requestId": "request-\(step)", - ] - }) - } - - @Test - func `pi scanner keeps unique event ids additive when usage is large and monotonic`() throws { - try self.assertLargeMonotonicPiUsage( - relativePath: "2026-07-11T10-00-00-000Z_unique-events-monotonic.jsonl", - expected: (21_000_000, 2_100_000, 21000, 23_121_000), - entryFields: { step in ["id": "event-\(step)"] }) - } - - @Test - func `pi scanner keeps rows without lineage raw even when large and monotonic`() throws { - try self.assertLargeMonotonicPiUsage( - relativePath: "2026-07-11T10-00-00-000Z_no-lineage-monotonic.jsonl", - expected: (21_000_000, 2_100_000, 21000, 23_121_000)) - } - - private func assertLargeMonotonicPiUsage( - relativePath: String, - expected: (input: Int, cacheRead: Int, output: Int, total: Int), - entryFields: (Int) -> [String: Any] = { _ in [:] }, - messageFields: (Int) -> [String: Any] = { _ in [:] }) throws - { - let env = try CostUsageTestEnvironment() - defer { env.cleanup() } - - let day = try env.makeLocalNoon(year: 2026, month: 7, day: 11) - - func assistant(step: Int) -> [String: Any] { - let timestamp = day.addingTimeInterval(TimeInterval(step)) - let input = step * 1_000_000 - let cacheRead = step * 100_000 - let output = step * 1000 - var message: [String: Any] = [ - "role": "assistant", - "provider": "openai-codex", - "model": "gpt-5.6-sol", - "timestamp": Int(timestamp.timeIntervalSince1970 * 1000), - "usage": [ - "input": input, - "cacheRead": cacheRead, - "output": output, - "totalTokens": input + cacheRead + output, - ], - ] - message.merge(messageFields(step)) { _, replacement in replacement } - var entry: [String: Any] = [ - "type": "message", - "timestamp": env.isoString(for: timestamp), - "message": message, - ] - entry.merge(entryFields(step)) { _, replacement in replacement } - return entry - } - - _ = try env.writePiSessionFile( - relativePath: relativePath, - contents: env.jsonl((1...6).map(assistant(step:)))) - - let report = PiSessionCostScanner.loadDailyReport( - provider: .codex, - since: day, - until: day, - now: day, - options: PiSessionCostScanner.Options( - piSessionsRoot: env.piSessionsRoot, - cacheRoot: env.cacheRoot, - refreshMinIntervalSeconds: 0)) - - #expect(report.data.count == 1) - #expect(report.data.first?.inputTokens == expected.input) - #expect(report.data.first?.cacheReadTokens == expected.cacheRead) - #expect(report.data.first?.outputTokens == expected.output) - #expect(report.data.first?.totalTokens == expected.total) - } - @Test func `pi scanner reprices unchanged files when catalog rates change`() throws { let env = try CostUsageTestEnvironment()