From 26b885bbec3aa64579f75ec2a3742540863f59c4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 04:11:03 -0700 Subject: [PATCH 1/7] test(finops): define bounded usage evidence contract --- tests/test_usage_evidence.py | 168 +++++++++++++++++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 tests/test_usage_evidence.py diff --git a/tests/test_usage_evidence.py b/tests/test_usage_evidence.py new file mode 100644 index 000000000..7043cf389 --- /dev/null +++ b/tests/test_usage_evidence.py @@ -0,0 +1,168 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for deterministic, provenance-distinct usage evidence.""" + +from __future__ import annotations + +import hashlib +import json + +import pytest + +from pg_llm_batch.usage_evidence import ( + UsageAuthority, + UsageEvidenceError, + build_usage_evidence, +) + + +def _valid_arguments() -> dict[str, object]: + return { + "authority": UsageAuthority.LOCAL_MEASURED, + "tenant_scope_id": "tenant-7f3a", + "source_id": "local-token-count-v1", + "request_count": 3, + "input_token_count": 120, + "output_token_count": 45, + "provider_alias": "openai-direct", + "endpoint_alias": "batch-v1", + "remote_batch_id": "batch_123", + } + + +def test_usage_authority_vocabulary_is_closed() -> None: + assert [authority.value for authority in UsageAuthority] == [ + "LOCAL_MEASURED", + "PROVIDER_REPORTED", + "HOST_RATE_ESTIMATE", + "RECONCILED", + ] + + +def test_build_usage_evidence_is_canonical_and_deterministic() -> None: + arguments = _valid_arguments() + + first_json, first_digest = build_usage_evidence(**arguments) + second_json, second_digest = build_usage_evidence(**arguments) + + assert first_json == second_json + assert first_digest == second_digest + assert first_digest == hashlib.sha256(first_json.encode("utf-8")).hexdigest() + assert first_json == json.dumps( + { + "authority": "LOCAL_MEASURED", + "endpoint_alias": "batch-v1", + "input_token_count": 120, + "output_token_count": 45, + "provider_alias": "openai-direct", + "remote_batch_id": "batch_123", + "request_count": 3, + "source_id": "local-token-count-v1", + "tenant_scope_id": "tenant-7f3a", + }, + sort_keys=True, + separators=(",", ":"), + ) + + +def test_optional_usage_dimensions_are_explicit_nulls() -> None: + canonical_json, _digest = build_usage_evidence( + authority=UsageAuthority.PROVIDER_REPORTED, + tenant_scope_id="tenant-7f3a", + source_id="provider-response-9", + request_count=1, + ) + + assert json.loads(canonical_json) == { + "authority": "PROVIDER_REPORTED", + "endpoint_alias": None, + "input_token_count": None, + "output_token_count": None, + "provider_alias": None, + "remote_batch_id": None, + "request_count": 1, + "source_id": "provider-response-9", + "tenant_scope_id": "tenant-7f3a", + } + + +def test_authority_and_source_identity_change_the_evidence_digest() -> None: + arguments = _valid_arguments() + _base_json, base_digest = build_usage_evidence(**arguments) + + authority_arguments = dict(arguments) + authority_arguments["authority"] = UsageAuthority.RECONCILED + _authority_json, authority_digest = build_usage_evidence(**authority_arguments) + + source_arguments = dict(arguments) + source_arguments["source_id"] = "reconciliation-export-2" + _source_json, source_digest = build_usage_evidence(**source_arguments) + + assert len({base_digest, authority_digest, source_digest}) == 3 + + +@pytest.mark.parametrize("authority", ["LOCAL_MEASURED", None, object()]) +def test_authority_requires_the_closed_enum(authority: object) -> None: + arguments = _valid_arguments() + arguments["authority"] = authority + + with pytest.raises(UsageEvidenceError, match="^invalid usage evidence authority$"): + build_usage_evidence(**arguments) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("tenant_scope_id", ""), + ("tenant_scope_id", "tenant scope"), + ("tenant_scope_id", "t" * 129), + ("tenant_scope_id", object()), + ("source_id", "source\ncontent"), + ("provider_alias", "provider?query"), + ("endpoint_alias", b"batch-v1"), + ("remote_batch_id", {}), + ], +) +def test_identifiers_reject_unbounded_or_content_bearing_values( + field_name: str, + value: object, +) -> None: + arguments = _valid_arguments() + arguments[field_name] = value + + with pytest.raises(UsageEvidenceError, match="^invalid usage evidence identifier$"): + build_usage_evidence(**arguments) + + +@pytest.mark.parametrize( + ("field_name", "value"), + [ + ("request_count", True), + ("request_count", -1), + ("request_count", 2**63), + ("input_token_count", 1.5), + ("input_token_count", float("inf")), + ("output_token_count", object()), + ], +) +def test_counts_reject_invalid_or_unbounded_values(field_name: str, value: object) -> None: + arguments = _valid_arguments() + arguments[field_name] = value + + with pytest.raises(UsageEvidenceError, match="^invalid usage evidence count$"): + build_usage_evidence(**arguments) + + +def test_count_boundary_accepts_zero_and_signed_bigint_maximum() -> None: + canonical_json, _digest = build_usage_evidence( + authority=UsageAuthority.HOST_RATE_ESTIMATE, + tenant_scope_id="tenant-7f3a", + source_id="rate-card-2026-08-27", + request_count=0, + input_token_count=2**63 - 1, + output_token_count=0, + ) + + evidence = json.loads(canonical_json) + assert evidence["request_count"] == 0 + assert evidence["input_token_count"] == 2**63 - 1 + assert evidence["output_token_count"] == 0 From fc80cd5f1095cb77788fc4da6c6fa9cc3339557c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 04:13:32 -0700 Subject: [PATCH 2/7] feat(finops): emit bounded usage evidence identity --- pg_llm_batch/usage_evidence.py | 86 ++++++++++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 pg_llm_batch/usage_evidence.py diff --git a/pg_llm_batch/usage_evidence.py b/pg_llm_batch/usage_evidence.py new file mode 100644 index 000000000..c4d2e69e7 --- /dev/null +++ b/pg_llm_batch/usage_evidence.py @@ -0,0 +1,86 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Build bounded deterministic usage-authority evidence without billing.""" + +from __future__ import annotations + +import hashlib +import json +import re +from enum import Enum + +_MAX_COUNT = 2**63 - 1 +_MAX_IDENTIFIER_LENGTH = 128 +_IDENTIFIER_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/+\-]{0,127}\Z") + + +class UsageEvidenceError(ValueError): + """Report a fail-closed usage-evidence validation error.""" + + +class UsageAuthority(str, Enum): + """Identify the authority class for one usage-evidence record.""" + + LOCAL_MEASURED = "LOCAL_MEASURED" + PROVIDER_REPORTED = "PROVIDER_REPORTED" + HOST_RATE_ESTIMATE = "HOST_RATE_ESTIMATE" + RECONCILED = "RECONCILED" + + +def _require_identifier(value: object) -> str: + if type(value) is not str: + raise UsageEvidenceError("invalid usage evidence identifier") + if len(value) > _MAX_IDENTIFIER_LENGTH: + raise UsageEvidenceError("invalid usage evidence identifier") + if _IDENTIFIER_RE.fullmatch(value) is None: + raise UsageEvidenceError("invalid usage evidence identifier") + return value + + +def _optional_identifier(value: object | None) -> str | None: + if value is None: + return None + return _require_identifier(value) + + +def _require_count(value: object) -> int: + if type(value) is not int or value < 0 or value > _MAX_COUNT: + raise UsageEvidenceError("invalid usage evidence count") + return value + + +def _optional_count(value: object | None) -> int | None: + if value is None: + return None + return _require_count(value) + + +def build_usage_evidence( + *, + authority: UsageAuthority, + tenant_scope_id: str, + source_id: str, + request_count: int, + input_token_count: int | None = None, + output_token_count: int | None = None, + provider_alias: str | None = None, + endpoint_alias: str | None = None, + remote_batch_id: str | None = None, +) -> tuple[str, str]: + """Return canonical JSON and SHA-256 identity for bounded usage evidence.""" + if type(authority) is not UsageAuthority: + raise UsageEvidenceError("invalid usage evidence authority") + + payload = { + "authority": authority.value, + "endpoint_alias": _optional_identifier(endpoint_alias), + "input_token_count": _optional_count(input_token_count), + "output_token_count": _optional_count(output_token_count), + "provider_alias": _optional_identifier(provider_alias), + "remote_batch_id": _optional_identifier(remote_batch_id), + "request_count": _require_count(request_count), + "source_id": _require_identifier(source_id), + "tenant_scope_id": _require_identifier(tenant_scope_id), + } + canonical_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) + evidence_sha256 = hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() + return canonical_json, evidence_sha256 From c990318345013ef24a619eff07f9f3bbb5698a2a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Thu, 27 Aug 2026 04:16:22 -0700 Subject: [PATCH 3/7] fix(finops): document usage evidence validators --- pg_llm_batch/usage_evidence.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pg_llm_batch/usage_evidence.py b/pg_llm_batch/usage_evidence.py index c4d2e69e7..0e573df4e 100644 --- a/pg_llm_batch/usage_evidence.py +++ b/pg_llm_batch/usage_evidence.py @@ -27,6 +27,7 @@ class UsageAuthority(str, Enum): def _require_identifier(value: object) -> str: + """Return one bounded opaque identifier or fail closed.""" if type(value) is not str: raise UsageEvidenceError("invalid usage evidence identifier") if len(value) > _MAX_IDENTIFIER_LENGTH: @@ -37,18 +38,21 @@ def _require_identifier(value: object) -> str: def _optional_identifier(value: object | None) -> str | None: + """Return a validated optional opaque identifier.""" if value is None: return None return _require_identifier(value) def _require_count(value: object) -> int: + """Return a bounded non-negative integer count or fail closed.""" if type(value) is not int or value < 0 or value > _MAX_COUNT: raise UsageEvidenceError("invalid usage evidence count") return value def _optional_count(value: object | None) -> int | None: + """Return a validated optional bounded count.""" if value is None: return None return _require_count(value) From 040f9d9e07373f9ab9cb68ac532b0e451e0d415d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 18 Sep 2026 05:06:23 +0900 Subject: [PATCH 4/7] test(finops): expose usage completeness ambiguity --- tests/test_usage_evidence.py | 124 ++++++++++++++++++++++++++++++++++- 1 file changed, 123 insertions(+), 1 deletion(-) diff --git a/tests/test_usage_evidence.py b/tests/test_usage_evidence.py index 7043cf389..23b4f30d9 100644 --- a/tests/test_usage_evidence.py +++ b/tests/test_usage_evidence.py @@ -10,6 +10,7 @@ from pg_llm_batch.usage_evidence import ( UsageAuthority, + UsageCompleteness, UsageEvidenceError, build_usage_evidence, ) @@ -22,7 +23,9 @@ def _valid_arguments() -> dict[str, object]: "source_id": "local-token-count-v1", "request_count": 3, "input_token_count": 120, + "input_token_completeness": UsageCompleteness.COMPLETE, "output_token_count": 45, + "output_token_completeness": UsageCompleteness.COMPLETE, "provider_alias": "openai-direct", "endpoint_alias": "batch-v1", "remote_batch_id": "batch_123", @@ -38,6 +41,15 @@ def test_usage_authority_vocabulary_is_closed() -> None: ] +def test_usage_completeness_vocabulary_is_closed() -> None: + assert [completeness.value for completeness in UsageCompleteness] == [ + "COMPLETE", + "PARTIAL", + "UNAVAILABLE", + "MIXED", + ] + + def test_build_usage_evidence_is_canonical_and_deterministic() -> None: arguments = _valid_arguments() @@ -51,7 +63,9 @@ def test_build_usage_evidence_is_canonical_and_deterministic() -> None: { "authority": "LOCAL_MEASURED", "endpoint_alias": "batch-v1", + "input_token_completeness": "COMPLETE", "input_token_count": 120, + "output_token_completeness": "COMPLETE", "output_token_count": 45, "provider_alias": "openai-direct", "remote_batch_id": "batch_123", @@ -64,18 +78,22 @@ def test_build_usage_evidence_is_canonical_and_deterministic() -> None: ) -def test_optional_usage_dimensions_are_explicit_nulls() -> None: +def test_unavailable_usage_dimensions_are_explicit_and_not_zero() -> None: canonical_json, _digest = build_usage_evidence( authority=UsageAuthority.PROVIDER_REPORTED, tenant_scope_id="tenant-7f3a", source_id="provider-response-9", request_count=1, + input_token_completeness=UsageCompleteness.UNAVAILABLE, + output_token_completeness=UsageCompleteness.UNAVAILABLE, ) assert json.loads(canonical_json) == { "authority": "PROVIDER_REPORTED", "endpoint_alias": None, + "input_token_completeness": "UNAVAILABLE", "input_token_count": None, + "output_token_completeness": "UNAVAILABLE", "output_token_count": None, "provider_alias": None, "remote_batch_id": None, @@ -85,6 +103,54 @@ def test_optional_usage_dimensions_are_explicit_nulls() -> None: } +def test_measured_zero_is_distinct_from_unavailable_usage() -> None: + measured_json, measured_digest = build_usage_evidence( + authority=UsageAuthority.PROVIDER_REPORTED, + tenant_scope_id="tenant-7f3a", + source_id="provider-response-10", + request_count=1, + input_token_count=0, + input_token_completeness=UsageCompleteness.COMPLETE, + output_token_count=0, + output_token_completeness=UsageCompleteness.COMPLETE, + ) + unavailable_json, unavailable_digest = build_usage_evidence( + authority=UsageAuthority.PROVIDER_REPORTED, + tenant_scope_id="tenant-7f3a", + source_id="provider-response-10", + request_count=1, + input_token_completeness=UsageCompleteness.UNAVAILABLE, + output_token_completeness=UsageCompleteness.UNAVAILABLE, + ) + + measured = json.loads(measured_json) + unavailable = json.loads(unavailable_json) + assert measured["input_token_count"] == 0 + assert measured["input_token_completeness"] == "COMPLETE" + assert unavailable["input_token_count"] is None + assert unavailable["input_token_completeness"] == "UNAVAILABLE" + assert measured_digest != unavailable_digest + + +def test_reconciled_usage_can_preserve_mixed_and_partial_completeness() -> None: + canonical_json, _digest = build_usage_evidence( + authority=UsageAuthority.RECONCILED, + tenant_scope_id="tenant-7f3a", + source_id="reconciliation-export-3", + request_count=4, + input_token_count=200, + input_token_completeness=UsageCompleteness.MIXED, + output_token_count=55, + output_token_completeness=UsageCompleteness.PARTIAL, + ) + + evidence = json.loads(canonical_json) + assert evidence["input_token_count"] == 200 + assert evidence["input_token_completeness"] == "MIXED" + assert evidence["output_token_count"] == 55 + assert evidence["output_token_completeness"] == "PARTIAL" + + def test_authority_and_source_identity_change_the_evidence_digest() -> None: arguments = _valid_arguments() _base_json, base_digest = build_usage_evidence(**arguments) @@ -109,6 +175,58 @@ def test_authority_requires_the_closed_enum(authority: object) -> None: build_usage_evidence(**arguments) +@pytest.mark.parametrize("completeness", ["COMPLETE", None, object()]) +def test_completeness_requires_the_closed_enum(completeness: object) -> None: + arguments = _valid_arguments() + arguments["input_token_completeness"] = completeness + + with pytest.raises(UsageEvidenceError, match="^invalid usage evidence completeness$"): + build_usage_evidence(**arguments) + + +@pytest.mark.parametrize( + ("count_field", "completeness_field", "count", "completeness"), + [ + ( + "input_token_count", + "input_token_completeness", + 0, + UsageCompleteness.UNAVAILABLE, + ), + ( + "output_token_count", + "output_token_completeness", + None, + UsageCompleteness.COMPLETE, + ), + ( + "input_token_count", + "input_token_completeness", + None, + UsageCompleteness.PARTIAL, + ), + ( + "output_token_count", + "output_token_completeness", + None, + UsageCompleteness.MIXED, + ), + ], +) +def test_count_and_completeness_fail_closed_on_lossy_combinations( + count_field: str, + completeness_field: str, + count: object, + completeness: UsageCompleteness, +) -> None: + arguments = _valid_arguments() + arguments[count_field] = count + arguments[completeness_field] = completeness + + with pytest.raises(UsageEvidenceError, match="^inconsistent usage evidence completeness$"): + build_usage_evidence(**arguments) + + @pytest.mark.parametrize( ("field_name", "value"), [ @@ -159,10 +277,14 @@ def test_count_boundary_accepts_zero_and_signed_bigint_maximum() -> None: source_id="rate-card-2026-08-27", request_count=0, input_token_count=2**63 - 1, + input_token_completeness=UsageCompleteness.COMPLETE, output_token_count=0, + output_token_completeness=UsageCompleteness.COMPLETE, ) evidence = json.loads(canonical_json) assert evidence["request_count"] == 0 assert evidence["input_token_count"] == 2**63 - 1 + assert evidence["input_token_completeness"] == "COMPLETE" assert evidence["output_token_count"] == 0 + assert evidence["output_token_completeness"] == "COMPLETE" From 38ecfd8ab7509ee59edc1241b5a79ab6042691fc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 18 Sep 2026 05:06:43 +0900 Subject: [PATCH 5/7] feat(finops): preserve usage measurement completeness --- pg_llm_batch/usage_evidence.py | 43 +++++++++++++++++++++++++++++++--- 1 file changed, 40 insertions(+), 3 deletions(-) diff --git a/pg_llm_batch/usage_evidence.py b/pg_llm_batch/usage_evidence.py index 0e573df4e..8074a2a2e 100644 --- a/pg_llm_batch/usage_evidence.py +++ b/pg_llm_batch/usage_evidence.py @@ -18,7 +18,7 @@ class UsageEvidenceError(ValueError): class UsageAuthority(str, Enum): - """Identify the authority class for one usage-evidence record.""" + """Identify the provenance authority for one usage-evidence record.""" LOCAL_MEASURED = "LOCAL_MEASURED" PROVIDER_REPORTED = "PROVIDER_REPORTED" @@ -26,6 +26,15 @@ class UsageAuthority(str, Enum): RECONCILED = "RECONCILED" +class UsageCompleteness(str, Enum): + """Identify how completely one token dimension covers the represented usage.""" + + COMPLETE = "COMPLETE" + PARTIAL = "PARTIAL" + UNAVAILABLE = "UNAVAILABLE" + MIXED = "MIXED" + + def _require_identifier(value: object) -> str: """Return one bounded opaque identifier or fail closed.""" if type(value) is not str: @@ -58,6 +67,22 @@ def _optional_count(value: object | None) -> int | None: return _require_count(value) +def _count_with_completeness( + value: object | None, + completeness: object, +) -> tuple[int | None, UsageCompleteness]: + """Validate one token count and its explicit measurement-completeness state.""" + if type(completeness) is not UsageCompleteness: + raise UsageEvidenceError("invalid usage evidence completeness") + count = _optional_count(value) + if completeness is UsageCompleteness.UNAVAILABLE: + if count is not None: + raise UsageEvidenceError("inconsistent usage evidence completeness") + elif count is None: + raise UsageEvidenceError("inconsistent usage evidence completeness") + return count, completeness + + def build_usage_evidence( *, authority: UsageAuthority, @@ -65,7 +90,9 @@ def build_usage_evidence( source_id: str, request_count: int, input_token_count: int | None = None, + input_token_completeness: UsageCompleteness, output_token_count: int | None = None, + output_token_completeness: UsageCompleteness, provider_alias: str | None = None, endpoint_alias: str | None = None, remote_batch_id: str | None = None, @@ -74,11 +101,21 @@ def build_usage_evidence( if type(authority) is not UsageAuthority: raise UsageEvidenceError("invalid usage evidence authority") + validated_input_count, validated_input_completeness = _count_with_completeness( + input_token_count, + input_token_completeness, + ) + validated_output_count, validated_output_completeness = _count_with_completeness( + output_token_count, + output_token_completeness, + ) payload = { "authority": authority.value, "endpoint_alias": _optional_identifier(endpoint_alias), - "input_token_count": _optional_count(input_token_count), - "output_token_count": _optional_count(output_token_count), + "input_token_completeness": validated_input_completeness.value, + "input_token_count": validated_input_count, + "output_token_completeness": validated_output_completeness.value, + "output_token_count": validated_output_count, "provider_alias": _optional_identifier(provider_alias), "remote_batch_id": _optional_identifier(remote_batch_id), "request_count": _require_count(request_count), From de5cca77f556b4a2126d8e15f00aa1194e8cd75c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 21 Sep 2026 10:10:21 +0900 Subject: [PATCH 6/7] test(finops): prove JSON integer interoperability bound --- ...st_usage_evidence_json_interoperability.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 tests/test_usage_evidence_json_interoperability.py diff --git a/tests/test_usage_evidence_json_interoperability.py b/tests/test_usage_evidence_json_interoperability.py new file mode 100644 index 000000000..94c17732c --- /dev/null +++ b/tests/test_usage_evidence_json_interoperability.py @@ -0,0 +1,63 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Regression tests for interoperable integer authority in JSON usage evidence.""" + +from __future__ import annotations + +import json + +import pytest + +from pg_llm_batch.usage_evidence import ( + UsageAuthority, + UsageCompleteness, + UsageEvidenceError, + build_usage_evidence, +) + + +_MAX_INTEROPERABLE_JSON_INTEGER = 2**53 - 1 + + +def _valid_arguments() -> dict[str, object]: + """Return one complete bounded usage-evidence input set.""" + return { + "authority": UsageAuthority.LOCAL_MEASURED, + "tenant_scope_id": "tenant-7f3a", + "source_id": "local-token-count-v1", + "request_count": 3, + "input_token_count": 120, + "input_token_completeness": UsageCompleteness.COMPLETE, + "output_token_count": 45, + "output_token_completeness": UsageCompleteness.COMPLETE, + "provider_alias": "openai-direct", + "endpoint_alias": "batch-v1", + "remote_batch_id": "batch_123", + } + + +@pytest.mark.parametrize( + "field_name", + ["request_count", "input_token_count", "output_token_count"], +) +def test_usage_counts_reject_values_above_json_interoperability_range( + field_name: str, +) -> None: + arguments = _valid_arguments() + arguments[field_name] = _MAX_INTEROPERABLE_JSON_INTEGER + 1 + + with pytest.raises(UsageEvidenceError, match="^invalid usage evidence count$"): + build_usage_evidence(**arguments) + + +@pytest.mark.parametrize( + "field_name", + ["request_count", "input_token_count", "output_token_count"], +) +def test_usage_counts_accept_json_interoperability_boundary(field_name: str) -> None: + arguments = _valid_arguments() + arguments[field_name] = _MAX_INTEROPERABLE_JSON_INTEGER + + canonical_json, _digest = build_usage_evidence(**arguments) + + evidence = json.loads(canonical_json) + assert evidence[field_name] == _MAX_INTEROPERABLE_JSON_INTEGER From 157d0ac9dcf23a54a05d010044344e0f650b8698 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 21 Sep 2026 18:05:46 +0900 Subject: [PATCH 7/7] fix(finops): bound usage evidence JSON integers --- pg_llm_batch/usage_evidence.py | 2 +- tests/test_usage_evidence.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pg_llm_batch/usage_evidence.py b/pg_llm_batch/usage_evidence.py index 8074a2a2e..ef2c0adc7 100644 --- a/pg_llm_batch/usage_evidence.py +++ b/pg_llm_batch/usage_evidence.py @@ -8,7 +8,7 @@ import re from enum import Enum -_MAX_COUNT = 2**63 - 1 +_MAX_COUNT = 2**53 - 1 _MAX_IDENTIFIER_LENGTH = 128 _IDENTIFIER_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/+\-]{0,127}\Z") diff --git a/tests/test_usage_evidence.py b/tests/test_usage_evidence.py index 23b4f30d9..9f682a3cb 100644 --- a/tests/test_usage_evidence.py +++ b/tests/test_usage_evidence.py @@ -270,13 +270,13 @@ def test_counts_reject_invalid_or_unbounded_values(field_name: str, value: objec build_usage_evidence(**arguments) -def test_count_boundary_accepts_zero_and_signed_bigint_maximum() -> None: +def test_count_boundary_accepts_zero_and_json_interoperability_maximum() -> None: canonical_json, _digest = build_usage_evidence( authority=UsageAuthority.HOST_RATE_ESTIMATE, tenant_scope_id="tenant-7f3a", source_id="rate-card-2026-08-27", request_count=0, - input_token_count=2**63 - 1, + input_token_count=2**53 - 1, input_token_completeness=UsageCompleteness.COMPLETE, output_token_count=0, output_token_completeness=UsageCompleteness.COMPLETE, @@ -284,7 +284,7 @@ def test_count_boundary_accepts_zero_and_signed_bigint_maximum() -> None: evidence = json.loads(canonical_json) assert evidence["request_count"] == 0 - assert evidence["input_token_count"] == 2**63 - 1 + assert evidence["input_token_count"] == 2**53 - 1 assert evidence["input_token_completeness"] == "COMPLETE" assert evidence["output_token_count"] == 0 assert evidence["output_token_completeness"] == "COMPLETE"