Skip to content

fix: revert unsafe Pi cumulative reconciliation #2059

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
steipete merged 1 commit into from
Jul 11, 2026
+49 −326
Merged

fix: revert unsafe Pi cumulative reconciliation #2059

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
229 changes: 49 additions & 180 deletions Sources/CodexBarCore/PiSessionCostScanner.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?
Expand All @@ -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(
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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 {
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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"]
Expand Down Expand Up @@ -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,
Expand Down
Loading