diff --git a/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift b/Sources/CodexBarCore/Generated/CodexParserHash.generated.swift index 3802864e25..6e66e0fd77 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 = "577571954a2d36bc" + static let value = "60b2fc9bcc6dd4c7" } diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CodexSubagentRolloutShape.swift b/Sources/CodexBarCore/Vendored/CostUsage/CodexSubagentRolloutShape.swift new file mode 100644 index 0000000000..22d51a1d05 --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CodexSubagentRolloutShape.swift @@ -0,0 +1,162 @@ +import Foundation + +extension CostUsageScanner { + enum CodexSubagentCounterSemantics: Equatable { + case independent + case copiedPrefix + } + + /// Subagent source is lineage evidence, not counter semantics. The first session metadata + /// owns leaf identity; only embedded metadata proves that this rollout copied an ancestor + /// prefix. Do not restore a blanket "all subagents are independent/inherited" rule. + struct CodexSubagentRolloutShape { + let counterSemantics: CodexSubagentCounterSemantics + let ownedSuffix: CodexSubagentOwnedSuffix? + let inferredParentSessionID: String? + + struct CodexSubagentOwnedSuffix { + let startLineIndex: Int + let rawTotalsBaseline: CostUsageCodexTotals + } + + struct Observation { + let lineIndex: Int + let kind: Kind + + enum Kind { + case sessionMetadata(id: String?) + case turnContext + case interAgentCommunication(triggerTurn: Bool) + case tokenCount(total: CostUsageCodexTotals?, last: CostUsageCodexTotals?) + } + } + + static func classify( + leafSessionID: String?, + observedSessionIDs: [String?]) -> Self + { + let normalizedLeafID = Self.normalizedSessionID(leafSessionID) + + let hasEmbeddedAncestor: Bool = if let normalizedLeafID { + observedSessionIDs.contains { Self.normalizedSessionID($0) != normalizedLeafID } + } else { + observedSessionIDs.count > 1 || observedSessionIDs.contains { Self.normalizedSessionID($0) != nil } + } + let distinctAncestorIDs = Set(observedSessionIDs + .compactMap(Self.normalizedSessionID) + .filter { normalizedLeafID == nil || $0 != normalizedLeafID }) + let inferredParentSessionID = distinctAncestorIDs.count == 1 ? distinctAncestorIDs.first : nil + + return Self( + counterSemantics: hasEmbeddedAncestor ? .copiedPrefix : .independent, + ownedSuffix: nil, + inferredParentSessionID: inferredParentSessionID) + } + + static func classify( + leafSessionID: String?, + observations: [Observation]) -> Self + { + let metadataIDs = observations.reduce(into: [String?]()) { result, observation in + guard case let .sessionMetadata(id) = observation.kind else { return } + result.append(id) + } + let metadataShape = Self.classify( + leafSessionID: leafSessionID, + observedSessionIDs: metadataIDs) + guard metadataShape.counterSemantics == .copiedPrefix else { return metadataShape } + + let normalizedLeafID = Self.normalizedSessionID(leafSessionID) + var lastRawTotals: CostUsageCodexTotals? + var pendingTurnContext: (lineIndex: Int, baseline: CostUsageCodexTotals)? + var ownedSuffix: CodexSubagentOwnedSuffix? + var inspectedOwnedSuffixFirstTotal = false + var observedAuthoritativeMetadata = false + + for observation in observations { + switch observation.kind { + case let .sessionMetadata(id): + let normalizedID = Self.normalizedSessionID(id) + let isEmbeddedAncestor: Bool = if !observedAuthoritativeMetadata { + false + } else if let normalizedLeafID { + normalizedID != normalizedLeafID + } else { + true + } + observedAuthoritativeMetadata = true + if isEmbeddedAncestor { + // A later ancestor meta proves that any earlier candidate boundary was replay. + ownedSuffix = nil + inspectedOwnedSuffixFirstTotal = false + } + pendingTurnContext = nil + + case .turnContext: + pendingTurnContext = lastRawTotals.map { (observation.lineIndex, $0) } + + case let .interAgentCommunication(triggerTurn): + if ownedSuffix == nil, + triggerTurn, + let pendingTurnContext, + observation.lineIndex == pendingTurnContext.lineIndex + 1 + { + ownedSuffix = Self.CodexSubagentOwnedSuffix( + startLineIndex: pendingTurnContext.lineIndex, + rawTotalsBaseline: pendingTurnContext.baseline) + inspectedOwnedSuffixFirstTotal = false + } + pendingTurnContext = nil + + case let .tokenCount(total, last): + if !inspectedOwnedSuffixFirstTotal, + let suffix = ownedSuffix, + let total + { + inspectedOwnedSuffixFirstTotal = true + if let last, + Self.totalsEqual(total, last), + !Self.totalsAtLeast(total, suffix.rawTotalsBaseline) + { + // Some future protocol may copy history and then restart its counter. + // Require both a strong boundary and total==last reset evidence. + ownedSuffix = Self.CodexSubagentOwnedSuffix( + startLineIndex: suffix.startLineIndex, + rawTotalsBaseline: .init(input: 0, cached: 0, output: 0)) + } + } + if let total { + lastRawTotals = total + } + pendingTurnContext = nil + } + } + + return Self( + counterSemantics: .copiedPrefix, + ownedSuffix: ownedSuffix, + inferredParentSessionID: metadataShape.inferredParentSessionID) + } + + static func sameConcreteSessionID(_ lhs: String?, _ rhs: String?) -> Bool { + guard let lhs = normalizedSessionID(lhs), + let rhs = normalizedSessionID(rhs) + else { return false } + return lhs == rhs + } + + private static func totalsEqual(_ lhs: CostUsageCodexTotals, _ rhs: CostUsageCodexTotals) -> Bool { + lhs.input == rhs.input && lhs.cached == rhs.cached && lhs.output == rhs.output + } + + private static func totalsAtLeast(_ lhs: CostUsageCodexTotals, _ rhs: CostUsageCodexTotals) -> Bool { + lhs.input >= rhs.input && lhs.cached >= rhs.cached && lhs.output >= rhs.output + } + + private static func normalizedSessionID(_ value: String?) -> String? { + guard let value else { return nil } + let trimmed = value.trimmingCharacters(in: .whitespacesAndNewlines) + return trimmed.isEmpty ? nil : trimmed + } + } +} diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CacheHelpers.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CacheHelpers.swift index 27ee4b0e21..8bdbeec25d 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CacheHelpers.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CacheHelpers.swift @@ -907,9 +907,11 @@ extension CostUsageScanner { } if let parentSessionId = cached.forkedFromId { guard let cachedDependencyKey = cached.forkBaselineDependencyKey else { return false } - let currentDependencyKey = try context.resources.inheritedResolver - .currentDependencyKey(for: parentSessionId) - guard cachedDependencyKey == currentDependencyKey else { return false } + if cachedDependencyKey != Self.codexForkDependencyNotRequiredKey { + let currentDependencyKey = try context.resources.inheritedResolver + .currentDependencyKey(for: parentSessionId) + guard cachedDependencyKey == currentDependencyKey else { return false } + } } if sessionAlreadyContributed { @@ -974,6 +976,14 @@ extension CostUsageScanner { if Self.cachedCodexRowsNeedIdentityRescan(cached) { return false } + // Subagent shape depends on the complete lineage prefix. Appended metadata can change an + // independent counter into a copied-prefix rollout, so a tail-only parse is not sound. + if try Self.codexFileIsSubagentThread( + fileURL: input.fileURL, + checkCancellation: context.checkCancellation) + { + return false + } let startOffset = cached.parsedBytes ?? cached.size let initialCountedTotals = cached.lastCountedTotals ?? cached.lastTotals let initialRawTotalsBaseline = cached.lastRawTotalsBaseline ?? cached.lastTotals @@ -1131,11 +1141,10 @@ extension CostUsageScanner { range: context.range, inheritedTotalsResolver: context.resources.inheritedResolver.inheritedTotals(for:atOrBefore:), checkCancellation: context.checkCancellation) - let forkBaselineDependencyKey: String? = if let parentSessionId = parsed.forkedFromId { - context.resources.inheritedResolver.dependencyKeyUsed(for: parentSessionId) - } else { - nil - } + let forkBaselineDependencyKey = Self.codexForkBaselineDependencyKey( + parentSessionId: parsed.forkedFromId, + dependsOnParentTotals: parsed.dependsOnParentTotals, + inheritedResolver: context.resources.inheritedResolver) let sessionId = parsed.sessionId ?? input.cached?.sessionId let projectPath = parsed.projectPath ?? input.cached?.projectPath let canonicalProjectPath = parsed.projectPath.map { @@ -1237,6 +1246,19 @@ extension CostUsageScanner { state: &state) } + static func codexForkBaselineDependencyKey( + parentSessionId: String?, + dependsOnParentTotals: Bool, + inheritedResolver: CodexInheritedTotalsResolver) -> String? + { + guard let parentSessionId else { return nil } + guard dependsOnParentTotals else { return Self.codexForkDependencyNotRequiredKey } + + // A nil key means the parent changed while its snapshots were read (or no stable + // snapshot was resolved). Preserve nil so the child cannot be reused on the next scan. + return inheritedResolver.dependencyKeyUsed(for: parentSessionId) + } + static func mergeFileDays( existing: inout [String: [String: [Int]]], delta: [String: [String: [Int]]]) diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexFastJSON.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexFastJSON.swift index 4d48a84a81..9080839149 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexFastJSON.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexFastJSON.swift @@ -56,6 +56,17 @@ extension CostUsageScanner { } } + static func extractJSONByteBoolField( + _ field: [UInt8], + from bytes: UnsafeBufferPointer, + in range: Range, + atDepth targetDepth: Int) -> Bool? + { + self.extractJSONByteField(field, from: bytes, in: range, atDepth: targetDepth) { valueIndex in + self.parseJSONByteBool(in: bytes, index: &valueIndex, limit: range.upperBound) + } + } + private static func extractJSONByteField( _ field: [UInt8], from bytes: UnsafeBufferPointer, @@ -187,6 +198,33 @@ extension CostUsageScanner { return sawDigit ? (sign == -1 ? -value : value) : nil } + private static func parseJSONByteBool( + in bytes: UnsafeBufferPointer, + index: inout Int, + limit: Int) -> Bool? + { + if index + 4 <= limit, + bytes[index] == 0x74, + bytes[index + 1] == 0x72, + bytes[index + 2] == 0x75, + bytes[index + 3] == 0x65 + { + index += 4 + return true + } + if index + 5 <= limit, + bytes[index] == 0x66, + bytes[index + 1] == 0x61, + bytes[index + 2] == 0x6C, + bytes[index + 3] == 0x73, + bytes[index + 4] == 0x65 + { + index += 5 + return false + } + return nil + } + private static func skipJSONByteWhitespace( in bytes: UnsafeBufferPointer, index: inout Int, diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexTruncatedPrefix.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexTruncatedPrefix.swift index 45946748f7..f736c387e5 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexTruncatedPrefix.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner+CodexTruncatedPrefix.swift @@ -1,12 +1,31 @@ import Foundation extension CostUsageScanner { - static func extractCodexTurnContextModel(from bytes: Data) -> String? { - guard let text = truncatedUTF8String(from: bytes) else { return nil } + static func extractCodexTruncatedSessionMetadata(from bytes: Data) -> + (isSessionMetadata: Bool, sessionID: String?) + { + guard let text = truncatedUTF8String(from: bytes) else { return (false, nil) } + let object = text[...] + guard Self.extractJSONStringField("type", from: object, atDepth: 1) == "session_meta" else { + return (false, nil) + } + guard let payloadText = Self.extractJSONObjectField("payload", from: object, atDepth: 1) else { + return (true, nil) + } + let sessionID = Self.extractJSONStringField("id", from: payloadText, atDepth: 1) + ?? Self.extractJSONStringField("session_id", from: payloadText, atDepth: 1) + ?? Self.extractJSONStringField("sessionId", from: payloadText, atDepth: 1) + return (true, sessionID) + } + + static func extractCodexTruncatedTurnContext(from bytes: Data) -> (isValid: Bool, model: String?) { + guard let text = truncatedUTF8String(from: bytes) else { return (false, nil) } let object = text[...] guard Self.extractJSONStringField("type", from: object, atDepth: 1) == "turn_context", + let timestamp = Self.extractJSONStringField("timestamp", from: object, atDepth: 1), + Self.dayKeyFromTimestamp(timestamp) ?? Self.dayKeyFromParsedISO(timestamp) != nil, let payloadText = Self.extractJSONObjectField("payload", from: object, atDepth: 1) - else { return nil } + else { return (false, nil) } let infoText = Self.extractJSONObjectField("info", from: payloadText, atDepth: 1) let model = Self.codexTurnContextModel( @@ -18,8 +37,8 @@ extension CostUsageScanner { infoModelName: infoText.flatMap { Self.extractJSONStringFieldAllowingEmpty("model_name", from: $0, atDepth: 1) }) - guard let model, model.isEmpty else { return model } - return Self.isCompleteJSONObject(payloadText) ? "" : nil + guard let model, model.isEmpty else { return (true, model) } + return (true, Self.isCompleteJSONObject(payloadText) ? "" : nil) } static func truncatedUTF8String(from bytes: Data) -> String? { @@ -44,8 +63,12 @@ extension CostUsageScanner { case "}": depth -= 1 text.formIndex(after: &index) - if depth == 0 { return true } - if depth < 0 { return false } + if depth == 0 { + return true + } + if depth < 0 { + return false + } case "\"": guard Self.parseJSONString(in: text, index: &index) != nil else { return false } default: diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner.swift index de337ed815..b4695bbc8b 100644 --- a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner.swift +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageScanner.swift @@ -14,6 +14,9 @@ enum CostUsageScanner { static let log = CodexBarLog.logger(LogCategories.tokenCost) static let codexActiveSessionLookbackDays = 30 static let costScale = 1_000_000_000.0 + /// Reserved cache marker. Resolver-produced dependencies use `file|...` or `missing:...`; + /// this value records that lineage exists but this rollout owns its counter or suffix. + static let codexForkDependencyNotRequiredKey = "mode:lineage-only:v1" enum ClaudeLogProviderFilter { case all @@ -62,6 +65,7 @@ enum CostUsageScanner { let lastCodexTurnID: String? let sessionId: String? let forkedFromId: String? + let dependsOnParentTotals: Bool let projectPath: String? let rows: [CodexUsageRow] } @@ -1306,8 +1310,23 @@ enum CostUsageScanner { private enum CodexFastLine { case sessionMeta(CodexSessionMetadata) case turnContext(model: String?) + case interAgentCommunication(triggerTurn: Bool) case taskStarted(turnID: String?) case tokenCount(CodexTokenCountRecord) + + var requiresValidTimestamp: Bool { + switch self { + case .sessionMeta: + false + case .turnContext, .interAgentCommunication, .taskStarted, .tokenCount: + true + } + } + } + + private struct CodexBufferedFastLine { + let lineIndex: Int + let line: CodexFastLine } private static let codexJSONFieldCachedInputTokens = Array("cached_input_tokens".utf8) @@ -1330,6 +1349,7 @@ enum CostUsageScanner { private static let codexJSONFieldSessionIdCamel = Array("sessionId".utf8) private static let codexJSONFieldTimestamp = Array("timestamp".utf8) private static let codexJSONFieldTotalTokenUsage = Array("total_token_usage".utf8) + private static let codexJSONFieldTriggerTurn = Array("trigger_turn".utf8) private static let codexJSONFieldTurnId = Array("turn_id".utf8) private static let codexJSONFieldTurnIdCamel = Array("turnId".utf8) private static let codexJSONFieldType = Array("type".utf8) @@ -1516,6 +1536,24 @@ enum CostUsageScanner { return CostUsageCodexTotals(input: input, cached: cached, output: output) } + private static func codexInterAgentCommunication( + from bytes: UnsafeBufferPointer, + in objectRange: Range) -> CodexFastLine? + { + guard let payloadRange = extractJSONByteObjectField( + codexJSONFieldPayload, + from: bytes, + in: objectRange, + atDepth: 1), + let triggerTurn = extractJSONByteBoolField( + codexJSONFieldTriggerTurn, + from: bytes, + in: payloadRange, + atDepth: 1) + else { return nil } + return .interAgentCommunication(triggerTurn: triggerTurn) + } + private static func parseCodexFastLine(_ bytes: Data) -> CodexFastLine? { bytes.withUnsafeBytes { rawBytes in let rawBuffer = rawBytes.bindMemory(to: UInt8.self) @@ -1593,6 +1631,11 @@ enum CostUsageScanner { }) return .turnContext(model: model) + case "inter_agent_communication_metadata": + // Compact Codex JSONL uses this exact spelling. Whitespace/escaped variants fall + // through to Foundation so a fast-path miss cannot change boundary semantics. + return Self.codexInterAgentCommunication(from: rawBuffer, in: objectRange) + case "event_msg": guard let payloadRange = Self.extractJSONByteObjectField( Self.codexJSONFieldPayload, @@ -1670,6 +1713,20 @@ enum CostUsageScanner { } } + private static func codexFastLineTimestampValidity(_ bytes: Data) -> Bool? { + let timestamp = bytes.withUnsafeBytes { rawBytes in + let rawBuffer = rawBytes.bindMemory(to: UInt8.self) + guard !rawBuffer.isEmpty else { return nil as String? } + return Self.extractJSONByteStringField( + Self.codexJSONFieldTimestamp, + from: rawBuffer, + in: 0.. String? @@ -1790,6 +1847,15 @@ enum CostUsageScanner { return nil } + static func codexFileIsSubagentThread( + fileURL: URL, + checkCancellation: CancellationCheck? = nil) throws -> Bool + { + try self.parseCodexSessionMetadata( + fileURL: fileURL, + checkCancellation: checkCancellation)?.isSubagentThread == true + } + private static func parseCodexTokenSnapshots( fileURL: URL, checkCancellation: CancellationCheck? = nil) throws -> ( @@ -1838,7 +1904,7 @@ enum CostUsageScanner { } case let .tokenCount(record): appendSnapshot(timestamp: record.timestamp, last: record.last, total: record.total) - case .turnContext, .taskStarted: + case .turnContext, .interAgentCommunication, .taskStarted: break } return @@ -1942,6 +2008,7 @@ enum CostUsageScanner { lastCodexTurnID: initialCodexTurnID, sessionId: nil, forkedFromId: nil, + dependsOnParentTotals: false, projectPath: nil, rows: []) } @@ -1969,6 +2036,11 @@ enum CostUsageScanner { var forkedFromId: String? var projectPath: String? var isSubagentThread = false + var didCaptureLeafMetadata = false + var forkTimestamp: String? + var subagentCounterSemantics: CodexSubagentCounterSemantics? + var usesLocalSubagentBoundary = false + var suppressUnownedCopiedPrefix = false var inheritedTotals: CostUsageCodexTotals? var remainingInheritedTotals: CostUsageCodexTotals? var forkBaselineResolved = false @@ -2015,35 +2087,54 @@ enum CostUsageScanner { } } + func configureForkAccountingIfReady() throws { + guard let forkedFromId else { return } + if isSubagentThread, subagentCounterSemantics == nil { + return + } + if subagentCounterSemantics == .independent || usesLocalSubagentBoundary { + forkBaselineResolved = true + inheritedTotals = nil + remainingInheritedTotals = nil + hasUnresolvedForkBaseline = false + return + } + try resolveForkBaseline( + parentSessionId: forkedFromId, + forkedAt: forkTimestamp ?? "") + } + func handleSessionMetadata(_ metadata: CodexSessionMetadata) throws { - if sessionId == nil { - sessionId = metadata.sessionId - } - if forkedFromId == nil { - forkedFromId = metadata.forkedFromId - } - if projectPath == nil { - projectPath = metadata.projectPath - } - isSubagentThread = isSubagentThread || metadata.isSubagentThread - if let forkedFromId { - if isSubagentThread { - // Codex subagent rollouts own an independent cumulative counter; their parent - // identifiers describe lineage, not a token baseline to subtract (#2193). - forkBaselineResolved = true - inheritedTotals = nil - remainingInheritedTotals = nil - hasUnresolvedForkBaseline = false - } else { - try resolveForkBaseline(parentSessionId: forkedFromId, forkedAt: metadata.forkTimestamp ?? "") + // The first parsed session_meta is the authoritative leaf. Copied prefixes can + // contain many embedded ancestor metas; they are shape evidence, never new identity. + if didCaptureLeafMetadata { + // A same-leaf restart may add metadata that was absent from the initial record. + // Enrich missing fork/project fields without allowing an ancestor to replace identity. + guard CodexSubagentRolloutShape.sameConcreteSessionID(metadata.sessionId, sessionId) else { return } + if forkedFromId == nil, let enrichedParentID = metadata.forkedFromId { + forkedFromId = enrichedParentID + forkTimestamp = metadata.forkTimestamp ?? forkTimestamp + try configureForkAccountingIfReady() + } + if projectPath == nil { + projectPath = metadata.projectPath } + return } + didCaptureLeafMetadata = true + sessionId = metadata.sessionId + forkedFromId = metadata.forkedFromId + forkTimestamp = metadata.forkTimestamp + projectPath = metadata.projectPath + isSubagentThread = metadata.isSubagentThread + try configureForkAccountingIfReady() } // swiftlint:disable:next function_body_length func handleTokenCount(_ record: CodexTokenCountRecord) throws { guard let dayKey = Self.dayKeyFromTimestamp(record.timestamp) ?? Self.dayKeyFromParsedISO(record.timestamp) else { return } + guard !suppressUnownedCopiedPrefix else { return } let model = Self.codexModelEvidence(currentModel) ?? Self.codexModelEvidence(record.model) @@ -2077,9 +2168,8 @@ enum CostUsageScanner { return adjusted } - // Generic fork children are measured against their parent-inherited baseline so every - // cumulative comparison in this file happens on a single scale. Subagents use their - // independent counter directly. + // Fork totals are normalized against the selected baseline. Classified independent + // counters and locally delimited suffixes intentionally bypass the parent baseline. let adjustedTotal: CostUsageCodexTotals? = total.map { rawTotals in guard let inheritedTotals, !hasUnresolvedForkBaseline else { return rawTotals } return CostUsageCodexTotals( @@ -2248,7 +2338,7 @@ enum CostUsageScanner { } } - func handleFastLine(_ fastLine: CodexFastLine) throws { + func processFastLine(_ fastLine: CodexFastLine) throws { switch fastLine { case let .sessionMeta(metadata): try handleSessionMetadata(metadata) @@ -2256,6 +2346,8 @@ enum CostUsageScanner { if let model { currentModel = model } + case .interAgentCommunication: + break case let .taskStarted(turnID): currentTurnID = turnID case let .tokenCount(record): @@ -2266,15 +2358,31 @@ enum CostUsageScanner { let maxLineBytes = 256 * 1024 let prefixBytes = maxLineBytes + var pendingSubagentLines: [CodexBufferedFastLine]? + if startOffset == 0, let metadata = try Self.parseCodexSessionMetadata( fileURL: fileURL, checkCancellation: checkCancellation) { try handleSessionMetadata(metadata) + if metadata.isSubagentThread { + // Subagent provenance can omit a fork id. Buffer parsed events, not JSON, so + // classification remains one disk pass and reuses the existing totals reducer. + pendingSubagentLines = [] + } + } + + func routeFastLine(_ fastLine: CodexFastLine, lineIndex: Int) throws { + if pendingSubagentLines != nil { + pendingSubagentLines?.append(Self.CodexBufferedFastLine(lineIndex: lineIndex, line: fastLine)) + } else { + try processFastLine(fastLine) + } } var parsedBytes: Int64 + var physicalLineIndex = 0 do { parsedBytes = try CostUsageJsonl.scan( fileURL: fileURL, @@ -2283,14 +2391,42 @@ enum CostUsageScanner { prefixBytes: prefixBytes, checkCancellation: checkCancellation, onLine: { line in + let lineIndex = physicalLineIndex + physicalLineIndex += 1 if deferredError != nil { return } guard !line.bytes.isEmpty else { return } if line.wasTruncated { // `turn_context` can carry very large prompts, but its model usually appears near the start. - if let model = Self.extractCodexTurnContextModel(from: line.bytes) { - currentModel = model + // A truncated line cannot be structurally validated with Foundation, so + // only accept the canonical root discriminator to avoid prompt-text hits. + let truncatedTurnContext = Self.extractCodexTruncatedTurnContext(from: line.bytes) + if truncatedTurnContext.isValid { + do { + try routeFastLine( + .turnContext(model: truncatedTurnContext.model), + lineIndex: lineIndex) + } catch { + deferredError = error + } + } + if pendingSubagentLines != nil { + let truncatedMetadata = Self.extractCodexTruncatedSessionMetadata(from: line.bytes) + if truncatedMetadata.isSessionMetadata { + do { + try routeFastLine( + .sessionMeta(CodexSessionMetadata( + sessionId: truncatedMetadata.sessionID, + forkedFromId: nil, + forkTimestamp: nil, + projectPath: nil, + isSubagentThread: false)), + lineIndex: lineIndex) + } catch { + deferredError = error + } + } } return } @@ -2298,7 +2434,11 @@ enum CostUsageScanner { guard line.bytes.containsAscii(#""type":"event_msg""#) || line.bytes.containsAscii(#""type":"turn_context""#) + || line.bytes.containsAscii(#""turn_context""#) || line.bytes.containsAscii(#""type":"session_meta""#) + || line.bytes.containsAscii(#""session_meta""#) + || line.bytes.containsAscii(#""type":"inter_agent_communication_metadata""#) + || line.bytes.containsAscii(#""inter_agent_communication_metadata""#) else { return } if line.bytes.containsAscii(#""type":"event_msg""#), @@ -2309,12 +2449,20 @@ enum CostUsageScanner { } if let fastLine = Self.parseCodexFastLine(line.bytes) { - do { - try handleFastLine(fastLine) - } catch { - deferredError = error + let timestampValidity = fastLine.requiresValidTimestamp + ? Self.codexFastLineTimestampValidity(line.bytes) + : true + if timestampValidity == true { + do { + try routeFastLine(fastLine, lineIndex: lineIndex) + } catch { + deferredError = error + } + return + } + if timestampValidity == false { + return } - return } autoreleasepool { @@ -2326,7 +2474,7 @@ enum CostUsageScanner { if type == "session_meta" { guard let metadata = Self.codexSessionMetadata(from: obj) else { return } do { - try handleSessionMetadata(metadata) + try routeFastLine(.sessionMeta(metadata), lineIndex: lineIndex) } catch { deferredError = error } @@ -2337,17 +2485,32 @@ enum CostUsageScanner { guard Self.dayKeyFromTimestamp(tsText) ?? Self.dayKeyFromParsedISO(tsText) != nil else { return } + if type == "inter_agent_communication_metadata" { + let payload = obj["payload"] as? [String: Any] + do { + try routeFastLine( + .interAgentCommunication(triggerTurn: payload?["trigger_turn"] as? Bool == true), + lineIndex: lineIndex) + } catch { + deferredError = error + } + return + } + if type == "turn_context" { + var model: String? if let payload = obj["payload"] as? [String: Any] { let info = payload["info"] as? [String: Any] - if let model = Self.codexTurnContextModel( + model = Self.codexTurnContextModel( payloadModel: payload["model"] as? String, payloadModelName: payload["model_name"] as? String, infoModel: info?["model"] as? String, infoModelName: info?["model_name"] as? String) - { - currentModel = model - } + } + do { + try routeFastLine(.turnContext(model: model), lineIndex: lineIndex) + } catch { + deferredError = error } return } @@ -2355,7 +2518,13 @@ enum CostUsageScanner { guard type == "event_msg" else { return } guard let payload = obj["payload"] as? [String: Any] else { return } if (payload["type"] as? String) == "task_started" { - currentTurnID = Self.codexTurnID(from: payload) + do { + try routeFastLine( + .taskStarted(turnID: Self.codexTurnID(from: payload)), + lineIndex: lineIndex) + } catch { + deferredError = error + } return } guard (payload["type"] as? String) == "token_count" else { return } @@ -2387,7 +2556,7 @@ enum CostUsageScanner { last: (info?["last_token_usage"] as? [String: Any]).map(tokenTotals), total: (info?["total_token_usage"] as? [String: Any]).map(tokenTotals)) do { - try handleTokenCount(record) + try routeFastLine(.tokenCount(record), lineIndex: lineIndex) } catch { deferredError = error } @@ -2396,6 +2565,90 @@ enum CostUsageScanner { if let deferredError { throw deferredError } + + if let pendingSubagentLines { + // Same-leaf metadata can fill lineage fields after the opening record. Collect it + // before replay so copied-prefix totals never run once on the wrong baseline, and + // so an owned-suffix filter cannot discard the only fork identifier. + for buffered in pendingSubagentLines { + guard case let .sessionMeta(metadata) = buffered.line, + CodexSubagentRolloutShape.sameConcreteSessionID(metadata.sessionId, sessionId) + else { continue } + if forkedFromId == nil, let enrichedParentID = metadata.forkedFromId { + forkedFromId = enrichedParentID + forkTimestamp = metadata.forkTimestamp ?? forkTimestamp + } + if projectPath == nil { + projectPath = metadata.projectPath + } + } + let observations = pendingSubagentLines.compactMap { buffered -> CodexSubagentRolloutShape + .Observation? in + let kind: CodexSubagentRolloutShape.Observation.Kind + switch buffered.line { + case let .sessionMeta(metadata): + kind = .sessionMetadata(id: metadata.sessionId) + case .turnContext: + kind = .turnContext + case let .interAgentCommunication(triggerTurn): + kind = .interAgentCommunication(triggerTurn: triggerTurn) + case let .tokenCount(record): + kind = .tokenCount(total: record.total, last: record.last) + case .taskStarted: + return nil + } + return Self.CodexSubagentRolloutShape.Observation( + lineIndex: buffered.lineIndex, + kind: kind) + } + let shape = CodexSubagentRolloutShape.classify( + leafSessionID: sessionId, + observations: observations) + subagentCounterSemantics = shape.counterSemantics + if forkedFromId == nil { + forkedFromId = shape.inferredParentSessionID + } + suppressUnownedCopiedPrefix = shape.counterSemantics == .copiedPrefix + && shape.ownedSuffix == nil + && forkedFromId == nil + if let ownedSuffix = shape.ownedSuffix { + usesLocalSubagentBoundary = true + previousTotals = nil + // Keep totals-derived accounting after the boundary. Real flat-total rows + // repeat the previous token payload with a fresh outer timestamp; their + // non-zero `last` is replay evidence, not new usage (#2037). + rawTotalsBaseline = ownedSuffix.rawTotalsBaseline + sawDivergentTotals = false + tracker = CodexTotalsTracker( + watermark: ownedSuffix.rawTotalsBaseline, + seenRawTotals: [], + sawInterleavedTotals: false) + currentModel = nil + currentTurnID = nil + unresolvedForkTotalWatermark = nil + } + self.log.debug( + "Codex cost usage classified subagent rollout counter semantics", + metadata: [ + "sessionId": sessionId ?? "unknown", + "semantics": subagentCounterSemantics == .copiedPrefix ? "copiedPrefix" : "independent", + "localBoundary": shape.ownedSuffix == nil ? "false" : "true", + "suppressedUnownedPrefix": suppressUnownedCopiedPrefix ? "true" : "false", + "sessionMetadataCount": String(observations.count(where: { + if case .sessionMetadata = $0.kind { + true + } else { + false + } + })), + ]) + try configureForkAccountingIfReady() + for buffered in pendingSubagentLines + where shape.ownedSuffix.map({ buffered.lineIndex >= $0.startLineIndex }) ?? true + { + try processFastLine(buffered.line) + } + } } catch is CancellationError { throw CancellationError() } catch { @@ -2421,6 +2674,9 @@ enum CostUsageScanner { lastCodexTurnID: currentTurnID, sessionId: sessionId, forkedFromId: forkedFromId, + dependsOnParentTotals: forkedFromId != nil + && subagentCounterSemantics != .independent + && !usesLocalSubagentBoundary, projectPath: projectPath, rows: rows) } diff --git a/Tests/CodexBarTests/CodexSubagentAccountingIntegrationTests.swift b/Tests/CodexBarTests/CodexSubagentAccountingIntegrationTests.swift new file mode 100644 index 0000000000..74343bc9c6 --- /dev/null +++ b/Tests/CodexBarTests/CodexSubagentAccountingIntegrationTests.swift @@ -0,0 +1,617 @@ +import Foundation +import Testing +@testable import CodexBarCore + +struct CodexSubagentAccountingIntegrationTests { + private typealias Usage = (input: Int, cached: Int, output: Int) + + @Test + func `copied parent prefix keeps the inherited baseline after late lineage metadata`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let forkTimestamp = env.isoString(for: day) + let parentModel = "openai/gpt-5.3" + let leafModel = "openai/gpt-5.4" + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(forkTimestamp)-child-session.jsonl", + contents: env.jsonl([ + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "child-session", + "timestamp": forkTimestamp, + "source": [ + "subagent": [ + "thread_spawn": ["parent_thread_id": "parent-session"], + ], + ], + ], + ], + self.turnContext(timestamp: forkTimestamp, model: parentModel), + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: parentModel, + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "child-session", + "forked_from_id": "parent-session", + "timestamp": forkTimestamp, + ], + ], + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "parent-session", + "timestamp": forkTimestamp, + ], + ], + self.turnContext(timestamp: forkTimestamp, model: leafModel), + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: leafModel, + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ])) + + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { parentSessionID, _ in + resolvedParentBaseline = true + #expect(parentSessionID == "parent-session") + return .resolved(.init(input: 1000, cached: 900, output: 100)) + }) + + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let normalizedLeafModel = CostUsagePricing.normalizeCodexModel(leafModel) + #expect(parsed.days[dayKey]?[normalizedLeafModel] == [50, 10, 5]) + #expect(parsed.days[dayKey]?[CostUsagePricing.normalizeCodexModel(parentModel)] == nil) + #expect(resolvedParentBaseline) + #expect(parsed.dependsOnParentTotals) + } + + @Test + func `local marker owns only its suffix and persists lineage-only cache mode`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let forkTimestamp = env.isoString(for: day) + let parentModel = "openai/gpt-5.3" + let leafModel = "openai/gpt-5.4" + let fastContents = try env.jsonl([ + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "marker-child", + "timestamp": forkTimestamp, + "source": [ + "subagent": [ + "thread_spawn": ["parent_thread_id": "parent-session"], + ], + ], + ], + ], + self.turnContext(timestamp: forkTimestamp, model: parentModel), + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: parentModel, + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "marker-child", + "forked_from_id": "parent-session", + "timestamp": forkTimestamp, + ], + ], + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": ["id": "ancestor-session"], + ], + self.turnContext(timestamp: env.isoString(for: day.addingTimeInterval(2)), model: leafModel), + [ + "type": "inter_agent_communication_metadata", + "timestamp": env.isoString(for: day.addingTimeInterval(2)), + "payload": ["trigger_turn": true], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(2.5)), + model: parentModel, + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(3)), + model: leafModel, + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ]) + let fastFileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(forkTimestamp)-marker-child.jsonl", + contents: fastContents) + let fallbackFileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(forkTimestamp)-marker-child-fallback.jsonl", + contents: fastContents + .replacingOccurrences(of: "marker-child", with: "marker-child-fallback") + .replacingOccurrences( + of: "\"type\":\"session_meta\"", + with: "\"ty\\u0070e\":\"session_meta\"") + .replacingOccurrences( + of: "\"type\":\"turn_context\"", + with: "\"ty\\u0070e\":\"turn_context\"") + .replacingOccurrences( + of: "\"type\":\"inter_agent_communication_metadata\"", + with: "\"ty\\u0070e\":\"inter_agent_communication_metadata\"")) + let escapedTimestampFileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(forkTimestamp)-marker-child-escaped-timestamp.jsonl", + contents: fastContents + .replacingOccurrences(of: "marker-child", with: "marker-child-escaped-timestamp") + .replacingOccurrences(of: "\"timestamp\":", with: "\"time\\u0073tamp\":")) + + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let normalizedLeafModel = CostUsagePricing.normalizeCodexModel(leafModel) + for fileURL in [fastFileURL, fallbackFileURL, escapedTimestampFileURL] { + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { _, _ in + resolvedParentBaseline = true + return .resolved(.init(input: 10, cached: 0, output: 0)) + }) + #expect(parsed.days[dayKey]?[normalizedLeafModel] == [50, 10, 5]) + #expect(parsed.days[dayKey]?[CostUsagePricing.normalizeCodexModel(parentModel)] == nil) + #expect(!parsed.dependsOnParentTotals) + #expect(!resolvedParentBaseline) + } + + var options = CostUsageScanner.Options( + codexSessionsRoot: env.codexSessionsRoot, + claudeProjectsRoots: nil, + cacheRoot: env.cacheRoot) + options.refreshMinIntervalSeconds = 0 + let report = CostUsageScanner.loadDailyReport( + provider: .codex, + since: day, + until: day, + now: day, + options: options) + #expect(report.data.first?.totalTokens == 165) + + let cache = CostUsageCacheIO.load(provider: .codex, cacheRoot: env.cacheRoot) + let childUsages = cache.files.values.filter { $0.sessionId?.hasPrefix("marker-child") == true } + #expect(childUsages.count == 3) + #expect(childUsages.allSatisfy { + $0.forkBaselineDependencyKey == CostUsageScanner.codexForkDependencyNotRequiredKey + }) + } + + @Test + func `copied prefix infers its parent and ignores a spoofed trigger outside the payload`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let forkTimestamp = env.isoString(for: day) + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(forkTimestamp)-inferred-parent.jsonl", + contents: env.jsonl([ + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": [ + "id": "inferred-child", + "timestamp": forkTimestamp, + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + [ + "type": "session_meta", + "timestamp": forkTimestamp, + "payload": ["id": "inferred-parent"], + ], + self.turnContext( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: "openai/gpt-5.4"), + [ + "type": "inter_agent_communication_metadata", + "timestamp": env.isoString(for: day.addingTimeInterval(2)), + "trigger_turn": true, + "payload": ["trigger_turn": false], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(3)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ])) + + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { parentSessionID, _ in + resolvedParentBaseline = true + #expect(parentSessionID == "inferred-parent") + return .resolved(.init(input: 1000, cached: 900, output: 100)) + }) + + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let model = CostUsagePricing.normalizeCodexModel("openai/gpt-5.4") + #expect(parsed.days[dayKey]?[model] == [50, 10, 5]) + #expect(parsed.forkedFromId == "inferred-parent") + #expect(parsed.dependsOnParentTotals) + #expect(resolvedParentBaseline) + } + + @Test + func `oversized ancestor metadata remains conservative copied-prefix evidence`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let timestamp = env.isoString(for: day) + let opening = try env.jsonl([ + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": [ + "id": "oversized-child", + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + ]) + let oversizedAncestor = "{\"type\":\"session_meta\",\"timestamp\":\"\(timestamp)\"," + + "\"payload\":{\"id\":\"oversized-parent\",\"padding\":\"" + + String(repeating: "x", count: 300_000) + "\"}}\n" + let tail = try env.jsonl([ + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ]) + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-oversized-ancestor.jsonl", + contents: opening + oversizedAncestor + tail) + + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { parentSessionID, _ in + resolvedParentBaseline = true + #expect(parentSessionID == "oversized-parent") + return .resolved(.init(input: 1000, cached: 900, output: 100)) + }) + + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let model = CostUsagePricing.normalizeCodexModel("openai/gpt-5.4") + #expect(parsed.days[dayKey]?[model] == [50, 10, 5]) + #expect(parsed.forkedFromId == "oversized-parent") + #expect(parsed.dependsOnParentTotals) + #expect(resolvedParentBaseline) + } + + @Test + func `invalid timestamp suffix markers preserve parent dependency on both parser paths`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let timestamp = env.isoString(for: day) + let contents = try env.jsonl([ + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": [ + "id": "invalid-marker-child", + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": ["id": "invalid-marker-parent"], + ], + [ + "type": "turn_context", + "payload": ["model": "openai/gpt-5.4"], + ], + [ + "type": "inter_agent_communication_metadata", + "payload": ["trigger_turn": true], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ]) + let fastFileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-invalid-marker.jsonl", + contents: contents) + let fallbackFileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-invalid-marker-fallback.jsonl", + contents: contents + .replacingOccurrences(of: "invalid-marker-child", with: "invalid-marker-child-fallback") + .replacingOccurrences(of: "\"type\":\"turn_context\"", with: "\"ty\\u0070e\":\"turn_context\"") + .replacingOccurrences( + of: "\"type\":\"inter_agent_communication_metadata\"", + with: "\"ty\\u0070e\":\"inter_agent_communication_metadata\"")) + + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let model = CostUsagePricing.normalizeCodexModel("openai/gpt-5.4") + for fileURL in [fastFileURL, fallbackFileURL] { + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { parentSessionID, _ in + resolvedParentBaseline = true + #expect(parentSessionID == "invalid-marker-parent") + return .resolved(.init(input: 1000, cached: 900, output: 100)) + }) + #expect(parsed.days[dayKey]?[model] == [50, 10, 5]) + #expect(parsed.dependsOnParentTotals) + #expect(resolvedParentBaseline) + } + } + + @Test + func `oversized invalid suffix markers preserve parent dependency`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let timestamp = env.isoString(for: day) + let opening = try env.jsonl([ + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": [ + "id": "oversized-marker-child", + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100), + last: (input: 50, cached: 10, output: 5)), + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": ["id": "oversized-marker-parent"], + ], + ]) + let padding = String(repeating: "x", count: 300_000) + let invalidTimestamp = "{\"type\":\"turn_context\",\"timestamp\":\"invalid\"," + + "\"payload\":{\"model\":\"openai/gpt-5.4\",\"padding\":\"\(padding)\"}}\n" + let nestedType = "{\"type\":\"event_msg\",\"timestamp\":\"\(timestamp)\"," + + "\"payload\":{\"type\":\"turn_context\",\"padding\":\"\(padding)\"}}\n" + let tail = try env.jsonl([ + [ + "type": "inter_agent_communication_metadata", + "timestamp": env.isoString(for: day.addingTimeInterval(2)), + "payload": ["trigger_turn": true], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(3)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ]) + + let files = try [invalidTimestamp, nestedType].enumerated().map { index, marker in + try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-oversized-invalid-marker-\(index).jsonl", + contents: opening + marker + tail) + } + let dayKey = CostUsageScanner.CostUsageDayRange.dayKey(from: day) + let model = CostUsagePricing.normalizeCodexModel("openai/gpt-5.4") + for fileURL in files { + var resolvedParentBaseline = false + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day), + inheritedTotalsResolver: { parentSessionID, _ in + resolvedParentBaseline = true + #expect(parentSessionID == "oversized-marker-parent") + return .resolved(.init(input: 1000, cached: 900, output: 100)) + }) + #expect(parsed.days[dayKey]?[model] == [50, 10, 5]) + #expect(parsed.dependsOnParentTotals) + #expect(resolvedParentBaseline) + } + } + + @Test + func `idless copied prefix without a parent or local marker is suppressed`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let timestamp = env.isoString(for: day) + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-ambiguous-prefix.jsonl", + contents: env.jsonl([ + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": [ + "id": "ambiguous-child", + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100)), + ["type": "session_meta", "timestamp": timestamp, "payload": [:]], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105)), + ])) + + let parsed = CostUsageScanner.parseCodexFile( + fileURL: fileURL, + range: CostUsageScanner.CostUsageDayRange(since: day, until: day)) + + #expect(parsed.days.isEmpty) + #expect(parsed.rows.isEmpty) + } + + @Test + func `appended ancestor metadata reclassifies the complete subagent rollout`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let day = try env.makeLocalNoon(year: 2026, month: 7, day: 16) + let timestamp = env.isoString(for: day) + let fileURL = try env.writeCodexSessionFile( + day: day, + filename: "rollout-\(timestamp)-growing-subagent.jsonl", + contents: env.jsonl([ + [ + "type": "session_meta", + "timestamp": timestamp, + "payload": [ + "id": "growing-child", + "source": ["subagent": ["thread_spawn": [:]]], + ], + ], + self.turnContext(timestamp: timestamp, model: "openai/gpt-5.3"), + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(1)), + model: "openai/gpt-5.3", + total: (input: 1000, cached: 900, output: 100)), + ])) + + var options = CostUsageScanner.Options( + codexSessionsRoot: env.codexSessionsRoot, + claudeProjectsRoots: nil, + cacheRoot: env.cacheRoot) + options.refreshMinIntervalSeconds = 0 + let first = CostUsageScanner.loadDailyReport( + provider: .codex, + since: day, + until: day, + now: day, + options: options) + #expect(first.data.first?.totalTokens == 1100) + + let appended = try env.jsonl([ + ["type": "session_meta", "timestamp": timestamp, "payload": ["id": "growing-parent"]], + self.turnContext( + timestamp: env.isoString(for: day.addingTimeInterval(2)), + model: "openai/gpt-5.4"), + [ + "type": "inter_agent_communication_metadata", + "timestamp": env.isoString(for: day.addingTimeInterval(2)), + "payload": ["trigger_turn": true], + ], + self.tokenCount( + timestamp: env.isoString(for: day.addingTimeInterval(3)), + model: "openai/gpt-5.4", + total: (input: 1050, cached: 910, output: 105), + last: (input: 50, cached: 10, output: 5)), + ]) + let handle = try FileHandle(forWritingTo: fileURL) + try handle.seekToEnd() + try handle.write(contentsOf: Data(appended.utf8)) + try handle.close() + + let second = CostUsageScanner.loadDailyReport( + provider: .codex, + since: day, + until: day, + now: day.addingTimeInterval(1), + options: options) + #expect(second.data.first?.totalTokens == 55) + + let cache = CostUsageCacheIO.load(provider: .codex, cacheRoot: env.cacheRoot) + let usage = try #require(cache.files.values.first { $0.sessionId == "growing-child" }) + #expect(usage.sessionId == "growing-child") + #expect(usage.forkedFromId == "growing-parent") + #expect(usage.forkBaselineDependencyKey == CostUsageScanner.codexForkDependencyNotRequiredKey) + } + + private func turnContext(timestamp: String, model: String) -> [String: Any] { + [ + "type": "turn_context", + "timestamp": timestamp, + "payload": ["model": model], + ] + } + + private func tokenCount( + timestamp: String, + model: String, + total: Usage? = nil, + last: Usage? = nil) -> [String: Any] + { + var info: [String: Any] = ["model": model] + if let total { + info["total_token_usage"] = [ + "input_tokens": total.input, + "cached_input_tokens": total.cached, + "output_tokens": total.output, + ] + } + if let last { + info["last_token_usage"] = [ + "input_tokens": last.input, + "cached_input_tokens": last.cached, + "output_tokens": last.output, + ] + } + return [ + "type": "event_msg", + "timestamp": timestamp, + "payload": [ + "type": "token_count", + "info": info, + ], + ] + } +} diff --git a/Tests/CodexBarTests/CodexSubagentRolloutShapeTests.swift b/Tests/CodexBarTests/CodexSubagentRolloutShapeTests.swift new file mode 100644 index 0000000000..d97355e5f1 --- /dev/null +++ b/Tests/CodexBarTests/CodexSubagentRolloutShapeTests.swift @@ -0,0 +1,190 @@ +import Testing +@testable import CodexBarCore + +struct CodexSubagentRolloutShapeTests { + @Test + func `single leaf metadata means an independent counter`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observedSessionIDs: ["leaf"]) + + #expect(shape.counterSemantics == .independent) + } + + @Test + func `embedded ancestor metadata means a copied prefix`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observedSessionIDs: ["leaf", "parent"]) + + #expect(shape.counterSemantics == .copiedPrefix) + #expect(shape.inferredParentSessionID == "parent") + } + + @Test + func `multiple ancestors do not infer an ambiguous parent`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observedSessionIDs: ["leaf", "parent", "grandparent"]) + + #expect(shape.counterSemantics == .copiedPrefix) + #expect(shape.inferredParentSessionID == nil) + } + + @Test + func `repeated leaf metadata does not invent an ancestor`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observedSessionIDs: ["leaf", "leaf"]) + + #expect(shape.counterSemantics == .independent) + } + + @Test + func `unknown leaf followed by a concrete metadata id is copied`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: nil, + observedSessionIDs: [nil, "parent"]) + + #expect(shape.counterSemantics == .copiedPrefix) + #expect(shape.inferredParentSessionID == "parent") + } + + @Test + func `idless metadata after a known leaf is conservatively copied`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observedSessionIDs: ["leaf", nil]) + + #expect(shape.counterSemantics == .copiedPrefix) + #expect(shape.inferredParentSessionID == nil) + } + + @Test + func `only concrete normalized ids identify the same leaf`() { + #expect(CostUsageScanner.CodexSubagentRolloutShape.sameConcreteSessionID(" leaf ", "leaf")) + #expect(!CostUsageScanner.CodexSubagentRolloutShape.sameConcreteSessionID(nil, nil)) + #expect(!CostUsageScanner.CodexSubagentRolloutShape.sameConcreteSessionID("", "")) + } + + @Test + func `adjacent trigger after the final ancestor opens an owned suffix`() throws { + let baseline = CostUsageCodexTotals(input: 1000, cached: 900, output: 100) + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observations: [ + .init(lineIndex: 0, kind: .sessionMetadata(id: "leaf")), + .init(lineIndex: 4, kind: .tokenCount(total: baseline, last: nil)), + .init(lineIndex: 5, kind: .sessionMetadata(id: "parent")), + .init(lineIndex: 8, kind: .turnContext), + .init(lineIndex: 9, kind: .interAgentCommunication(triggerTurn: true)), + ]) + + let suffix = try #require(shape.ownedSuffix) + #expect(shape.counterSemantics == .copiedPrefix) + #expect(suffix.startLineIndex == 8) + #expect(suffix.rawTotalsBaseline.input == 1000) + #expect(suffix.rawTotalsBaseline.cached == 900) + #expect(suffix.rawTotalsBaseline.output == 100) + } + + @Test + func `nonadjacent trigger does not invent an owned suffix`() { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observations: [ + .init( + lineIndex: 0, + kind: .tokenCount( + total: .init(input: 1000, cached: 900, output: 100), + last: nil)), + .init(lineIndex: 1, kind: .sessionMetadata(id: "parent")), + .init(lineIndex: 3, kind: .turnContext), + .init(lineIndex: 5, kind: .interAgentCommunication(triggerTurn: true)), + ]) + + #expect(shape.counterSemantics == .copiedPrefix) + #expect(shape.ownedSuffix == nil) + } + + @Test + func `copied prefix can restart only with strong reset evidence`() throws { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observations: [ + .init(lineIndex: 0, kind: .sessionMetadata(id: "leaf")), + .init( + lineIndex: 2, + kind: .tokenCount( + total: .init(input: 1000, cached: 900, output: 100), + last: nil)), + .init(lineIndex: 3, kind: .sessionMetadata(id: "parent")), + .init(lineIndex: 5, kind: .turnContext), + .init(lineIndex: 6, kind: .interAgentCommunication(triggerTurn: true)), + .init( + lineIndex: 7, + kind: .tokenCount( + total: .init(input: 50, cached: 10, output: 5), + last: .init(input: 50, cached: 10, output: 5))), + ]) + + let suffix = try #require(shape.ownedSuffix) + #expect(suffix.rawTotalsBaseline.input == 0) + #expect(suffix.rawTotalsBaseline.cached == 0) + #expect(suffix.rawTotalsBaseline.output == 0) + } + + @Test + func `first valid leaf marker owns later leaf turns`() throws { + let firstBaseline = CostUsageCodexTotals(input: 1000, cached: 900, output: 100) + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observations: [ + .init(lineIndex: 0, kind: .sessionMetadata(id: "leaf")), + .init(lineIndex: 1, kind: .sessionMetadata(id: "parent")), + .init(lineIndex: 2, kind: .tokenCount(total: firstBaseline, last: nil)), + .init(lineIndex: 4, kind: .turnContext), + .init(lineIndex: 5, kind: .interAgentCommunication(triggerTurn: true)), + .init( + lineIndex: 6, + kind: .tokenCount( + total: .init(input: 1050, cached: 910, output: 105), + last: nil)), + .init(lineIndex: 8, kind: .turnContext), + .init(lineIndex: 9, kind: .interAgentCommunication(triggerTurn: true)), + ]) + + let suffix = try #require(shape.ownedSuffix) + #expect(suffix.startLineIndex == 4) + #expect(suffix.rawTotalsBaseline.input == firstBaseline.input) + } + + @Test + func `later ancestor invalidates a tentative marker`() throws { + let shape = CostUsageScanner.CodexSubagentRolloutShape.classify( + leafSessionID: "leaf", + observations: [ + .init(lineIndex: 0, kind: .sessionMetadata(id: "leaf")), + .init(lineIndex: 1, kind: .sessionMetadata(id: "parent")), + .init( + lineIndex: 2, + kind: .tokenCount( + total: .init(input: 1000, cached: 900, output: 100), + last: nil)), + .init(lineIndex: 3, kind: .turnContext), + .init(lineIndex: 4, kind: .interAgentCommunication(triggerTurn: true)), + .init(lineIndex: 5, kind: .sessionMetadata(id: "grandparent")), + .init( + lineIndex: 6, + kind: .tokenCount( + total: .init(input: 2000, cached: 1800, output: 200), + last: nil)), + .init(lineIndex: 8, kind: .turnContext), + .init(lineIndex: 9, kind: .interAgentCommunication(triggerTurn: true)), + ]) + + let suffix = try #require(shape.ownedSuffix) + #expect(suffix.startLineIndex == 8) + #expect(suffix.rawTotalsBaseline.input == 2000) + } +} diff --git a/Tests/CodexBarTests/CostUsageScannerBreakdownTests.swift b/Tests/CodexBarTests/CostUsageScannerBreakdownTests.swift index dd9a6063b8..dd40efcbf7 100644 --- a/Tests/CodexBarTests/CostUsageScannerBreakdownTests.swift +++ b/Tests/CodexBarTests/CostUsageScannerBreakdownTests.swift @@ -4266,6 +4266,64 @@ struct CostUsageScannerBreakdownTests { #expect(resolver.dependencyKeyUsed(for: parentSessionId) == parsedDependencyKey) } + @Test + func `codex unstable parent snapshot keeps fork dependency uncached`() throws { + let env = try CostUsageTestEnvironment() + defer { env.cleanup() } + + let parentDay = try env.makeLocalNoon(year: 2026, month: 2, day: 1) + let parentSessionId = "sess-parent-unstable" + let model = "openai/gpt-5.2-codex" + let metadata: [String: Any] = [ + "type": "session_meta", + "payload": ["id": parentSessionId], + ] + let firstContents = try env.jsonl([ + metadata, + self.codexTurnContext(timestamp: env.isoString(for: parentDay), model: model), + self.codexTokenCount( + timestamp: env.isoString(for: parentDay.addingTimeInterval(1)), + model: model, + total: (input: 20, cached: 5, output: 2)), + ]) + let secondContents = try env.jsonl([ + metadata, + self.codexTurnContext(timestamp: env.isoString(for: parentDay), model: model), + self.codexTokenCount( + timestamp: env.isoString(for: parentDay.addingTimeInterval(1)), + model: model, + total: (input: 21, cached: 5, output: 2)), + ]) + let parentURL = try env.writeCodexSessionFile( + day: parentDay, + filename: "rollout-2026-02-01T12-00-00-\(parentSessionId).jsonl", + contents: firstContents) + let fileIndex = CostUsageScanner.CodexSessionFileIndex( + files: [parentURL], + roots: [], + cachedSessionFiles: [parentSessionId: parentURL]) + var mutationCount = 0 + let resolver = CostUsageScanner.CodexInheritedTotalsResolver( + fileIndex: fileIndex, + checkCancellation: { + mutationCount += 1 + let contents = mutationCount.isMultiple(of: 2) ? firstContents : secondContents + try contents.write(to: parentURL, atomically: true, encoding: .utf8) + }) + + let baseline = try resolver.inheritedTotals( + for: parentSessionId, + atOrBefore: env.isoString(for: parentDay.addingTimeInterval(2))) + if case .resolved = baseline { + Issue.record("Expected an unstable parent snapshot to stay unresolved") + } + #expect(resolver.dependencyKeyUsed(for: parentSessionId) == nil) + #expect(CostUsageScanner.codexForkBaselineDependencyKey( + parentSessionId: parentSessionId, + dependsOnParentTotals: true, + inheritedResolver: resolver) == nil) + } + @Test func `codex forked child skips cumulative totals when parent session is missing`() throws { let env = try CostUsageTestEnvironment() @@ -4774,15 +4832,28 @@ struct CostUsageScannerBreakdownTests { claudeProjectsRoots: nil, cacheRoot: env.cacheRoot) options.refreshMinIntervalSeconds = 0 + let coldReport = CostUsageScanner.loadDailyReport( + provider: .codex, + since: day, + until: day, + now: forkDate.addingTimeInterval(3), + options: options) + let warmReport = CostUsageScanner.loadDailyReport( + provider: .codex, + since: day, + until: day, + now: forkDate.addingTimeInterval(4), + options: options) options.forceRescan = true - let report = CostUsageScanner.loadDailyReport( provider: .codex, since: day, until: day, - now: forkDate.addingTimeInterval(3), + now: forkDate.addingTimeInterval(5), options: options) + #expect(coldReport.data.first?.totalTokens == 63_065_400) + #expect(warmReport.data.first?.totalTokens == 63_065_400) #expect(report.data.count == 1) #expect(report.data[0].inputTokens == 60_062_200) #expect(report.data[0].cacheReadTokens == 48_051_000) @@ -4802,6 +4873,8 @@ struct CostUsageScannerBreakdownTests { #expect(abs((report.data[0].costUSD ?? 0) - expectedCost) < 0.000001) let cache = CostUsageCacheIO.load(provider: .codex, cacheRoot: env.cacheRoot) + let childUsage = try #require(cache.files.values.first(where: { $0.sessionId == "child-session" })) + #expect(childUsage.forkBaselineDependencyKey == CostUsageScanner.codexForkDependencyNotRequiredKey) let projects = CostUsageScanner.buildCodexProjectBreakdownsFromCache( cache: cache, range: CostUsageScanner.CostUsageDayRange(since: day, until: day),