"""Model-agnostic token-efficiency accounting for benchmark artifacts. The harness does not assume a tokenizer or provider. It reports a reproducible upper-bound-like estimate of one token per four UTF-8 bytes so reports can be compared across runs without adding a model-specific dependency. """ from __future__ import annotations import math from collections.abc import Iterable, Mapping from typing import Any from .metrics import distribution_summary TOKEN_ESTIMATOR = "ceil(utf8_bytes/4)" def estimate_text_tokens(text: str) -> int: """Return a deterministic, model-agnostic estimate for UTF-8 text.""" return math.ceil(len(text.encode("utf-8")) / 4) def _metric_summary(bytes_samples: list[int], findings: list[int]) -> dict[str, Any]: tokens = [math.ceil(value / 4) for value in bytes_samples] density = [ 1000.0 * count / token for count, token in zip(findings, tokens, strict=True) if token > 0 ] return { "bytes": distribution_summary(bytes_samples), "estimated_tokens": distribution_summary(tokens), "findings_per_1k_estimated_tokens": distribution_summary(density), } def _nonnegative_int(record: Mapping[str, Any], key: str) -> int: value = record.get(key, 0) if isinstance(value, bool) or not isinstance(value, (int, float)): return 0 try: return max(0, int(value)) except (OverflowError, ValueError): return 0 def artifact_token_efficiency(runs: Iterable[Mapping[str, Any]]) -> dict[str, Any]: """Summarize driver and SARIF artifact token costs for completed runs.""" records = list(runs) findings = [_nonnegative_int(record, "finding_count") for record in records] return { "estimator": TOKEN_ESTIMATOR, "driver_output": _metric_summary( [_nonnegative_int(record, "output_bytes") for record in records], findings ), "sarif_report": _metric_summary( [_nonnegative_int(record, "report_bytes") for record in records], findings ), }