diff --git a/Scripts/regenerate-codex-parser-hash.sh b/Scripts/regenerate-codex-parser-hash.sh index 91e4863e73..4516f8e89a 100755 --- a/Scripts/regenerate-codex-parser-hash.sh +++ b/Scripts/regenerate-codex-parser-hash.sh @@ -32,10 +32,12 @@ fi FILE_LIST="$(mktemp)" trap 'rm -f "$FILE_LIST"' EXIT +# Storage-only files have their own schema-version gate and do not change parser semantics. find "$SOURCE_DIR" \ -type f \ -name '*.swift' \ ! -name '*Claude*' \ + ! -name 'CostUsageStore*.swift' \ -print | sed "s#^${ROOT_DIR}/##" | LC_ALL=C sort >"$FILE_LIST" diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Reads.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Reads.swift new file mode 100644 index 0000000000..712b11185e --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Reads.swift @@ -0,0 +1,518 @@ +import Foundation + +#if canImport(SQLite3) +import SQLite3 +#elseif canImport(CSQLite3) +import CSQLite3 +#endif + +// MARK: - Typed reads + +extension CostUsageStore { + func fetchFile(path: String) -> CostUsageStoreFile? { + self.withDatabase(default: nil) { database in + let statement = try Self.prepare(database, Self.fileSelectSQL + " WHERE path = ?") + defer { sqlite3_finalize(statement) } + Self.bind(path, to: statement, at: 1) + guard sqlite3_step(statement) == SQLITE_ROW else { return nil } + return try Self.decodeFile(statement) + } + } + + func fetchTokenSnapshots(path: String) -> [CostUsageStoreTokenSnapshot] { + self.withDatabase(default: []) { database in + try Self.readTokenSnapshots(database, path: path) + } + } + + func fetchDayAggregates(sinceDay: String, untilDay: String) -> [CostUsageStoreDayAggregate] { + guard sinceDay <= untilDay else { return [] } + return self.withDatabase(default: []) { database in + try Self.readDayAggregates(database, sinceDay: sinceDay, untilDay: untilDay) + } + } + + func fetchFileDayAggregates(path: String) -> [CostUsageStoreDayAggregate] { + self.withDatabase(default: []) { database in + try Self.readFileDayAggregates(database, path: path).map(\.aggregate) + } + } + + func fetchForkLineage(path: String) -> CostUsageStoreForkLineage? { + self.withDatabase(default: nil) { database in + let values = try Self.readForkLineage(database, path: path) + return values.first + } + } + + func fetchBufferedLines( + path: String, + kind: CostUsageStoreBufferedLineKind? = nil) -> [CostUsageStoreBufferedLine] + { + self.withDatabase(default: []) { database in + try Self.readBufferedLines(database, path: path, kind: kind) + } + } + + func fetchDiscoveryState() -> CostUsageStoreDiscoveryState? { + self.readSingleton(CostUsageStoreDiscoveryState.self, table: "discovery_state") + } + + func fetchLookbackState() -> CostUsageStoreLookbackState? { + self.readSingleton(CostUsageStoreLookbackState.self, table: "lookback_state") + } + + func fetchMetadata() -> CostUsageStoreMetadata { + self.readSingleton(CostUsageStoreMetadata.self, table: "scan_metadata") ?? .empty + } + + func fetchAccumulator(path: String) -> CostUsageStoreAccumulator? { + self.withDatabase(default: nil) { database in + try Self.readAccumulators(database, path: path).first + } + } + + func readReport(sinceDay: String, untilDay: String) -> CostUsageStoreReport { + guard sinceDay <= untilDay else { + return CostUsageStoreReport(metadata: .empty, aggregates: []) + } + return self.withDatabase(default: CostUsageStoreReport(metadata: .empty, aggregates: [])) { database in + try Self.inReadTransaction(database) { + let metadata = try Self.readSingleton( + CostUsageStoreMetadata.self, + database: database, + table: "scan_metadata") ?? .empty + let aggregates = try Self.readDayAggregates( + database, + sinceDay: sinceDay, + untilDay: untilDay) + return CostUsageStoreReport(metadata: metadata, aggregates: aggregates) + } + } + } + + func readSnapshot() -> CostUsageStoreSnapshot { + let empty = CostUsageStoreSnapshot( + metadata: .empty, + files: [], + tokenSnapshots: [], + fileDayAggregates: [], + dayAggregates: [], + forkLineage: [], + bufferedLines: [], + discoveryState: nil, + lookbackState: nil, + accumulators: []) + return self.withDatabase(default: empty) { database in + try Self.inReadTransaction(database) { + try CostUsageStoreSnapshot( + metadata: Self.readSingleton( + CostUsageStoreMetadata.self, + database: database, + table: "scan_metadata") ?? .empty, + files: Self.readFiles(database), + tokenSnapshots: Self.readTokenSnapshots(database, path: nil), + fileDayAggregates: Self.readFileDayAggregates(database, path: nil), + dayAggregates: Self.readDayAggregates(database, sinceDay: nil, untilDay: nil), + forkLineage: Self.readForkLineage(database, path: nil), + bufferedLines: Self.readBufferedLines(database, path: nil, kind: nil), + discoveryState: Self.readSingleton( + CostUsageStoreDiscoveryState.self, + database: database, + table: "discovery_state"), + lookbackState: Self.readSingleton( + CostUsageStoreLookbackState.self, + database: database, + table: "lookback_state"), + accumulators: Self.readAccumulators(database, path: nil)) + } + } + } + + func configuration() -> CostUsageStoreConfiguration? { + self.withDatabase(default: nil) { database in + try CostUsageStoreConfiguration( + journalMode: Self.scalarText(database, "PRAGMA journal_mode") ?? "", + busyTimeoutMilliseconds: Int(Self.scalarInt(database, "PRAGMA busy_timeout")), + foreignKeysEnabled: Self.scalarInt(database, "PRAGMA foreign_keys") == 1, + autoVacuumMode: Int(Self.scalarInt(database, "PRAGMA auto_vacuum")), + userVersion: Int(Self.scalarInt(database, "PRAGMA user_version"))) + } + } + + private func readSingleton(_ type: Value.Type, table: String) -> Value? { + self.withDatabase(default: nil) { database in + try Self.readSingleton(type, database: database, table: table) + } + } +} + +// MARK: - Read implementations + +extension CostUsageStore { + private static let fileSelectSQL = """ + SELECT path, inode, mtime_ms, size, parsed_bytes, anchor_indexed_bytes, + anchor_window_start, anchor_sha256, scan_state, scan_target_size, + scan_complete, session_id, coverage_since_day, coverage_until_day, updated_at_ms + FROM files + """ + + static func readFiles(_ database: OpaquePointer) throws -> [CostUsageStoreFile] { + let statement = try self.prepare(database, self.fileSelectSQL + " ORDER BY path") + defer { sqlite3_finalize(statement) } + var values: [CostUsageStoreFile] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + try values.append(self.decodeFile(statement)) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + static func decodeFile(_ statement: OpaquePointer) throws -> CostUsageStoreFile { + guard let path = self.columnText(statement, at: 0), + let stateData = self.columnData(statement, at: 8) + else { throw StoreError.invalidData } + var state = try JSONDecoder().decode(CostUsageStoreScanState.self, from: stateData) + state.targetSize = self.columnInt64(statement, at: 9) + state.isComplete = sqlite3_column_int(statement, 10) == 1 + let anchor: CostUsageStoreValidationAnchor? = if let sha = self.columnText(statement, at: 7), + let indexedBytes = self.columnInt64(statement, at: 5), + let windowStart = self.columnInt64(statement, at: 6) + { + .init(indexedBytes: indexedBytes, windowStart: windowStart, sha256: sha) + } else { + nil + } + return CostUsageStoreFile( + path: path, + inode: self.columnInt64(statement, at: 1), + mtimeUnixMs: sqlite3_column_int64(statement, 2), + size: sqlite3_column_int64(statement, 3), + parsedBytes: self.columnInt64(statement, at: 4), + anchor: anchor, + scanState: state, + sessionID: self.columnText(statement, at: 11), + coverageSinceDay: self.columnText(statement, at: 12), + coverageUntilDay: self.columnText(statement, at: 13), + updatedAtUnixMs: sqlite3_column_int64(statement, 14)) + } + + static func readTokenSnapshots( + _ database: OpaquePointer, + path: String?) throws -> [CostUsageStoreTokenSnapshot] + { + var sql = """ + SELECT f.path, t.event_index, t.timestamp, t.timestamp_ms, t.day, + t.last_input, t.last_cached, t.last_output, t.last_reasoning, + t.total_input, t.total_cached, t.total_output, t.total_reasoning, t.end_offset + FROM token_snapshots t JOIN files f ON f.id = t.file_id + """ + if path != nil { + sql += " WHERE f.path = ?" + } + sql += " ORDER BY f.path, t.event_index" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + if let path { + self.bind(path, to: statement, at: 1) + } + var values: [CostUsageStoreTokenSnapshot] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0), + let eventIndex = Int(exactly: sqlite3_column_int64(statement, 1)), + let timestamp = self.columnText(statement, at: 2) + else { throw StoreError.invalidData } + values.append(CostUsageStoreTokenSnapshot( + path: path, + eventIndex: eventIndex, + timestamp: timestamp, + timestampUnixMs: self.columnInt64(statement, at: 3), + day: self.columnText(statement, at: 4), + last: self.decodeTotals(statement, startingAt: 5), + total: self.decodeTotals(statement, startingAt: 9), + endOffset: sqlite3_column_int64(statement, 13))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + static func readDayAggregates( + _ database: OpaquePointer, + sinceDay: String?, + untilDay: String?) throws -> [CostUsageStoreDayAggregate] + { + var sql = """ + SELECT day, model, input_tokens, cached_tokens, output_tokens, reasoning_tokens, + request_count, known_cost_nanos, unpriced_tokens, standard_cost_nanos, + priority_cost_nanos, standard_tokens, priority_tokens + FROM day_aggregates + """ + if sinceDay != nil, untilDay != nil { + sql += " WHERE day >= ? AND day <= ?" + } + sql += " ORDER BY day, model" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + if let sinceDay, let untilDay { + self.bind(sinceDay, to: statement, at: 1) + self.bind(untilDay, to: statement, at: 2) + } + var values: [CostUsageStoreDayAggregate] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let day = self.columnText(statement, at: 0), + let model = self.columnText(statement, at: 1) + else { throw StoreError.invalidData } + values.append(CostUsageStoreDayAggregate( + day: day, + model: model, + inputTokens: sqlite3_column_int64(statement, 2), + cachedTokens: sqlite3_column_int64(statement, 3), + outputTokens: sqlite3_column_int64(statement, 4), + reasoningTokens: sqlite3_column_int64(statement, 5), + requestCount: sqlite3_column_int64(statement, 6), + knownCostNanos: sqlite3_column_int64(statement, 7), + unpricedTokens: sqlite3_column_int64(statement, 8), + standardCostNanos: sqlite3_column_int64(statement, 9), + priorityCostNanos: sqlite3_column_int64(statement, 10), + standardTokens: sqlite3_column_int64(statement, 11), + priorityTokens: sqlite3_column_int64(statement, 12))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + static func readFileDayAggregates( + _ database: OpaquePointer, + path: String?) throws -> [CostUsageStoreFileDayAggregate] + { + var sql = """ + SELECT f.path, a.day, a.model, a.input_tokens, a.cached_tokens, a.output_tokens, + a.reasoning_tokens, a.request_count, a.known_cost_nanos, a.unpriced_tokens, + a.standard_cost_nanos, a.priority_cost_nanos, a.standard_tokens, a.priority_tokens + FROM file_day_aggregates a JOIN files f ON f.id = a.file_id + """ + if path != nil { + sql += " WHERE f.path = ?" + } + sql += " ORDER BY f.path, a.day, a.model" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + if let path { + self.bind(path, to: statement, at: 1) + } + var values: [CostUsageStoreFileDayAggregate] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0), + let day = self.columnText(statement, at: 1), + let model = self.columnText(statement, at: 2) + else { throw StoreError.invalidData } + values.append(CostUsageStoreFileDayAggregate( + path: path, + aggregate: CostUsageStoreDayAggregate( + day: day, + model: model, + inputTokens: sqlite3_column_int64(statement, 3), + cachedTokens: sqlite3_column_int64(statement, 4), + outputTokens: sqlite3_column_int64(statement, 5), + reasoningTokens: sqlite3_column_int64(statement, 6), + requestCount: sqlite3_column_int64(statement, 7), + knownCostNanos: sqlite3_column_int64(statement, 8), + unpricedTokens: sqlite3_column_int64(statement, 9), + standardCostNanos: sqlite3_column_int64(statement, 10), + priorityCostNanos: sqlite3_column_int64(statement, 11), + standardTokens: sqlite3_column_int64(statement, 12), + priorityTokens: sqlite3_column_int64(statement, 13)))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } +} + +extension CostUsageStore { + static func readForkLineage( + _ database: OpaquePointer, + path: String?) throws -> [CostUsageStoreForkLineage] + { + var sql = """ + SELECT f.path, l.session_id, l.forked_from_id, l.fork_timestamp, + l.dependency_key, l.subagent_state, l.accounting_state + FROM fork_lineage l JOIN files f ON f.id = l.file_id + """ + if path != nil { + sql += " WHERE f.path = ?" + } + sql += " ORDER BY f.path" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + if let path { + self.bind(path, to: statement, at: 1) + } + var values: [CostUsageStoreForkLineage] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0) else { throw StoreError.invalidData } + values.append(CostUsageStoreForkLineage( + path: path, + sessionID: self.columnText(statement, at: 1), + forkedFromID: self.columnText(statement, at: 2), + forkTimestamp: self.columnText(statement, at: 3), + dependencyKey: self.columnText(statement, at: 4), + subagentState: self.columnData(statement, at: 5), + accountingState: self.columnData(statement, at: 6))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + static func readBufferedLines( + _ database: OpaquePointer, + path: String?, + kind: CostUsageStoreBufferedLineKind?) throws -> [CostUsageStoreBufferedLine] + { + var sql = """ + SELECT f.path, b.kind, b.line_index, b.ordinal, b.end_offset, b.payload + FROM buffered_lines b JOIN files f ON f.id = b.file_id + """ + var clauses: [String] = [] + if path != nil { + clauses.append("f.path = ?") + } + if kind != nil { + clauses.append("b.kind = ?") + } + if !clauses.isEmpty { + sql += " WHERE " + clauses.joined(separator: " AND ") + } + sql += " ORDER BY f.path, b.kind, b.line_index" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + var index: Int32 = 1 + if let path { + self.bind(path, to: statement, at: index) + index += 1 + } + if let kind { + self.bind(kind.rawValue, to: statement, at: index) + } + var values: [CostUsageStoreBufferedLine] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0), + let rawKind = self.columnText(statement, at: 1), + let kind = CostUsageStoreBufferedLineKind(rawValue: rawKind), + let lineIndex = Int(exactly: sqlite3_column_int64(statement, 2)), + let payload = self.columnData(statement, at: 5) + else { throw StoreError.invalidData } + values.append(CostUsageStoreBufferedLine( + path: path, + kind: kind, + lineIndex: lineIndex, + ordinal: self.columnInt64(statement, at: 3).flatMap(Int.init(exactly:)), + endOffset: self.columnInt64(statement, at: 4), + payload: payload)) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + static func readAccumulators( + _ database: OpaquePointer, + path: String?) throws -> [CostUsageStoreAccumulator] + { + var sql = """ + SELECT f.path, a.event_count, a.next_usage_row_index, + a.counted_input, a.counted_cached, a.counted_output, a.counted_reasoning, + a.baseline_input, a.baseline_cached, a.baseline_output, a.baseline_reasoning, + a.watermark_input, a.watermark_cached, a.watermark_output, a.watermark_reasoning, + a.saw_divergent, a.saw_interleaved, a.seen_raw_totals, a.updated_at_ms + FROM accumulators a JOIN files f ON f.id = a.file_id + """ + if path != nil { + sql += " WHERE f.path = ?" + } + sql += " ORDER BY f.path" + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + if let path { + self.bind(path, to: statement, at: 1) + } + var values: [CostUsageStoreAccumulator] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0), + let eventCount = Int(exactly: sqlite3_column_int64(statement, 1)), + let seenData = self.columnData(statement, at: 17) + else { throw StoreError.invalidData } + let seen = try JSONDecoder().decode([CostUsageStoreTotals].self, from: seenData) + values.append(CostUsageStoreAccumulator( + path: path, + eventCount: eventCount, + nextUsageRowIndex: self.columnInt64(statement, at: 2).flatMap(Int.init(exactly:)), + countedTotals: self.decodeTotals(statement, startingAt: 3), + rawTotalsBaseline: self.decodeTotals(statement, startingAt: 7), + rawTotalsWatermark: self.decodeTotals(statement, startingAt: 11), + sawDivergentTotals: sqlite3_column_int(statement, 15) == 1, + sawInterleavedTotals: sqlite3_column_int(statement, 16) == 1, + seenRawTotals: seen, + updatedAtUnixMs: sqlite3_column_int64(statement, 18))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } +} + +// MARK: - Read helpers + +extension CostUsageStore { + static func readSingleton( + _ type: Value.Type, + database: OpaquePointer, + table: String) throws -> Value? + { + let statement = try self.prepare(database, "SELECT payload FROM \(table) WHERE id = 1") + defer { sqlite3_finalize(statement) } + let result = sqlite3_step(statement) + if result == SQLITE_DONE { + return nil + } + guard result == SQLITE_ROW, let payload = self.columnData(statement, at: 0) else { + throw StoreError.sqlite(result) + } + return try JSONDecoder().decode(type, from: payload) + } + + static func decodeTotals(_ statement: OpaquePointer, startingAt index: Int32) -> CostUsageStoreTotals? { + guard let input = self.columnInt64(statement, at: index), + let cached = self.columnInt64(statement, at: index + 1), + let output = self.columnInt64(statement, at: index + 2) + else { return nil } + return CostUsageStoreTotals( + input: input, + cached: cached, + output: output, + reasoning: self.columnInt64(statement, at: index + 3)) + } + + static func inReadTransaction(_ database: OpaquePointer, _ operation: () throws -> T) throws -> T { + try self.execute(database, "BEGIN") + do { + let value = try operation() + try self.execute(database, "COMMIT") + return value + } catch { + try? self.execute(database, "ROLLBACK") + throw error + } + } +} diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Retention.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Retention.swift new file mode 100644 index 0000000000..3f1fcd11e7 --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Retention.swift @@ -0,0 +1,279 @@ +import Foundation + +#if canImport(SQLite3) +import SQLite3 +#elseif canImport(CSQLite3) +import CSQLite3 +#endif + +// MARK: - Retention + +extension CostUsageStore { + @discardableResult + func retainDayWindow(sinceDay: String, untilDay: String) -> CostUsageStoreRetentionResult { + guard sinceDay <= untilDay else { + return CostUsageStoreRetentionResult( + deletedFiles: 0, + deletedTokenSnapshots: 0, + deletedFileDayAggregates: 0, + deletedDayAggregates: 0) + } + let fallback = CostUsageStoreRetentionResult( + deletedFiles: 0, + deletedTokenSnapshots: 0, + deletedFileDayAggregates: 0, + deletedDayAggregates: 0) + return self.withDatabase(default: fallback) { database in + try Self.prune(database, sinceDay: sinceDay, untilDay: untilDay) + } + } + + @discardableResult + func deleteFile(path: String) -> Bool { + self.withDatabase(default: false) { database in + let statement = try Self.prepare(database, "DELETE FROM files WHERE path = ?") + defer { sqlite3_finalize(statement) } + Self.bind(path, to: statement, at: 1) + try Self.stepDone(statement, database: database) + return sqlite3_changes(database) > 0 + } + } + + private static func prune( + _ database: OpaquePointer, + sinceDay: String, + untilDay: String) throws -> CostUsageStoreRetentionResult + { + try self.inTransaction(database) { + let beforeFiles = try self.scalarInt(database, "SELECT COUNT(*) FROM files") + let beforeSnapshots = try self.scalarInt(database, "SELECT COUNT(*) FROM token_snapshots") + let beforeFileAggregates = try self.scalarInt(database, "SELECT COUNT(*) FROM file_day_aggregates") + let beforeAggregates = try self.scalarInt(database, "SELECT COUNT(*) FROM day_aggregates") + + let candidates = try self.retentionCandidates( + database, + sinceDay: sinceDay, + untilDay: untilDay) + let deleteFile = try self.prepare(database, "DELETE FROM files WHERE path = ?") + defer { sqlite3_finalize(deleteFile) } + for candidate in candidates { + sqlite3_reset(deleteFile) + sqlite3_clear_bindings(deleteFile) + self.bind(candidate.path, to: deleteFile, at: 1) + try self.stepDone(deleteFile, database: database) + } + + let deleteSnapshots = try self.prepare(database, """ + DELETE FROM token_snapshots WHERE day IS NOT NULL AND (day < ? OR day > ?) + """) + defer { sqlite3_finalize(deleteSnapshots) } + self.bind(sinceDay, to: deleteSnapshots, at: 1) + self.bind(untilDay, to: deleteSnapshots, at: 2) + try self.stepDone(deleteSnapshots, database: database) + + let deleteFileAggregates = try self.prepare( + database, + "DELETE FROM file_day_aggregates WHERE day < ? OR day > ?") + defer { sqlite3_finalize(deleteFileAggregates) } + self.bind(sinceDay, to: deleteFileAggregates, at: 1) + self.bind(untilDay, to: deleteFileAggregates, at: 2) + try self.stepDone(deleteFileAggregates, database: database) + + let deleteAggregates = try self.prepare( + database, + "DELETE FROM day_aggregates WHERE day < ? OR day > ?") + defer { sqlite3_finalize(deleteAggregates) } + self.bind(sinceDay, to: deleteAggregates, at: 1) + self.bind(untilDay, to: deleteAggregates, at: 2) + try self.stepDone(deleteAggregates, database: database) + + try self.pruneDiscovery(database, candidates: candidates) + var metadata = try self.readSingleton( + CostUsageStoreMetadata.self, + database: database, + table: "scan_metadata") ?? .empty + metadata.scanSinceDay = sinceDay + metadata.scanUntilDay = untilDay + try self.writeSingleton(metadata, database: database, table: "scan_metadata") + + let afterFiles = try self.scalarInt(database, "SELECT COUNT(*) FROM files") + let afterSnapshots = try self.scalarInt(database, "SELECT COUNT(*) FROM token_snapshots") + let afterFileAggregates = try self.scalarInt(database, "SELECT COUNT(*) FROM file_day_aggregates") + let afterAggregates = try self.scalarInt(database, "SELECT COUNT(*) FROM day_aggregates") + return CostUsageStoreRetentionResult( + deletedFiles: Int(beforeFiles - afterFiles), + deletedTokenSnapshots: Int(beforeSnapshots - afterSnapshots), + deletedFileDayAggregates: Int(beforeFileAggregates - afterFileAggregates), + deletedDayAggregates: Int(beforeAggregates - afterAggregates)) + } + } + + private struct RetentionCandidate { + var path: String + var sessionID: String? + } + + private static func retentionCandidates( + _ database: OpaquePointer, + sinceDay: String, + untilDay: String) throws -> [RetentionCandidate] + { + let statement = try self.prepare(database, """ + SELECT f.path, f.session_id + FROM files f + WHERE f.scan_complete = 1 + AND f.coverage_since_day IS NOT NULL + AND f.coverage_until_day IS NOT NULL + AND (f.coverage_until_day < ? OR f.coverage_since_day > ?) + AND NOT EXISTS (SELECT 1 FROM buffered_lines b WHERE b.file_id = f.id) + AND ( + f.session_id IS NULL OR NOT EXISTS ( + SELECT 1 FROM fork_lineage l + WHERE l.forked_from_id = f.session_id AND l.file_id != f.id + ) + ) + ORDER BY f.updated_at_ms, f.path + """) + defer { sqlite3_finalize(statement) } + self.bind(sinceDay, to: statement, at: 1) + self.bind(untilDay, to: statement, at: 2) + var values: [RetentionCandidate] = [] + var result = sqlite3_step(statement) + while result == SQLITE_ROW { + guard let path = self.columnText(statement, at: 0) else { throw StoreError.invalidData } + values.append(RetentionCandidate(path: path, sessionID: self.columnText(statement, at: 1))) + result = sqlite3_step(statement) + } + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + return values + } + + private static func pruneDiscovery( + _ database: OpaquePointer, + candidates: [RetentionCandidate]) throws + { + guard !candidates.isEmpty, + var state = try self.readSingleton( + CostUsageStoreDiscoveryState.self, + database: database, + table: "discovery_state") + else { return } + let paths = Set(candidates.map(\.path)) + let sessionIDs = Set(candidates.compactMap(\.sessionID)) + state.filePaths.removeAll(where: paths.contains) + state.pendingSessionIDs.removeAll(where: sessionIDs.contains) + state.missingSessionIDs.removeAll(where: sessionIDs.contains) + state.filePathBySessionID = state.filePathBySessionID.filter { + !sessionIDs.contains($0.key) && !paths.contains($0.value) + } + state.nextFileIndex = min(state.nextFileIndex, state.filePaths.count) + try self.writeSingleton(state, database: database, table: "discovery_state") + } +} + +// MARK: - Budgets and vacuum + +extension CostUsageStore { + func enforceBudgets(maxRows: Int, maxFileBytes: Int64) -> CostUsageStoreBudgetResult { + let fallback = CostUsageStoreBudgetResult(deletedRows: 0, rowCount: 0, fileBytes: 0) + return self.withDatabase(default: fallback) { database in + let initialRows = try Self.rowCount(database) + if let metadata = try Self.readSingleton( + CostUsageStoreMetadata.self, + database: database, + table: "scan_metadata"), + let sinceDay = metadata.scanSinceDay, + let untilDay = metadata.scanUntilDay, + sinceDay <= untilDay + { + _ = try Self.prune(database, sinceDay: sinceDay, untilDay: untilDay) + } + + let rowLimit = max(0, maxRows) + while try Self.rowCount(database) > Int64(rowLimit) { + guard try Self.deleteOldestRetainedRow(database) else { break } + } + try Self.reclaimFreePages(database) + + let byteLimit = max(0, maxFileBytes) + var fileBytes = Self.fileSize(at: self.databaseURL) + while fileBytes > byteLimit { + guard try Self.deleteOldestRetainedRow(database) else { break } + try Self.reclaimFreePages(database) + fileBytes = Self.fileSize(at: self.databaseURL) + } + let finalRows = try Self.rowCount(database) + return CostUsageStoreBudgetResult( + deletedRows: Int(max(0, initialRows - finalRows)), + rowCount: Int(finalRows), + fileBytes: fileBytes) + } + } + + func fileSizeBytes() -> Int64 { + self.withDatabase(default: 0) { database in + try Self.reclaimFreePages(database) + return Self.fileSize(at: self.databaseURL) + } + } + + private static func deleteOldestRetainedRow(_ database: OpaquePointer) throws -> Bool { + let statements = [ + "DELETE FROM files WHERE id = (SELECT id FROM files ORDER BY updated_at_ms, id LIMIT 1)", + """ + DELETE FROM day_aggregates WHERE rowid = ( + SELECT rowid FROM day_aggregates ORDER BY day, model LIMIT 1 + ) + """, + ] + for sql in statements { + try self.execute(database, sql) + if sqlite3_changes(database) > 0 { + return true + } + } + return false + } + + private static func rowCount(_ database: OpaquePointer) throws -> Int64 { + try self.scalarInt(database, """ + SELECT + (SELECT COUNT(*) FROM files) + + (SELECT COUNT(*) FROM token_snapshots) + + (SELECT COUNT(*) FROM file_day_aggregates) + + (SELECT COUNT(*) FROM day_aggregates) + + (SELECT COUNT(*) FROM fork_lineage) + + (SELECT COUNT(*) FROM buffered_lines) + + (SELECT COUNT(*) FROM accumulators) + """) + } + + private static func reclaimFreePages(_ database: OpaquePointer) throws { + try self.execute(database, "PRAGMA incremental_vacuum(1000000)") + try self.execute(database, "PRAGMA wal_checkpoint(TRUNCATE)") + } + + private static func fileSize(at url: URL) -> Int64 { + let attributes = try? FileManager.default.attributesOfItem(atPath: url.path) + return (attributes?[.size] as? NSNumber)?.int64Value ?? 0 + } +} + +// MARK: - Singleton write helper + +extension CostUsageStore { + private static func writeSingleton( + _ value: some Encodable, + database: OpaquePointer, + table: String) throws + { + let payload = try JSONEncoder().encode(value) + let statement = try self.prepare(database, """ + INSERT INTO \(table)(id, payload) VALUES (1, ?) + ON CONFLICT(id) DO UPDATE SET payload = excluded.payload + """) + defer { sqlite3_finalize(statement) } + self.bind(payload, to: statement, at: 1) + try self.stepDone(statement, database: database) + } +} diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Writes.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Writes.swift new file mode 100644 index 0000000000..bed2e362da --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore+Writes.swift @@ -0,0 +1,380 @@ +import Foundation + +#if canImport(SQLite3) +import SQLite3 +#elseif canImport(CSQLite3) +import CSQLite3 +#endif + +// MARK: - Files and token deltas + +extension CostUsageStore { + @discardableResult + func upsertFile(_ file: CostUsageStoreFile) -> Bool { + self.withDatabase(default: false) { database in + let state = try JSONEncoder().encode(file.scanState) + let statement = try Self.prepare(database, """ + INSERT INTO files ( + path, inode, mtime_ms, size, parsed_bytes, anchor_indexed_bytes, + anchor_window_start, anchor_sha256, scan_state, scan_target_size, + scan_complete, session_id, coverage_since_day, coverage_until_day, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + inode = excluded.inode, + mtime_ms = excluded.mtime_ms, + size = excluded.size, + parsed_bytes = excluded.parsed_bytes, + anchor_indexed_bytes = excluded.anchor_indexed_bytes, + anchor_window_start = excluded.anchor_window_start, + anchor_sha256 = excluded.anchor_sha256, + scan_state = excluded.scan_state, + scan_target_size = excluded.scan_target_size, + scan_complete = excluded.scan_complete, + session_id = excluded.session_id, + coverage_since_day = excluded.coverage_since_day, + coverage_until_day = excluded.coverage_until_day, + updated_at_ms = excluded.updated_at_ms + """) + defer { sqlite3_finalize(statement) } + Self.bind(file.path, to: statement, at: 1) + Self.bind(file.inode, to: statement, at: 2) + sqlite3_bind_int64(statement, 3, file.mtimeUnixMs) + sqlite3_bind_int64(statement, 4, file.size) + Self.bind(file.parsedBytes, to: statement, at: 5) + Self.bind(file.anchor?.indexedBytes, to: statement, at: 6) + Self.bind(file.anchor?.windowStart, to: statement, at: 7) + Self.bind(file.anchor?.sha256, to: statement, at: 8) + Self.bind(state, to: statement, at: 9) + Self.bind(file.scanState.targetSize, to: statement, at: 10) + sqlite3_bind_int(statement, 11, file.scanState.isComplete ? 1 : 0) + Self.bind(file.sessionID, to: statement, at: 12) + Self.bind(file.coverageSinceDay, to: statement, at: 13) + Self.bind(file.coverageUntilDay, to: statement, at: 14) + sqlite3_bind_int64(statement, 15, file.updatedAtUnixMs) + try Self.stepDone(statement, database: database) + return true + } + } + + @discardableResult + func appendTokenSnapshots(_ snapshots: [CostUsageStoreTokenSnapshot]) -> Bool { + guard !snapshots.isEmpty else { return true } + return self.withDatabase(default: false) { database in + try Self.inTransaction(database) { + let statement = try Self.prepare(database, """ + INSERT INTO token_snapshots ( + file_id, event_index, timestamp, timestamp_ms, day, + last_input, last_cached, last_output, last_reasoning, + total_input, total_cached, total_output, total_reasoning, end_offset + ) VALUES ((SELECT id FROM files WHERE path = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(file_id, event_index) DO UPDATE SET + timestamp = excluded.timestamp, + timestamp_ms = excluded.timestamp_ms, + day = excluded.day, + last_input = excluded.last_input, + last_cached = excluded.last_cached, + last_output = excluded.last_output, + last_reasoning = excluded.last_reasoning, + total_input = excluded.total_input, + total_cached = excluded.total_cached, + total_output = excluded.total_output, + total_reasoning = excluded.total_reasoning, + end_offset = excluded.end_offset + """) + defer { sqlite3_finalize(statement) } + for snapshot in snapshots { + sqlite3_reset(statement) + sqlite3_clear_bindings(statement) + Self.bind(snapshot.path, to: statement, at: 1) + sqlite3_bind_int64(statement, 2, Int64(snapshot.eventIndex)) + Self.bind(snapshot.timestamp, to: statement, at: 3) + Self.bind(snapshot.timestampUnixMs, to: statement, at: 4) + Self.bind(snapshot.day, to: statement, at: 5) + Self.bindTotals(snapshot.last, to: statement, startingAt: 6) + Self.bindTotals(snapshot.total, to: statement, startingAt: 10) + sqlite3_bind_int64(statement, 14, snapshot.endOffset) + try Self.stepDone(statement, database: database) + } + } + return true + } + } + + @discardableResult + func replaceFileDayAggregates( + path: String, + aggregates: [CostUsageStoreDayAggregate]) -> Bool + { + self.withDatabase(default: false) { database in + try Self.inTransaction(database) { + let delete = try Self.prepare(database, """ + DELETE FROM file_day_aggregates + WHERE file_id = (SELECT id FROM files WHERE path = ?) + """) + defer { sqlite3_finalize(delete) } + Self.bind(path, to: delete, at: 1) + try Self.stepDone(delete, database: database) + + let insert = try Self.prepare(database, """ + INSERT INTO file_day_aggregates ( + file_id, day, model, input_tokens, cached_tokens, output_tokens, + reasoning_tokens, request_count, known_cost_nanos, unpriced_tokens, + standard_cost_nanos, priority_cost_nanos, standard_tokens, priority_tokens + ) VALUES ((SELECT id FROM files WHERE path = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """) + defer { sqlite3_finalize(insert) } + for aggregate in aggregates { + sqlite3_reset(insert) + sqlite3_clear_bindings(insert) + Self.bind(path, to: insert, at: 1) + Self.bind(aggregate.day, to: insert, at: 2) + Self.bind(aggregate.model, to: insert, at: 3) + Self.bindAggregateValues(aggregate, to: insert, startingAt: 4) + try Self.stepDone(insert, database: database) + } + } + return true + } + } + + @discardableResult + func mergeDayAggregates(_ deltas: [CostUsageStoreDayAggregate]) -> Bool { + guard !deltas.isEmpty else { return true } + return self.withDatabase(default: false) { database in + try Self.inTransaction(database) { + let statement = try Self.prepare(database, """ + INSERT INTO day_aggregates ( + day, model, input_tokens, cached_tokens, output_tokens, reasoning_tokens, + request_count, known_cost_nanos, unpriced_tokens, standard_cost_nanos, + priority_cost_nanos, standard_tokens, priority_tokens + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(day, model) DO UPDATE SET + input_tokens = input_tokens + excluded.input_tokens, + cached_tokens = cached_tokens + excluded.cached_tokens, + output_tokens = output_tokens + excluded.output_tokens, + reasoning_tokens = reasoning_tokens + excluded.reasoning_tokens, + request_count = request_count + excluded.request_count, + known_cost_nanos = known_cost_nanos + excluded.known_cost_nanos, + unpriced_tokens = unpriced_tokens + excluded.unpriced_tokens, + standard_cost_nanos = standard_cost_nanos + excluded.standard_cost_nanos, + priority_cost_nanos = priority_cost_nanos + excluded.priority_cost_nanos, + standard_tokens = standard_tokens + excluded.standard_tokens, + priority_tokens = priority_tokens + excluded.priority_tokens + """) + defer { sqlite3_finalize(statement) } + for delta in deltas { + sqlite3_reset(statement) + sqlite3_clear_bindings(statement) + Self.bind(delta.day, to: statement, at: 1) + Self.bind(delta.model, to: statement, at: 2) + Self.bindAggregateValues(delta, to: statement, startingAt: 3) + try Self.stepDone(statement, database: database) + } + } + return true + } + } +} + +// MARK: - Lineage, buffers, discovery, and accumulators + +extension CostUsageStore { + @discardableResult + func upsertForkLineage(_ lineage: CostUsageStoreForkLineage) -> Bool { + self.withDatabase(default: false) { database in + let statement = try Self.prepare(database, """ + INSERT INTO fork_lineage ( + file_id, session_id, forked_from_id, fork_timestamp, dependency_key, + subagent_state, accounting_state + ) VALUES ((SELECT id FROM files WHERE path = ?), ?, ?, ?, ?, ?, ?) + ON CONFLICT(file_id) DO UPDATE SET + session_id = excluded.session_id, + forked_from_id = excluded.forked_from_id, + fork_timestamp = excluded.fork_timestamp, + dependency_key = excluded.dependency_key, + subagent_state = excluded.subagent_state, + accounting_state = excluded.accounting_state + """) + defer { sqlite3_finalize(statement) } + Self.bind(lineage.path, to: statement, at: 1) + Self.bind(lineage.sessionID, to: statement, at: 2) + Self.bind(lineage.forkedFromID, to: statement, at: 3) + Self.bind(lineage.forkTimestamp, to: statement, at: 4) + Self.bind(lineage.dependencyKey, to: statement, at: 5) + Self.bind(lineage.subagentState, to: statement, at: 6) + Self.bind(lineage.accountingState, to: statement, at: 7) + try Self.stepDone(statement, database: database) + return true + } + } + + @discardableResult + func replaceBufferedLines( + path: String, + kind: CostUsageStoreBufferedLineKind, + lines: [CostUsageStoreBufferedLine]) -> Bool + { + guard lines.allSatisfy({ $0.path == path && $0.kind == kind }) else { return false } + return self.withDatabase(default: false) { database in + try Self.inTransaction(database) { + let delete = try Self.prepare(database, """ + DELETE FROM buffered_lines + WHERE file_id = (SELECT id FROM files WHERE path = ?) AND kind = ? + """) + Self.bind(path, to: delete, at: 1) + Self.bind(kind.rawValue, to: delete, at: 2) + defer { sqlite3_finalize(delete) } + try Self.stepDone(delete, database: database) + + let insert = try Self.prepare(database, """ + INSERT INTO buffered_lines (file_id, kind, line_index, ordinal, end_offset, payload) + VALUES ((SELECT id FROM files WHERE path = ?), ?, ?, ?, ?, ?) + """) + defer { sqlite3_finalize(insert) } + for line in lines { + sqlite3_reset(insert) + sqlite3_clear_bindings(insert) + Self.bind(line.path, to: insert, at: 1) + Self.bind(line.kind.rawValue, to: insert, at: 2) + sqlite3_bind_int64(insert, 3, Int64(line.lineIndex)) + Self.bind(line.ordinal, to: insert, at: 4) + Self.bind(line.endOffset, to: insert, at: 5) + Self.bind(line.payload, to: insert, at: 6) + try Self.stepDone(insert, database: database) + } + } + return true + } + } + + @discardableResult + func setDiscoveryState(_ state: CostUsageStoreDiscoveryState?) -> Bool { + self.setSingleton(state, table: "discovery_state") + } + + @discardableResult + func setLookbackState(_ state: CostUsageStoreLookbackState?) -> Bool { + self.setSingleton(state, table: "lookback_state") + } + + @discardableResult + func setMetadata(_ metadata: CostUsageStoreMetadata) -> Bool { + self.setSingleton(metadata, table: "scan_metadata") + } + + @discardableResult + func upsertAccumulator(_ accumulator: CostUsageStoreAccumulator) -> Bool { + self.withDatabase(default: false) { database in + let seen = try JSONEncoder().encode(accumulator.seenRawTotals) + let statement = try Self.prepare(database, """ + INSERT INTO accumulators ( + file_id, event_count, next_usage_row_index, + counted_input, counted_cached, counted_output, counted_reasoning, + baseline_input, baseline_cached, baseline_output, baseline_reasoning, + watermark_input, watermark_cached, watermark_output, watermark_reasoning, + saw_divergent, saw_interleaved, seen_raw_totals, updated_at_ms + ) VALUES ((SELECT id FROM files WHERE path = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(file_id) DO UPDATE SET + event_count = excluded.event_count, + next_usage_row_index = excluded.next_usage_row_index, + counted_input = excluded.counted_input, + counted_cached = excluded.counted_cached, + counted_output = excluded.counted_output, + counted_reasoning = excluded.counted_reasoning, + baseline_input = excluded.baseline_input, + baseline_cached = excluded.baseline_cached, + baseline_output = excluded.baseline_output, + baseline_reasoning = excluded.baseline_reasoning, + watermark_input = excluded.watermark_input, + watermark_cached = excluded.watermark_cached, + watermark_output = excluded.watermark_output, + watermark_reasoning = excluded.watermark_reasoning, + saw_divergent = excluded.saw_divergent, + saw_interleaved = excluded.saw_interleaved, + seen_raw_totals = excluded.seen_raw_totals, + updated_at_ms = excluded.updated_at_ms + """) + defer { sqlite3_finalize(statement) } + Self.bind(accumulator.path, to: statement, at: 1) + sqlite3_bind_int64(statement, 2, Int64(accumulator.eventCount)) + Self.bind(accumulator.nextUsageRowIndex, to: statement, at: 3) + Self.bindTotals(accumulator.countedTotals, to: statement, startingAt: 4) + Self.bindTotals(accumulator.rawTotalsBaseline, to: statement, startingAt: 8) + Self.bindTotals(accumulator.rawTotalsWatermark, to: statement, startingAt: 12) + sqlite3_bind_int(statement, 16, accumulator.sawDivergentTotals ? 1 : 0) + sqlite3_bind_int(statement, 17, accumulator.sawInterleavedTotals ? 1 : 0) + Self.bind(seen, to: statement, at: 18) + sqlite3_bind_int64(statement, 19, accumulator.updatedAtUnixMs) + try Self.stepDone(statement, database: database) + return true + } + } + + private func setSingleton(_ value: (some Encodable)?, table: String) -> Bool { + self.withDatabase(default: false) { database in + if let value { + let payload = try JSONEncoder().encode(value) + let statement = try Self.prepare(database, """ + INSERT INTO \(table)(id, payload) VALUES (1, ?) + ON CONFLICT(id) DO UPDATE SET payload = excluded.payload + """) + defer { sqlite3_finalize(statement) } + Self.bind(payload, to: statement, at: 1) + try Self.stepDone(statement, database: database) + } else { + try Self.execute(database, "DELETE FROM \(table) WHERE id = 1") + } + return true + } + } +} + +// MARK: - Write helpers + +extension CostUsageStore { + static func inTransaction(_ database: OpaquePointer, _ operation: () throws -> T) throws -> T { + try self.execute(database, "BEGIN IMMEDIATE") + do { + let value = try operation() + try self.execute(database, "COMMIT") + return value + } catch { + try? self.execute(database, "ROLLBACK") + throw error + } + } + + static func bindTotals( + _ totals: CostUsageStoreTotals?, + to statement: OpaquePointer, + startingAt index: Int32) + { + self.bind(totals?.input, to: statement, at: index) + self.bind(totals?.cached, to: statement, at: index + 1) + self.bind(totals?.output, to: statement, at: index + 2) + self.bind(totals?.reasoning, to: statement, at: index + 3) + } + + static func bindAggregateValues( + _ aggregate: CostUsageStoreDayAggregate, + to statement: OpaquePointer, + startingAt index: Int32) + { + let values = [ + aggregate.inputTokens, + aggregate.cachedTokens, + aggregate.outputTokens, + aggregate.reasoningTokens, + aggregate.requestCount, + aggregate.knownCostNanos, + aggregate.unpricedTokens, + aggregate.standardCostNanos, + aggregate.priorityCostNanos, + aggregate.standardTokens, + aggregate.priorityTokens, + ] + for (offset, value) in values.enumerated() { + sqlite3_bind_int64(statement, index + Int32(offset), value) + } + } +} diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift new file mode 100644 index 0000000000..bef9c58974 --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStore.swift @@ -0,0 +1,422 @@ +import Foundation + +#if canImport(SQLite3) +import SQLite3 +#elseif canImport(CSQLite3) +import CSQLite3 +#endif + +/// Single-writer persistence for Codex cost scanning. The actor owns the only writable +/// connection; Phase 2 can keep its existing scan-queue serialization while independent +/// app and CLI readers use WAL snapshots through separate read-only connections. +actor CostUsageStore { + private final class SQLiteConnection: @unchecked Sendable { + private(set) var handle: OpaquePointer? + + init(handle: OpaquePointer) { + self.handle = handle + } + + func close() { + guard let handle else { return } + sqlite3_close_v2(handle) + self.handle = nil + } + + deinit { + self.close() + } + } + + static let databaseFilename = "cost-usage.sqlite" + static let baseSchemaVersion = 1 + static let schemaVersion = CostUsageStore.combinedSchemaVersion( + base: CostUsageStore.baseSchemaVersion, + parserHash: CodexParserHash.value) + + let databaseURL: URL + private let expectedSchemaVersion: Int32 + private let expectedParserHash: String + private var connection: SQLiteConnection? + private(set) var rebuildCount = 0 + + init( + cacheRoot: URL? = nil, + schemaVersion: Int32 = CostUsageStore.schemaVersion, + parserHash: String = CodexParserHash.value) + { + let root = cacheRoot ?? FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask).first! + self.databaseURL = root + .appendingPathComponent("cost-usage", isDirectory: true) + .appendingPathComponent(Self.databaseFilename, isDirectory: false) + self.expectedSchemaVersion = schemaVersion + self.expectedParserHash = parserHash + } + + static func combinedSchemaVersion(base: Int, parserHash: String) -> Int32 { + var hash: UInt32 = 2_166_136_261 + for byte in parserHash.utf8 { + hash ^= UInt32(byte) + hash &*= 16_777_619 + } + let combined = (UInt32(truncatingIfNeeded: base) & 0x7F) << 24 | (hash & 0x00FF_FFFF) + return Int32(combined) + } +} + +// MARK: - Connection lifecycle + +extension CostUsageStore { + enum StoreError: Error { + case sqlite(Int32) + case invalidData + case incompatibleSchema + } + + func withDatabase(default fallback: T, _ operation: (OpaquePointer) throws -> T) -> T { + do { + let database = try self.ensureDatabase() + return try operation(database) + } catch { + self.rebuildDatabase() + do { + let database = try self.ensureDatabase() + return try operation(database) + } catch { + self.rebuildDatabase() + return fallback + } + } + } + + func ensureDatabase() throws -> OpaquePointer { + if let database = self.connection?.handle { + return database + } + do { + let opened = try self.openDatabase() + self.connection = SQLiteConnection(handle: opened) + return opened + } catch { + self.rebuildDatabase() + guard let database = self.connection?.handle else { throw error } + return database + } + } + + private func openDatabase() throws -> OpaquePointer { + let directory = self.databaseURL.deletingLastPathComponent() + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + let existed = FileManager.default.fileExists(atPath: self.databaseURL.path) + var opened: OpaquePointer? + let flags = SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX + let result = sqlite3_open_v2(self.databaseURL.path, &opened, flags, nil) + guard result == SQLITE_OK, let opened else { + if let opened { + sqlite3_close_v2(opened) + } + throw StoreError.sqlite(result) + } + do { + try Self.configure(opened) + if existed { + try self.validateExistingDatabase(opened) + } else { + try Self.execute(opened, "PRAGMA auto_vacuum=INCREMENTAL") + try Self.execute(opened, "VACUUM") + try self.createSchema(opened) + } + return opened + } catch { + sqlite3_close_v2(opened) + throw error + } + } + + private func validateExistingDatabase(_ database: OpaquePointer) throws { + guard try Self.scalarInt(database, "PRAGMA user_version") == Int64(self.expectedSchemaVersion) else { + throw StoreError.incompatibleSchema + } + guard try Self.scalarText( + database, + "SELECT value FROM meta WHERE key = 'parser_hash'") == self.expectedParserHash + else { throw StoreError.incompatibleSchema } + guard try Self.scalarText(database, "PRAGMA quick_check") == "ok" else { + throw StoreError.invalidData + } + guard try Self.scalarInt(database, "PRAGMA auto_vacuum") == 2 else { + throw StoreError.incompatibleSchema + } + } + + private func createSchema(_ database: OpaquePointer) throws { + try Self.execute(database, Self.schemaSQL) + try Self.execute(database, "PRAGMA user_version = \(self.expectedSchemaVersion)") + let statement = try Self.prepare(database, "INSERT INTO meta(key, value) VALUES ('parser_hash', ?)") + defer { sqlite3_finalize(statement) } + Self.bind(self.expectedParserHash, to: statement, at: 1) + try Self.stepDone(statement, database: database) + } + + private func rebuildDatabase() { + self.connection?.close() + self.connection = nil + for suffix in ["", "-wal", "-shm"] { + let path = self.databaseURL.path + suffix + if FileManager.default.fileExists(atPath: path) { + try? FileManager.default.removeItem(atPath: path) + } + } + self.rebuildCount += 1 + if let database = try? self.openDatabase() { + self.connection = SQLiteConnection(handle: database) + } + } +} + +// MARK: - Schema + +extension CostUsageStore { + private static let schemaSQL = """ + CREATE TABLE meta ( + key TEXT PRIMARY KEY NOT NULL, + value TEXT NOT NULL + ); + CREATE TABLE scan_metadata ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload BLOB NOT NULL + ); + CREATE TABLE files ( + id INTEGER PRIMARY KEY, + path TEXT NOT NULL UNIQUE, + inode INTEGER, + mtime_ms INTEGER NOT NULL, + size INTEGER NOT NULL, + parsed_bytes INTEGER, + anchor_indexed_bytes INTEGER, + anchor_window_start INTEGER, + anchor_sha256 TEXT, + scan_state BLOB NOT NULL, + scan_target_size INTEGER, + scan_complete INTEGER NOT NULL, + session_id TEXT, + coverage_since_day TEXT, + coverage_until_day TEXT, + updated_at_ms INTEGER NOT NULL + ); + CREATE INDEX files_path_idx ON files(path); + CREATE INDEX files_session_idx ON files(session_id); + CREATE INDEX files_coverage_idx ON files(coverage_since_day, coverage_until_day); + CREATE INDEX files_updated_idx ON files(updated_at_ms); + CREATE TABLE token_snapshots ( + file_id INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE, + event_index INTEGER NOT NULL, + timestamp TEXT NOT NULL, + timestamp_ms INTEGER, + day TEXT, + last_input INTEGER, + last_cached INTEGER, + last_output INTEGER, + last_reasoning INTEGER, + total_input INTEGER, + total_cached INTEGER, + total_output INTEGER, + total_reasoning INTEGER, + end_offset INTEGER NOT NULL, + PRIMARY KEY(file_id, event_index) + ); + CREATE INDEX token_snapshots_day_idx ON token_snapshots(day); + CREATE INDEX token_snapshots_timestamp_idx ON token_snapshots(file_id, timestamp_ms, event_index); + CREATE TABLE file_day_aggregates ( + file_id INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE, + day TEXT NOT NULL, + model TEXT NOT NULL, + input_tokens INTEGER NOT NULL, + cached_tokens INTEGER NOT NULL, + output_tokens INTEGER NOT NULL, + reasoning_tokens INTEGER NOT NULL, + request_count INTEGER NOT NULL, + known_cost_nanos INTEGER NOT NULL, + unpriced_tokens INTEGER NOT NULL, + standard_cost_nanos INTEGER NOT NULL, + priority_cost_nanos INTEGER NOT NULL, + standard_tokens INTEGER NOT NULL, + priority_tokens INTEGER NOT NULL, + PRIMARY KEY(file_id, day, model) + ); + CREATE INDEX file_day_aggregates_day_idx ON file_day_aggregates(day); + CREATE INDEX file_day_aggregates_model_day_idx ON file_day_aggregates(model, day); + CREATE TABLE day_aggregates ( + day TEXT NOT NULL, + model TEXT NOT NULL, + input_tokens INTEGER NOT NULL, + cached_tokens INTEGER NOT NULL, + output_tokens INTEGER NOT NULL, + reasoning_tokens INTEGER NOT NULL, + request_count INTEGER NOT NULL, + known_cost_nanos INTEGER NOT NULL, + unpriced_tokens INTEGER NOT NULL, + standard_cost_nanos INTEGER NOT NULL, + priority_cost_nanos INTEGER NOT NULL, + standard_tokens INTEGER NOT NULL, + priority_tokens INTEGER NOT NULL, + PRIMARY KEY(day, model) + ); + CREATE INDEX day_aggregates_day_idx ON day_aggregates(day); + CREATE INDEX day_aggregates_model_idx ON day_aggregates(model); + CREATE INDEX day_aggregates_model_day_idx ON day_aggregates(model, day); + CREATE TABLE fork_lineage ( + file_id INTEGER PRIMARY KEY REFERENCES files(id) ON DELETE CASCADE, + session_id TEXT, + forked_from_id TEXT, + fork_timestamp TEXT, + dependency_key TEXT, + subagent_state BLOB, + accounting_state BLOB + ); + CREATE INDEX fork_lineage_parent_idx ON fork_lineage(forked_from_id); + CREATE TABLE buffered_lines ( + file_id INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE, + kind TEXT NOT NULL, + line_index INTEGER NOT NULL, + ordinal INTEGER, + end_offset INTEGER, + payload BLOB NOT NULL, + PRIMARY KEY(file_id, kind, line_index) + ); + CREATE INDEX buffered_lines_file_idx ON buffered_lines(file_id, kind, line_index); + CREATE TABLE discovery_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload BLOB NOT NULL + ); + CREATE TABLE lookback_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload BLOB NOT NULL + ); + CREATE TABLE accumulators ( + file_id INTEGER PRIMARY KEY REFERENCES files(id) ON DELETE CASCADE, + event_count INTEGER NOT NULL, + next_usage_row_index INTEGER, + counted_input INTEGER, + counted_cached INTEGER, + counted_output INTEGER, + counted_reasoning INTEGER, + baseline_input INTEGER, + baseline_cached INTEGER, + baseline_output INTEGER, + baseline_reasoning INTEGER, + watermark_input INTEGER, + watermark_cached INTEGER, + watermark_output INTEGER, + watermark_reasoning INTEGER, + saw_divergent INTEGER NOT NULL, + saw_interleaved INTEGER NOT NULL, + seen_raw_totals BLOB NOT NULL, + updated_at_ms INTEGER NOT NULL + ); + CREATE INDEX accumulators_updated_idx ON accumulators(updated_at_ms); + """ +} + +// MARK: - SQLite primitives + +extension CostUsageStore { + static func configure(_ database: OpaquePointer) throws { + guard sqlite3_busy_timeout(database, 5000) == SQLITE_OK else { + throw self.sqliteError(database) + } + try self.execute(database, "PRAGMA foreign_keys=ON") + try self.execute(database, "PRAGMA journal_mode=WAL") + } + + static func execute(_ database: OpaquePointer, _ sql: String) throws { + var message: UnsafeMutablePointer? + let result = sqlite3_exec(database, sql, nil, nil, &message) + if let message { + sqlite3_free(message) + } + guard result == SQLITE_OK else { throw StoreError.sqlite(result) } + } + + static func prepare(_ database: OpaquePointer, _ sql: String) throws -> OpaquePointer { + var statement: OpaquePointer? + let result = sqlite3_prepare_v2(database, sql, -1, &statement, nil) + guard result == SQLITE_OK, let statement else { throw StoreError.sqlite(result) } + return statement + } + + static func stepDone(_ statement: OpaquePointer, database: OpaquePointer) throws { + let result = sqlite3_step(statement) + guard result == SQLITE_DONE else { throw StoreError.sqlite(result) } + } + + static func scalarInt(_ database: OpaquePointer, _ sql: String) throws -> Int64 { + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + let result = sqlite3_step(statement) + guard result == SQLITE_ROW else { throw StoreError.sqlite(result) } + return sqlite3_column_int64(statement, 0) + } + + static func scalarText(_ database: OpaquePointer, _ sql: String) throws -> String? { + let statement = try self.prepare(database, sql) + defer { sqlite3_finalize(statement) } + let result = sqlite3_step(statement) + guard result == SQLITE_ROW else { throw StoreError.sqlite(result) } + return self.columnText(statement, at: 0) + } + + static func bind(_ value: String?, to statement: OpaquePointer, at index: Int32) { + guard let value else { + sqlite3_bind_null(statement, index) + return + } + sqlite3_bind_text(statement, index, value, -1, self.transientDestructor) + } + + static func bind(_ value: Int64?, to statement: OpaquePointer, at index: Int32) { + guard let value else { + sqlite3_bind_null(statement, index) + return + } + sqlite3_bind_int64(statement, index, value) + } + + static func bind(_ value: Int?, to statement: OpaquePointer, at index: Int32) { + self.bind(value.map(Int64.init), to: statement, at: index) + } + + static func bind(_ value: Data?, to statement: OpaquePointer, at index: Int32) { + guard let value else { + sqlite3_bind_null(statement, index) + return + } + _ = value.withUnsafeBytes { bytes in + sqlite3_bind_blob(statement, index, bytes.baseAddress, Int32(bytes.count), self.transientDestructor) + } + } + + static func columnText(_ statement: OpaquePointer, at index: Int32) -> String? { + guard let value = sqlite3_column_text(statement, index) else { return nil } + return String(cString: value) + } + + static func columnInt64(_ statement: OpaquePointer, at index: Int32) -> Int64? { + sqlite3_column_type(statement, index) == SQLITE_NULL ? nil : sqlite3_column_int64(statement, index) + } + + static func columnData(_ statement: OpaquePointer, at index: Int32) -> Data? { + guard let bytes = sqlite3_column_blob(statement, index) else { + return sqlite3_column_type(statement, index) == SQLITE_NULL ? nil : Data() + } + return Data(bytes: bytes, count: Int(sqlite3_column_bytes(statement, index))) + } + + static func sqliteError(_ database: OpaquePointer) -> StoreError { + StoreError.sqlite(sqlite3_extended_errcode(database)) + } + + static var transientDestructor: sqlite3_destructor_type { + unsafeBitCast(-1, to: sqlite3_destructor_type.self) + } +} diff --git a/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStoreModels.swift b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStoreModels.swift new file mode 100644 index 0000000000..659fa0f531 --- /dev/null +++ b/Sources/CodexBarCore/Vendored/CostUsage/CostUsageStoreModels.swift @@ -0,0 +1,225 @@ +import Foundation + +struct CostUsageStoreTotals: Codable, Equatable, Sendable { + var input: Int64 + var cached: Int64 + var output: Int64 + var reasoning: Int64? + + static let zero = Self(input: 0, cached: 0, output: 0, reasoning: nil) +} + +struct CostUsageStoreValidationAnchor: Codable, Equatable, Sendable { + var indexedBytes: Int64 + var windowStart: Int64 + var sha256: String +} + +struct CostUsageStoreScanState: Codable, Equatable, Sendable { + var targetSize: Int64? + var isComplete: Bool + var resumePayload: Data? + var tokenTimestampsMonotonic: Bool? + var nextUsageRowIndex: Int? + var lastModel: String? + var lastTurnID: String? +} + +struct CostUsageStoreFile: Codable, Equatable, Sendable { + var path: String + var inode: Int64? + var mtimeUnixMs: Int64 + var size: Int64 + var parsedBytes: Int64? + var anchor: CostUsageStoreValidationAnchor? + var scanState: CostUsageStoreScanState + var sessionID: String? + var coverageSinceDay: String? + var coverageUntilDay: String? + var updatedAtUnixMs: Int64 +} + +struct CostUsageStoreTokenSnapshot: Codable, Equatable, Sendable { + var path: String + var eventIndex: Int + var timestamp: String + var timestampUnixMs: Int64? + var day: String? + var last: CostUsageStoreTotals? + var total: CostUsageStoreTotals? + var endOffset: Int64 +} + +struct CostUsageStoreDayAggregate: Codable, Equatable, Sendable { + var day: String + var model: String + var inputTokens: Int64 + var cachedTokens: Int64 + var outputTokens: Int64 + var reasoningTokens: Int64 + var requestCount: Int64 + var knownCostNanos: Int64 + var unpricedTokens: Int64 + var standardCostNanos: Int64 + var priorityCostNanos: Int64 + var standardTokens: Int64 + var priorityTokens: Int64 + + static func zero(day: String, model: String) -> Self { + Self( + day: day, + model: model, + inputTokens: 0, + cachedTokens: 0, + outputTokens: 0, + reasoningTokens: 0, + requestCount: 0, + knownCostNanos: 0, + unpricedTokens: 0, + standardCostNanos: 0, + priorityCostNanos: 0, + standardTokens: 0, + priorityTokens: 0) + } +} + +struct CostUsageStoreFileDayAggregate: Codable, Equatable, Sendable { + var path: String + var aggregate: CostUsageStoreDayAggregate +} + +struct CostUsageStoreForkLineage: Codable, Equatable, Sendable { + var path: String + var sessionID: String? + var forkedFromID: String? + var forkTimestamp: String? + var dependencyKey: String? + var subagentState: Data? + var accountingState: Data? +} + +enum CostUsageStoreBufferedLineKind: String, Codable, CaseIterable, Sendable { + case subagent + case unresolvedFork + case deferredReplay +} + +struct CostUsageStoreBufferedLine: Codable, Equatable, Sendable { + var path: String + var kind: CostUsageStoreBufferedLineKind + var lineIndex: Int + var ordinal: Int? + var endOffset: Int64? + var payload: Data +} + +struct CostUsageStoreDiscoveryState: Codable, Equatable, Sendable { + var roots: [String] + var generation: String? + var directoryPaths: [String] + var nextDirectoryIndex: Int + var filePaths: [String] + var nextFileIndex: Int + var filePathBySessionID: [String: String] + var missingSessionIDs: [String] + var pendingSessionIDs: [String] + var validationDirectoryIndex: Int + var isComplete: Bool + var payload: Data? +} + +struct CostUsageStoreLookbackState: Codable, Equatable, Sendable { + var scanSinceDay: String + var rootPaths: [String] + var nextDayByRoot: [String: String] + var completedRootPaths: [String] + var pendingFilePaths: [String] + var legacyRecursivePendingRootPaths: [String] +} + +struct CostUsageStoreAccumulator: Codable, Equatable, Sendable { + var path: String + var eventCount: Int + var nextUsageRowIndex: Int? + var countedTotals: CostUsageStoreTotals? + var rawTotalsBaseline: CostUsageStoreTotals? + var rawTotalsWatermark: CostUsageStoreTotals? + var sawDivergentTotals: Bool + var sawInterleavedTotals: Bool + var seenRawTotals: [CostUsageStoreTotals] + var updatedAtUnixMs: Int64 +} + +struct CostUsageStoreMetadata: Codable, Equatable, Sendable { + var lastScanUnixMs: Int64 + var scanSinceDay: String? + var scanUntilDay: String? + var timeZoneIdentifier: String? + var pricingKey: String? + var priorityMetadataKey: String? + var catchUpPending: Bool + var processedBytes: Int64? + var totalBytes: Int64? + var completedFiles: Int? + var totalFiles: Int? + var rootMtimes: [String: Int64]? + var previousReportPayload: Data? + var priorityTurnStatePayload: Data? + var projectMetadataVersion: Int? + + static let empty = Self( + lastScanUnixMs: 0, + scanSinceDay: nil, + scanUntilDay: nil, + timeZoneIdentifier: nil, + pricingKey: nil, + priorityMetadataKey: nil, + catchUpPending: false, + processedBytes: nil, + totalBytes: nil, + completedFiles: nil, + totalFiles: nil, + rootMtimes: nil, + previousReportPayload: nil, + priorityTurnStatePayload: nil, + projectMetadataVersion: nil) +} + +struct CostUsageStoreReport: Equatable, Sendable { + var metadata: CostUsageStoreMetadata + var aggregates: [CostUsageStoreDayAggregate] +} + +struct CostUsageStoreSnapshot: Equatable, Sendable { + var metadata: CostUsageStoreMetadata + var files: [CostUsageStoreFile] + var tokenSnapshots: [CostUsageStoreTokenSnapshot] + var fileDayAggregates: [CostUsageStoreFileDayAggregate] + var dayAggregates: [CostUsageStoreDayAggregate] + var forkLineage: [CostUsageStoreForkLineage] + var bufferedLines: [CostUsageStoreBufferedLine] + var discoveryState: CostUsageStoreDiscoveryState? + var lookbackState: CostUsageStoreLookbackState? + var accumulators: [CostUsageStoreAccumulator] +} + +struct CostUsageStoreRetentionResult: Equatable, Sendable { + var deletedFiles: Int + var deletedTokenSnapshots: Int + var deletedFileDayAggregates: Int + var deletedDayAggregates: Int +} + +struct CostUsageStoreBudgetResult: Equatable, Sendable { + var deletedRows: Int + var rowCount: Int + var fileBytes: Int64 +} + +struct CostUsageStoreConfiguration: Equatable, Sendable { + var journalMode: String + var busyTimeoutMilliseconds: Int + var foreignKeysEnabled: Bool + var autoVacuumMode: Int + var userVersion: Int +} diff --git a/Tests/CodexBarTests/CostUsageStoreTests.swift b/Tests/CodexBarTests/CostUsageStoreTests.swift new file mode 100644 index 0000000000..37e3e120a3 --- /dev/null +++ b/Tests/CodexBarTests/CostUsageStoreTests.swift @@ -0,0 +1,866 @@ +import Foundation +import Testing +@testable import CodexBarCore + +#if canImport(SQLite3) +import SQLite3 +#elseif canImport(CSQLite3) +import CSQLite3 +#endif + +struct CostUsageStoreTests { + @Test + func `database lives beside the legacy artifact directory`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + + #expect(await store.databaseURL.lastPathComponent == "cost-usage.sqlite") + #expect(await store.databaseURL.deletingLastPathComponent().lastPathComponent == "cost-usage") + } + + @Test + func `new database uses WAL`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let configuration = await CostUsageStore(cacheRoot: fixture.root).configuration() + + #expect(configuration?.journalMode.lowercased() == "wal") + } + + @Test + func `new database configures busy timeout`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let configuration = await CostUsageStore(cacheRoot: fixture.root).configuration() + + #expect(configuration?.busyTimeoutMilliseconds == 5000) + } + + @Test + func `new database enables foreign keys`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let configuration = await CostUsageStore(cacheRoot: fixture.root).configuration() + + #expect(configuration?.foreignKeysEnabled == true) + } + + @Test + func `new database uses incremental auto vacuum`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let configuration = await CostUsageStore(cacheRoot: fixture.root).configuration() + + #expect(configuration?.autoVacuumMode == 2) + } + + @Test + func `user version combines schema and parser hash`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let configuration = await CostUsageStore(cacheRoot: fixture.root).configuration() + + #expect(configuration?.userVersion == Int(CostUsageStore.schemaVersion)) + #expect(CostUsageStore.combinedSchemaVersion(base: 1, parserHash: "a") != + CostUsageStore.combinedSchemaVersion(base: 1, parserHash: "b")) + } + + @Test + func `schema contains phase two query indexes`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + _ = await store.configuration() + + let indexes = try await SQLiteTestConnection.indexNames(at: store.databaseURL) + #expect(indexes.contains("files_path_idx")) + #expect(indexes.contains("file_day_aggregates_day_idx")) + #expect(indexes.contains("file_day_aggregates_model_day_idx")) + #expect(indexes.contains("day_aggregates_day_idx")) + #expect(indexes.contains("day_aggregates_model_idx")) + #expect(indexes.contains("day_aggregates_model_day_idx")) + } +} + +extension CostUsageStoreTests { + @Test + func `file state round trips all validation fields`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + + #expect(await store.upsertFile(file)) + #expect(await store.fetchFile(path: file.path) == file) + } + + @Test + func `file upsert replaces mutable state`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + var file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + file.size = 999 + file.parsedBytes = 800 + file.scanState.isComplete = false + file.updatedAtUnixMs = 20 + + #expect(await store.upsertFile(file)) + #expect(await store.fetchFile(path: file.path) == file) + } + + @Test + func `missing file returns nil`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + + #expect(await CostUsageStore(cacheRoot: fixture.root).fetchFile(path: "/missing") == nil) + } + + @Test + func `deleting file cascades dependent tables`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + #expect(await store.appendTokenSnapshots([Self.snapshot(path: file.path, eventIndex: 0)])) + #expect(await store.replaceFileDayAggregates( + path: file.path, + aggregates: [Self.aggregate(day: "2026-08-01", model: "model-a", scale: 1)])) + #expect(await store.upsertForkLineage(Self.lineage(path: file.path))) + #expect(await store.upsertAccumulator(Self.accumulator(path: file.path))) + + #expect(await store.deleteFile(path: file.path)) + let snapshot = await store.readSnapshot() + #expect(snapshot.files.isEmpty) + #expect(snapshot.tokenSnapshots.isEmpty) + #expect(snapshot.fileDayAggregates.isEmpty) + #expect(snapshot.forkLineage.isEmpty) + #expect(snapshot.accumulators.isEmpty) + } +} + +extension CostUsageStoreTests { + @Test + func `token snapshot deltas round trip`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + let snapshots = [ + Self.snapshot(path: file.path, eventIndex: 0), + Self.snapshot(path: file.path, eventIndex: 1, day: "2026-08-02"), + ] + + #expect(await store.appendTokenSnapshots(snapshots)) + #expect(await store.fetchTokenSnapshots(path: file.path) == snapshots) + } + + @Test + func `token snapshot append is idempotent by event index`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + var snapshot = Self.snapshot(path: file.path, eventIndex: 0) + #expect(await store.appendTokenSnapshots([snapshot])) + snapshot.endOffset = 222 + + #expect(await store.appendTokenSnapshots([snapshot])) + #expect(await store.fetchTokenSnapshots(path: file.path) == [snapshot]) + } + + @Test + func `day aggregate inserts and reads by model`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let aggregate = Self.aggregate(day: "2026-08-01", model: "gpt-5.6-sol", scale: 1) + + #expect(await store.mergeDayAggregates([aggregate])) + #expect(await store.fetchDayAggregates(sinceDay: "2026-08-01", untilDay: "2026-08-01") == [aggregate]) + } + + @Test + func `per file aggregates replace and round trip`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + let first = Self.aggregate(day: "2026-08-01", model: "model-a", scale: 1) + let replacement = Self.aggregate(day: "2026-08-02", model: "model-b", scale: 2) + #expect(await store.upsertFile(file)) + #expect(await store.replaceFileDayAggregates(path: file.path, aggregates: [first])) + #expect(await store.replaceFileDayAggregates(path: file.path, aggregates: [replacement])) + + #expect(await store.fetchFileDayAggregates(path: file.path) == [replacement]) + } + + @Test + func `day aggregate merge adds every metric`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let first = Self.aggregate(day: "2026-08-01", model: "gpt-5.6-sol", scale: 1) + let second = Self.aggregate(day: "2026-08-01", model: "gpt-5.6-sol", scale: 2) + + #expect(await store.mergeDayAggregates([first, second])) + #expect(await store.fetchDayAggregates(sinceDay: "2026-08-01", untilDay: "2026-08-01") == [ + Self.aggregate(day: "2026-08-01", model: "gpt-5.6-sol", scale: 3), + ]) + } + + @Test + func `day aggregate range is inclusive and sorted`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let values = [ + Self.aggregate(day: "2026-08-03", model: "model-b", scale: 1), + Self.aggregate(day: "2026-08-01", model: "model-a", scale: 1), + Self.aggregate(day: "2026-08-02", model: "model-a", scale: 1), + ] + #expect(await store.mergeDayAggregates(values)) + + let fetched = await store.fetchDayAggregates(sinceDay: "2026-08-01", untilDay: "2026-08-02") + #expect(fetched.map(\.day) == ["2026-08-01", "2026-08-02"]) + } + + @Test + func `negative aggregate delta subtracts prior contribution`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let first = Self.aggregate(day: "2026-08-01", model: "model-a", scale: 3) + let delta = Self.aggregate(day: "2026-08-01", model: "model-a", scale: -1) + #expect(await store.mergeDayAggregates([first, delta])) + + #expect(await store.fetchDayAggregates(sinceDay: "2026-08-01", untilDay: "2026-08-01") == [ + Self.aggregate(day: "2026-08-01", model: "model-a", scale: 2), + ]) + } +} + +extension CostUsageStoreTests { + @Test + func `fork lineage round trips typed and opaque state`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/child.jsonl", day: "2026-08-01") + let lineage = Self.lineage(path: file.path) + #expect(await store.upsertFile(file)) + + #expect(await store.upsertForkLineage(lineage)) + #expect(await store.fetchForkLineage(path: file.path) == lineage) + } + + @Test + func `buffered lines round trip every retry kind`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/child.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + var expected: [CostUsageStoreBufferedLine] = [] + for (index, kind) in CostUsageStoreBufferedLineKind.allCases.enumerated() { + let line = Self.bufferedLine(path: file.path, kind: kind, index: index) + expected.append(line) + #expect(await store.replaceBufferedLines(path: file.path, kind: kind, lines: [line])) + } + + #expect(await store.fetchBufferedLines(path: file.path) == expected.sorted { + $0.kind.rawValue < $1.kind.rawValue + }) + } + + @Test + func `buffer replacement is scoped by retry kind`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/child.jsonl", day: "2026-08-01") + #expect(await store.upsertFile(file)) + let subagent = Self.bufferedLine(path: file.path, kind: .subagent, index: 1) + let fork = Self.bufferedLine(path: file.path, kind: .unresolvedFork, index: 2) + #expect(await store.replaceBufferedLines(path: file.path, kind: .subagent, lines: [subagent])) + #expect(await store.replaceBufferedLines(path: file.path, kind: .unresolvedFork, lines: [fork])) + + #expect(await store.replaceBufferedLines(path: file.path, kind: .subagent, lines: [])) + #expect(await store.fetchBufferedLines(path: file.path) == [fork]) + } + + @Test + func `discovery state round trips and clears`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let state = Self.discoveryState(paths: ["/a", "/b"]) + + #expect(await store.setDiscoveryState(state)) + #expect(await store.fetchDiscoveryState() == state) + #expect(await store.setDiscoveryState(nil)) + #expect(await store.fetchDiscoveryState() == nil) + } + + @Test + func `active lookback state round trips`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let state = CostUsageStoreLookbackState( + scanSinceDay: "2026-07-01", + rootPaths: ["/root"], + nextDayByRoot: ["/root": "2026-07-10"], + completedRootPaths: [], + pendingFilePaths: ["/pending"], + legacyRecursivePendingRootPaths: ["/legacy"]) + + #expect(await store.setLookbackState(state)) + #expect(await store.fetchLookbackState() == state) + } + + @Test + func `scan metadata round trips`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let metadata = Self.metadata() + + #expect(await store.setMetadata(metadata)) + #expect(await store.fetchMetadata() == metadata) + } + + @Test + func `terminal accumulator round trips all state`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + let accumulator = Self.accumulator(path: file.path) + #expect(await store.upsertFile(file)) + + #expect(await store.upsertAccumulator(accumulator)) + #expect(await store.fetchAccumulator(path: file.path) == accumulator) + } +} + +extension CostUsageStoreTests { + @Test + func `full snapshot reads every table from one transaction`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + let token = Self.snapshot(path: file.path, eventIndex: 0) + let aggregate = Self.aggregate(day: "2026-08-01", model: "model-a", scale: 1) + let lineage = Self.lineage(path: file.path) + let line = Self.bufferedLine(path: file.path, kind: .subagent, index: 0) + let discovery = Self.discoveryState(paths: [file.path]) + let lookback = CostUsageStoreLookbackState( + scanSinceDay: "2026-08-01", + rootPaths: ["/root"], + nextDayByRoot: [:], + completedRootPaths: [], + pendingFilePaths: [], + legacyRecursivePendingRootPaths: []) + let accumulator = Self.accumulator(path: file.path) + let metadata = Self.metadata() + #expect(await store.upsertFile(file)) + #expect(await store.appendTokenSnapshots([token])) + #expect(await store.replaceFileDayAggregates(path: file.path, aggregates: [aggregate])) + #expect(await store.mergeDayAggregates([aggregate])) + #expect(await store.upsertForkLineage(lineage)) + #expect(await store.replaceBufferedLines(path: file.path, kind: .subagent, lines: [line])) + #expect(await store.setDiscoveryState(discovery)) + #expect(await store.setLookbackState(lookback)) + #expect(await store.upsertAccumulator(accumulator)) + #expect(await store.setMetadata(metadata)) + + let snapshot = await store.readSnapshot() + #expect(snapshot == CostUsageStoreSnapshot( + metadata: metadata, + files: [file], + tokenSnapshots: [token], + fileDayAggregates: [CostUsageStoreFileDayAggregate(path: file.path, aggregate: aggregate)], + dayAggregates: [aggregate], + forkLineage: [lineage], + bufferedLines: [line], + discoveryState: discovery, + lookbackState: lookback, + accumulators: [accumulator])) + } + + @Test + func `report readback includes metadata and bounded aggregates`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let metadata = Self.metadata() + let inside = Self.aggregate(day: "2026-08-02", model: "model-a", scale: 1) + let outside = Self.aggregate(day: "2026-07-31", model: "model-a", scale: 1) + #expect(await store.setMetadata(metadata)) + #expect(await store.mergeDayAggregates([inside, outside])) + + #expect(await store.readReport(sinceDay: "2026-08-01", untilDay: "2026-08-03") == + CostUsageStoreReport(metadata: metadata, aggregates: [inside])) + } + + @Test + func `database persists across store instances`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let first = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/a.jsonl", day: "2026-08-01") + #expect(await first.upsertFile(file)) + let second = CostUsageStore(cacheRoot: fixture.root) + + #expect(await second.fetchFile(path: file.path) == file) + #expect(await second.rebuildCount == 0) + } +} + +extension CostUsageStoreTests { + @Test + func `version mismatch drops and recreates`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let url = fixture.databaseURL + try FileManager.default.createDirectory(at: url.deletingLastPathComponent(), withIntermediateDirectories: true) + try SQLiteTestConnection.execute(at: url, sql: "PRAGMA user_version = 7") + let store = CostUsageStore(cacheRoot: fixture.root) + + #expect(await store.fetchMetadata() == .empty) + #expect(await store.rebuildCount == 1) + #expect(await store.configuration()?.userVersion == Int(CostUsageStore.schemaVersion)) + } + + @Test + func `parser hash mismatch drops and recreates`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let url = fixture.databaseURL + try FileManager.default.createDirectory(at: url.deletingLastPathComponent(), withIntermediateDirectories: true) + try SQLiteTestConnection.execute(at: url, sql: """ + PRAGMA auto_vacuum=INCREMENTAL; + VACUUM; + CREATE TABLE meta(key TEXT PRIMARY KEY, value TEXT NOT NULL); + INSERT INTO meta(key, value) VALUES ('parser_hash', 'old-parser'); + PRAGMA user_version = \(CostUsageStore.schemaVersion); + """) + let store = CostUsageStore(cacheRoot: fixture.root) + + #expect(await store.fetchMetadata() == .empty) + #expect(await store.rebuildCount == 1) + } + + @Test + func `garbage database recovers by rebuild`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let url = fixture.databaseURL + try FileManager.default.createDirectory(at: url.deletingLastPathComponent(), withIntermediateDirectories: true) + try Data("not a sqlite database".utf8).write(to: url) + let store = CostUsageStore(cacheRoot: fixture.root) + + #expect(await store.fetchMetadata() == .empty) + #expect(await store.rebuildCount == 1) + #expect(await store.configuration()?.journalMode.lowercased() == "wal") + } + + @Test + func `runtime sqlite error degrades to a fresh store`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + #expect(await store.mergeDayAggregates([Self.aggregate(day: "2026-08-01", model: "model-a", scale: 1)])) + try await SQLiteTestConnection.execute(at: store.databaseURL, sql: "DROP TABLE day_aggregates") + + #expect(await store.fetchDayAggregates(sinceDay: "2026-08-01", untilDay: "2026-08-01").isEmpty) + #expect(await store.rebuildCount == 1) + } +} + +extension CostUsageStoreTests { + @Test + func `retention keeps inclusive window edges`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let days = ["2026-07-31", "2026-08-01", "2026-08-03", "2026-08-04"] + for (index, day) in days.enumerated() { + let file = Self.file(path: "/rollouts/\(index).jsonl", day: day) + #expect(await store.upsertFile(file)) + #expect(await store.appendTokenSnapshots([Self.snapshot(path: file.path, eventIndex: 0, day: day)])) + let aggregate = Self.aggregate(day: day, model: "model-a", scale: 1) + #expect(await store.replaceFileDayAggregates(path: file.path, aggregates: [aggregate])) + #expect(await store.mergeDayAggregates([aggregate])) + } + + let result = await store.retainDayWindow(sinceDay: "2026-08-01", untilDay: "2026-08-03") + #expect(result.deletedFiles == 2) + #expect(result.deletedFileDayAggregates == 2) + #expect(await (store.readSnapshot()).files.map(\.coverageSinceDay) == ["2026-08-01", "2026-08-03"]) + #expect(await (store.readSnapshot()).dayAggregates.map(\.day) == ["2026-08-01", "2026-08-03"]) + } + + @Test + func `retention preserves incomplete out of window file`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + var file = Self.file(path: "/rollouts/incomplete.jsonl", day: "2026-07-01") + file.scanState.isComplete = false + #expect(await store.upsertFile(file)) + + _ = await store.retainDayWindow(sinceDay: "2026-08-01", untilDay: "2026-08-03") + #expect(await store.fetchFile(path: file.path) != nil) + } + + @Test + func `retention preserves file with buffered retry lines`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + let file = Self.file(path: "/rollouts/buffered.jsonl", day: "2026-07-01") + let line = Self.bufferedLine(path: file.path, kind: .unresolvedFork, index: 0) + #expect(await store.upsertFile(file)) + #expect(await store.replaceBufferedLines(path: file.path, kind: .unresolvedFork, lines: [line])) + + _ = await store.retainDayWindow(sinceDay: "2026-08-01", untilDay: "2026-08-03") + #expect(await store.fetchFile(path: file.path) != nil) + } + + @Test + func `retention preserves parent referenced by surviving child`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + var parent = Self.file(path: "/rollouts/parent.jsonl", day: "2026-07-01") + parent.sessionID = "parent-session" + let child = Self.file(path: "/rollouts/child.jsonl", day: "2026-08-02") + var lineage = Self.lineage(path: child.path) + lineage.forkedFromID = "parent-session" + #expect(await store.upsertFile(parent)) + #expect(await store.upsertFile(child)) + #expect(await store.upsertForkLineage(lineage)) + + _ = await store.retainDayWindow(sinceDay: "2026-08-01", untilDay: "2026-08-03") + #expect(await store.fetchFile(path: parent.path) != nil) + } + + @Test + func `retention prunes discovery references for removed files`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + var old = Self.file(path: "/rollouts/old.jsonl", day: "2026-07-01") + old.sessionID = "old-session" + #expect(await store.upsertFile(old)) + #expect(await store.setDiscoveryState(Self.discoveryState(paths: [old.path]))) + + _ = await store.retainDayWindow(sinceDay: "2026-08-01", untilDay: "2026-08-03") + let discovery = try #require(await store.fetchDiscoveryState()) + #expect(discovery.filePaths.isEmpty) + #expect(discovery.pendingSessionIDs.isEmpty) + #expect(discovery.filePathBySessionID.isEmpty) + } +} + +extension CostUsageStoreTests { + @Test + func `row budget deletes oldest rows to cap`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + for index in 0..<10 { + let file = Self.file(path: "/rollouts/\(index).jsonl", day: "2026-08-01", updatedAt: Int64(index)) + #expect(await store.upsertFile(file)) + } + + let result = await store.enforceBudgets(maxRows: 3, maxFileBytes: Int64.max) + #expect(result.rowCount <= 3) + #expect(result.deletedRows >= 7) + } + + @Test + func `file size budget reclaims bytes after incremental vacuum`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + for index in 0..<8 { + let file = Self.file(path: "/rollouts/\(index).jsonl", day: "2026-08-01", updatedAt: Int64(index)) + #expect(await store.upsertFile(file)) + let line = CostUsageStoreBufferedLine( + path: file.path, + kind: .deferredReplay, + lineIndex: 0, + ordinal: nil, + endOffset: nil, + payload: Data(repeating: UInt8(index), count: 256 * 1024)) + #expect(await store.replaceBufferedLines(path: file.path, kind: .deferredReplay, lines: [line])) + } + let before = await store.fileSizeBytes() + let limit = max(1, before / 2) + + let result = await store.enforceBudgets(maxRows: .max, maxFileBytes: limit) + #expect(result.fileBytes < before) + #expect(result.deletedRows > 0) + } + + @Test + func `empty append and merge are no ops`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + + #expect(await store.appendTokenSnapshots([])) + #expect(await store.mergeDayAggregates([])) + #expect(await (store.readSnapshot()).files.isEmpty) + } + + @Test + func `read only WAL reader keeps a consistent snapshot during write`() async throws { + let fixture = try StoreFixture() + defer { fixture.remove() } + let store = CostUsageStore(cacheRoot: fixture.root) + #expect(await store.upsertFile(Self.file(path: "/rollouts/one.jsonl", day: "2026-08-01"))) + let reader = try await SQLiteTestConnection(url: store.databaseURL, readOnly: true) + try reader.execute("BEGIN") + #expect(try reader.scalarInt("SELECT COUNT(*) FROM files") == 1) + + #expect(await store.upsertFile(Self.file(path: "/rollouts/two.jsonl", day: "2026-08-01"))) + #expect(try reader.scalarInt("SELECT COUNT(*) FROM files") == 1) + try reader.execute("COMMIT") + #expect(try reader.scalarInt("SELECT COUNT(*) FROM files") == 2) + } +} + +// MARK: - Fixtures + +extension CostUsageStoreTests { + private static func file( + path: String, + day: String, + updatedAt: Int64 = 10) -> CostUsageStoreFile + { + CostUsageStoreFile( + path: path, + inode: 42, + mtimeUnixMs: 1000, + size: 500, + parsedBytes: 400, + anchor: CostUsageStoreValidationAnchor(indexedBytes: 400, windowStart: 144, sha256: "abc123"), + scanState: CostUsageStoreScanState( + targetSize: 500, + isComplete: true, + resumePayload: Data([1, 2, 3]), + tokenTimestampsMonotonic: true, + nextUsageRowIndex: 7, + lastModel: "gpt-5.6-sol", + lastTurnID: "turn-1"), + sessionID: "session-\(path)", + coverageSinceDay: day, + coverageUntilDay: day, + updatedAtUnixMs: updatedAt) + } + + private static func snapshot( + path: String, + eventIndex: Int, + day: String = "2026-08-01") -> CostUsageStoreTokenSnapshot + { + CostUsageStoreTokenSnapshot( + path: path, + eventIndex: eventIndex, + timestamp: "2026-08-01T12:00:00Z", + timestampUnixMs: 1_754_046_000_000 + Int64(eventIndex), + day: day, + last: CostUsageStoreTotals(input: 2, cached: 1, output: 3, reasoning: 1), + total: CostUsageStoreTotals(input: 20, cached: 10, output: 30, reasoning: 5), + endOffset: 100 + Int64(eventIndex)) + } + + private static func aggregate( + day: String, + model: String, + scale: Int64) -> CostUsageStoreDayAggregate + { + CostUsageStoreDayAggregate( + day: day, + model: model, + inputTokens: 10 * scale, + cachedTokens: 2 * scale, + outputTokens: 3 * scale, + reasoningTokens: 1 * scale, + requestCount: 1 * scale, + knownCostNanos: 1000 * scale, + unpricedTokens: 4 * scale, + standardCostNanos: 600 * scale, + priorityCostNanos: 400 * scale, + standardTokens: 9 * scale, + priorityTokens: 6 * scale) + } + + private static func lineage(path: String) -> CostUsageStoreForkLineage { + CostUsageStoreForkLineage( + path: path, + sessionID: "child-session", + forkedFromID: "parent-session", + forkTimestamp: "2026-08-01T11:00:00Z", + dependencyKey: "parent-key", + subagentState: Data([4, 5]), + accountingState: Data([6, 7])) + } + + private static func bufferedLine( + path: String, + kind: CostUsageStoreBufferedLineKind, + index: Int) -> CostUsageStoreBufferedLine + { + CostUsageStoreBufferedLine( + path: path, + kind: kind, + lineIndex: index, + ordinal: index + 10, + endOffset: Int64(index + 100), + payload: Data([UInt8(index), 9, 8])) + } + + private static func discoveryState(paths: [String]) -> CostUsageStoreDiscoveryState { + CostUsageStoreDiscoveryState( + roots: ["/root"], + generation: "generation-1", + directoryPaths: ["/root/2026/08/01"], + nextDirectoryIndex: 1, + filePaths: paths, + nextFileIndex: paths.count, + filePathBySessionID: paths.isEmpty ? [:] : ["old-session": paths[0]], + missingSessionIDs: ["old-session"], + pendingSessionIDs: ["old-session"], + validationDirectoryIndex: 2, + isComplete: false, + payload: Data([1, 3, 5])) + } + + private static func accumulator(path: String) -> CostUsageStoreAccumulator { + CostUsageStoreAccumulator( + path: path, + eventCount: 12, + nextUsageRowIndex: 9, + countedTotals: CostUsageStoreTotals(input: 10, cached: 2, output: 3, reasoning: 1), + rawTotalsBaseline: CostUsageStoreTotals(input: 8, cached: 1, output: 2, reasoning: nil), + rawTotalsWatermark: CostUsageStoreTotals(input: 12, cached: 3, output: 4, reasoning: 1), + sawDivergentTotals: true, + sawInterleavedTotals: true, + seenRawTotals: [CostUsageStoreTotals(input: 9, cached: 1, output: 2, reasoning: nil)], + updatedAtUnixMs: 99) + } + + private static func metadata() -> CostUsageStoreMetadata { + CostUsageStoreMetadata( + lastScanUnixMs: 100, + scanSinceDay: "2026-08-01", + scanUntilDay: "2026-08-03", + timeZoneIdentifier: "UTC", + pricingKey: "pricing-v1", + priorityMetadataKey: "priority-v1", + catchUpPending: true, + processedBytes: 100, + totalBytes: 200, + completedFiles: 2, + totalFiles: 4, + rootMtimes: ["/root": 123], + previousReportPayload: Data([2, 4, 6]), + priorityTurnStatePayload: Data([1, 3, 5]), + projectMetadataVersion: 2) + } +} + +private struct StoreFixture: Sendable { + let root: URL + + init() throws { + self.root = FileManager.default.temporaryDirectory + .appendingPathComponent("CodexBar-CostUsageStoreTests-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: self.root, withIntermediateDirectories: true) + } + + var databaseURL: URL { + self.root + .appendingPathComponent("cost-usage", isDirectory: true) + .appendingPathComponent("cost-usage.sqlite", isDirectory: false) + } + + func remove() { + try? FileManager.default.removeItem(at: self.root) + } +} + +private final class SQLiteTestConnection: @unchecked Sendable { + enum TestError: Error { + case sqlite(Int32) + } + + private var database: OpaquePointer? + + init(url: URL, readOnly: Bool = false) throws { + let flags = readOnly ? SQLITE_OPEN_READONLY : SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE + let result = sqlite3_open_v2(url.path, &self.database, flags, nil) + guard result == SQLITE_OK else { throw TestError.sqlite(result) } + sqlite3_busy_timeout(self.database, 5000) + } + + deinit { + if let database { + sqlite3_close_v2(database) + } + } + + func execute(_ sql: String) throws { + let result = sqlite3_exec(self.database, sql, nil, nil, nil) + guard result == SQLITE_OK else { throw TestError.sqlite(result) } + } + + func scalarInt(_ sql: String) throws -> Int64 { + var statement: OpaquePointer? + let prepare = sqlite3_prepare_v2(self.database, sql, -1, &statement, nil) + guard prepare == SQLITE_OK, let statement else { throw TestError.sqlite(prepare) } + defer { sqlite3_finalize(statement) } + let result = sqlite3_step(statement) + guard result == SQLITE_ROW else { throw TestError.sqlite(result) } + return sqlite3_column_int64(statement, 0) + } + + static func execute(at url: URL, sql: String) throws { + try Self(url: url).execute(sql) + } + + static func indexNames(at url: URL) throws -> Set { + let connection = try Self(url: url, readOnly: true) + var statement: OpaquePointer? + let result = sqlite3_prepare_v2( + connection.database, + "SELECT name FROM sqlite_master WHERE type = 'index'", + -1, + &statement, + nil) + guard result == SQLITE_OK, let statement else { throw TestError.sqlite(result) } + defer { sqlite3_finalize(statement) } + var names: Set = [] + var step = sqlite3_step(statement) + while step == SQLITE_ROW { + if let value = sqlite3_column_text(statement, 0) { + names.insert(String(cString: value)) + } + step = sqlite3_step(statement) + } + guard step == SQLITE_DONE else { throw TestError.sqlite(step) } + return names + } +}