From 1097b0a035713ee530f1c34e305341266e9a2385 Mon Sep 17 00:00:00 2001 From: Bryan Font Date: Mon, 13 Jul 2026 14:42:06 -0400 Subject: [PATCH] Bound Codex lineage reconciliation --- .../Providers/Codex/CodexLineageEngine.swift | 429 ++++++++++++++++++ .../CodexLineageEngineTests.swift | 208 +++++++++ 2 files changed, 637 insertions(+) create mode 100644 Sources/CodexBarCore/Providers/Codex/CodexLineageEngine.swift create mode 100644 Tests/CodexBarTests/CodexLineageEngineTests.swift diff --git a/Sources/CodexBarCore/Providers/Codex/CodexLineageEngine.swift b/Sources/CodexBarCore/Providers/Codex/CodexLineageEngine.swift new file mode 100644 index 0000000000..8e9e3c921c --- /dev/null +++ b/Sources/CodexBarCore/Providers/Codex/CodexLineageEngine.swift @@ -0,0 +1,429 @@ +#if canImport(CryptoKit) +import CryptoKit +#else +import Crypto +#endif +import Foundation + +/// Pure, in-memory preparation and reuse seam for lineage families. +/// +/// Callers retain the previous cache until `publish` succeeds. This keeps cancellation and failed +/// reconciliation from exposing a partially rebuilt set of families. +enum CodexLineageEngine { + private static let algorithmVersion = 1 + + struct Fingerprint: Equatable, Hashable, Sendable { + let value: String + } + + struct PreparedFamily: Equatable, Sendable { + let stableID: String + let inputFingerprint: Fingerprint + let documents: [CodexLineageLedger.Document] + let unresolvedParents: Set + let observationCount: Int + } + + struct FamilyResult: Equatable, Sendable { + let stableID: String + let inputFingerprint: Fingerprint + let familyFingerprint: Fingerprint + let quality: CodexLineageLedger.FamilyQuality + let report: CodexLineageLedger.Report + } + + struct Cache: Equatable, Sendable { + let algorithmVersion: Int + let familiesByInputFingerprint: [Fingerprint: FamilyResult] + + static let empty = Self(algorithmVersion: CodexLineageEngine.algorithmVersion, familiesByInputFingerprint: [:]) + } + + struct Diagnostics: Equatable, Sendable { + let familyCount: Int + let recomputedFamilyCount: Int + let reusedFamilyCount: Int + let observationCount: Int + let peakFamilyObservationCount: Int + let peakAcceptedFingerprintCount: Int + } + + struct Result: Equatable, Sendable { + let report: CodexLineageLedger.Report + let families: [FamilyResult] + let candidateCache: Cache + let diagnostics: Diagnostics + } + + static func prepareFamilies( + documents: [CodexLineageLedger.Document], + unresolvedParents: Set = [], + checkCancellation: CostUsageScanner.CancellationCheck? = nil) throws -> [PreparedFamily] + { + var graph = DisjointSet() + for document in documents { + try checkCancellation?() + let owner = Self.scoped(document.ownerID, scopeID: document.scopeID) + graph.insert(owner) + if let metadata = Self.nonEmpty(document.metadataSessionID) { + graph.union(owner, Self.scoped(metadata, scopeID: document.scopeID)) + } + if let parent = Self.nonEmpty(document.parentSessionID) { + graph.union(owner, Self.scoped(parent, scopeID: document.scopeID)) + } + } + + var documentsByRoot: [String: [CodexLineageLedger.Document]] = [:] + for document in documents { + try checkCancellation?() + let root = graph.find(Self.scoped(document.ownerID, scopeID: document.scopeID)) + documentsByRoot[root, default: []].append(document) + } + + var families: [PreparedFamily] = [] + families.reserveCapacity(documentsByRoot.count) + for familyDocuments in documentsByRoot.values { + try checkCancellation?() + var keyedDocuments: [(document: CodexLineageLedger.Document, key: Fingerprint)] = [] + keyedDocuments.reserveCapacity(familyDocuments.count) + for document in familyDocuments { + try checkCancellation?() + try keyedDocuments.append(( + document: document, + key: Self.documentFingerprint(document, checkCancellation: checkCancellation))) + } + keyedDocuments.sort { $0.key.value < $1.key.value } + let sortedDocuments = keyedDocuments.map(\.document) + let identities = Set(sortedDocuments.flatMap { document in + [document.ownerID, document.metadataSessionID, document.parentSessionID].compactMap { value in + Self.nonEmpty(value).map { Self.scoped($0, scopeID: document.scopeID) } + } + }) + let familyUnresolved = unresolvedParents.filter { parent in + identities.contains(Self.scoped(parent.sessionID, scopeID: parent.scopeID)) + } + let stableID = identities.min() ?? "" + let fingerprint = try Self.inputFingerprint( + documents: sortedDocuments, + unresolvedParents: familyUnresolved, + checkCancellation: checkCancellation) + families.append(PreparedFamily( + stableID: stableID, + inputFingerprint: fingerprint, + documents: sortedDocuments, + unresolvedParents: familyUnresolved, + observationCount: sortedDocuments.reduce(0) { $0 + $1.observations.count })) + } + return families.sorted { $0.stableID < $1.stableID } + } + + static func reconcile( + families: [PreparedFamily], + previousCache: Cache? = nil, + localTimeZone: TimeZone, + checkCancellation: CostUsageScanner.CancellationCheck? = nil) throws -> Result + { + let reusable = previousCache?.algorithmVersion == Self.algorithmVersion + ? previousCache?.familiesByInputFingerprint ?? [:] + : [:] + var results: [FamilyResult] = [] + results.reserveCapacity(families.count) + var recomputed = 0 + var reused = 0 + var peakAccepted = 0 + + for family in families.sorted(by: { $0.stableID < $1.stableID }) { + try checkCancellation?() + let cacheFingerprint = Self.cacheFingerprint( + input: family.inputFingerprint, + localTimeZone: localTimeZone) + if let cached = reusable[cacheFingerprint], + cached.stableID == family.stableID, + cached.inputFingerprint == family.inputFingerprint + { + results.append(cached) + reused += 1 + peakAccepted = max(peakAccepted, cached.report.acceptedObservationCount) + continue + } + let conservative = try CodexLineageLedger.reconcileConservatively( + documents: family.documents, + unresolvedParents: family.unresolvedParents, + localTimeZone: localTimeZone, + checkCancellation: checkCancellation) + let quality = conservative.families.first?.quality ?? .primary + let familyFingerprint = Self.familyFingerprint(input: cacheFingerprint, quality: quality) + let result = FamilyResult( + stableID: family.stableID, + inputFingerprint: family.inputFingerprint, + familyFingerprint: familyFingerprint, + quality: quality, + report: conservative.primary) + results.append(result) + recomputed += 1 + peakAccepted = max(peakAccepted, result.report.acceptedObservationCount) + } + + results.sort { $0.stableID < $1.stableID } + let report = try Self.compose(results.map(\.report), checkCancellation: checkCancellation) + let candidate = Cache( + algorithmVersion: Self.algorithmVersion, + familiesByInputFingerprint: Dictionary(uniqueKeysWithValues: results.map { + (Self.cacheFingerprint(input: $0.inputFingerprint, localTimeZone: localTimeZone), $0) + })) + try checkCancellation?() + return Result( + report: report, + families: results, + candidateCache: candidate, + diagnostics: Diagnostics( + familyCount: families.count, + recomputedFamilyCount: recomputed, + reusedFamilyCount: reused, + observationCount: families.reduce(0) { $0 + $1.observationCount }, + peakFamilyObservationCount: families.map(\.observationCount).max() ?? 0, + peakAcceptedFingerprintCount: peakAccepted)) + } + + static func publish( + _ candidate: Cache, + to cache: inout Cache, + checkCancellation: CostUsageScanner.CancellationCheck? = nil) throws + { + try checkCancellation?() + cache = candidate + } + + private static func compose( + _ reports: [CodexLineageLedger.Report], + checkCancellation: CostUsageScanner.CancellationCheck?) throws -> CodexLineageLedger.Report + { + var utcDays: [String: CodexLineageLedger.Totals] = [:] + var localDays: [String: CodexLineageLedger.Totals] = [:] + var utcRows: [RowKey: RowValue] = [:] + var localRows: [RowKey: RowValue] = [:] + for report in reports { + try checkCancellation?() + try Self.mergeDays(report.utcDays, into: &utcDays, checkCancellation: checkCancellation) + try Self.mergeDays(report.localDays, into: &localDays, checkCancellation: checkCancellation) + try Self.mergeRows(report.utcRows, into: &utcRows, checkCancellation: checkCancellation) + try Self.mergeRows(report.localRows, into: &localRows, checkCancellation: checkCancellation) + } + return CodexLineageLedger.Report( + utcDays: utcDays, + localDays: localDays, + utcRows: Self.rows(utcRows), + localRows: Self.rows(localRows), + componentCount: reports.reduce(0) { $0 + $1.componentCount }, + acceptedObservationCount: reports.reduce(0) { $0 + $1.acceptedObservationCount }, + duplicateObservationCount: reports.reduce(0) { $0 + $1.duplicateObservationCount }) + } + + private struct RowKey: Hashable { + let day: String + let model: String + } + + private struct RowValue { + var totals = CodexLineageLedger.Totals.zero + var costUSD = 0.0 + var isPriced = true + } + + private static func mergeDays( + _ source: [String: CodexLineageLedger.Totals], + into destination: inout [String: CodexLineageLedger.Totals], + checkCancellation: CostUsageScanner.CancellationCheck?) throws + { + for (day, totals) in source { + try checkCancellation?() + var combined = destination[day] ?? .zero + combined.add(totals) + destination[day] = combined + } + } + + private static func mergeRows( + _ source: [CodexLineageLedger.DailyRow], + into destination: inout [RowKey: RowValue], + checkCancellation: CostUsageScanner.CancellationCheck?) throws + { + for row in source { + try checkCancellation?() + let key = RowKey(day: row.day, model: row.model) + var value = destination[key] ?? RowValue() + value.totals.add(row.totals) + if let cost = row.costUSD { + value.costUSD += cost + } else { + value.isPriced = false + } + destination[key] = value + } + } + + private static func rows(_ values: [RowKey: RowValue]) -> [CodexLineageLedger.DailyRow] { + values.map { key, value in + .init(day: key.day, model: key.model, totals: value.totals, costUSD: value.isPriced ? value.costUSD : nil) + }.sorted { ($0.day, $0.model) < ($1.day, $1.model) } + } + + private static func inputFingerprint( + documents: [CodexLineageLedger.Document], + unresolvedParents: Set, + checkCancellation: CostUsageScanner.CancellationCheck?) throws -> Fingerprint + { + var digest = DigestBuilder() + digest.append("codex-lineage-family") + digest.append(String(Self.algorithmVersion)) + for document in documents { + try checkCancellation?() + digest.append(contentsOf: [ + "document", document.scopeID, Self.canonical(document.ownerID), + document.metadataSessionID.map(Self.canonical) ?? "", + document.parentSessionID.map(Self.canonical) ?? "", + String(document.incompleteObservationCount), + ]) + let observations = document.observations.sorted(by: Self.observationComesBefore) + for observation in observations { + try checkCancellation?() + digest.append(contentsOf: [ + "observation", observation.timestamp, observation.model, + String(observation.last.input), String(observation.last.cached), String(observation.last.output), + String(observation.total.input), String(observation.total.cached), String(observation.total.output), + ]) + } + } + for parent in unresolvedParents.sorted(by: { ($0.scopeID, $0.sessionID) < ($1.scopeID, $1.sessionID) }) { + digest.append(contentsOf: ["unresolved", parent.scopeID, Self.canonical(parent.sessionID)]) + } + return digest.finalize() + } + + private static func familyFingerprint( + input: Fingerprint, + quality: CodexLineageLedger.FamilyQuality) -> Fingerprint + { + let qualityFields = switch quality { + case .primary: + ["primary"] + case .incompleteProvenance: + ["incompleteProvenance"] + case let .contained(reasons): + ["contained"] + reasons.map(\.rawValue).sorted() + } + return Self.digest(["codex-lineage-family-result", input.value] + qualityFields) + } + + private static func cacheFingerprint(input: Fingerprint, localTimeZone: TimeZone) -> Fingerprint { + self.digest([ + "codex-lineage-family-projection", + input.value, + localTimeZone.identifier, + ]) + } + + private static func digest(_ fields: [String]) -> Fingerprint { + var digest = DigestBuilder() + for field in fields { + digest.append(field) + } + return digest.finalize() + } + + private static func documentFingerprint( + _ document: CodexLineageLedger.Document, + checkCancellation: CostUsageScanner.CancellationCheck?) throws -> Fingerprint + { + let observations = document.observations.sorted(by: Self.observationComesBefore) + var digest = DigestBuilder() + digest.append(contentsOf: [ + document.scopeID, Self.canonical(document.ownerID), + document.metadataSessionID.map(Self.canonical) ?? "", + document.parentSessionID.map(Self.canonical) ?? "", + String(document.incompleteObservationCount), + ]) + for observation in observations { + try checkCancellation?() + digest.append(contentsOf: [ + observation.timestamp, observation.model, + String(observation.last.input), String(observation.last.cached), String(observation.last.output), + String(observation.total.input), String(observation.total.cached), String(observation.total.output), + ]) + } + return digest.finalize() + } + + private static func observationComesBefore( + _ lhs: CodexLineageLedger.Observation, + _ rhs: CodexLineageLedger.Observation) -> Bool + { + self.observationKey(lhs) < self.observationKey(rhs) + } + + private static func observationKey(_ observation: CodexLineageLedger.Observation) -> String { + [ + observation.timestamp, observation.model, + String(observation.last.input), String(observation.last.cached), String(observation.last.output), + String(observation.total.input), String(observation.total.cached), String(observation.total.output), + ].joined(separator: "\u{0}") + } + + private static func nonEmpty(_ value: String?) -> String? { + guard let value, !value.isEmpty else { return nil } + return value + } + + private static func canonical(_ value: String) -> String { + UUID(uuidString: value)?.uuidString.lowercased() ?? value + } + + private static func scoped(_ value: String, scopeID: String) -> String { + scopeID + "\u{0}" + self.canonical(value) + } + + private struct DisjointSet { + var parents: [String: String] = [:] + + mutating func insert(_ value: String) { + self.parents[value] = self.parents[value] ?? value + } + + mutating func find(_ value: String) -> String { + self.insert(value) + guard let parent = self.parents[value], parent != value else { return value } + let root = self.find(parent) + self.parents[value] = root + return root + } + + mutating func union(_ lhs: String, _ rhs: String) { + let lhsRoot = self.find(lhs) + let rhsRoot = self.find(rhs) + guard lhsRoot != rhsRoot else { return } + let first = min(lhsRoot, rhsRoot) + let second = max(lhsRoot, rhsRoot) + self.parents[second] = first + } + } + + private struct DigestBuilder { + private var hasher = SHA256() + + mutating func append(_ field: String) { + var length = UInt64(field.utf8.count).bigEndian + withUnsafeBytes(of: &length) { self.hasher.update(bufferPointer: $0) } + self.hasher.update(data: Data(field.utf8)) + } + + mutating func append(contentsOf fields: [String]) { + for field in fields { + self.append(field) + } + } + + consuming func finalize() -> Fingerprint { + Fingerprint(value: self.hasher.finalize().map { String(format: "%02x", $0) }.joined()) + } + } +} diff --git a/Tests/CodexBarTests/CodexLineageEngineTests.swift b/Tests/CodexBarTests/CodexLineageEngineTests.swift new file mode 100644 index 0000000000..2ad505de2e --- /dev/null +++ b/Tests/CodexBarTests/CodexLineageEngineTests.swift @@ -0,0 +1,208 @@ +import Foundation +import Testing +@testable import CodexBarCore + +struct CodexLineageEngineTests { + @Test + func `family fingerprints and totals are deterministic under input permutation`() throws { + let first = Self.document(owner: "first", observations: [ + Self.observation(timestamp: "2026-07-09T12:01:00Z", input: 20, total: 30), + Self.observation(timestamp: "2026-07-09T12:00:00Z", input: 10, total: 10), + ]) + let child = Self.document(owner: "child", parent: "first", observations: [ + Self.observation(timestamp: "2026-07-09T12:02:00Z", input: 5, total: 35), + ]) + let other = Self.document(owner: "other", observations: [ + Self.observation(timestamp: "2026-07-09T13:00:00Z", input: 7, total: 7), + ]) + + let forwardFamilies = try CodexLineageEngine.prepareFamilies(documents: [first, child, other]) + let reverseFamilies = try CodexLineageEngine.prepareFamilies(documents: [other, child, first]) + let forward = try CodexLineageEngine.reconcile(families: forwardFamilies, localTimeZone: .gmt) + let reverse = try CodexLineageEngine.reconcile(families: reverseFamilies, localTimeZone: .gmt) + + #expect(forward.families.map(\.familyFingerprint) == reverse.families.map(\.familyFingerprint)) + #expect(forward.report == reverse.report) + #expect(forward.report.utcDays["2026-07-09"]?.input == 42) + } + + @Test + func `warm reconciliation reuses unchanged families and recomputes only a changed family`() throws { + let first = Self.document(owner: "first", observations: [Self.observation(input: 10, total: 10)]) + let second = Self.document(owner: "second", observations: [Self.observation(input: 20, total: 20)]) + let initialFamilies = try CodexLineageEngine.prepareFamilies(documents: [first, second]) + let initial = try CodexLineageEngine.reconcile(families: initialFamilies, localTimeZone: .gmt) + let warm = try CodexLineageEngine.reconcile( + families: initialFamilies, + previousCache: initial.candidateCache, + localTimeZone: .gmt) + + let changed = Self.document(owner: "second", observations: [Self.observation(input: 25, total: 25)]) + let changedFamilies = try CodexLineageEngine.prepareFamilies(documents: [first, changed]) + let updated = try CodexLineageEngine.reconcile( + families: changedFamilies, + previousCache: initial.candidateCache, + localTimeZone: .gmt) + + #expect(warm.diagnostics.reusedFamilyCount == 2) + #expect(warm.diagnostics.recomputedFamilyCount == 0) + #expect(updated.diagnostics.reusedFamilyCount == 1) + #expect(updated.diagnostics.recomputedFamilyCount == 1) + #expect(updated.report.utcDays["2026-07-09"]?.input == 35) + } + + @Test + func `cache entries with mismatched family identity are recomputed`() throws { + let first = Self.document(owner: "first", observations: [Self.observation(input: 10, total: 10)]) + let second = Self.document(owner: "second", observations: [Self.observation(input: 20, total: 20)]) + let families = try CodexLineageEngine.prepareFamilies(documents: [first, second]) + let initial = try CodexLineageEngine.reconcile(families: families, localTimeZone: .gmt) + let entries = initial.candidateCache.familiesByInputFingerprint.sorted { $0.key.value < $1.key.value } + let mismatched = CodexLineageEngine.Cache( + algorithmVersion: initial.candidateCache.algorithmVersion, + familiesByInputFingerprint: [entries[0].key: entries[1].value, entries[1].key: entries[0].value]) + + let rebuilt = try CodexLineageEngine.reconcile( + families: families, + previousCache: mismatched, + localTimeZone: .gmt) + + #expect(rebuilt.diagnostics.reusedFamilyCount == 0) + #expect(rebuilt.diagnostics.recomputedFamilyCount == 2) + #expect(rebuilt.report.utcDays["2026-07-09"]?.input == 30) + } + + @Test + func `changing projection timezone recomputes families and does not reuse stale local days`() throws { + let document = Self.document(owner: "first", observations: [ + Self.observation(timestamp: "2026-07-10T02:00:00Z", input: 10, total: 10), + ]) + let families = try CodexLineageEngine.prepareFamilies(documents: [document]) + let utc = try CodexLineageEngine.reconcile(families: families, localTimeZone: .gmt) + let newYork = try CodexLineageEngine.reconcile( + families: families, + previousCache: utc.candidateCache, + localTimeZone: #require(TimeZone(identifier: "America/New_York"))) + + #expect(newYork.diagnostics.reusedFamilyCount == 0) + #expect(newYork.diagnostics.recomputedFamilyCount == 1) + #expect(utc.report.localDays["2026-07-10"]?.input == 10) + #expect(newYork.report.localDays["2026-07-09"]?.input == 10) + } + + @Test + func `new parent merges only affected families while unrelated family remains reusable`() throws { + let child = Self.document( + owner: "child", + parent: "parent", + observations: [Self.observation(input: 10, total: 10)]) + let parent = Self.document(owner: "parent", observations: [Self.observation(input: 20, total: 20)]) + let unrelated = Self.document(owner: "other", observations: [Self.observation(input: 5, total: 5)]) + let unresolved: Set = [.init(sessionID: "parent")] + let initialFamilies = try CodexLineageEngine.prepareFamilies( + documents: [child, unrelated], + unresolvedParents: unresolved) + let initial = try CodexLineageEngine.reconcile(families: initialFamilies, localTimeZone: .gmt) + + let resolvedFamilies = try CodexLineageEngine.prepareFamilies(documents: [child, parent, unrelated]) + let resolved = try CodexLineageEngine.reconcile( + families: resolvedFamilies, + previousCache: initial.candidateCache, + localTimeZone: .gmt) + + #expect(resolved.diagnostics.familyCount == 2) + #expect(resolved.diagnostics.reusedFamilyCount == 1) + #expect(resolved.diagnostics.recomputedFamilyCount == 1) + #expect(resolved.report.utcDays["2026-07-09"]?.input == 35) + } + + @Test + func `cancelled publication leaves the prior cache unchanged`() throws { + let family = try CodexLineageEngine.prepareFamilies(documents: [ + Self.document(owner: "first", observations: [Self.observation(input: 10, total: 10)]), + ]) + let result = try CodexLineageEngine.reconcile(families: family, localTimeZone: .gmt) + var published = CodexLineageEngine.Cache.empty + + #expect(throws: CancellationError.self) { + try CodexLineageEngine.publish( + result.candidateCache, + to: &published, + checkCancellation: { throw CancellationError() }) + } + #expect(published == .empty) + + try CodexLineageEngine.publish(result.candidateCache, to: &published) + #expect(published == result.candidateCache) + } + + @Test + func `structural diagnostics bound reconciliation scratch state to the largest family`() throws { + let repeated = (0..<2000).map { index in + Self.observation( + timestamp: String(format: "2026-07-09T12:%02d:%02dZ", (index / 60) % 60, index % 60), + input: 1, + total: index + 1) + } + let documents = [ + Self.document(owner: "large", observations: repeated), + Self.document(owner: "small", observations: Array(repeated.prefix(100))), + ] + let families = try CodexLineageEngine.prepareFamilies(documents: documents) + let result = try CodexLineageEngine.reconcile(families: families, localTimeZone: .gmt) + + #expect(result.diagnostics.observationCount == 2100) + #expect(result.diagnostics.peakFamilyObservationCount == 2000) + #expect(result.diagnostics.peakAcceptedFingerprintCount <= 2000) + } + + @Test + func `cancellation propagates during preparation and reconciliation without a candidate`() throws { + let documents = (0..<20).map { index in + Self.document(owner: "owner-\(index)", observations: [Self.observation(input: index + 1, total: index + 1)]) + } + var preparationChecks = 0 + #expect(throws: CancellationError.self) { + _ = try CodexLineageEngine.prepareFamilies(documents: documents) { + preparationChecks += 1 + if preparationChecks == 5 { + throw CancellationError() + } + } + } + + let families = try CodexLineageEngine.prepareFamilies(documents: documents) + var reconciliationChecks = 0 + #expect(throws: CancellationError.self) { + _ = try CodexLineageEngine.reconcile(families: families, localTimeZone: .gmt) { + reconciliationChecks += 1 + if reconciliationChecks == 5 { + throw CancellationError() + } + } + } + } + + private static func document( + owner: String, + parent: String? = nil, + observations: [CodexLineageLedger.Observation]) -> CodexLineageLedger.Document + { + .init( + ownerID: owner, + metadataSessionID: owner, + parentSessionID: parent, + observations: observations) + } + + private static func observation( + timestamp: String = "2026-07-09T12:00:00Z", + input: Int, + total: Int) -> CodexLineageLedger.Observation + { + .init( + timestamp: timestamp, + last: .init(input: input, cached: 0, output: 0), + total: .init(input: total, cached: 0, output: 0)) + } +}