diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7542947..ce55753 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -116,6 +116,7 @@ jobs: run: | go test ./internal/spend -run '^$' -fuzz '^FuzzReaderBoundedFormats$' -fuzztime 1000x -parallel 2 -timeout 2m go test ./internal/spend -run '^$' -fuzz '^FuzzSessionBudget$' -fuzztime 1000x -parallel 2 -timeout 2m + go test ./internal/spend -run '^$' -fuzz '^FuzzDailyCounts$' -fuzztime 10000x -parallel 2 -timeout 2m go test ./internal/compare -run '^$' -fuzz '^FuzzCanonicalSummary$' -fuzztime 10000x -parallel 2 -timeout 2m go test ./internal/compare -run '^$' -fuzz '^FuzzEmbeddedSketch$' -fuzztime 10000x -parallel 2 -timeout 2m go test ./internal/compare -run '^$' -fuzz '^FuzzScanSummary$' -fuzztime 10000x -parallel 2 -timeout 2m diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 49a001c..4d14aea 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -114,7 +114,7 @@ jobs: run: go test ./internal/cli -run '^TestReleaseBinary' -count=1 -v - name: Pinned SQL export through the packaged importer if: matrix.os == 'linux' && matrix.arch == 'amd64' - timeout-minutes: 5 + timeout-minutes: 10 env: FLEETDIFF_BIN: ${{ github.workspace }}/unpacked/fleetdiff FLEETDIFF_SPEND_DOCKER: '1' diff --git a/docs/ROW_COVERAGE.md b/docs/ROW_COVERAGE.md new file mode 100644 index 0000000..10516a7 --- /dev/null +++ b/docs/ROW_COVERAGE.md @@ -0,0 +1,107 @@ + + + +# Are All Requests In The Spend Log? + +**8 exported rows versus 9 daily requests: 1 missing row.** This synthetic +fixture reproduces a real response-ID collision through pinned LiteLLM's proxy +and database writers. The next UTC day matches: 1 row and 1 daily request. +The check leaves token and spend totals exactly as recorded in the export. + +This optional check is unreleased. Build this branch to try it: + +```sh +go build -o bin/fleetdiff ./cmd/fleetdiff +``` + +## Export And Compare + +Use the [read-only SQL recipe](../examples/litellm-spend/export.sql) with +read-only access to both `LiteLLM_SpendLogs` and `LiteLLM_DailyUserSpend`. +Configure the `litellm_readonly` PostgreSQL service through your existing +credential mechanism. The [coverage export helper](../examples/litellm-spend/export-coverage.sh) +creates a new private directory and exports both files in one repeatable-read, +read-only transaction: + +```sh +umask 077 +export_dir="$HOME/litellm-coverage-export" +sh examples/litellm-spend/export-coverage.sh "$export_dir" \ + 2026-09-23T00:00:00Z 2026-10-09T00:00:00Z +bin/fleetdiff investigate --litellm-spend "$export_dir/spend.csv" \ + --litellm-daily-spend "$export_dir/daily.csv" --group-by key +``` + +Replace the historical bounds with your interval of complete UTC days. The +16-day export leaves room for the default two complete seven-day periods. +Use `--before-period` and `--after-period` for explicit periods. Session review +also accepts the option with `--period` or `--session-budget-usd`. + +The daily projection sums `api_requests`, `successful_requests` and +`failed_requests` across all daily user/model/key rows for each date. Its CSV +contains only those three counts and `date`; no identities or content. The +request export is unchanged. The helper refuses an existing directory and uses +distinct inherited file descriptors so filename aliases cannot overwrite an +export. Its `daily_csv_file` SQL variable is internal to the helper, not a +general output-path option. Use the existing `export.sql` commands for a single +request export. Omit `--litellm-daily-spend` to keep the existing report byte for byte. + +## Reading Coverage + +Run the recorded fixture from the repository root: + +```sh +bin/fleetdiff investigate \ + --litellm-spend examples/litellm-spend/coverage_runtime_spend.csv \ + --litellm-daily-spend examples/litellm-spend/coverage_runtime_daily.csv \ + --before-period 2026-10-07 --after-period 2026-10-08 +``` + +The report begins: + +```text +Spend-log row coverage (relative to LiteLLM daily request counts): + 2026-10-07 UTC: 8 exported rows; 9 daily requests (8 successful, 1 failed); 1 missing row. + 2026-10-08 UTC: 1 exported row; 1 daily request (1 successful, 0 failed); counts match. +Token and spend totals below use exported rows only. +``` + +All selected request rows count here, including failed records and call types +excluded from token analysis. UTC days follow request `startTime`, not receipt +or export time. A positive difference is reported as missing rows relative to +the daily request count. Extra rows are a discrepancy to review, never a +negative missing count. No token, request, spend or ranking total is imputed. + +JSON adds `row_coverage.days`: `exported_rows`, `api_requests`, +`successful_requests`, `failed_requests`, `missing_rows` and `status`, with +`date_utc`. An absent daily row has null daily counts and unknown coverage. +Partial UTC days have null `missing_rows`; full-day counters cannot establish +coverage for a selected part of that day. A real exported zero remains zero. +Empty request exports can still be checked with explicit periods; the usage +comparison then says it has no analyzable requests. + +## Limits + +- Match application, date bounds, filters and retention in both inputs. Use one + LiteLLM database scope and allow its asynchronous writers to settle before + exporting completed days. A shared database snapshot aligns reads, not the + completion times of separate writes. +- The daily table is a second accounting path, not an independent request + oracle. Logging configuration, retention and lost writes can affect either + path. Matching counts cannot reveal requests absent from both, establish + complete token usage, or certify a provider invoice. A shortfall identifies + records to investigate, not their cause. +- In the pinned runtime, disabling failure logging (`disable_error_logs=true`) + omitted the failed request from **both** tables. Successful response-cache + hits counted in both despite making no provider call. These are logged + spend-transaction counts, not universal ingress or provider-send counts. +- The projection and runtime definitions are pinned to LiteLLM **v1.104.0** in + [images.json](../examples/litellm-spend/images.json). Recheck other versions, + logging configurations and routes before interpreting their counts. This + check compares request records, not provider sends or retry attempts. +- Daily inputs are capped at 3,660 dates, 1 MiB and 512 bytes per CSV line. + Duplicate dates, malformed counts and extra columns fail with source-safe + errors. Dates with no daily row remain unknown rather than being filled in. + +See [verification](../examples/litellm-spend/TESTING.md) and the +[request-level accounting contract](../examples/litellm-spend/README.md#data-and-limits). diff --git a/examples/litellm-spend/README.md b/examples/litellm-spend/README.md index cdb7dc4..b766175 100644 --- a/examples/litellm-spend/README.md +++ b/examples/litellm-spend/README.md @@ -7,7 +7,7 @@ increase, and two keys account for all of it.** The report separates request count from request size where the usage records support that comparison. -Install [fleetdiff v0.6.0](../../README.md#2-install-v060), then run this fixture +Install [fleetdiff v0.8.0](../../README.md#2-install-v080), then run this fixture from the repository root with the verified binary: ```sh @@ -79,6 +79,9 @@ Use `--group-by key|team|user|end-user|session`; model breakdowns are automatic. set your own strong secret in an environment variable and pass `--hash-secret-env NAME`. Add `--show-hashes` to display them. +To check for rows missing relative to LiteLLM's daily counters, use the optional +[daily row coverage check](../../docs/ROW_COVERAGE.md) (unreleased). + ### Usage Details (Default On) The SQL includes these three projection aliases by default. Add diff --git a/examples/litellm-spend/TESTING.md b/examples/litellm-spend/TESTING.md index cb099da..42379d5 100644 --- a/examples/litellm-spend/TESTING.md +++ b/examples/litellm-spend/TESTING.md @@ -60,6 +60,43 @@ manually; do not prune unrelated Docker resources. ## Coverage +### Daily Row Reconciliation + +`test_row_coverage_runtime.py` runs the actual pinned proxy, a loopback mock +provider, and both asynchronous PostgreSQL writers against the image's full +Prisma schema. PostgreSQL has `--network none`; the proxy, mock provider and +client share only that container's loopback namespace. No paid provider, +external network, source-only inference or directly inserted spend/daily row +is used. Request clock references are controlled to test midnight; persistence +and accounting code are unchanged. Error logging is tested in its default, +explicitly enabled and disabled configurations. + +| Runtime case | SpendLogs row increment | api_requests | successful_requests | failed_requests | +| --- | ---: | ---: | ---: | ---: | +| Successful non-streaming request | 1 | 1 | 1 | 0 | +| Successful streaming request | 1 | 1 | 1 | 0 | +| Response-cache hit (no provider forward) | 1 | 1 | 1 | 0 | +| Failed request, default/enabled error logging | 1 | 1 | 0 | 1 | +| Failed request, error logging disabled | 0 | 0 | 0 | 0 | +| Two forwards with the same provider response ID, combined | 1 | 2 | 2 | 0 | +| Request with no user identity | 1 | 1 | 1 | 0 | + +A request starting before midnight and ending afterward increments the start +day's UTC counters. The second collision request returns HTTP 200, reaches the +provider and increments the daily table while its duplicate SpendLogs primary +key loses a row. The [recorded fixture](coverage_runtime_expected.json) binds +the observations and synthetic exports to image, source, SQL and harness hashes. +`coverage_runtime_fixture.py` checks the exported fixture; `test_row_coverage.py` +checks both CLI modes, unchanged accounting and source-safe output. The optional +daily SQL export uses a SELECT-only role and the same read-only snapshot as the +request export. Omitting the CLI option preserves the v0.8.0 output goldens. + +See [row coverage and its limits](../../docs/ROW_COVERAGE.md). These runtime +cases pin one chat-completions profile with retries disabled, not every endpoint +or deployment configuration. + +### Request-Level Accounting + - The original 20-row fixture has identical CSV, JSON, and JSONL records. Request, token, zero-only, and failure arithmetic is checked separately from the actual importer output. diff --git a/examples/litellm-spend/coverage_runtime_daily.csv b/examples/litellm-spend/coverage_runtime_daily.csv new file mode 100644 index 0000000..0ab2631 --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_daily.csv @@ -0,0 +1,3 @@ +date,api_requests,successful_requests,failed_requests +2026-10-07,9,8,1 +2026-10-08,1,1,0 diff --git a/examples/litellm-spend/coverage_runtime_daily.csv.license b/examples/litellm-spend/coverage_runtime_daily.csv.license new file mode 100644 index 0000000..501395f --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_daily.csv.license @@ -0,0 +1,3 @@ +SPDX-License-Identifier: Apache-2.0 +SPDX-FileCopyrightText: 2026 Vijay and Codex +Fresh synthetic counters from the same pinned offline runtime database, exported by export-coverage.sh. diff --git a/examples/litellm-spend/coverage_runtime_expected.json b/examples/litellm-spend/coverage_runtime_expected.json new file mode 100644 index 0000000..c24bec5 --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_expected.json @@ -0,0 +1,266 @@ +{ + "license": "Apache-2.0", + "synthetic": true, + "provenance": { + "images": { + "litellm": "ghcr.io/berriai/litellm@sha256:625981c83410a3ea68eb0697590a57ec1d764d634514d54fa5db0591077ee839", + "litellm_version": "1.104.0", + "litellm_source_commit": "79645770fedc7ec2627e6468d31062f20f82aecc", + "schema_sha256": "dbb02f20858ed397ea83b0dfabe2d3cfe59ef658976fdcde373c5baad33df387", + "writer_sha256": "e4e468859ad5b42a17a1683782a0bb7671c7122d519801c25b29e088a20e280b", + "postgres": "postgres@sha256:3645570cccdfa447589da9f57dd740faa29b30938e861289a5574b6ca6b03826", + "postgres_version": "17.11", + "postgres_source_commit": "2603e26e245e558218728ee14e0a42dcb020dc7f", + "postgres_resolved_tag": "17-bookworm", + "postgres_verified_date": "2026-10-10" + }, + "installed": { + "version": "1.104.0", + "source_sha256": { + "schema": "dbb02f20858ed397ea83b0dfabe2d3cfe59ef658976fdcde373c5baad33df387", + "writer": "e4e468859ad5b42a17a1683782a0bb7671c7122d519801c25b29e088a20e280b", + "db_writer": "c48dbdf704dba3fccb2c9e392e0c6c66ecda46d38c01ce84bd1251bc8e48fde6", + "daily_upsert": "54f104fee52e9d1011fac17c815555858627f4967b38b97ba3c34883781b9690", + "request_utils": "86ae14eaecc554e7dedc1b807d534b7dd16f09761e39b51d096fdf178759f0eb", + "logging": "28f8f13a0913d01f09e1b641d11ec53ed4d7b8696f07589c8a79a704de59b82f", + "request_processing": "c68c5007d120c5923d9f249aafe988b8e9b4aa206df5a7e683526c4cf766facd", + "cost_callback": "4dac983d1fd0881dea462b72603cfdb3ecdd91dbd3704e431642839894f7555f", + "caching": "618217a2cda110f4e59a148c684ec463fa7ded7e565131e512f21b3a15ec9974" + } + }, + "export_source_sha256": { + "export.sql": "4324cbbdb0c538d24273621a0138a5a31f757e64413e3d813aafec06ee924b17", + "daily-projection.sql": "f72e96674a760b7ccf2664bc5f11f5be38af935be99a72015f0dd5476f36f9e5", + "export-coverage.sh": "d44304d135a6785358503ed16b04782f8eb31034b6477519cb642e11256b7688" + }, + "runtime_harness_sha256": { + "coverage_runtime_probe.py": "b629b42453080a453e7b47e5878b44ce9ff0b549716a357f6dfdf20bb458f951", + "test_row_coverage_runtime.py": "5dcabcc935403777dd5a669c974889c49ea6d1cdf031ff08c640b16f59b337b5", + "coverage_runtime_pg_service.conf": "1477e5d37ec583fd693d585666af6ba67e754178a834605f6b9a5105558d42bd" + }, + "fixture_generator_sha256": "c85c13d53bdd7a15decf512702758a28f3bee71903eca8ec83657bd37b1423fb", + "export_helper": "export-coverage.sh; one psql invocation with inherited distinct file descriptors", + "export_safety": { + "reader_role": "fixture_reader", + "default_transaction_read_only": true, + "file_modes": { + "directory": "700", + "spend.csv": "600", + "daily.csv": "600", + "private-psql.out": "600", + "private-psql.err": "600" + }, + "distinct_output_inodes": true, + "rejected_before_output": [ + "path_alias", + "literal_paths", + "same_descriptor", + "same_descriptor_leading_zero", + "daily_jsonl" + ], + "existing_directory_refused": true, + "directory_symlink_refused": true, + "dangling_symlink_refused": true, + "private_diagnostics": true + }, + "isolation": "PostgreSQL network=none; proxy, provider and client share its loopback namespace; no published ports", + "clock_control": "LiteLLM request timing only; no persistence patch", + "persistence": "Unmodified full image schema, proxy callbacks and asynchronous DB writers; no inserted spend or daily rows", + "failure_logging": "default (disable_error_logs omitted)" + }, + "files": { + "spend": { + "path": "coverage_runtime_spend.csv", + "sha256": "7ebb6dd44eb1bad22c0d0b5843d19e47652a167d05d51498e7deeb5fbee1fb5a" + }, + "daily": { + "path": "coverage_runtime_daily.csv", + "sha256": "d2950abecf25174677be4aa465ce29977ebaa4113e331e520c323b1b74559f1b" + } + }, + "cases": [ + { + "case": "success", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "failure", + "http_status": 500, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 0, + "failed_requests": 1 + } + }, + { + "case": "stream", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "cache_fill", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "cache_hit", + "http_status": 200, + "provider_forwards": 0, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "collision_first", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "collision_second", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 0, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "absent_user", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "cross_midnight", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + { + "case": "after_midnight", + "http_status": 200, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + } + ], + "expected": { + "proxy_requests": 10, + "provider_forwards": 9, + "spend_rows": 9, + "api_requests": 10, + "successful_requests": 9, + "failed_requests": 1, + "days": { + "2026-10-07": { + "spend_rows": 8, + "api_requests": 9, + "successful_requests": 8, + "failed_requests": 1 + }, + "2026-10-08": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 1, + "failed_requests": 0 + } + }, + "collision": { + "upstream_response_id": "chatcmpl-coverage-collision", + "provider_forwards": 2, + "spend_rows": 1, + "api_requests": 2 + }, + "cross_midnight": { + "request_id": "chatcmpl-coverage-cross_midnight", + "startTime": "2026-10-07T23:59:59.000Z", + "endTime": "2026-10-08T00:00:01.000Z", + "daily_date": "2026-10-07" + }, + "failure_logging_profiles": { + "default": { + "http_status": 500, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 0, + "failed_requests": 1 + } + }, + "enabled": { + "http_status": 500, + "provider_forwards": 1, + "delta": { + "spend_rows": 1, + "api_requests": 1, + "successful_requests": 0, + "failed_requests": 1 + } + }, + "disabled": { + "http_status": 500, + "provider_forwards": 1, + "delta": { + "spend_rows": 0, + "api_requests": 0, + "successful_requests": 0, + "failed_requests": 0 + } + } + } + }, + "limits": [ + "One pinned version and synthetic chat-completions profile, retries disabled", + "Daily counters are spend-transaction counts, not universal ingress or provider-call counts", + "Identity-free daily sums require the same application and operation population as the export", + "Identical missing records on both sides, delayed writes, retention and configuration differences remain distinct limitations", + "Only the request clock is controlled; real transport latency is not measured" + ] +} diff --git a/examples/litellm-spend/coverage_runtime_expected.json.license b/examples/litellm-spend/coverage_runtime_expected.json.license new file mode 100644 index 0000000..a2ed2c4 --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_expected.json.license @@ -0,0 +1,3 @@ +SPDX-License-Identifier: Apache-2.0 +SPDX-FileCopyrightText: 2026 Vijay and Codex +Expected counts, guarded-export checks, and exact source hashes from a fresh synthetic offline runtime run. diff --git a/examples/litellm-spend/coverage_runtime_fixture.py b/examples/litellm-spend/coverage_runtime_fixture.py new file mode 100644 index 0000000..a247252 --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_fixture.py @@ -0,0 +1,119 @@ +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +"""Publish/check the small synthetic exports from an accepted offline runtime run. + +python3 -B coverage_runtime_fixture.py --record RUN_DIR +python3 -B coverage_runtime_fixture.py +Only the allowlisted exports and compact provenance are published, not proxy logs. +""" +import argparse +import csv +import hashlib +import json +from pathlib import Path +import shutil + +ROOT = Path(__file__).resolve().parent +COUNTERS = ("api_requests", "successful_requests", "failed_requests") + + +def digest(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def read_csv(path): + with path.open(newline="") as handle: + return list(csv.DictReader(handle)) + + +def record(source): + observed = json.loads((source / "default/observed.json").read_text()) + assert observed["export_rows"] == 9 + indexed = {case["case"]: case for case in observed["cases"]} + first, second = indexed["collision_first"], indexed["collision_second"] + assert first["response_ids"] == second["response_ids"] == ["chatcmpl-coverage-collision"] + assert len(first["provider_calls"]) == len(second["provider_calls"]) == 1 + assert first["delta"]["spend_rows"] == 1 and second["delta"]["spend_rows"] == 0 + assert first["delta"]["api_requests"] == second["delta"]["api_requests"] == 1 + failures = {} + for profile in ("default", "enabled", "disabled"): + run = json.loads((source / profile / "observed.json").read_text()) + assert run["export_safety"] == observed["export_safety"] + case = next(case for case in run["cases"] if case["case"] == "failure") + failures[profile] = {"http_status": case["http_status"], + "provider_forwards": len(case["provider_calls"]), "delta": case["delta"]} + files = {"spend": "coverage_runtime_spend.csv", "daily": "coverage_runtime_daily.csv"} + shutil.copyfile(source / "default/spend-with-daily.csv", ROOT / files["spend"]) + shutil.copyfile(source / "default/daily.csv", ROOT / files["daily"]) + expected = { + "license": "Apache-2.0", "synthetic": True, + "provenance": { + "images": observed["images"], "installed": observed["installed"], + "export_source_sha256": observed["export_source_sha256"], + "runtime_harness_sha256": observed["harness_sha256"], + "fixture_generator_sha256": digest(Path(__file__)), + "export_helper": "export-coverage.sh; one psql invocation with inherited distinct file descriptors", + "export_safety": observed["export_safety"], + "isolation": "PostgreSQL network=none; proxy, provider and client share its loopback namespace; no published ports", + "clock_control": observed["clock_control"], + "persistence": "Unmodified full image schema, proxy callbacks and asynchronous DB writers; no inserted spend or daily rows", + "failure_logging": "default (disable_error_logs omitted)", + }, + "files": {kind: {"path": name, "sha256": digest(ROOT / name)} for kind, name in files.items()}, + "cases": [{"case": case["case"], "http_status": case["http_status"], + "provider_forwards": len(case["provider_calls"]), "delta": case["delta"]} + for case in observed["cases"]], + "expected": { + "proxy_requests": 10, "provider_forwards": 9, "spend_rows": 9, + "api_requests": 10, "successful_requests": 9, "failed_requests": 1, + "days": {"2026-10-07": {"spend_rows": 8, "api_requests": 9, "successful_requests": 8, + "failed_requests": 1}, + "2026-10-08": {"spend_rows": 1, "api_requests": 1, "successful_requests": 1, + "failed_requests": 0}}, + "collision": {"upstream_response_id": "chatcmpl-coverage-collision", "provider_forwards": 2, + "spend_rows": 1, "api_requests": 2}, + "cross_midnight": {"request_id": "chatcmpl-coverage-cross_midnight", + "startTime": "2026-10-07T23:59:59.000Z", "endTime": "2026-10-08T00:00:01.000Z", + "daily_date": "2026-10-07"}, + "failure_logging_profiles": failures, + }, + "limits": ["One pinned version and synthetic chat-completions profile, retries disabled", + "Daily counters are spend-transaction counts, not universal ingress or provider-call counts", + "Identity-free daily sums require the same application and operation population as the export", + "Identical missing records on both sides, delayed writes, retention and configuration differences remain distinct limitations", + "Only the request clock is controlled; real transport latency is not measured"], + } + (ROOT / "coverage_runtime_expected.json").write_text(json.dumps(expected, indent=2) + "\n") + + +def check(): + fixture = json.loads((ROOT / "coverage_runtime_expected.json").read_text()) + for item in fixture["files"].values(): + assert digest(ROOT / item["path"]) == item["sha256"] + spend = read_csv(ROOT / fixture["files"]["spend"]["path"]) + daily = read_csv(ROOT / fixture["files"]["daily"]["path"]) + assert tuple(daily[0]) == ("date", *COUNTERS) + assert len(spend) == fixture["expected"]["spend_rows"] + for day, counts in fixture["expected"]["days"].items(): + assert sum(row["startTime"].startswith(day) for row in spend) == counts["spend_rows"] + row = next(row for row in daily if row["date"] == day) + for field in COUNTERS: + assert int(row[field]) == counts[field] + for field in COUNTERS: + assert sum(int(row[field]) for row in daily) == fixture["expected"][field] + collision = fixture["expected"]["collision"] + assert sum(row["request_id"] == collision["upstream_response_id"] for row in spend) == 1 + crossing = fixture["expected"]["cross_midnight"] + row = next(row for row in spend if row["request_id"] == crossing["request_id"]) + assert row["startTime"] == crossing["startTime"] and row["endTime"] == crossing["endTime"] + assert next(row for row in spend if row["request_id"] == "chatcmpl-coverage-absent_user")["user"] == "" + print("PASS: 9 persisted SpendLogs rows vs 10 daily API requests; real two-forward response-ID collision; start-UTC day buckets") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--record", type=Path) + args = parser.parse_args() + if args.record: + record(args.record) + check() diff --git a/examples/litellm-spend/coverage_runtime_pg_service.conf b/examples/litellm-spend/coverage_runtime_pg_service.conf new file mode 100644 index 0000000..505224c --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_pg_service.conf @@ -0,0 +1,9 @@ +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +# Disposable network-none fixture only; no credentials or remote host. +[coverage_fixture] +host=/var/run/postgresql +port=5432 +dbname=fixture +user=fixture_reader +options=-c default_transaction_read_only=on diff --git a/examples/litellm-spend/coverage_runtime_probe.py b/examples/litellm-spend/coverage_runtime_probe.py new file mode 100644 index 0000000..d1aa82a --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_probe.py @@ -0,0 +1,241 @@ +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +"""Synthetic loopback provider and narrowly clock-controlled real proxy runner.""" +import argparse +import datetime +import hashlib +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import importlib.metadata +import json +import os +from pathlib import Path +import subprocess +import sys +import threading +import time +from types import SimpleNamespace +import urllib.error +import urllib.request + +ROOT = Path("/example") +STATE = Path("/tmp/coverage-state.json") +MASTER = "sk-coverage-synthetic-master" +PROVIDER_CALLS = [] + + +def write_state(state): + temporary = STATE.with_suffix(".tmp") + temporary.write_text(json.dumps(state)) + temporary.replace(STATE) + + +def request(path, body=None, key=MASTER, port=4000): + data = None if body is None else json.dumps(body).encode() + req = urllib.request.Request( + f"http://127.0.0.1:{port}{path}", data=data, + headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=60) as response: + return response.status, response.read(), dict(response.headers) + except urllib.error.HTTPError as error: + return error.code, error.read(), dict(error.headers) + + +class Provider(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + state = json.loads(STATE.read_text()) + case = state["case"] + response_id = "chatcmpl-coverage-" + ("collision" if case.startswith("collision") else case) + PROVIDER_CALLS.append({"case": case, "response_id": response_id, + "stream": bool(body.get("stream"))}) + state["now"] = state["end"] + write_state(state) + if case == "failure": + payload = json.dumps({"error": {"message": "Synthetic provider failure", + "type": "server_error", "code": "mock_failure"}}).encode() + self.send_response(500) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return + usage = {"prompt_tokens": 12, "completion_tokens": 3, "total_tokens": 15} + common = {"id": response_id, "created": 1791331200, "model": "gpt-4o-mini"} + if body.get("stream"): + chunks = [dict(common, object="chat.completion.chunk", choices=[ + {"index": 0, "delta": {"role": "assistant", "content": "synthetic ok"}, + "finish_reason": None}]), + dict(common, object="chat.completion.chunk", choices=[ + {"index": 0, "delta": {}, "finish_reason": "stop"}]), + dict(common, object="chat.completion.chunk", choices=[], usage=usage)] + payload = ("".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + + "data: [DONE]\n\n").encode() + content_type = "text/event-stream" + else: + payload = json.dumps(dict(common, object="chat.completion", usage=usage, + choices=[{"index": 0, "message": {"role": "assistant", "content": "synthetic ok"}, + "finish_reason": "stop"}])).encode() + content_type = "application/json" + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +def proxy(): + # Replace only LiteLLM request timing references, never the DB writer or queries. + # Wall/monotonic clocks used by HTTP, retries, Prisma and the scheduler stay real. + import litellm.utils + import litellm.litellm_core_utils.litellm_logging as logging_module + import litellm.proxy.common_request_processing as processing + import litellm.proxy.hooks.proxy_track_cost_callback as cost_callback + import litellm.caching.caching_handler as caching + + class RequestDateTime(datetime.datetime): + @classmethod + def now(cls, tz=None): + if not STATE.exists(): + return super().now(tz) + value = cls.fromisoformat(json.loads(STATE.read_text())["now"]) + return value.astimezone(tz) if tz is not None else value.replace(tzinfo=None) + + clock_module = SimpleNamespace(**{name: getattr(datetime, name) for name in dir(datetime)}) + clock_module.datetime = RequestDateTime + litellm.utils.datetime = clock_module + logging_module.datetime = clock_module + processing.datetime = RequestDateTime + cost_callback.datetime = RequestDateTime + caching.datetime = clock_module + from litellm.proxy.proxy_cli import run_server + sys.argv = ["litellm", "--config", "/tmp/coverage-config.json", "--host", "127.0.0.1", + "--port", "4000"] + run_server() + + +def serve(): + config = { + "model_list": [{"model_name": "coverage-model", "litellm_params": { + "model": "openai/gpt-4o-mini", "api_base": "http://127.0.0.1:8080/v1", + "api_key": "synthetic-provider-only", "max_retries": 0, "num_retries": 0, + "timeout": 20}}], + "router_settings": {"num_retries": 0}, + "litellm_settings": {"num_retries": 0, "cache": True, + "cache_params": {"type": "local"}, "turn_off_message_logging": True}, + "general_settings": {"master_key": MASTER, "proxy_batch_write_at": 1, + "disable_reset_budget": True}} + failure_logging = os.environ.get("COVERAGE_FAILURE_LOGGING", "default") + if failure_logging != "default": + config["general_settings"]["disable_error_logs"] = failure_logging == "disabled" + Path("/tmp/coverage-config.json").write_text(json.dumps(config)) + server = ThreadingHTTPServer(("127.0.0.1", 8080), Provider) + threading.Thread(target=server.serve_forever, daemon=True).start() + with Path("/tmp/coverage-proxy.log").open("wb") as log: + child = subprocess.Popen([sys.executable, "-B", __file__, "proxy"], stdout=log, stderr=log) + try: + deadline = time.monotonic() + 150 + while True: + if child.poll() is not None: + raise RuntimeError("proxy exited; inspect private coverage-proxy.log") + try: + if request("/health/liveliness")[0] == 200: + break + except OSError: + pass + if time.monotonic() > deadline: + raise RuntimeError("proxy startup timed out") + time.sleep(0.5) + Path("/tmp/coverage-ready").touch() + # A separate loopback control endpoint owns the mock oracle, never DB data. + class Control(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + payload = json.dumps(PROVIDER_CALLS).encode() + self.send_response(200) + self.end_headers() + self.wfile.write(payload) + ThreadingHTTPServer(("127.0.0.1", 8081), Control).serve_forever() + finally: + child.terminate() + try: + child.wait(timeout=20) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + server.shutdown() + + +def exercise(case): + start = "2026-10-07T12:00:00+00:00" + end = "2026-10-07T12:00:01+00:00" + if case == "cross_midnight": + start, end = "2026-10-07T23:59:59+00:00", "2026-10-08T00:00:01+00:00" + elif case == "after_midnight": + start, end = "2026-10-08T00:00:01+00:00", "2026-10-08T00:00:02+00:00" + write_state({"case": case, "now": start, "end": end}) + key_path = Path("/tmp/coverage-keys.json") + if not key_path.exists(): + keys = {} + for label, body in (("user", {"user_id": "coverage-user"}), ("absent", {})): + status, response, _ = request("/key/generate", dict(body, models=["coverage-model"])) + if status != 200: + raise RuntimeError(f"synthetic key creation failed: {status}: {response.decode()}") + keys[label] = json.loads(response)["key"] + key_path.write_text(json.dumps(keys)) + keys = json.loads(key_path.read_text()) + before = json.loads(request("/", port=8081)[1]) + cache_case = case in ("cache_fill", "cache_hit") + body = {"model": "coverage-model", "messages": [{"role": "user", "content": + "synthetic cache" if cache_case else "synthetic " + case}], "temperature": 0} + if not cache_case: + body["cache"] = {"no-cache": True, "no-store": True} + if case == "stream": + body.update(stream=True, stream_options={"include_usage": True}) + status, response, headers = request("/v1/chat/completions", body, + key=keys["absent" if case == "absent_user" else "user"]) + after = json.loads(request("/", port=8081)[1]) + response_ids = [] + if status == 200: + if case == "stream": + for line in response.decode().splitlines(): + if line.startswith("data: ") and line != "data: [DONE]": + response_ids.append(json.loads(line[6:])["id"]) + else: + response_ids.append(json.loads(response)["id"]) + print(json.dumps({"case": case, "http_status": status, + "start_clock": start, "end_clock": end, + "provider_calls": after[len(before):], + "response_ids": sorted(set(response_ids)), + "cache_header": headers.get("x-litellm-cache-hit"), + "stream_done": case == "stream" and b"data: [DONE]" in response})) + + +def pins(): + base = Path(importlib.metadata.distribution("litellm").locate_file("litellm")) + files = {"schema": "proxy/schema.prisma", "writer": "proxy/spend_tracking/spend_tracking_utils.py", + "db_writer": "proxy/db/db_spend_update_writer.py", + "daily_upsert": "proxy/db/daily_spend_bulk_upsert.py", + "request_utils": "utils.py", "logging": "litellm_core_utils/litellm_logging.py", + "request_processing": "proxy/common_request_processing.py", + "cost_callback": "proxy/hooks/proxy_track_cost_callback.py", + "caching": "caching/caching_handler.py"} + print(json.dumps({"version": importlib.metadata.version("litellm"), + "source_sha256": {key: hashlib.sha256((base / path).read_bytes()).hexdigest() + for key, path in files.items()}})) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("mode", choices=("serve", "proxy", "exercise", "pins")) + parser.add_argument("case", nargs="?") + args = parser.parse_args() + if args.mode == "exercise": + exercise(args.case) + else: + globals()[args.mode]() diff --git a/examples/litellm-spend/coverage_runtime_spend.csv b/examples/litellm-spend/coverage_runtime_spend.csv new file mode 100644 index 0000000..3e408af --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_spend.csv @@ -0,0 +1,10 @@ +request_id,call_type,api_key,user,team_id,organization_id,end_user,model,model_id,model_group,custom_llm_provider,prompt_tokens,completion_tokens,total_tokens,spend,status,session_id,agent_id,rejection_kind,startTime,endTime,cache_read_input_tokens,cache_write_input_tokens,reasoning_output_tokens +3199c0e4-129f-4abe-8b86-125cb553dd9b,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,13,0,13,0,failure,4d01ea73-d124-405b-955f-a8440b3b6a45,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,, +chatcmpl-coverage-absent_user,acompletion,7a8dad512531765ebf78ccfd53f662a9fcd19dd2e661a5527a41a0170c474e44,"","","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,2e992118-9e78-4bf9-bb02-09f36843bde3,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,, +chatcmpl-coverage-cache_fill,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,9089078c-500e-4f7b-8aa1-a5bc539e1ce0,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,, +chatcmpl-coverage-cache_fill_cache_hit1791997595.4835541,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,0,success,cd6ce630-6658-4854-a6a4-832c8e50091f,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:00.000Z,,, +chatcmpl-coverage-collision,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,cc9445b3-47ca-420d-9781-ebe689d687de,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,, +chatcmpl-coverage-stream,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,c6343ad6-2196-4158-b4fb-f63f97e693c3,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,,0 +chatcmpl-coverage-success,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,752e686b-4ac3-4148-8da6-ffc71a72c20c,,,2026-10-07T12:00:00.000Z,2026-10-07T12:00:01.000Z,,, +chatcmpl-coverage-cross_midnight,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,5c34cdea-6b13-4ef7-bd5a-a8c49f698ab1,,,2026-10-07T23:59:59.000Z,2026-10-08T00:00:01.000Z,,, +chatcmpl-coverage-after_midnight,acompletion,39dcf56c6a879ccf4f7f341b1f4dac26f25f0a14048c93fb262e451febbb5c26,coverage-user,"","","",openai/gpt-4o-mini,b839acdde25eef275df507f390befc41d80285472acf1cfc8698eb19ab7f0ac7,coverage-model,openai,12,3,15,3.6e-06,success,38059217-ff89-4f42-865c-430ed089c042,,,2026-10-08T00:00:01.000Z,2026-10-08T00:00:02.000Z,,, diff --git a/examples/litellm-spend/coverage_runtime_spend.csv.license b/examples/litellm-spend/coverage_runtime_spend.csv.license new file mode 100644 index 0000000..0282ed0 --- /dev/null +++ b/examples/litellm-spend/coverage_runtime_spend.csv.license @@ -0,0 +1,3 @@ +SPDX-License-Identifier: Apache-2.0 +SPDX-FileCopyrightText: 2026 Vijay and Codex +Fresh synthetic data from the pinned offline mock-provider runtime, exported by export-coverage.sh. diff --git a/examples/litellm-spend/daily-projection.sql b/examples/litellm-spend/daily-projection.sql new file mode 100644 index 0000000..4c42a8f --- /dev/null +++ b/examples/litellm-spend/daily-projection.sql @@ -0,0 +1,15 @@ +-- SPDX-License-Identifier: Apache-2.0 +-- Code authors: Vijay and Codex +-- Included by export.sql inside its read-only, repeatable-read transaction. +-- Sum all user/model/key rows. Do not export identifiers or synthesize absent days. +COPY ( + SELECT date, + sum(api_requests)::bigint AS api_requests, + sum(successful_requests)::bigint AS successful_requests, + sum(failed_requests)::bigint AS failed_requests + FROM public."LiteLLM_DailyUserSpend" + WHERE date >= to_char(:'start_utc'::timestamptz, 'YYYY-MM-DD') + AND date < to_char(:'end_utc'::timestamptz, 'YYYY-MM-DD') + GROUP BY date +) TO STDOUT WITH (FORMAT CSV, HEADER true) +\g :daily_csv_file diff --git a/examples/litellm-spend/export-coverage.sh b/examples/litellm-spend/export-coverage.sh new file mode 100644 index 0000000..8f92b12 --- /dev/null +++ b/examples/litellm-spend/export-coverage.sh @@ -0,0 +1,36 @@ +#!/bin/sh +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +set -eu +set -C +umask 077 +if [ "$#" -lt 3 ] || [ "$#" -gt 4 ]; then + printf '%s\n' 'Usage: sh export-coverage.sh NEW_DIRECTORY START_UTC END_UTC [PG_SERVICE]' >&2 + exit 2 +fi +directory=$1 +start=$2 +end=$3 +service=${4:-litellm_readonly} +script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +if ! command -v psql >/dev/null 2>&1; then + printf '%s\n' 'Install psql before exporting.' >&2 + exit 1 +fi +# A new private directory and exclusive opens prevent overwriting existing exports. +# psql receives handles, not caller-selected paths that could alias one another. +if ! mkdir -m 700 -- "$directory" 2>/dev/null; then + printf '%s\n' 'Choose a new output directory with an existing private parent.' >&2 + exit 1 +fi +exec 3>"$directory/spend.csv" 4>"$directory/daily.csv" +if ! psql "service=$service" -Xq -v ON_ERROR_STOP=1 \ + -v "start_utc=$start" -v "end_utc=$end" \ + -v csv_file=/dev/fd/3 -v daily_csv_file=/dev/fd/4 \ + -f "$script_dir/export.sql" "$directory/private-psql.out" 2>"$directory/private-psql.err"; then + printf '%s\n' 'Coverage export failed; inspect private-psql.err in the new output directory.' >&2 + exit 1 +fi +exec 3>&- 4>&- +printf '%s\n' 'Exported spend.csv and daily.csv in the new private output directory.' diff --git a/examples/litellm-spend/export.sql b/examples/litellm-spend/export.sql index 77d19b9..b6191f4 100644 --- a/examples/litellm-spend/export.sql +++ b/examples/litellm-spend/export.sql @@ -23,6 +23,29 @@ SELECT 1 / CASE WHEN THEN 1 ELSE 0 END AS valid_bounds \gset +-- Optional coverage is invoked by export-coverage.sh using distinct inherited +-- descriptors. Do not accept arbitrary filenames that might alias each other. +\if :{?daily_csv_file} +SELECT 1 / CASE WHEN + :'start_utc'::timestamptz = date_trunc('day', :'start_utc'::timestamptz) + AND :'end_utc'::timestamptz = date_trunc('day', :'end_utc'::timestamptz) + AND :'end_utc'::timestamptz - :'start_utc'::timestamptz <= interval '3660 days' + THEN 1 ELSE 0 END AS valid_daily_bounds +\gset +\if :{?jsonl_file} +-- The guarded helper exports CSV only. +SELECT 1 / 0 AS unsupported_daily_jsonl +\gset +\else +SELECT 1 / CASE WHEN + :'daily_csv_file' ~ '^/dev/fd/[0-9]+$' + AND :'csv_file' ~ '^/dev/fd/[0-9]+$' + AND substring(:'daily_csv_file' FROM 9)::bigint <> substring(:'csv_file' FROM 9)::bigint + THEN 1 ELSE 0 END AS separate_daily_file +\gset +\endif +\endif + -- Default-on, pinned-writer paths only; usage_details=false omits these columns. -- Missing details stay NULL, never zero. SELECT CASE WHEN :'usage_details'::boolean THEN $details$ @@ -88,4 +111,7 @@ SELECT row_to_json(r) FROM (:export_query) AS r COPY (:export_query) TO STDOUT WITH (FORMAT CSV, HEADER true) \g :csv_file \endif +\if :{?daily_csv_file} +\ir daily-projection.sql +\endif COMMIT; diff --git a/examples/litellm-spend/test_docs.py b/examples/litellm-spend/test_docs.py index 0ac9ef2..3da7052 100644 --- a/examples/litellm-spend/test_docs.py +++ b/examples/litellm-spend/test_docs.py @@ -67,6 +67,7 @@ def test_document_links(self): ROOT / "docs/releases/v0.6.0.md", ROOT / "docs/releases/v0.7.0.md", ROOT / "docs/releases/v0.8.0.md", ROOT / "docs/SESSION_BUDGET.md", ROOT / "docs/CACHE.md", Path(__file__).with_name("README.md")] + paths.append(ROOT / "docs/ROW_COVERAGE.md") for path in paths: for link in re.findall(r"\]\(([^)\s]+)\)", path.read_text()): parsed = urlsplit(link) diff --git a/examples/litellm-spend/test_row_coverage.py b/examples/litellm-spend/test_row_coverage.py new file mode 100644 index 0000000..fd1f35f --- /dev/null +++ b/examples/litellm-spend/test_row_coverage.py @@ -0,0 +1,134 @@ +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +"""Pinned dropped-row fixture through the built local CLI; no runtime required.""" +import csv +import json +import os +from pathlib import Path +import tempfile +import unittest + +import coverage_runtime_fixture +from test_spend import CLI as BIN, ROOT, SECRET, require, run + + +class CoverageExportSafetyTests(unittest.TestCase): + def test_fresh_private_outputs_and_existing_aliases(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + commands = root / "commands" + commands.mkdir() + client = commands / "psql" + client.write_text("#!/bin/sh\nset -eu\nprintf 'request_id,call_type\\n' >&3\n" + "printf 'date,api_requests,successful_requests,failed_requests\\n' >&4\n") + client.chmod(0o700) + env = dict(os.environ, PATH=str(commands) + ":/usr/bin:/bin") + output = root / "PRIVATE output" + command = ["sh", str(ROOT / "export-coverage.sh"), str(output), + "2026-10-07T00:00:00Z", "2026-10-09T00:00:00Z"] + result = run(command, env=env) + require(result.returncode == 0, "guarded export failed") + require(str(output).encode() not in result.stdout + result.stderr, "export path escaped") + self.assertEqual(output.stat().st_mode & 0o777, 0o700) + expected = {name: (output / name).read_bytes() for name in + ("spend.csv", "daily.csv", "private-psql.out", "private-psql.err")} + self.assertNotEqual(expected["spend.csv"], expected["daily.csv"]) + for name in expected: + self.assertEqual((output / name).stat().st_mode & 0o777, 0o600) + alias = root / "PRIVATE_alias" + alias.symlink_to(output, target_is_directory=True) + for destination in (str(output), str(output) + "/.", str(alias)): + retry = command.copy() + retry[2] = destination + result = run(retry, env=env) + require(result.returncode != 0 and not result.stdout, + "existing or aliased output directory accepted") + require(destination.encode() not in result.stderr, "error exposed output path") + self.assertEqual(expected, {name: (output / name).read_bytes() for name in expected}) + + def test_client_diagnostics_stay_private(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + commands = root / "commands" + commands.mkdir() + client = commands / "psql" + client.write_text("#!/bin/sh\nprintf 'PRIVATE_CONNECTION_SENTINEL\\n' >&2\nexit 1\n") + client.chmod(0o700) + output = root / "PRIVATE_output" + result = run(["sh", str(ROOT / "export-coverage.sh"), str(output), + "2026-10-07T00:00:00Z", "2026-10-09T00:00:00Z"], + env=dict(os.environ, PATH=str(commands) + ":/usr/bin:/bin")) + require(result.returncode != 0 and not result.stdout, "client failure ignored") + require(b"PRIVATE_CONNECTION_SENTINEL" not in result.stderr, "client diagnostics escaped") + self.assertIn(b"PRIVATE_CONNECTION_SENTINEL", (output / "private-psql.err").read_bytes()) + + +class RowCoverageTests(unittest.TestCase): + def test_recorded_fixture(self): + coverage_runtime_fixture.check() + expected = json.loads((ROOT / "coverage_runtime_expected.json").read_text()) + for name, digest in expected["provenance"]["export_source_sha256"].items(): + self.assertEqual(coverage_runtime_fixture.digest(ROOT / name), digest) + for name, digest in expected["provenance"]["runtime_harness_sha256"].items(): + self.assertEqual(coverage_runtime_fixture.digest(ROOT / name), digest) + + @unittest.skipUnless(BIN, "set FLEETDIFF_BIN for built CLI tests") + def test_collision_fixture_and_safe_output(self): + expected = json.loads((ROOT / "coverage_runtime_expected.json").read_text()) + path = ROOT / "coverage_runtime_spend.csv" + daily = ROOT / "coverage_runtime_daily.csv" + with path.open(newline="") as source: + rows = list(csv.DictReader(source)) + base = [str(BIN), "investigate", "--litellm-spend", str(path), + "--hash-secret-env", "FLEETDIFF_TEST_SECRET"] + env = dict(os.environ, FLEETDIFF_TEST_SECRET=SECRET) + for mode in ("comparison", "session"): + mode_args = (["--before-period", "2026-10-07", "--after-period", "2026-10-08"] + if mode == "comparison" else ["--period", "2026-10-07"]) + for fmt in ("text", "json"): + command = base + mode_args + ["--format", fmt] + result = run(command + ["--litellm-daily-spend", str(daily)], env=env) + require(result.returncode == 0 and not result.stderr, "coverage CLI failed") + for row in rows: + for field in ("request_id", "api_key", "user", "team_id", "session_id", "agent_id", "model_id", "model"): + if row[field]: + require(row[field].encode() not in result.stdout + result.stderr, + "coverage CLI leaked a raw fixture identifier") + require(str(daily).encode() not in result.stdout + result.stderr, + "coverage CLI leaked an input path") + if fmt == "text": + require(b"2026-10-07 UTC: 8 exported rows; 9 daily requests (8 successful, 1 failed); 1 missing row." in result.stdout, + "missing-row result changed") + require(result.stdout.startswith(b"Spend-log row coverage"), + "coverage was hidden below the usage headline") + else: + report = json.loads(result.stdout) + for day in report["row_coverage"]["days"]: + want = expected["expected"]["days"][day["date_utc"]] + require(day["exported_rows"] == want["spend_rows"], "wrong exported count") + for key in ("api_requests", "successful_requests", "failed_requests"): + require(day[key] == want[key], "wrong daily count") + require(day["missing_rows"] == want["api_requests"] - want["spend_rows"], + "wrong missing-row difference") + plain = run(command, env=env) + require(plain.returncode == 0, "baseline comparison failed") + baseline = json.loads(plain.stdout) + report.pop("row_coverage") + require(report == baseline, "coverage adjusted accounting or identities") + + @unittest.skipUnless(BIN, "set FLEETDIFF_BIN for built CLI tests") + def test_missing_reference_is_unknown(self): + with tempfile.TemporaryDirectory() as directory: + daily = Path(directory) / "PRIVATE_daily.csv" + daily.write_text("date,api_requests,successful_requests,failed_requests\n") + result = run([str(BIN), "investigate", "--litellm-spend", str(ROOT / "coverage_runtime_spend.csv"), + "--before-period", "2026-10-07", "--after-period", "2026-10-08", + "--litellm-daily-spend", str(daily), "--format", "json"]) + require(result.returncode == 0, "unknown coverage failed") + for day in json.loads(result.stdout)["row_coverage"]["days"]: + require(day["api_requests"] is None and day["missing_rows"] is None + and day["status"] == "daily_counts_missing", "missing reference became zero") + + +if __name__ == "__main__": + unittest.main() diff --git a/examples/litellm-spend/test_row_coverage_runtime.py b/examples/litellm-spend/test_row_coverage_runtime.py new file mode 100644 index 0000000..77182bd --- /dev/null +++ b/examples/litellm-spend/test_row_coverage_runtime.py @@ -0,0 +1,330 @@ +# SPDX-License-Identifier: Apache-2.0 +# Code authors: Vijay and Codex +"""Opt-in pinned proxy -> loopback provider -> real PostgreSQL coverage gate. + +FLEETDIFF_SPEND_DOCKER=1 python3 -B test_row_coverage_runtime.py -v +Optional FLEETDIFF_COVERAGE_OUTPUT saves fresh synthetic evidence to a new directory. +No source-only check or manually inserted spend/daily row counts as runtime evidence. +""" +from contextlib import contextmanager +import csv +import hashlib +import io +import json +import os +from pathlib import Path +import time +import unittest +import uuid + +from test_spend import DOCKER, PINS, ROOT, docker, psql, require, run + +CASES = ("success", "failure", "stream", "cache_fill", "cache_hit", "collision_first", + "collision_second", "absent_user", "cross_midnight", "after_midnight") +SPEND_QUERY = '''SELECT COALESCE(json_agg(t), '[]'::json) FROM ( + SELECT request_id, call_type, "user", "startTime", "endTime", status, cache_hit, + prompt_tokens, completion_tokens, total_tokens, + metadata->>'status' AS metadata_status + FROM public."LiteLLM_SpendLogs" ORDER BY "startTime", request_id) t;''' +DAILY_QUERY = '''SELECT COALESCE(json_agg(t), '[]'::json) FROM ( + SELECT date, user_id, model, custom_llm_provider, endpoint, + api_requests, successful_requests, failed_requests, + prompt_tokens, completion_tokens + FROM public."LiteLLM_DailyUserSpend" ORDER BY date, user_id, model) t;''' + + +def query(container, sql): + return json.loads(psql(container, "-At", "-c", sql)) + + +def totals(snapshot): + return {"spend_rows": len(snapshot["spend"]), **{ + field: sum(row[field] for row in snapshot["daily"]) + for field in ("api_requests", "successful_requests", "failed_requests")}} + + +@contextmanager +def runtime(output, failure_logging="default"): + name = "fleetdiff-coverage-" + uuid.uuid4().hex + gateway, volume = name + "-proxy", name + "-data" + docker("volume", "create", volume) + try: + docker("run", "-d", "--name", name, "--pull=never", "--network=none", "--read-only", + "--tmpfs", "/tmp:rw,noexec,nosuid,size=64m", + "--tmpfs", "/var/run/postgresql:rw,noexec,nosuid,size=16m", + "--mount", f"type=volume,src={volume},dst=/var/lib/postgresql/data", + "--mount", f"type=bind,src={ROOT},dst=/example,readonly", + "--pids-limit=128", "--memory=512m", "--log-driver=none", + "-e", "POSTGRES_DB=fixture", "-e", "POSTGRES_HOST_AUTH_METHOD=trust", + PINS["postgres"], "-c", "listen_addresses=127.0.0.1", + "-c", "log_statement=none", "-c", "log_min_error_statement=panic") + deadline = time.monotonic() + 60 + while True: + process = run(["docker", "exec", name, "cat", "/proc/1/comm"]) + ready = run(["docker", "exec", name, "pg_isready", "-U", "postgres", "-d", "fixture"]) + if process.stdout.strip() == b"postgres" and ready.returncode == 0: + break + require(time.monotonic() < deadline, "runtime PostgreSQL startup timed out") + time.sleep(0.25) + require(PINS["postgres_version"].encode() in docker("exec", name, "postgres", "--version"), + "PostgreSQL pin mismatch") + common = ["--pull=never", "--network=container:" + name, "--read-only", "--cap-drop=ALL", + "--security-opt=no-new-privileges", "--tmpfs", "/tmp:rw,nosuid,size=512m", + "--memory=2g", "--pids-limit=256", "--log-driver=none", + "--mount", f"type=bind,src={ROOT},dst=/example,readonly", + "-e", "DATABASE_URL=postgresql://postgres@127.0.0.1:5432/fixture", + "-e", "HOME=/tmp", "-e", "PYTHONDONTWRITEBYTECODE=1", "-e", "TZ=UTC", + "-e", "LITELLM_LOCAL_MODEL_COST_MAP=True", "-e", "LITELLM_TELEMETRY=False", + "-e", "STORE_MODEL_IN_DB=False", "-e", "DISABLE_SCHEMA_UPDATE=true"] + common.extend(("-e", "COVERAGE_FAILURE_LOGGING=" + failure_logging)) + # Full upstream schema, installed offline CLI and engines. No handwritten DB schema. + result = run(["docker", "run", "--rm", *common, "--entrypoint", "prisma", PINS["litellm"], + "db", "push", "--schema", "/app/schema.prisma", "--skip-generate"], timeout=150) + if output: + (output / "schema-init.log").write_bytes(result.stdout + result.stderr) + require(result.returncode == 0, "offline Prisma schema initialization failed") + docker("run", "-d", "--name", gateway, *common, "--entrypoint", "python", + PINS["litellm"], "-B", "/example/coverage_runtime_probe.py", "serve") + deadline = time.monotonic() + 180 + while run(["docker", "exec", gateway, "/bin/sh", "-c", "test -f /tmp/coverage-ready"]).returncode: + require(time.monotonic() < deadline, "runtime proxy startup timed out") + state = json.loads(docker("inspect", gateway))[0]["State"] + require(state["Running"], "runtime proxy exited before readiness") + time.sleep(0.5) + yield name, gateway + finally: + if output: + result = run(["docker", "exec", gateway, "cat", "/tmp/coverage-proxy.log"]) + if result.returncode == 0: + (output / "proxy.log").write_bytes(result.stdout) + run(["docker", "rm", "-f", gateway]) + run(["docker", "rm", "-f", name]) + require(run(["docker", "volume", "rm", volume]).returncode == 0, + "runtime named volume cleanup failed") + + +class RuntimeCoverageTests(unittest.TestCase): + def export_with_helper(self, database, output): + parent = "/tmp/coverage-export-private" + directory = parent + "/paired" + service_env = ["-e", "PGSERVICEFILE=/example/coverage_runtime_pg_service.conf"] + reader = ["docker", "exec", *service_env, database, "psql", "service=coverage_fixture", + "-Xq", "-v", "ON_ERROR_STOP=1"] + identity = run([*reader, "-At", "-c", '''SELECT current_user, + current_setting('default_transaction_read_only'), + has_table_privilege(current_user, 'public."LiteLLM_SpendLogs"', 'SELECT'), + has_table_privilege(current_user, 'public."LiteLLM_SpendLogs"', 'INSERT'), + has_table_privilege(current_user, 'public."LiteLLM_DailyUserSpend"', 'INSERT');''']) + self.assertEqual(identity.returncode, 0) + self.assertEqual(identity.stdout.strip(), b"fixture_reader|on|t|f|f") + docker("exec", database, "mkdir", "-m", "700", parent) + helper = ["docker", "exec", *service_env, database, "sh", "/example/export-coverage.sh"] + bounds = ["2026-10-07T00:00:00Z", "2026-10-09T00:00:00Z"] + completed = run([*helper, directory, *bounds, "coverage_fixture"]) + if output: + (output / "helper.stdout").write_bytes(completed.stdout) + (output / "helper.stderr").write_bytes(completed.stderr) + self.assertEqual(completed.returncode, 0, "guarded export helper failed") + self.assertFalse(completed.stderr) + self.assertNotIn(directory.encode(), completed.stdout) + names = ("spend.csv", "daily.csv", "private-psql.out", "private-psql.err") + self.assertEqual(sorted(docker("exec", database, "ls", "-1", directory).decode().splitlines()), + sorted(names)) + modes = {"directory": docker("exec", database, "stat", "-c", "%a", directory).decode().strip()} + files = {} + inodes = [] + for name in names: + path = directory + "/" + name + files[name] = docker("exec", database, "cat", path) + modes[name] = docker("exec", database, "stat", "-c", "%a", path).decode().strip() + inodes.append(docker("exec", database, "stat", "-c", "%d:%i", path).decode().strip()) + if output and name.startswith("private-"): + (output / name).write_bytes(files[name]) + self.assertEqual(modes, {"directory": "700", **dict.fromkeys(names, "600")}) + self.assertEqual(len(set(inodes)), len(names)) + self.assertFalse(files["private-psql.out"]) + self.assertFalse(files["private-psql.err"]) + + # The previous overwrite exploit must fail before either output is opened. + rejected = [] + literal = parent + "/alias-output.csv" + for label, arguments in ( + ("path_alias", ["csv_file=" + literal, "daily_csv_file=" + parent + "/./alias-output.csv"]), + ("literal_paths", ["csv_file=" + literal, "daily_csv_file=" + parent + "/other.csv"]), + ("same_descriptor", ["csv_file=/dev/fd/3", "daily_csv_file=/dev/fd/3"]), + ("same_descriptor_leading_zero", ["csv_file=/dev/fd/3", "daily_csv_file=/dev/fd/03"]), + ("daily_jsonl", ["csv_file=/dev/fd/3", "daily_csv_file=/dev/fd/4", + "jsonl_file=" + literal]), + ): + command = [*reader, "-v", "start_utc=" + bounds[0], "-v", "end_utc=" + bounds[1]] + for argument in arguments: + command.extend(("-v", argument)) + result = run([*command, "-f", "/example/export.sql"]) + self.assertNotEqual(result.returncode, 0, label) + self.assertFalse(result.stdout, label) + rejected.append(label) + if output: + (output / (label + ".private.err")).write_bytes(result.stderr) + self.assertEqual(docker("exec", database, "ls", "-1", parent).decode().splitlines(), ["paired"]) + + alias = parent + "/directory-link" + dangling = parent + "/dangling-link" + docker("exec", database, "ln", "-s", directory, alias) + docker("exec", database, "ln", "-s", parent + "/absent-target", dangling) + for target in (directory, alias, dangling): + result = run([*helper, target, *bounds, "coverage_fixture"]) + self.assertNotEqual(result.returncode, 0) + self.assertFalse(result.stdout) + self.assertNotIn(target.encode(), result.stderr) + for name in names: + self.assertEqual(docker("exec", database, "cat", directory + "/" + name), files[name]) + self.assertNotEqual(run(["docker", "exec", database, "stat", parent + "/absent-target"]).returncode, 0) + + # DB diagnostics are retained privately, never copied into the public message. + failed_directory = parent + "/failed-service" + sentinel = "coverage_missing_service_sentinel" + result = run([*helper, failed_directory, *bounds, sentinel]) + self.assertNotEqual(result.returncode, 0) + self.assertFalse(result.stdout) + self.assertNotIn(sentinel.encode(), result.stderr) + self.assertNotIn(failed_directory.encode(), result.stderr) + diagnostic = docker("exec", database, "cat", failed_directory + "/private-psql.err") + self.assertIn(sentinel.encode(), diagnostic) + self.assertEqual(docker("exec", database, "stat", "-c", "%a", + failed_directory + "/private-psql.err").strip(), b"600") + if output: + (output / "failed-service.private.err").write_bytes(diagnostic) + safety = {"reader_role": "fixture_reader", "default_transaction_read_only": True, + "file_modes": modes, "distinct_output_inodes": True, + "rejected_before_output": rejected, "existing_directory_refused": True, + "directory_symlink_refused": True, "dangling_symlink_refused": True, + "private_diagnostics": True} + return files["spend.csv"], files["daily.csv"], safety + + @unittest.skipUnless(DOCKER, "set FLEETDIFF_SPEND_DOCKER=1 for the real pinned offline runtime") + def test_real_proxy_persistence(self): + self.run_cases(CASES, "default") + + @unittest.skipUnless(DOCKER, "set FLEETDIFF_SPEND_DOCKER=1 for the real pinned offline runtime") + def test_failure_logging_enabled(self): + self.run_cases(("success", "failure"), "enabled") + + @unittest.skipUnless(DOCKER, "set FLEETDIFF_SPEND_DOCKER=1 for the real pinned offline runtime") + def test_failure_logging_disabled(self): + self.run_cases(("success", "failure"), "disabled") + + def run_cases(self, cases, failure_logging): + output = (Path(os.environ["FLEETDIFF_COVERAGE_OUTPUT"]) / failure_logging + if os.environ.get("FLEETDIFF_COVERAGE_OUTPUT") else None) + if output: + output.mkdir(mode=0o700, parents=True, exist_ok=False) + evidence = {"images": PINS, "failure_logging": failure_logging, + "clock_control": "LiteLLM request timing only; no persistence patch", + "cases": []} + with runtime(output, failure_logging) as (database, gateway): + pins = json.loads(docker("exec", gateway, "python", "-B", "/example/coverage_runtime_probe.py", "pins")) + self.assertEqual(pins["version"], PINS["litellm_version"]) + for kind in ("schema", "writer"): + self.assertEqual(pins["source_sha256"][kind], PINS[kind + "_sha256"]) + evidence["installed"] = pins + evidence["network"] = {container: json.loads(docker("inspect", container))[0]["HostConfig"]["NetworkMode"] + for container in (database, gateway)} + previous = {"spend_rows": 0, "api_requests": 0, "successful_requests": 0, "failed_requests": 0} + for case in cases: + result = json.loads(docker("exec", gateway, "python", "-B", "/example/coverage_runtime_probe.py", + "exercise", case, timeout=90)) + self.assertEqual(result["http_status"], 500 if case == "failure" else 200, case) + # Both real background writers must settle; no direct writer invocation. + started = time.monotonic() + deadline, stable_since, last = started + 75, None, None + disabled_failure = case == "failure" and failure_logging == "disabled" + expected = {"spend_rows": int(not disabled_failure and case != "collision_second"), + "api_requests": int(not disabled_failure), + "successful_requests": int(case != "failure"), + "failed_requests": int(case == "failure" and not disabled_failure)} + while True: + snapshot = {"spend": query(database, SPEND_QUERY), "daily": query(database, DAILY_QUERY)} + if snapshot == last: + stable_since = stable_since or time.monotonic() + else: + stable_since = None + current = totals(snapshot) + delta = {key: current[key] - previous[key] for key in current} + if (stable_since and time.monotonic() - stable_since >= 8 and delta == expected + and (not disabled_failure or time.monotonic() - started >= 20)): + break + require(time.monotonic() < deadline, "runtime persistence did not settle") + last = snapshot + time.sleep(0.5) + result["delta"] = {key: current[key] - previous[key] for key in current} + self.assertEqual(result["delta"], expected, case) + self.assertEqual(len(result["provider_calls"]), 0 if case == "cache_hit" else 1, case) + if case == "stream": + self.assertTrue(result["stream_done"]) + result["snapshot"] = snapshot + previous = current + evidence["cases"].append(result) + print(json.dumps({"profile": failure_logging, "case": case, "http": result["http_status"], + "forwards": len(result["provider_calls"]), "delta": result["delta"]}), flush=True) + if output: + (output / "observed.json").write_text(json.dumps(evidence, indent=2) + "\n") + if failure_logging == "default": + indexed = {entry["case"]: entry for entry in evidence["cases"]} + self.assertEqual(indexed["collision_first"]["response_ids"], + indexed["collision_second"]["response_ids"]) + self.assertEqual(indexed["cache_fill"]["response_ids"], indexed["cache_hit"]["response_ids"]) + rows = {row["request_id"]: row for row in snapshot["spend"]} + self.assertEqual(rows["chatcmpl-coverage-absent_user"]["user"], "") + self.assertTrue(any(row["user_id"] == "" and row["api_requests"] == 1 + for row in snapshot["daily"])) + boundary = rows["chatcmpl-coverage-cross_midnight"] + self.assertTrue(boundary["startTime"].startswith("2026-10-07T23:59:59")) + self.assertTrue(boundary["endTime"].startswith("2026-10-08T00:00:01")) + self.assertEqual(sum(row["api_requests"] for row in indexed["cross_midnight"]["snapshot"]["daily"] + if row["date"] == "2026-10-08"), 0) + self.assertEqual(sum(row["api_requests"] for row in snapshot["daily"] + if row["date"] == "2026-10-08"), 1) + # Independent SELECT of daily counters and actual repository SpendLogs export. + psql(database, data=b'''CREATE ROLE fixture_reader LOGIN; + GRANT USAGE ON SCHEMA public TO fixture_reader; + GRANT SELECT ON public."LiteLLM_SpendLogs" TO fixture_reader; + GRANT SELECT (date, api_requests, successful_requests, failed_requests) + ON public."LiteLLM_DailyUserSpend" TO fixture_reader;''') + from test_spend import export_database + exports = export_database(database) + exported = list(csv.DictReader(io.StringIO(exports["csv"].decode()))) + evidence["export_rows"] = len(exported) + evidence["export_rows_by_start_day"] = {day: sum(row["startTime"].startswith(day) for row in exported) + for day in ("2026-10-07", "2026-10-08")} + self.assertEqual(len(exported), len(snapshot["spend"])) + spend_bytes, daily_bytes, safety = self.export_with_helper(database, output) + evidence["export_safety"] = safety + daily_export = list(csv.DictReader(io.StringIO(daily_bytes.decode()))) + exported_again = list(csv.DictReader(io.StringIO(spend_bytes.decode()))) + self.assertEqual(sorted(exported, key=lambda r: r["request_id"]), + sorted(exported_again, key=lambda r: r["request_id"])) + independently_summed = {} + for row in snapshot["daily"]: + counters = independently_summed.setdefault(row["date"], dict.fromkeys( + ("api_requests", "successful_requests", "failed_requests"), 0)) + for field in counters: + counters[field] += row[field] + self.assertEqual({row["date"]: {field: int(row[field]) for field in + ("api_requests", "successful_requests", "failed_requests")} for row in daily_export}, + independently_summed) + evidence["daily_export"] = daily_export + evidence["export_source_sha256"] = {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() + for name in ("export.sql", "daily-projection.sql", "export-coverage.sh")} + evidence["harness_sha256"] = {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() + for name in ("coverage_runtime_probe.py", "test_row_coverage_runtime.py", + "coverage_runtime_pg_service.conf")} + if output: + for fmt, data in exports.items(): + (output / ("spend." + fmt)).write_bytes(data) + (output / "daily.csv").write_bytes(daily_bytes) + (output / "spend-with-daily.csv").write_bytes(spend_bytes) + (output / "observed.json").write_text(json.dumps(evidence, indent=2) + "\n") + + +if __name__ == "__main__": + unittest.main() diff --git a/internal/cli/row_coverage.go b/internal/cli/row_coverage.go new file mode 100644 index 0000000..15b7917 --- /dev/null +++ b/internal/cli/row_coverage.go @@ -0,0 +1,50 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package cli + +import ( + "fmt" + "io" + + "github.com/llm-measurement/fleetdiff/internal/spend" +) + +func renderRowCoverage(out io.Writer, coverage *spend.RowCoverage) { + if coverage == nil { + return + } + fmt.Fprintln(out, "Spend-log row coverage (relative to LiteLLM daily request counts):") + for _, day := range coverage.Days { + suffix := "s" + if day.ExportedRows == 1 { + suffix = "" + } + fmt.Fprintf(out, " %s UTC: %s exported row%s", day.Date, spendNumber(day.ExportedRows), suffix) + if day.APIRequests != nil { + suffix := "s" + if *day.APIRequests == 1 { + suffix = "" + } + fmt.Fprintf(out, "; %s daily request%s (%s successful, %s failed)", spendNumber(*day.APIRequests), suffix, spendNumber(*day.SuccessfulRequests), spendNumber(*day.FailedRequests)) + } + switch day.Status { + case "matched": + fmt.Fprintln(out, "; counts match.") + case "missing_rows": + suffix := "s" + if *day.MissingRows == 1 { + suffix = "" + } + fmt.Fprintf(out, "; %s missing row%s.\n", spendNumber(*day.MissingRows), suffix) + case "excess_rows": + fmt.Fprintln(out, "; exported rows exceed daily counts; check scope and export timing.") + case "partial_day": + fmt.Fprintln(out, "; partial UTC day selected; coverage unknown.") + case "daily_counts_missing": + fmt.Fprintln(out, "; daily counts not exported; coverage unknown.") + } + } + fmt.Fprintln(out, "Token and spend totals below use exported rows only.") + fmt.Fprintln(out) +} diff --git a/internal/cli/row_coverage_test.go b/internal/cli/row_coverage_test.go new file mode 100644 index 0000000..de86c09 --- /dev/null +++ b/internal/cli/row_coverage_test.go @@ -0,0 +1,105 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package cli + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestRowCoverageCLIAndPrivacy(t *testing.T) { + t.Setenv("ROW_COVERAGE_KEY", "synthetic-row-coverage-sentinel-secret") + dir := t.TempDir() + daily := filepath.Join(dir, "PRIVATE_daily.csv") + const header = "date,api_requests,successful_requests,failed_requests\n" + if err := os.WriteFile(daily, []byte(header+"2026-10-07,1000,900,100\n2026-10-08,1000,1000,0\n"), 0600); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, "PRIVATE_spend.jsonl") + const sentinel = "PRIVATE-identity@example.invalid" + var source bytes.Buffer + enc := json.NewEncoder(&source) + for i, stamp := range []string{"2026-10-07T12:00:00Z", "2026-10-08T12:00:00Z"} { + row := map[string]any{"request_id": sentinel + string(rune('a'+i)), "call_type": "completion", "startTime": stamp, "endTime": stamp, "prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110, "spend": "0.03", "status": "success"} + for _, name := range []string{"api_key", "user", "team_id", "end_user", "session_id", "agent_id", "model"} { + row[name] = sentinel + } + if err := enc.Encode(row); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(path, source.Bytes(), 0600); err != nil { + t.Fatal(err) + } + base := []string{"investigate", "--litellm-spend", path, "--litellm-daily-spend", daily, "--hash-secret-env", "ROW_COVERAGE_KEY"} + for _, format := range []string{"text", "json"} { + for _, group := range []string{"key", "team", "user", "end-user", "session", "session-review"} { + for _, hashes := range []bool{false, true} { + args := append([]string{}, base...) + args = append(args, "--format", format) + if group == "session-review" { + args = append(args, "--period", "2026-10-07", "--session-budget-usd", "1") + } else { + args = append(args, "--before-period", "2026-10-07", "--after-period", "2026-10-08", "--group-by", group) + } + if hashes { + args = append(args, "--show-hashes") + } + var out, errout bytes.Buffer + if code := Run(args, &out, &errout); code != 0 { + t.Fatal(code, errout.String()) + } + for _, secret := range []string{sentinel, daily, path, os.Getenv("ROW_COVERAGE_KEY")} { + if strings.Contains(out.String()+errout.String(), secret) { + t.Fatal("privacy sentinel escaped") + } + } + if format == "text" { + if !strings.HasPrefix(out.String(), "Spend-log row coverage") || !strings.Contains(out.String(), "1 exported row; 1,000 daily requests (900 successful, 100 failed); 999 missing rows.") { + t.Fatal(out.String()) + } + } else { + var report struct { + Coverage struct { + Days []struct { + Missing uint64 `json:"missing_rows"` + } `json:"days"` + } `json:"row_coverage"` + } + if err := json.Unmarshal(out.Bytes(), &report); err != nil || report.Coverage.Days[0].Missing != 999 { + t.Fatal("missing JSON coverage", err) + } + } + } + } + } + for _, body := range []string{header + sentinel + ",1,1,0\n", header + "2026-10-07," + sentinel + ",1,0\n", "date,api_requests," + sentinel + ",failed_requests\n"} { + if err := os.WriteFile(daily, []byte(body), 0600); err != nil { + t.Fatal(err) + } + var out, errout bytes.Buffer + args := append(append([]string{}, base...), "--before-period", "2026-10-07", "--after-period", "2026-10-08") + if code := Run(args, &out, &errout); code == 0 || out.Len() != 0 || strings.Contains(errout.String(), sentinel) || strings.Contains(errout.String(), daily) { + t.Fatal("unsafe error or partial report") + } + } +} + +func TestRowCoverageFlagIsolation(t *testing.T) { + for _, args := range [][]string{ + {"investigate", "--litellm-daily-spend", "PRIVATE.csv"}, + {"compare", "--litellm-spend", "PRIVATE.csv", "--litellm-daily-spend", "PRIVATE.csv"}, + {"investigate", "--litellm-spend", "PRIVATE.csv", "--litellm-daily-spend", ""}, + {"investigate", "--before", "PRIVATE", "--after", "PRIVATE", "--litellm-daily-spend", "PRIVATE.csv"}, + } { + var out, errout bytes.Buffer + if code := Run(args, &out, &errout); code != 2 || out.Len() != 0 || strings.Contains(errout.String(), "PRIVATE") { + t.Fatal(code, errout.String()) + } + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index e913da6..c91ed2c 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -59,6 +59,7 @@ Options: --allow-partial Permit missing producers or incomplete observation intervals --show-hashes Include pseudonymous, linkable hashes in output --litellm-spend FILE Read a request-level SQL export: CSV, JSON array or JSONL + --litellm-daily-spend FILE Compare row counts with the daily SQL-projection CSV --before-period PERIOD UTC date or RFC3339 start/end (requires after-period) --after-period PERIOD UTC date or RFC3339 start/end (equal duration) --group-by KIND key (default), team, user, end-user or session @@ -124,6 +125,7 @@ func RunWithInput(args []string, in io.Reader, out, errout io.Writer) int { partial := flags.Bool("allow-partial", false, "") hashes := flags.Bool("show-hashes", false, "") spendPath := flags.String("litellm-spend", "", "") + dailySpendPath := flags.String("litellm-daily-spend", "", "") beforePeriod := flags.String("before-period", "", "") afterPeriod := flags.String("after-period", "", "") group := flags.String("group-by", "key", "") @@ -148,8 +150,12 @@ func RunWithInput(args []string, in io.Reader, out, errout io.Writer) int { spendMode, incompatible := false, false sessionMode, windowSet, emptySessionOption, twoPeriods := false, false, false, false groupExplicit := false + emptyDailyOption := false flags.Visit(func(f *flag.Flag) { switch f.Name { + case "litellm-daily-spend": + spendMode = true + emptyDailyOption = f.Value.String() == "" case "period", "session-budget-usd": spendMode, sessionMode = true, true emptySessionOption = emptySessionOption || f.Value.String() == "" @@ -168,6 +174,9 @@ func RunWithInput(args []string, in io.Reader, out, errout io.Writer) int { } }) if spendMode { + if emptyDailyOption { + return fail(2, "litellm-daily-spend needs a daily SQL-projection CSV file") + } if args[0] != "investigate" || *spendPath == "" || incompatible || flags.NArg() != 0 { return fail(2, "use investigate --litellm-spend FILE with period flags, not summary paths or expected producers") } @@ -175,9 +184,9 @@ func RunWithInput(args []string, in io.Reader, out, errout io.Writer) int { if !sessionMode || groupExplicit && *group != "session" || twoPeriods || emptySessionOption || *counterWindow <= 0 { return fail(2, "session review needs --period or --session-budget-usd; grouping is session; do not combine with other groups or two-period flags") } - return runSessionBudget(*spendPath, *period, *budget, *secret, *format, *top, *hashes, *counterWindow, out, errout) + return runSessionBudget(*spendPath, *period, *budget, *secret, *format, *dailySpendPath, *top, *hashes, *counterWindow, out, errout) } - return runSpend(*spendPath, *beforePeriod, *afterPeriod, *group, *secret, *format, *top, *hashes, out, errout) + return runSpend(*spendPath, *beforePeriod, *afterPeriod, *group, *secret, *format, *dailySpendPath, *top, *hashes, out, errout) } for _, required := range []struct{ name, value string }{{"before", *before}, {"after", *after}, {"expected", *expected}} { if strings.TrimSpace(required.value) == "" { diff --git a/internal/cli/session_budget.go b/internal/cli/session_budget.go index 75b54d1..2d1e0e1 100644 --- a/internal/cli/session_budget.go +++ b/internal/cli/session_budget.go @@ -14,12 +14,12 @@ import ( "github.com/llm-measurement/fleetdiff/internal/spend" ) -func runSessionBudget(path, period, budget, secret, format string, top int, hashes bool, window time.Duration, out, errout io.Writer) int { +func runSessionBudget(path, period, budget, secret, format, daily string, top int, hashes bool, window time.Duration, out, errout io.Writer) int { fail := func(code int, message string) int { fmt.Fprintln(errout, message); return code } if format != "text" && format != "json" || top < 1 || top > 100 { return fail(2, "use text or json output and top between 1 and 100") } - r, err := spend.ReviewSessions(path, spend.SessionOptions{Period: period, BudgetUSD: budget, SecretEnv: secret, CounterWindow: window, Top: top, ShowHashes: hashes}) + r, err := spend.ReviewSessions(path, spend.SessionOptions{Period: period, BudgetUSD: budget, SecretEnv: secret, CounterWindow: window, Top: top, ShowHashes: hashes, DailySpendPath: daily}) if err != nil { return fail(1, err.Error()) } @@ -48,6 +48,7 @@ func sessionUSD(s string) string { } func renderSessionBudget(out io.Writer, r spend.SessionReport) { + renderRowCoverage(out, r.RowCoverage) if r.UnsupportedRows != 0 { fmt.Fprintf(out, "Session review is incomplete: %s selected records have unsupported call types.\n", spendNumber(r.UnsupportedRows)) } diff --git a/internal/cli/spend.go b/internal/cli/spend.go index c155f8a..d2706c6 100644 --- a/internal/cli/spend.go +++ b/internal/cli/spend.go @@ -15,12 +15,12 @@ import ( "github.com/llm-measurement/fleetdiff/internal/spend" ) -func runSpend(path, before, after, group, secret, format string, top int, hashes bool, out, errout io.Writer) int { +func runSpend(path, before, after, group, secret, format, daily string, top int, hashes bool, out, errout io.Writer) int { fail := func(code int, msg string) int { fmt.Fprintln(errout, msg); return code } if format != "text" && format != "json" || top < 1 || top > 100 { return fail(2, "use text or json output and top between 1 and 100") } - r, err := spend.Investigate(path, spend.Options{BeforePeriod: before, AfterPeriod: after, GroupBy: group, SecretEnv: secret, Top: top, ShowHashes: hashes}) + r, err := spend.Investigate(path, spend.Options{BeforePeriod: before, AfterPeriod: after, GroupBy: group, SecretEnv: secret, Top: top, ShowHashes: hashes, DailySpendPath: daily}) if err != nil { return fail(1, err.Error()) } @@ -41,6 +41,12 @@ func runSpend(path, before, after, group, secret, format string, top int, hashes } func renderSpend(out io.Writer, r spend.Report) { + renderRowCoverage(out, r.RowCoverage) + if r.AnalysisGap != "" { + fmt.Fprintln(out, "No analyzable model requests in either selected period; usage changes cannot be compared.") + fmt.Fprintln(out, "Use explicit periods covering model requests for a usage comparison.") + return + } if r.Incomplete { fmt.Fprintln(out, "Comparison incomplete: some selected call types are not analyzed.") for _, c := range r.CallTypes { @@ -163,6 +169,11 @@ func renderSpend(out io.Writer, r spend.Report) { } fmt.Fprintln(out, "\nHow complete is the export?") problems := r.Incomplete + if r.RowCoverage != nil { + for _, day := range r.RowCoverage.Days { + problems = problems || day.Status != "matched" + } + } if r.Incomplete { fmt.Fprintln(out, " Usage checks below cover analyzed calls only; unsupported token fields above remain unanalyzed.") } diff --git a/internal/spend/investigate.go b/internal/spend/investigate.go index 65fa23c..b548799 100644 --- a/internal/spend/investigate.go +++ b/internal/spend/investigate.go @@ -28,6 +28,7 @@ const MaxModels = 128 type Options struct { BeforePeriod, AfterPeriod, GroupBy, SecretEnv string + DailySpendPath string Top int ShowHashes bool Now time.Time @@ -175,6 +176,10 @@ func Investigate(path string, opts Options) (Report, error) { if err != nil { return Report{}, err } + coverage, err := newCoverage(opts.DailySpendPath, before, after) + if err != nil { + return Report{}, err + } windows := [2]*aggregate{newAggregate(before, true), newAggregate(after, true)} models := map[[32]byte][2]*aggregate{} modelOrder := map[[32]byte]int{} @@ -213,6 +218,7 @@ func Investigate(path string, opts Options) (Report, error) { r.OutsidePeriods++ return nil } + coverage.add(start) if e := recordCallType(&r, callTypes, row, side); e != nil { return e } @@ -295,7 +301,13 @@ func Investigate(path string, opts Options) (Report, error) { return Report{}, errors.New("export changed while reading; use a completed immutable export") } r.Before, r.After = windows[0].result(), windows[1].result() + r.RowCoverage = coverage.result() if r.Before.Requests == 0 && r.After.Requests == 0 { + if r.RowCoverage != nil { + r.Incomplete = true + r.AnalysisGap = "no analyzable model requests in either selected period" + return r, nil + } names := []string{} for _, c := range r.CallTypes { names = append(names, c.Name) diff --git a/internal/spend/report.go b/internal/spend/report.go index 58ae32b..c095de1 100644 --- a/internal/spend/report.go +++ b/internal/spend/report.go @@ -37,6 +37,8 @@ type IncreaseShare struct { } type Report struct { + AnalysisGap string `json:"analysis_unavailable,omitempty"` + RowCoverage *RowCoverage `json:"row_coverage,omitempty"` Cache compare.CacheView `json:"cache"` Incomplete bool `json:"comparison_incomplete"` CallTypes []CallType `json:"call_types"` diff --git a/internal/spend/row_coverage.go b/internal/spend/row_coverage.go new file mode 100644 index 0000000..7aee6f3 --- /dev/null +++ b/internal/spend/row_coverage.go @@ -0,0 +1,204 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package spend + +import ( + "bufio" + "bytes" + "encoding/csv" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strconv" + "time" + + "github.com/llm-measurement/fleetdiff/internal/localfile" +) + +const maxCoverageDays = 3660 +const maxCoverageBytes = 1 << 20 +const maxCoverageLine = 512 + +var errCoverageFile = errors.New("daily spend input must be a readable regular CSV file, not a symlink or special file") +var errCoverageCSV = errors.New("invalid daily spend CSV; use the four-column read-only daily SQL projection") +var errCoverageLimit = errors.New("daily spend coverage exceeds 3,660 days, 1 MiB or 512 bytes per line; select a shorter period") + +// RowCoverage compares records with a second accounting path. Matching counts +// do not establish that requests missing from both paths were recorded. +type RowCoverage struct { + Reference string `json:"reference"` + Days []CoverageDay `json:"days"` +} + +type CoverageDay struct { + Date string `json:"date_utc"` + Status string `json:"status"` + ExportedRows uint64 `json:"exported_rows"` + APIRequests *uint64 `json:"api_requests"` + SuccessfulRequests *uint64 `json:"successful_requests"` + FailedRequests *uint64 `json:"failed_requests"` + MissingRows *uint64 `json:"missing_rows"` + completeDay bool +} + +type dailyCounts struct { + api, successful, failed uint64 +} + +// The projection contains dates and counts only. Bound the whole file before +// parsing, then each physical CSV line before encoding/csv allocates fields. +func readDailyCounts(path string) (map[string]dailyCounts, error) { + root, err := os.OpenRoot(filepath.Dir(path)) + if err != nil { + return nil, errCoverageFile + } + defer root.Close() + data, err := localfile.Read(root, filepath.Base(path), maxCoverageBytes) + if err != nil { + return nil, fmt.Errorf("daily spend input: %w", err) + } + return parseDailyCounts(data) +} + +func parseDailyCounts(data []byte) (map[string]dailyCounts, error) { + if len(data) > maxCoverageBytes { + return nil, errCoverageLimit + } + scanner := bufio.NewScanner(bytes.NewReader(data)) + scanner.Buffer(make([]byte, maxCoverageLine+2), maxCoverageLine+2) + columns := map[string]int{} + counts := map[string]dailyCounts{} + for scanner.Scan() { + line := scanner.Bytes() + if len(line) > maxCoverageLine { + return nil, errCoverageLimit + } + if len(line) == 0 { + continue + } + r := csv.NewReader(bytes.NewReader(line)) + r.FieldsPerRecord = 4 + fields, err := r.Read() + if err != nil { + return nil, errCoverageCSV + } + if len(columns) == 0 { + for i, name := range fields { + switch name { + case "date", "api_requests", "successful_requests", "failed_requests": + default: + return nil, errCoverageCSV + } + if _, duplicate := columns[name]; duplicate { + return nil, errCoverageCSV + } + columns[name] = i + } + continue + } + date := fields[columns["date"]] + d, err := time.Parse(time.DateOnly, date) + if err != nil || d.Year() < 1970 || d.Year() > 2199 || d.Format(time.DateOnly) != date { + return nil, errCoverageCSV + } + if _, duplicate := counts[date]; duplicate { + return nil, errCoverageCSV + } + if len(counts) == maxCoverageDays { + return nil, errCoverageLimit + } + var values [3]uint64 + for i, column := range []string{"api_requests", "successful_requests", "failed_requests"} { + value := fields[columns[column]] + if value == "" { + return nil, errCoverageCSV + } + for _, c := range value { + if c < '0' || c > '9' { + return nil, errCoverageCSV + } + } + values[i], err = strconv.ParseUint(value, 10, 63) + if err != nil { + return nil, errCoverageCSV + } + } + counts[date] = dailyCounts{values[0], values[1], values[2]} + } + if scanner.Err() != nil { + return nil, errCoverageLimit + } + if len(columns) != 4 { + return nil, errCoverageCSV + } + return counts, nil +} + +type coverageBuilder struct { + days map[string]*CoverageDay +} + +func newCoverage(path string, periods ...Period) (*coverageBuilder, error) { + if path == "" { + return nil, nil + } + counts, err := readDailyCounts(path) + if err != nil { + return nil, err + } + b := &coverageBuilder{days: map[string]*CoverageDay{}} + for _, period := range periods { + for day := period.Start.UTC().Truncate(24 * time.Hour); day.Before(period.End); day = day.AddDate(0, 0, 1) { + date := day.Format(time.DateOnly) + d, exists := b.days[date] + if !exists { + if len(b.days) == maxCoverageDays { + return nil, errCoverageLimit + } + d = &CoverageDay{Date: date} + b.days[date] = d + if c, ok := counts[date]; ok { + d.APIRequests, d.SuccessfulRequests, d.FailedRequests = &c.api, &c.successful, &c.failed + } + } + d.completeDay = d.completeDay || (!period.Start.After(day) && !period.End.Before(day.AddDate(0, 0, 1))) + } + } + return b, nil +} + +func (b *coverageBuilder) add(start time.Time) { + if b != nil { + b.days[start.UTC().Format(time.DateOnly)].ExportedRows++ + } +} + +func (b *coverageBuilder) result() *RowCoverage { + if b == nil { + return nil + } + r := &RowCoverage{Reference: "LiteLLM_DailyUserSpend", Days: []CoverageDay{}} + for _, d := range b.days { + switch { + case !d.completeDay: + d.Status = "partial_day" + case d.APIRequests == nil: + d.Status = "daily_counts_missing" + case d.ExportedRows > *d.APIRequests: + d.Status = "excess_rows" + default: + missing := *d.APIRequests - d.ExportedRows + d.MissingRows = &missing + d.Status = "matched" + if missing > 0 { + d.Status = "missing_rows" + } + } + r.Days = append(r.Days, *d) + } + sort.Slice(r.Days, func(i, j int) bool { return r.Days[i].Date < r.Days[j].Date }) + return r +} diff --git a/internal/spend/row_coverage_test.go b/internal/spend/row_coverage_test.go new file mode 100644 index 0000000..8a8cdab --- /dev/null +++ b/internal/spend/row_coverage_test.go @@ -0,0 +1,231 @@ +// SPDX-License-Identifier: Apache-2.0 +// Code authors: Vijay and Codex + +package spend + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "strings" + "syscall" + "testing" + "time" +) + +const dailyHeader = "date,api_requests,successful_requests,failed_requests\n" + +func TestDailyCountsReader(t *testing.T) { + input := "failed_requests,date,successful_requests,api_requests\r\n1,2026-10-07,8,9\r\n0,2026-10-08,0,0\r\n" + path := readerWrite(t, ".csv", input) + got, err := readDailyCounts(path) + if err != nil || !reflect.DeepEqual(got, map[string]dailyCounts{"2026-10-07": {9, 8, 1}, "2026-10-08": {0, 0, 0}}) { + t.Fatalf("%v %v", got, err) + } + after, err := os.ReadFile(path) + if err != nil || string(after) != input { + t.Fatal("source changed") + } + for _, input := range []string{ + "", "date,api_requests,api_requests,failed_requests\n", + "date,api_requests,user,failed_requests\n", + dailyHeader + "2026-10-07,1,1\n", + dailyHeader + "2026-10-07,1,1,0,SECRET_SENTINEL\n", + dailyHeader + "2026-10-07,1,1,0\n2026-10-07,1,1,0\n", + dailyHeader + "2026-02-30,1,1,0\n", + dailyHeader + "1969-12-31,1,1,0\n", + dailyHeader + "2200-01-01,1,1,0\n", + dailyHeader + "2026-10-7,1,1,0\n", + } { + if _, err := parseDailyCounts([]byte(input)); err == nil { + t.Fatalf("accepted invalid input %q", input) + } + } + for _, value := range []string{"", "NULL", "-1", "+1", "1.0", "1e2", " 1", "9223372036854775808", "SECRET_SENTINEL"} { + for col := 0; col < 3; col++ { + values := []string{"1", "1", "0"} + values[col] = value + _, err := parseDailyCounts([]byte(dailyHeader + "2026-10-07," + strings.Join(values, ",") + "\n")) + if err == nil || strings.Contains(err.Error(), "SECRET_SENTINEL") { + t.Fatal("invalid value accepted or leaked") + } + } + } +} + +func TestDailyCountsBoundsAndSpecialFiles(t *testing.T) { + for _, data := range [][]byte{ + []byte(dailyHeader + strings.Repeat("x", maxCoverageLine+1)), + []byte(strings.Repeat("x", maxCoverageBytes+1)), + } { + if _, err := parseDailyCounts(data); !errors.Is(err, errCoverageLimit) { + t.Fatal(err) + } + } + var many strings.Builder + many.WriteString(dailyHeader) + start := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC) + for i := 0; i < maxCoverageDays; i++ { + fmt.Fprintf(&many, "%s,1,1,0\n", start.AddDate(0, 0, i).Format(time.DateOnly)) + } + if _, err := parseDailyCounts([]byte(many.String())); err != nil { + t.Fatal(err) + } + fmt.Fprintf(&many, "%s,1,1,0\n", start.AddDate(0, 0, maxCoverageDays).Format(time.DateOnly)) + if _, err := parseDailyCounts([]byte(many.String())); !errors.Is(err, errCoverageLimit) { + t.Fatal(err) + } + dir := t.TempDir() + regular := readerWrite(t, ".csv", dailyHeader) + link, fifo := filepath.Join(dir, "SECRET_LINK"), filepath.Join(dir, "SECRET_FIFO") + if err := os.Symlink(regular, link); err != nil { + t.Fatal(err) + } + if err := syscall.Mkfifo(fifo, 0600); err != nil { + t.Fatal(err) + } + for _, path := range []string{dir, link, fifo, filepath.Join(dir, "SECRET_MISSING")} { + if _, err := readDailyCounts(path); err == nil || strings.Contains(err.Error(), "SECRET") { + t.Fatal(err) + } + } + oversize := readerWrite(t, ".csv", strings.Repeat("x", maxCoverageBytes+1)) + if _, err := readDailyCounts(oversize); err == nil || !strings.Contains(err.Error(), "limit") { + t.Fatal(err) + } + if _, err := newCoverage(regular, Period{Start: start, End: start.AddDate(0, 0, maxCoverageDays+1)}); !errors.Is(err, errCoverageLimit) { + t.Fatal(err) + } +} + +func TestCoverageDaysAndUnknowns(t *testing.T) { + path := readerWrite(t, ".csv", dailyHeader+"2026-10-01,1,1,0\n2026-10-02,3,2,1\n2026-10-03,0,0,0\n2026-10-05,100,100,0\n") + period, err := ParsePeriod("2026-10-01T00:00:00Z/2026-10-05T12:00:00Z") + if err != nil { + t.Fatal(err) + } + b, err := newCoverage(path, period) + if err != nil { + t.Fatal(err) + } + for _, stamp := range []string{"2026-10-01T23:59:59Z", "2026-10-01T21:00:00-04:00", "2026-10-03T12:00:00Z"} { + start, _ := timestamp(stamp) + b.add(start) + } + r := b.result() + for i, want := range []string{"matched", "missing_rows", "excess_rows", "daily_counts_missing", "partial_day"} { + if r.Days[i].Status != want { + t.Fatalf("day %d: %+v", i, r.Days[i]) + } + } + if *r.Days[1].MissingRows != 2 || r.Days[2].MissingRows != nil || r.Days[3].APIRequests != nil || r.Days[4].MissingRows != nil { + t.Fatal(r) + } + if got, err := newCoverage(""); err != nil || got != nil || got.result() != nil { + t.Fatal("disabled coverage changed behavior") + } +} + +func TestCoverageDoesNotAlterAccounting(t *testing.T) { + t.Setenv("COVERAGE_TEST_KEY", "synthetic-coverage-key-never-a-credential") + path := readerWrite(t, ".csv", dailyHeader+"2026-10-07,100,100,0\n2026-10-08,100,100,0\n") + opts := Options{BeforePeriod: "2026-10-07", AfterPeriod: "2026-10-08", GroupBy: "key", SecretEnv: "COVERAGE_TEST_KEY", Top: 10, Now: time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC)} + before, err := Investigate("../../examples/litellm-spend/synthetic.csv", opts) + if err != nil { + t.Fatal(err) + } + opts.DailySpendPath = path + after, err := Investigate("../../examples/litellm-spend/synthetic.csv", opts) + if err != nil { + t.Fatal(err) + } + if after.RowCoverage == nil || after.RowCoverage.Days[0].Status != "missing_rows" { + t.Fatal(after.RowCoverage) + } + after.RowCoverage = nil + a, _ := json.Marshal(before) + b, _ := json.Marshal(after) + if string(a) != string(b) { + t.Fatal("coverage changed accounting") + } + sessionOpts := SessionOptions{Period: "2026-10-03", BudgetUSD: "1", SecretEnv: "COVERAGE_TEST_KEY", Top: 10, Now: opts.Now} + plain, err := ReviewSessions("../../examples/litellm-spend/session-budget.csv", sessionOpts) + if err != nil { + t.Fatal(err) + } + sessionOpts.DailySpendPath = path + covered, err := ReviewSessions("../../examples/litellm-spend/session-budget.csv", sessionOpts) + if err != nil || covered.RowCoverage == nil { + t.Fatal(err) + } + covered.RowCoverage = nil + if !reflect.DeepEqual(plain, covered) { + t.Fatal("coverage changed session accounting") + } +} + +func TestCoverageCountsAllRowsBeforeAnalysisFilters(t *testing.T) { + path := readerWrite(t, ".csv", readerHeader+",endTime,status\n"+ + "a,completion,2026-10-07T10:00:00Z,10,1,11,2026-10-07T10:00:01Z,success\n"+ + "b,completion,2026-10-07T11:00:00Z,0,0,0,2026-10-07T11:00:01Z,failure\n"+ + "c,embedding,2026-10-07T12:00:00Z,10,0,10,2026-10-07T12:00:01Z,success\n"+ + "d,responses,2026-10-08T10:00:00Z,10,1,11,2026-10-08T10:00:01Z,success\n"+ + "e,completion,2026-10-09T10:00:00Z,10,1,11,2026-10-09T10:00:01Z,success\n") + daily := readerWrite(t, ".csv", dailyHeader+"2026-10-07,3,2,1\n2026-10-08,1,1,0\n") + opts := Options{BeforePeriod: "2026-10-07", AfterPeriod: "2026-10-08", DailySpendPath: daily, Top: 10, Now: time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC)} + r, err := Investigate(path, opts) + if err != nil || r.RowCoverage == nil { + t.Fatal(err) + } + if r.RowCoverage.Days[0].ExportedRows != 3 || r.RowCoverage.Days[1].ExportedRows != 1 || r.RowCoverage.Days[0].Status != "matched" || !r.Incomplete || r.Before.Requests != 2 { + t.Fatalf("coverage accidentally used analyzed calls: %+v", r) + } + for _, period := range []string{"2026-10-07", "2026-10-07T10:00:00Z/2026-10-07T13:00:00Z"} { + r, err := ReviewSessions(path, SessionOptions{Period: period, DailySpendPath: daily, Top: 10, Now: opts.Now}) + if err != nil || r.RowCoverage.Days[0].ExportedRows != 3 { + t.Fatal(r.RowCoverage, err) + } + want := "matched" + if strings.Contains(period, "/") { + want = "partial_day" + } + if r.RowCoverage.Days[0].Status != want { + t.Fatal(r.RowCoverage) + } + } +} + +func TestCoverageWithEntirelyMissingSpendRows(t *testing.T) { + path := readerWrite(t, ".csv", readerHeader+"\n") + daily := readerWrite(t, ".csv", dailyHeader+"2026-10-07,3,2,1\n2026-10-08,1,1,0\n") + opts := Options{BeforePeriod: "2026-10-07", AfterPeriod: "2026-10-08", DailySpendPath: daily, Top: 10, Now: time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC)} + r, err := Investigate(path, opts) + if err != nil || r.RowCoverage == nil || r.AnalysisGap == "" || !r.Incomplete || *r.RowCoverage.Days[0].MissingRows != 3 { + t.Fatal(r, err) + } + s, err := ReviewSessions(path, SessionOptions{Period: "2026-10-07", DailySpendPath: daily, Top: 10, Now: opts.Now}) + if err != nil || s.RowCoverage == nil || *s.RowCoverage.Days[0].MissingRows != 3 || s.SelectedRows != 0 { + t.Fatal(s, err) + } + opts.DailySpendPath = "" + if _, err := Investigate(path, opts); err == nil { + t.Fatal("changed default empty-file behavior") + } +} + +func FuzzDailyCounts(f *testing.F) { + f.Add([]byte(dailyHeader + "2026-10-07,2,1,1\n")) + f.Add([]byte(dailyHeader + "\"unterminated\n")) + f.Fuzz(func(t *testing.T, data []byte) { + counts, err := parseDailyCounts(data) + if err == nil && len(counts) > maxCoverageDays { + t.Fatal("day cap bypassed") + } + if err != nil && err != errCoverageLimit && err != errCoverageCSV { + t.Fatal("untrusted parse error escaped") + } + }) +} diff --git a/internal/spend/session.go b/internal/spend/session.go index 07ec6c9..dafb057 100644 --- a/internal/spend/session.go +++ b/internal/spend/session.go @@ -22,6 +22,7 @@ const MaxSessionMemberships = 1_000_000 type SessionOptions struct { Period, BudgetUSD, SecretEnv string + DailySpendPath string CounterWindow time.Duration Top int ShowHashes bool @@ -57,6 +58,7 @@ type Session struct { } type SessionReport struct { + RowCoverage *RowCoverage `json:"row_coverage,omitempty"` Schema string `json:"schema"` Grouping string `json:"grouping"` Period Period `json:"period"` @@ -199,6 +201,10 @@ func ReviewSessions(path string, opts SessionOptions) (SessionReport, error) { if err != nil { return SessionReport{}, err } + coverage, err := newCoverage(opts.DailySpendPath, p) + if err != nil { + return SessionReport{}, err + } r := SessionReport{Schema: "fleetdiff-litellm-session-budget/v1", Grouping: "session_id_only", Period: p, AutomaticPeriod: opts.Period == "", Hashing: mode, CounterWindowSeconds: opts.CounterWindow.Seconds(), Rejections: map[string]uint64{}, Largest: []Session{}, CapReview: []Session{}, LongSpanReview: []Session{}, SharedCounterReview: []Session{}} r.FailureLoggingCompleteness = "unknown" r.ObservedSpan = r.AutomaticPeriod && p.End.Equal(last.Add(time.Nanosecond)) @@ -230,6 +236,7 @@ func ReviewSessions(path string, opts SessionOptions) (SessionReport, error) { return nil } r.SelectedRows++ + coverage.add(t) _, classified := v["rejection_kind"] if classified { r.ClassifiedRows++ @@ -385,7 +392,7 @@ func ReviewSessions(path string, opts SessionOptions) (SessionReport, error) { if seen.duplicated() { return SessionReport{}, errors.New("duplicate request_id; export unique request-level rows") } - if r.SelectedRows == 0 { + if r.SelectedRows == 0 && coverage == nil { return SessionReport{}, errors.New("no records in the selected period") } // Probe only already-retained session digests: outside-period identities do @@ -421,6 +428,7 @@ func ReviewSessions(path string, opts SessionOptions) (SessionReport, error) { last.Before(p.End) && p.End.Sub(s.lastActivity) <= opts.CounterWindow) } r.UnattributedSpend = costText(&unattributedSpend) + r.RowCoverage = coverage.result() return finishSessions(r, states, opts, cap), nil } diff --git a/scripts/restricted-smoke.sh b/scripts/restricted-smoke.sh index 5f3df57..8a1d15c 100644 --- a/scripts/restricted-smoke.sh +++ b/scripts/restricted-smoke.sh @@ -95,6 +95,15 @@ check_report investigate-spend '.before.recorded_tokens == 8000 and .after.recor .concentration_of_increase.tracked_contributors == 2 and .concentration_of_increase.share_of_net_recorded_increase.lower == 1' +check_case row-coverage 0 investigate --litellm-spend /examples/litellm-spend/coverage_runtime_spend.csv \ + --litellm-daily-spend /examples/litellm-spend/coverage_runtime_daily.csv \ + --before-period 2026-10-07 --after-period 2026-10-08 \ + --hash-secret-env FLEETDIFF_SMOKE_SECRET --format json +check_report row-coverage '.row_coverage.days[0].exported_rows == 8 and + .row_coverage.days[0].api_requests == 9 and .row_coverage.days[0].missing_rows == 1 and + .row_coverage.days[1].status == "matched" and + .before.recorded_tokens == 118 and .after.recorded_tokens == 15' + check_case session-budget 0 investigate --litellm-spend /examples/litellm-spend/session-budget.csv \ --period 2026-10-03 --group-by session --session-budget-usd 1.00 \ --hash-secret-env FLEETDIFF_SMOKE_SECRET --format json @@ -111,4 +120,4 @@ check_report diagnose '.status == "supported_safe" and .configuration_parsed == check_case scan 3 scan /windows --expected app --recent 2 --as-of 2026-10-04T00:30:00Z --format json check_report scan '.schema == "fleetdiff-scan/v1" and .unusual_windows == 2' -printf 'PASS restricted offline binary smoke: 7 cases (%s)\n' "$platform" +printf 'PASS restricted offline binary smoke: 8 cases (%s)\n' "$platform"