diff --git a/codenib/eval/agent_runner/query_sweep.py b/codenib/eval/agent_runner/query_sweep.py index 03cf9aa7..fe564e14 100644 --- a/codenib/eval/agent_runner/query_sweep.py +++ b/codenib/eval/agent_runner/query_sweep.py @@ -323,6 +323,7 @@ def run_query_sweep( for row, subset_id, skills, rep, cell_id, cell_path in plan: cell_started = time.time() + out: Optional[Dict[str, Any]] = None gt_blocks = collect_target_blocks(row) target_files, target_symbols = query_targets( row, simplified_symbols=cfg.gt_simplified_symbols @@ -378,9 +379,13 @@ def run_query_sweep( "trace_summary": out.get("trace_summary"), "answer": out["answer"], "total_turns": out["total_turns"], + "prompt_tokens": out["prompt_tokens"], + "completion_tokens": out["completion_tokens"], "total_tokens": out["total_tokens"], "cost_usd": out["cost_usd"], + "cost_provenance": out.get("cost_provenance"), "cache_read_input_tokens": out["cache_read_input_tokens"], + "cache_creation_input_tokens": out["cache_creation_input_tokens"], "elapsed_seconds": time.time() - cell_started, "error": None, } @@ -414,9 +419,35 @@ def run_query_sweep( "instance_id": instance_id, "query_id": row["query_id"], "category": row.get("category"), + "length_variant": row.get("length_variant"), + "source_config": row.get("source_config"), "subset_id": subset_id, + "skills": skills, + "model": cfg.model, + "rep": rep, + "language": language_key_for_query_row(row), "success": False, "metrics": {}, + "metrics_meaningful": bool(target_files or gt_blocks), + "query": row["query"], + "total_turns": out.get("total_turns") if out else None, + "prompt_tokens": out.get("prompt_tokens") if out else None, + "completion_tokens": ( + out.get("completion_tokens") if out else None + ), + "total_tokens": out.get("total_tokens") if out else None, + "cost_usd": out.get("cost_usd") if out else None, + "cost_provenance": ( + out.get("cost_provenance") + if out + else {"kind": "unavailable", "model": cfg.model} + ), + "cache_read_input_tokens": ( + out.get("cache_read_input_tokens") if out else None + ), + "cache_creation_input_tokens": ( + out.get("cache_creation_input_tokens") if out else None + ), "error": str(exc), "elapsed_seconds": time.time() - cell_started, } diff --git a/codenib/eval/agent_runner/sweep.py b/codenib/eval/agent_runner/sweep.py index 93ee5368..82f7a11a 100644 --- a/codenib/eval/agent_runner/sweep.py +++ b/codenib/eval/agent_runner/sweep.py @@ -24,6 +24,8 @@ from pathlib import Path from typing import Any, Dict, List, Mapping, Optional, Sequence +from codenib.llm.usage import runtime_cost_provenance + from .sweep_config import SweepConfig # language_group (HF column) -> classify() language key @@ -367,7 +369,6 @@ def _run_subagent(sys_prompt, q, mt): # Sum token usage across every run so verify/scatter arms are charged for # the closed loop, not just their final turn. usage_totals = accounting.usage_totals() - return { "nodes": observations["nodes"], "tool_calls": observations["tool_calls"], @@ -385,6 +386,7 @@ def _run_subagent(sys_prompt, q, mt): "completion_tokens": usage_totals["completion_tokens"], "total_tokens": usage_totals["total_tokens"], "cost_usd": usage_totals["cost_usd"], + "cost_provenance": runtime_cost_provenance(cfg.model, usage_totals["cost_usd"]), "cache_read_input_tokens": usage_totals["cache_read_input_tokens"], "cache_creation_input_tokens": usage_totals["cache_creation_input_tokens"], } @@ -563,6 +565,7 @@ def run_sweep(cfg: SweepConfig, output_dir: Path, *, resume: bool = True) -> Dic "completion_tokens": out["completion_tokens"], "total_tokens": out["total_tokens"], "cost_usd": out["cost_usd"], + "cost_provenance": out.get("cost_provenance"), "cache_read_input_tokens": out["cache_read_input_tokens"], "cache_creation_input_tokens": out["cache_creation_input_tokens"], "elapsed_seconds": time.time() - t, @@ -590,6 +593,16 @@ def run_sweep(cfg: SweepConfig, output_dir: Path, *, resume: bool = True) -> Dic "metrics": {}, "metrics_meaningful": gt_meaningful, "tool_calls": [], + "prompt_tokens": None, + "completion_tokens": None, + "total_tokens": None, + "cost_usd": None, + "cost_provenance": { + "kind": "unavailable", + "model": cfg.model, + }, + "cache_read_input_tokens": None, + "cache_creation_input_tokens": None, "error": str(exc), "elapsed_seconds": time.time() - t, } diff --git a/codenib/eval/pricing/anthropic-direct-2026-08-04.json b/codenib/eval/pricing/anthropic-direct-2026-08-04.json new file mode 100644 index 00000000..2f38e95c --- /dev/null +++ b/codenib/eval/pricing/anthropic-direct-2026-08-04.json @@ -0,0 +1,26 @@ +{ + "schema": "codenib.pricing-snapshot.v1", + "snapshot_id": "anthropic-direct-2026-08-04", + "currency": "USD", + "effective_date": "2026-08-04", + "retrieved_date": "2026-08-04", + "models": [ + { + "model_id": "anthropic/claude-haiku-4-5-20251001", + "aliases": [ + "anthropic/claude-haiku-4-5", + "claude-haiku-4-5-20251001" + ], + "channel": "anthropic-api-direct", + "token_semantics": "prompt_excludes_cache", + "cache_write_ttl": "5m", + "rates_per_million": { + "uncached_input": 1.0, + "output": 5.0, + "cache_read_input": 0.1, + "cache_write_input": 1.25 + }, + "source_url": "https://platform.claude.com/docs/en/about-claude/pricing" + } + ] +} diff --git a/codenib/eval/reports/pricing.py b/codenib/eval/reports/pricing.py new file mode 100644 index 00000000..45c5f4e0 --- /dev/null +++ b/codenib/eval/reports/pricing.py @@ -0,0 +1,350 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Immutable pricing snapshots for reproducible offline cost projection.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Any, Mapping, Optional, Sequence + +PRICING_SNAPSHOT_SCHEMA = "codenib.pricing-snapshot.v1" +DEFAULT_PRICING_SNAPSHOT = "anthropic-direct-2026-08-04.json" + +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_RATE_KEYS = ( + "uncached_input", + "output", + "cache_read_input", + "cache_write_input", +) +_TOKEN_SEMANTICS = {"prompt_excludes_cache", "prompt_includes_cache"} + + +class PricingSnapshotError(ValueError): + """Raised when a pricing snapshot is malformed or ambiguous.""" + + +@dataclass(frozen=True) +class ModelPrice: + """One provider-channel-specific token price contract.""" + + model_id: str + aliases: tuple[str, ...] + channel: str + token_semantics: str + cache_write_ttl: str + rates_per_million: Mapping[str, float] + source_url: str + + +@dataclass(frozen=True) +class PricingSnapshot: + """Validated immutable catalog plus its exact content hash.""" + + snapshot_id: str + sha256: str + currency: str + effective_date: str + retrieved_date: str + models: tuple[ModelPrice, ...] + source_file: Optional[str] = None + + def resolve_model(self, model: str) -> Optional[ModelPrice]: + """Resolve an exact id or alias; snapshots reject ambiguous aliases.""" + + selected = str(model or "").strip() + for entry in self.models: + if selected == entry.model_id or selected in entry.aliases: + return entry + return None + + def public_identity(self) -> dict[str, Any]: + """Return provenance fields suitable for a machine-readable report.""" + + return { + "schema": PRICING_SNAPSHOT_SCHEMA, + "snapshot_id": self.snapshot_id, + "sha256": self.sha256, + "currency": self.currency, + "effective_date": self.effective_date, + "retrieved_date": self.retrieved_date, + "source_file": self.source_file, + } + + +def bundled_pricing_snapshot_path() -> Path: + """Return the bundled direct-provider example catalog.""" + + return Path(__file__).resolve().parents[1] / "pricing" / DEFAULT_PRICING_SNAPSHOT + + +def load_pricing_snapshot(path: Path) -> PricingSnapshot: + """Load and validate one exact JSON pricing snapshot.""" + + source = Path(path).expanduser().resolve() + try: + raw = source.read_bytes() + value = json.loads(raw) + except OSError as exc: + raise PricingSnapshotError( + f"cannot read pricing snapshot {source}: {exc}" + ) from exc + except json.JSONDecodeError as exc: + raise PricingSnapshotError(f"invalid pricing JSON {source}: {exc}") from exc + return parse_pricing_snapshot( + value, + sha256="sha256:" + hashlib.sha256(raw).hexdigest(), + source_path=str(source), + ) + + +def parse_pricing_snapshot( + value: Mapping[str, Any], + *, + sha256: Optional[str] = None, + source_path: Optional[str] = None, +) -> PricingSnapshot: + """Validate a decoded snapshot and build its alias index.""" + + if not isinstance(value, Mapping): + raise PricingSnapshotError("pricing snapshot must be a JSON object") + if value.get("schema") != PRICING_SNAPSHOT_SCHEMA: + raise PricingSnapshotError( + f"pricing snapshot schema must be {PRICING_SNAPSHOT_SCHEMA!r}" + ) + + snapshot_id = _required_string(value, "snapshot_id") + currency = _required_string(value, "currency") + if currency != "USD": + raise PricingSnapshotError("only USD pricing snapshots are currently supported") + effective_date = _date(value, "effective_date") + retrieved_date = _date(value, "retrieved_date") + raw_models = value.get("models") + if not isinstance(raw_models, Sequence) or isinstance(raw_models, (str, bytes)): + raise PricingSnapshotError("pricing snapshot models must be an array") + if not raw_models: + raise PricingSnapshotError("pricing snapshot must contain at least one model") + + models = [] + names: dict[str, str] = {} + for index, raw_model in enumerate(raw_models): + if not isinstance(raw_model, Mapping): + raise PricingSnapshotError(f"models[{index}] must be an object") + model_id = _required_string(raw_model, "model_id", prefix=f"models[{index}].") + aliases_value = raw_model.get("aliases") or [] + if not isinstance(aliases_value, Sequence) or isinstance( + aliases_value, (str, bytes) + ): + raise PricingSnapshotError(f"models[{index}].aliases must be an array") + aliases = tuple( + _nonempty_string(alias, f"models[{index}].aliases") + for alias in aliases_value + ) + channel = _required_string(raw_model, "channel", prefix=f"models[{index}].") + token_semantics = _required_string( + raw_model, "token_semantics", prefix=f"models[{index}]." + ) + if token_semantics not in _TOKEN_SEMANTICS: + raise PricingSnapshotError( + f"models[{index}].token_semantics must be one of " + f"{sorted(_TOKEN_SEMANTICS)}" + ) + cache_write_ttl = _required_string( + raw_model, "cache_write_ttl", prefix=f"models[{index}]." + ) + rates = _rates(raw_model.get("rates_per_million"), index=index) + source_url = _required_string( + raw_model, "source_url", prefix=f"models[{index}]." + ) + if not source_url.startswith("https://"): + raise PricingSnapshotError(f"models[{index}].source_url must use https") + + for name in (model_id, *aliases): + previous = names.get(name) + if previous is not None: + raise PricingSnapshotError( + f"pricing alias {name!r} is ambiguous between " + f"{previous!r} and {model_id!r}" + ) + names[name] = model_id + models.append( + ModelPrice( + model_id=model_id, + aliases=aliases, + channel=channel, + token_semantics=token_semantics, + cache_write_ttl=cache_write_ttl, + rates_per_million=rates, + source_url=source_url, + ) + ) + + if sha256 is None: + encoded = json.dumps( + value, + ensure_ascii=True, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + sha256 = "sha256:" + hashlib.sha256(encoded).hexdigest() + if not re.fullmatch(r"sha256:[0-9a-f]{64}", sha256): + raise PricingSnapshotError("pricing snapshot hash must be sha256:<64 hex>") + + return PricingSnapshot( + snapshot_id=snapshot_id, + sha256=sha256, + currency=currency, + effective_date=effective_date, + retrieved_date=retrieved_date, + models=tuple(models), + source_file=Path(source_path).name if source_path else None, + ) + + +def project_cell_cost( + cell: Mapping[str, Any], snapshot: PricingSnapshot +) -> dict[str, Any]: + """Project one cell from complete raw token classes under ``snapshot``.""" + + model = str(cell.get("model") or "") + entry = snapshot.resolve_model(model) + if entry is None: + return _unavailable("model_not_in_snapshot", model=model) + + fields = { + "prompt_tokens": cell.get("prompt_tokens"), + "completion_tokens": cell.get("completion_tokens"), + "cache_read_input_tokens": cell.get("cache_read_input_tokens"), + "cache_creation_input_tokens": cell.get("cache_creation_input_tokens"), + } + missing = sorted(name for name, raw in fields.items() if raw is None) + if missing: + return _unavailable( + "missing_token_classes", + model=model, + missing_fields=missing, + ) + try: + tokens = {name: _nonnegative_number(raw, name) for name, raw in fields.items()} + except PricingSnapshotError as exc: + return _unavailable("invalid_token_classes", model=model, detail=str(exc)) + + prompt = tokens["prompt_tokens"] + cache_read = tokens["cache_read_input_tokens"] + cache_write = tokens["cache_creation_input_tokens"] + if entry.token_semantics == "prompt_includes_cache": + prompt -= cache_read + cache_write + if prompt < 0: + return _unavailable( + "inconsistent_token_classes", + model=model, + detail="cache tokens exceed prompt_tokens", + ) + + billable = { + "uncached_input": prompt, + "output": tokens["completion_tokens"], + "cache_read_input": cache_read, + "cache_write_input": cache_write, + } + total = sum( + billable[name] * entry.rates_per_million[name] / 1_000_000 + for name in _RATE_KEYS + ) + return { + "available": True, + "kind": "offline_projection", + "currency": snapshot.currency, + "total": total, + "model": model, + "pricing_model_id": entry.model_id, + "provider_channel": entry.channel, + "token_semantics": entry.token_semantics, + "cache_write_ttl": entry.cache_write_ttl, + "billable_tokens": billable, + "rates_per_million": dict(entry.rates_per_million), + "source_url": entry.source_url, + "snapshot": snapshot.public_identity(), + } + + +def _unavailable(reason: str, **details: Any) -> dict[str, Any]: + return { + "available": False, + "kind": "offline_projection", + "reason": reason, + **details, + } + + +def _required_string(value: Mapping[str, Any], key: str, *, prefix: str = "") -> str: + return _nonempty_string(value.get(key), f"{prefix}{key}") + + +def _nonempty_string(value: Any, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise PricingSnapshotError(f"{field} must be a non-empty string") + return value.strip() + + +def _date(value: Mapping[str, Any], key: str) -> str: + selected = _required_string(value, key) + if not _DATE_RE.fullmatch(selected): + raise PricingSnapshotError(f"{key} must use YYYY-MM-DD") + try: + date.fromisoformat(selected) + except ValueError as exc: + raise PricingSnapshotError(f"{key} must be a valid calendar date") from exc + return selected + + +def _rates(value: Any, *, index: int) -> dict[str, float]: + if not isinstance(value, Mapping): + raise PricingSnapshotError( + f"models[{index}].rates_per_million must be an object" + ) + missing = sorted(set(_RATE_KEYS) - set(value)) + extra = sorted(set(value) - set(_RATE_KEYS)) + if missing or extra: + raise PricingSnapshotError( + f"models[{index}].rates_per_million keys mismatch: " + f"missing={missing}, extra={extra}" + ) + return { + key: _nonnegative_number(value[key], f"models[{index}].rates_per_million.{key}") + for key in _RATE_KEYS + } + + +def _nonnegative_number(value: Any, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise PricingSnapshotError(f"{field} must be a non-negative number") + selected = float(value) + if ( + selected < 0 + or selected != selected + or selected in {float("inf"), float("-inf")} + ): + raise PricingSnapshotError(f"{field} must be a finite non-negative number") + return selected + + +__all__ = [ + "DEFAULT_PRICING_SNAPSHOT", + "PRICING_SNAPSHOT_SCHEMA", + "ModelPrice", + "PricingSnapshot", + "PricingSnapshotError", + "bundled_pricing_snapshot_path", + "load_pricing_snapshot", + "parse_pricing_snapshot", + "project_cell_cost", +] diff --git a/codenib/eval/reports/quality_cost.py b/codenib/eval/reports/quality_cost.py new file mode 100644 index 00000000..332d14ab --- /dev/null +++ b/codenib/eval/reports/quality_cost.py @@ -0,0 +1,1187 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Quality-constrained token and USD cost per successful localization.""" + +from __future__ import annotations + +import argparse +import json +import random +import statistics +import sys +from collections import defaultdict +from pathlib import Path +from typing import Any, Callable, Iterable, Mapping, Optional, Sequence + +from .pricing import PricingSnapshot, load_pricing_snapshot, project_cell_cost + +QUALITY_COST_REPORT_SCHEMA = "codenib.quality-cost-report.v1" +DEFAULT_BASELINE = "grep_only" +DEFAULT_QUALITY_SCOPE = "answer_blocks" +DEFAULT_QUALITY_FIELD = "recall" +DEFAULT_QUALITY_K = 5 +DEFAULT_SUCCESS_THRESHOLD = 1.0 +DEFAULT_NONINFERIORITY_MARGIN = 0.02 +DEFAULT_BOOTSTRAP_SAMPLES = 5000 + +_TOKEN_FIELDS = ( + "prompt_tokens", + "completion_tokens", + "total_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", +) + + +class QualityCostError(ValueError): + """Raised when cell identity or report denominators are not auditable.""" + + +def load_quality_cost_cells( + paths: Sequence[Path], +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Recursively load cell JSON while ignoring non-cell result metadata.""" + + if not paths: + raise QualityCostError("at least one cell file or directory is required") + candidates: dict[Path, set[str]] = defaultdict(set) + source_roots = [] + root_names = {} + for root_index, raw_path in enumerate(paths, start=1): + path = Path(raw_path).expanduser().resolve() + root_id = f"input-{root_index:02d}" + root_names[root_id] = path.name + source_roots.append({"id": root_id, "name": path.name}) + if path.is_file(): + candidates[path].add(root_id) + elif path.is_dir(): + for candidate in sorted(path.rglob("*.json")): + if candidate.is_file(): + candidates[candidate.resolve()].add(root_id) + else: + raise QualityCostError(f"cell input does not exist: {path}") + + cells = [] + retry_failures = [] + skipped_non_cells = 0 + for path in sorted(candidates): + try: + value = json.loads(path.read_text(encoding="utf-8")) + except OSError as exc: + raise QualityCostError(f"cannot read {path}: {exc}") from exc + except json.JSONDecodeError as exc: + raise QualityCostError(f"invalid cell JSON {path}: {exc}") from exc + if _looks_like_retry_failure(path, value): + retry = dict(value) + retry["_quality_cost_source"] = str(path) + retry["_quality_cost_root"] = _primary_source_root(candidates[path]) + retry_failures.append(retry) + continue + if not _looks_like_cell(value): + skipped_non_cells += 1 + continue + cell = dict(value) + cell["_quality_cost_source"] = str(path) + cell["_quality_cost_root"] = _primary_source_root(candidates[path]) + cells.append(cell) + + normalized = _validate_cells(cells) + if not normalized: + raise QualityCostError("no agent-runner cells found under the supplied inputs") + cells_by_id: dict[str, list[dict[str, Any]]] = defaultdict(list) + for cell in normalized: + if cell.get("cell_id"): + cells_by_id[str(cell["cell_id"])].append(cell) + linked_retries = 0 + for retry in retry_failures: + cell_id = str(retry.get("cell_id") or "") + candidates_for_retry = cells_by_id.get(cell_id, []) + same_root = [ + cell + for cell in candidates_for_retry + if cell.get("_quality_cost_root") == retry.get("_quality_cost_root") + ] + if len(same_root) == 1: + target = same_root[0] + elif len(candidates_for_retry) == 1: + target = candidates_for_retry[0] + elif not candidates_for_retry: + raise QualityCostError( + f"retry failure {retry['_quality_cost_source']} has no final cell " + f"with id {cell_id!r}" + ) + else: + raise QualityCostError( + f"retry failure {retry['_quality_cost_source']} ambiguously matches " + f"{len(candidates_for_retry)} final cells with id {cell_id!r}" + ) + retry["model"] = target["model"] + target.setdefault("_quality_cost_retry_attempts", []).append(retry) + linked_retries += 1 + roots = {} + for root_id, root_name in root_names.items(): + root_cells = [ + cell for cell in normalized if cell.get("_quality_cost_root") == root_id + ] + root_retries = [ + retry + for retry in retry_failures + if retry.get("_quality_cost_root") == root_id + ] + roots[root_id] = { + "name": root_name, + "cell_files": len(root_cells), + "models": { + model: sum(cell["model"] == model for cell in root_cells) + for model in sorted({cell["model"] for cell in root_cells}) + }, + "retry_failure_records": len(root_retries), + } + return normalized, { + "source_roots": source_roots, + "roots": roots, + "candidate_json_files": len(candidates), + "cell_files": len(normalized), + "skipped_non_cell_json": skipped_non_cells, + "retry_failure_records": len(retry_failures), + "linked_retry_failures": linked_retries, + "retry_failures_with_total_tokens": sum( + retry.get("total_tokens") is not None for retry in retry_failures + ), + "retry_failures_with_cost_usd": sum( + retry.get("cost_usd") is not None for retry in retry_failures + ), + "model_count": len({cell["model"] for cell in normalized}), + } + + +def analyze_quality_cost( + cells: Sequence[Mapping[str, Any]], + *, + baseline: str = DEFAULT_BASELINE, + quality_scope: str = DEFAULT_QUALITY_SCOPE, + quality_field: str = DEFAULT_QUALITY_FIELD, + quality_k: int = DEFAULT_QUALITY_K, + success_threshold: float = DEFAULT_SUCCESS_THRESHOLD, + noninferiority_margin: float = DEFAULT_NONINFERIORITY_MARGIN, + bootstrap_samples: int = DEFAULT_BOOTSTRAP_SAMPLES, + bootstrap_seed: int = 0, + strict_denominators: bool = True, + included_arms: Optional[Sequence[str]] = None, + pricing_snapshot: Optional[PricingSnapshot] = None, + usd_source: str = "auto", + shared_build_cost_usd: Optional[float] = None, + query_horizons: Sequence[int] = (), +) -> dict[str, Any]: + """Analyze paired arms without dropping failures or unpriced attempts.""" + + _validate_contract( + baseline=baseline, + quality_scope=quality_scope, + quality_field=quality_field, + quality_k=quality_k, + success_threshold=success_threshold, + noninferiority_margin=noninferiority_margin, + bootstrap_samples=bootstrap_samples, + usd_source=usd_source, + shared_build_cost_usd=shared_build_cost_usd, + query_horizons=query_horizons, + included_arms=included_arms, + ) + normalized = _validate_cells(cells) + input_cells = list(normalized) + models_in_input = sorted({cell["model"] for cell in input_cells}) + selected_arms = tuple(str(arm).strip() for arm in (included_arms or ())) + if selected_arms: + normalized = [cell for cell in normalized if cell["subset_id"] in selected_arms] + excluded_cells = [ + cell for cell in input_cells if cell["subset_id"] not in selected_arms + ] + else: + excluded_cells = [] + by_model: dict[str, list[dict[str, Any]]] = defaultdict(list) + for cell in normalized: + by_model[cell["model"]].append(cell) + + model_reports = {} + for model in models_in_input: + model_cells = by_model.get(model, []) + actual_arms = {cell["subset_id"] for cell in model_cells} + missing_arms = sorted(set(selected_arms) - actual_arms) + if missing_arms: + raise QualityCostError( + f"model {model!r} is missing requested arms {missing_arms}" + ) + model_reports[model] = _analyze_model( + model, + model_cells, + baseline=baseline, + quality_scope=quality_scope, + quality_field=quality_field, + quality_k=quality_k, + success_threshold=success_threshold, + noninferiority_margin=noninferiority_margin, + bootstrap_samples=bootstrap_samples, + bootstrap_seed=bootstrap_seed, + strict_denominators=strict_denominators, + pricing_snapshot=pricing_snapshot, + usd_source=usd_source, + ) + + return { + "schema": QUALITY_COST_REPORT_SCHEMA, + "contract": { + "unit": "query repetition", + "baseline": baseline, + "quality_metric": f"{quality_scope}.{quality_field}@{quality_k}", + "localization_success_threshold": success_threshold, + "noninferiority_margin": noninferiority_margin, + "misses_are_charged": True, + "infrastructure_failures_score_zero_quality": True, + "strict_denominators": strict_denominators, + "included_arms": list(selected_arms) if selected_arms else None, + "usd_source": usd_source, + "claim_boundary": "successful localization, not issue resolution", + }, + "bootstrap": { + "method": "repository-clustered paired percentile bootstrap", + "confidence": 0.95, + "samples": bootstrap_samples, + "seed": bootstrap_seed, + }, + "pricing_snapshot": ( + pricing_snapshot.public_identity() if pricing_snapshot else None + ), + "arm_selection_audit": { + "input_cell_count": len(input_cells), + "analyzed_cell_count": len(normalized), + "excluded_cell_count": len(excluded_cells), + "excluded_arms": { + model: { + arm: sum( + cell["model"] == model and cell["subset_id"] == arm + for cell in excluded_cells + ) + for arm in sorted( + { + cell["subset_id"] + for cell in excluded_cells + if cell["model"] == model + } + ) + } + for model in models_in_input + if any(cell["model"] == model for cell in excluded_cells) + }, + }, + "shared_build_cost": _build_cost_report(shared_build_cost_usd, query_horizons), + "models": model_reports, + } + + +def render_quality_cost_markdown(report: Mapping[str, Any]) -> str: + """Render a concise artifact- and paper-facing report.""" + + contract = report["contract"] + lines = ["# Quality-constrained cost per localization", ""] + lines.append( + "The unit is one attempted query repetition. A success reaches " + f"`{contract['quality_metric']} >= " + f"{contract['localization_success_threshold']:.3g}`. All paired attempts, " + "including misses and recorded infrastructure failures, remain in the " + "token and cost denominator. This is localization, not issue resolution." + ) + lines.append("") + lines.append( + "An arm qualifies when the lower bound of its repository-clustered paired " + f"quality-delta CI is greater than `-{contract['noninferiority_margin']:.3g}` " + f"relative to `{contract['baseline']}`." + ) + lines.append("") + + for model, model_report in report["models"].items(): + audit = model_report["denominator_audit"] + lines.extend( + [ + f"## `{model}`", + "", + f"Paired units: {audit['common_unit_count']}; queries: " + f"{audit['query_count']}; repositories: {audit['repository_count']}.", + "", + _render_optimum( + "Token optimum", + model_report["constrained_optima"]["tokens_per_success"], + formatter=lambda value: f"{value:,.0f} tokens/success", + ), + _render_optimum( + "USD optimum", + model_report["constrained_optima"]["usd_per_success"], + formatter=lambda value: f"${value:.6f}/success", + ), + "", + "| arm | attempts | infra | loc success | quality | delta [95% CI] " + "| qualified | tokens/attempt | tokens/success | USD source " + "| USD/attempt | USD/success |", + "| --- | ---: | ---: | ---: | ---: | ---: | :---: | ---: " + "| ---: | --- | ---: | ---: |", + ] + ) + for arm, arm_report in model_report["arms"].items(): + ci = arm_report["quality_delta"]["ci95"] + delta = arm_report["quality_delta"]["point"] + delta_text = ( + f"{delta:+.3f} [{ci['lower']:+.3f}, {ci['upper']:+.3f}]" + if delta is not None and ci["lower"] is not None + else "n/a" + ) + tokens = arm_report["tokens"] + usd = arm_report["usd"] + lines.append( + "| " + + " | ".join( + [ + arm, + str(arm_report["attempted"]), + f"{arm_report['infrastructure_success_rate']:.1%}", + f"{arm_report['localization_success_rate']:.1%}", + f"{arm_report['quality_mean']:.3f}", + delta_text, + "yes" if arm_report["quality_qualified"] else "no", + _format_number(tokens["per_attempt"], 0), + _format_number(tokens["per_success"], 0), + usd.get("kind") if usd.get("available") else "unavailable", + _format_money(usd.get("per_attempt")), + _format_money(usd.get("per_success")), + ] + ) + + " |" + ) + lines.append("") + unavailable = [ + f"`{arm}`: {values['usd'].get('reason')}" + for arm, values in model_report["arms"].items() + if not values["usd"].get("available") + ] + if unavailable: + lines.append("USD unavailable for " + "; ".join(unavailable) + ".") + lines.append("") + + shared = report.get("shared_build_cost") + if shared: + lines.extend( + [ + "## Shared build-cost amortization", + "", + "This cost is reported separately and is not added to model-call cost.", + "", + "| query horizon | shared build USD/query |", + "| ---: | ---: |", + ] + ) + for row in shared["horizons"]: + lines.append( + f"| {row['queries']} | {_format_money(row['usd_per_query'])} |" + ) + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +def _render_optimum( + label: str, + optimum: Mapping[str, Any], + *, + formatter: Callable[[float], str], +) -> str: + selected = optimum.get("selected_arm") + value = optimum.get("value") + if selected is None or value is None: + return f"{label}: unavailable ({optimum.get('reason', 'no eligible arm')})." + comparison = "" + savings = optimum.get("savings_fraction_vs_baseline") + if selected == optimum["baseline"]: + comparison = "; baseline remains the constrained optimum" + elif savings is not None and savings >= 0: + comparison = f"; {savings:.1%} less than `{optimum['baseline']}`" + elif savings is not None: + comparison = f"; {-savings:.1%} more than `{optimum['baseline']}`" + return f"{label}: `{selected}` at {formatter(value)}{comparison}." + + +def write_quality_cost_report( + *, + cells: Sequence[Mapping[str, Any]], + output_dir: Path, + load_audit: Optional[Mapping[str, Any]] = None, + **analysis_options: Any, +) -> dict[str, Any]: + """Write ``quality_cost.json`` and ``quality_cost.md``.""" + + report = analyze_quality_cost(cells, **analysis_options) + report["load_audit"] = dict(load_audit or {}) + output = Path(output_dir) + output.mkdir(parents=True, exist_ok=True) + (output / "quality_cost.json").write_text( + json.dumps(report, indent=2, sort_keys=True, ensure_ascii=True) + "\n", + encoding="utf-8", + ) + (output / "quality_cost.md").write_text( + render_quality_cost_markdown(report), encoding="utf-8" + ) + return report + + +def _analyze_model( + model: str, + cells: Sequence[dict[str, Any]], + *, + baseline: str, + quality_scope: str, + quality_field: str, + quality_k: int, + success_threshold: float, + noninferiority_margin: float, + bootstrap_samples: int, + bootstrap_seed: int, + strict_denominators: bool, + pricing_snapshot: Optional[PricingSnapshot], + usd_source: str, +) -> dict[str, Any]: + by_arm: dict[str, dict[tuple[str, int], dict[str, Any]]] = defaultdict(dict) + for cell in cells: + key = _unit_key(cell) + by_arm[cell["subset_id"]][key] = cell + if baseline not in by_arm: + raise QualityCostError(f"model {model!r} has no baseline arm {baseline!r}") + if len(by_arm) < 2: + raise QualityCostError(f"model {model!r} needs at least two arms") + + unit_sets = {arm: set(rows) for arm, rows in by_arm.items()} + union = set().union(*unit_sets.values()) + common = set.intersection(*unit_sets.values()) + mismatched = { + arm: { + "missing_count": len(union - keys), + "extra_vs_baseline_count": len(keys - unit_sets[baseline]), + "missing_examples": [ + _format_unit_key(key) for key in sorted(union - keys)[:10] + ], + } + for arm, keys in sorted(unit_sets.items()) + } + if strict_denominators and any( + keys != unit_sets[baseline] for keys in unit_sets.values() + ): + detail = ", ".join( + f"{arm}={len(keys)}" for arm, keys in sorted(unit_sets.items()) + ) + raise QualityCostError( + f"model {model!r} has unmatched paired denominators: {detail}; " + f"baseline={len(unit_sets[baseline])}" + ) + if not common: + raise QualityCostError(f"model {model!r} has no complete paired units") + + ordered_keys = sorted(common) + baseline_cells = [by_arm[baseline][key] for key in ordered_keys] + baseline_quality = [ + _quality_value( + cell, + scope=quality_scope, + field=quality_field, + k=quality_k, + ) + for cell in baseline_cells + ] + arm_reports = {} + for arm in sorted(by_arm, key=lambda name: (name != baseline, name)): + selected = [by_arm[arm][key] for key in ordered_keys] + quality = [ + _quality_value( + cell, + scope=quality_scope, + field=quality_field, + k=quality_k, + ) + for cell in selected + ] + delta_rows = [ + { + "repository": _repository_key(cell), + "delta": arm_value - baseline_value, + } + for cell, arm_value, baseline_value in zip( + selected, quality, baseline_quality, strict=True + ) + ] + point = statistics.fmean(row["delta"] for row in delta_rows) + samples = _repository_clustered_bootstrap( + delta_rows, + samples=bootstrap_samples, + seed=bootstrap_seed, + ) + lower = _percentile(samples, 0.025) + upper = _percentile(samples, 0.975) + localization_successes = sum( + bool(cell["success"] and value >= success_threshold) + for cell, value in zip(selected, quality, strict=True) + ) + arm_reports[arm] = { + "attempted": len(selected), + "infrastructure_successes": sum(bool(cell["success"]) for cell in selected), + "infrastructure_success_rate": sum( + bool(cell["success"]) for cell in selected + ) + / len(selected), + "localization_successes": localization_successes, + "localization_success_rate": localization_successes / len(selected), + "quality_mean": statistics.fmean(quality), + "quality_delta": { + "estimand": f"{arm} - {baseline}", + "point": point, + "ci95": {"lower": lower, "upper": upper}, + }, + "quality_qualified": bool( + arm == baseline + or (lower is not None and lower > -noninferiority_margin) + ), + "tokens": _token_report(selected, localization_successes), + "usd": _usd_report( + selected, + localization_successes, + pricing_snapshot=pricing_snapshot, + usd_source=usd_source, + ), + } + + constrained_optima = { + "tokens_per_success": _constrained_optimum( + arm_reports, + baseline=baseline, + report_key="tokens", + value_key="per_success", + ), + "usd_per_success": _constrained_optimum( + arm_reports, + baseline=baseline, + report_key="usd", + value_key="per_success", + ), + } + return { + "model": model, + "baseline": baseline, + "denominator_audit": { + "strict": strict_denominators, + "arm_unit_counts": { + arm: len(keys) for arm, keys in sorted(unit_sets.items()) + }, + "common_unit_count": len(common), + "union_unit_count": len(union), + "query_count": len({key[0] for key in common}), + "repository_count": len( + {_repository_key(by_arm[baseline][key]) for key in ordered_keys} + ), + "mismatches": mismatched, + }, + "arms": arm_reports, + "constrained_optima": constrained_optima, + } + + +def _constrained_optimum( + arms: Mapping[str, Mapping[str, Any]], + *, + baseline: str, + report_key: str, + value_key: str, +) -> dict[str, Any]: + """Select the least-cost arm only after the quality guard passes.""" + + candidates = [] + for arm, arm_report in arms.items(): + cost_report = arm_report[report_key] + value = cost_report.get(value_key) + if arm_report["quality_qualified"] and value is not None: + candidates.append((float(value), arm)) + candidates.sort(key=lambda row: (row[0], row[1])) + + baseline_value = arms[baseline][report_key].get(value_key) + if not candidates: + return { + "objective": f"minimize {report_key}.{value_key}", + "baseline": baseline, + "baseline_value": baseline_value, + "eligible_arms": [], + "selected_arm": None, + "value": None, + "ratio_to_baseline": None, + "savings_fraction_vs_baseline": None, + "reason": "no quality-qualified arm has a complete cost denominator", + } + + value, selected = candidates[0] + ratio = None + if baseline_value is not None and float(baseline_value) > 0: + ratio = value / float(baseline_value) + return { + "objective": f"minimize {report_key}.{value_key}", + "baseline": baseline, + "baseline_value": baseline_value, + "eligible_arms": [arm for _, arm in candidates], + "selected_arm": selected, + "value": value, + "ratio_to_baseline": ratio, + "savings_fraction_vs_baseline": 1.0 - ratio if ratio is not None else None, + "reason": None, + } + + +def _token_report( + cells: Sequence[Mapping[str, Any]], localization_successes: int +) -> dict[str, Any]: + attempts = list(_execution_attempts(cells)) + sums = {} + coverage = {} + for field in _TOKEN_FIELDS: + values = [_optional_nonnegative(cell.get(field), field) for cell in attempts] + known = [value for value in values if value is not None] + sums[field] = sum(known) + coverage[field] = len(known) / len(attempts) + complete = coverage["total_tokens"] == 1.0 + total = sums["total_tokens"] if complete else None + missing_retry_tokens = any( + cell.get("_quality_cost_is_retry") and cell.get("total_tokens") is None + for cell in attempts + ) + return { + "available": complete, + "reason": ( + None + if complete + else ( + "unmetered_retry_attempts" + if missing_retry_tokens + else "missing_total_tokens" + ) + ), + "field_coverage": coverage, + "known_sums": sums, + "query_repetitions": len(cells), + "execution_attempts": len(attempts), + "total": total, + "per_attempt": total / len(cells) if total is not None else None, + "per_success": ( + total / localization_successes + if total is not None and localization_successes + else None + ), + } + + +def _usd_report( + cells: Sequence[Mapping[str, Any]], + localization_successes: int, + *, + pricing_snapshot: Optional[PricingSnapshot], + usd_source: str, +) -> dict[str, Any]: + projected = _projected_usd(cells, pricing_snapshot) + recorded = _recorded_usd(cells) + if usd_source == "projected": + selected = projected + elif usd_source == "recorded": + selected = recorded + else: + selected = projected if projected.get("available") else recorded + result = dict(selected) + result["alternatives"] = { + "projected": _cost_availability(projected), + "recorded": _cost_availability(recorded), + } + if result.get("available"): + total = float(result["total"]) + result["per_attempt"] = total / len(cells) + result["per_success"] = ( + total / localization_successes if localization_successes else None + ) + else: + result["per_attempt"] = None + result["per_success"] = None + return result + + +def _projected_usd( + cells: Sequence[Mapping[str, Any]], snapshot: Optional[PricingSnapshot] +) -> dict[str, Any]: + if snapshot is None: + return { + "available": False, + "kind": "offline_projection", + "reason": "no_pricing_snapshot", + } + attempts = list(_execution_attempts(cells)) + rows = [project_cell_cost(cell, snapshot) for cell in attempts] + unavailable = [row for row in rows if not row.get("available")] + if unavailable: + reasons = sorted({str(row.get("reason")) for row in unavailable}) + return { + "available": False, + "kind": "offline_projection", + "reason": "+".join(reasons), + "priced_attempts": len(rows) - len(unavailable), + "attempted": len(rows), + "snapshot": snapshot.public_identity(), + } + return { + "available": True, + "kind": "offline_projection", + "currency": snapshot.currency, + "total": sum(float(row["total"]) for row in rows), + "priced_attempts": len(rows), + "attempted": len(rows), + "snapshot": snapshot.public_identity(), + "pricing_model_id": rows[0]["pricing_model_id"], + "provider_channel": rows[0]["provider_channel"], + "source_url": rows[0]["source_url"], + } + + +def _recorded_usd(cells: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + attempts = list(_execution_attempts(cells)) + values = [ + _optional_nonnegative(cell.get("cost_usd"), "cost_usd") for cell in attempts + ] + known = [value for value in values if value is not None] + if len(known) != len(attempts): + missing_retry_cost = any( + cell.get("_quality_cost_is_retry") and cell.get("cost_usd") is None + for cell in attempts + ) + return { + "available": False, + "kind": "recorded_runtime_estimate", + "reason": ( + "unmetered_retry_attempts" + if missing_retry_cost + else "missing_recorded_cost" + ), + "priced_attempts": len(known), + "attempted": len(attempts), + } + total = sum(known) + ambiguous_zero = any( + value == 0 and (cell.get("cost_provenance") or {}).get("kind") != "metered_zero" + for cell, value in zip(attempts, values, strict=True) + ) + if ambiguous_zero: + return { + "available": False, + "kind": "recorded_runtime_estimate", + "reason": "ambiguous_zero_runtime_estimates", + "priced_attempts": len(known), + "attempted": len(attempts), + } + provenance = [] + for cell in attempts: + raw = cell.get("cost_provenance") + if isinstance(raw, Mapping): + identity = { + key: raw.get(key) + for key in ("kind", "calculator", "calculator_version", "model") + if raw.get(key) is not None + } + if identity and identity not in provenance: + provenance.append(identity) + return { + "available": True, + "kind": "recorded_runtime_estimate", + "currency": "USD", + "total": total, + "priced_attempts": len(known), + "attempted": len(attempts), + "provenance": provenance or [{"kind": "legacy_cell_cost_usd"}], + "reproducibility": "unpinned calculator estimate", + } + + +def _cost_availability(value: Mapping[str, Any]) -> dict[str, Any]: + return { + key: value.get(key) + for key in ("available", "kind", "reason", "priced_attempts", "attempted") + if key in value + } + + +def _repository_clustered_bootstrap( + rows: Sequence[Mapping[str, Any]], *, samples: int, seed: int +) -> list[float]: + by_repository: dict[str, list[float]] = defaultdict(list) + for row in rows: + by_repository[str(row["repository"])].append(float(row["delta"])) + repositories = sorted(by_repository) + rng = random.Random(seed) + result = [] + for _ in range(samples): + values = [] + for repository in rng.choices(repositories, k=len(repositories)): + values.extend(by_repository[repository]) + result.append(statistics.fmean(values)) + return result + + +def _percentile(values: Sequence[float], quantile: float) -> Optional[float]: + if not values: + return None + ordered = sorted(float(value) for value in values) + if len(ordered) == 1: + return ordered[0] + position = (len(ordered) - 1) * quantile + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + weight = position - lower + return ordered[lower] * (1 - weight) + ordered[upper] * weight + + +def _quality_value(cell: Mapping[str, Any], *, scope: str, field: str, k: int) -> float: + if not cell["success"]: + return 0.0 + metrics = cell.get("metrics") + if not isinstance(metrics, Mapping): + raise QualityCostError(f"successful cell {_cell_label(cell)} has no metrics") + scoped = metrics.get(scope) + if not isinstance(scoped, Mapping): + raise QualityCostError( + f"successful cell {_cell_label(cell)} lacks metric scope {scope!r}" + ) + cutoff = scoped.get(k) or scoped.get(str(k)) + if not isinstance(cutoff, Mapping): + raise QualityCostError(f"successful cell {_cell_label(cell)} lacks {scope}@{k}") + value = cutoff.get(field) + numeric = _optional_nonnegative(value, f"{scope}.{field}@{k}") + if numeric is None or numeric > 1: + raise QualityCostError( + f"successful cell {_cell_label(cell)} has invalid {scope}.{field}@{k}" + ) + return numeric + + +def _validate_cells( + cells: Sequence[Mapping[str, Any]], +) -> list[dict[str, Any]]: + normalized = [] + identities: dict[tuple[str, str, str, int], str] = {} + query_identity: dict[tuple[str, str, int], tuple[str, Optional[str]]] = {} + for index, raw in enumerate(cells): + if not isinstance(raw, Mapping): + raise QualityCostError(f"cell {index} must be an object") + cell = dict(raw) + model = _required_cell_string(cell, "model", index) + query_id = _required_cell_string(cell, "query_id", index) + instance_id = _required_cell_string(cell, "instance_id", index) + arm = _required_cell_string(cell, "subset_id", index) + rep = cell.get("rep") + if isinstance(rep, bool) or not isinstance(rep, int) or rep < 1: + raise QualityCostError(f"cell {index} rep must be a positive integer") + if not isinstance(cell.get("success"), bool): + raise QualityCostError(f"cell {index} success must be boolean") + if cell.get("metrics_meaningful") is False: + raise QualityCostError( + f"cell {index} has no meaningful ground truth for quality reporting" + ) + cell.update( + model=model, + query_id=query_id, + instance_id=instance_id, + subset_id=arm, + rep=rep, + ) + identity = (model, query_id, arm, rep) + source = str(cell.get("_quality_cost_source") or cell.get("cell_id") or index) + if identity in identities: + raise QualityCostError( + f"duplicate cell {identity!r}: {identities[identity]} and {source}" + ) + identities[identity] = source + + query_key = (model, query_id, rep) + query_text = str(cell.get("query")) if cell.get("query") is not None else None + expected = query_identity.setdefault(query_key, (instance_id, query_text)) + if expected[0] != instance_id or ( + expected[1] is not None + and query_text is not None + and expected[1] != query_text + ): + raise QualityCostError( + f"query identity mismatch for {query_key!r}: " + f"expected {expected!r}, found {(instance_id, query_text)!r}" + ) + normalized.append(cell) + return normalized + + +def _validate_contract( + *, + baseline: str, + quality_scope: str, + quality_field: str, + quality_k: int, + success_threshold: float, + noninferiority_margin: float, + bootstrap_samples: int, + usd_source: str, + shared_build_cost_usd: Optional[float], + query_horizons: Sequence[int], + included_arms: Optional[Sequence[str]], +) -> None: + if not baseline.strip() or not quality_scope.strip() or not quality_field.strip(): + raise QualityCostError("baseline and quality metric names must be non-empty") + if quality_k < 1: + raise QualityCostError("quality_k must be positive") + if not 0 <= success_threshold <= 1: + raise QualityCostError("success_threshold must be in [0, 1]") + if not 0 <= noninferiority_margin < 1: + raise QualityCostError("noninferiority_margin must be in [0, 1)") + if bootstrap_samples < 1: + raise QualityCostError("bootstrap_samples must be positive") + if usd_source not in {"auto", "recorded", "projected"}: + raise QualityCostError("usd_source must be auto, recorded, or projected") + if shared_build_cost_usd is not None: + _nonnegative(shared_build_cost_usd, "shared_build_cost_usd") + for horizon in query_horizons: + if isinstance(horizon, bool) or not isinstance(horizon, int) or horizon < 1: + raise QualityCostError("query horizons must be positive integers") + if query_horizons and shared_build_cost_usd is None: + raise QualityCostError("query horizons require shared_build_cost_usd") + if included_arms is not None: + normalized_arms = [str(arm).strip() for arm in included_arms] + if any(not arm for arm in normalized_arms): + raise QualityCostError("included arms must be non-empty strings") + if len(set(normalized_arms)) != len(normalized_arms): + raise QualityCostError("included arms must be unique") + if baseline not in normalized_arms: + raise QualityCostError("included arms must contain the baseline") + if len(normalized_arms) < 2: + raise QualityCostError("at least two included arms are required") + + +def _build_cost_report( + cost: Optional[float], horizons: Sequence[int] +) -> Optional[dict[str, Any]]: + if cost is None: + return None + return { + "currency": "USD", + "total": float(cost), + "included_in_model_call_cost": False, + "horizons": [ + {"queries": horizon, "usd_per_query": float(cost) / horizon} + for horizon in sorted(set(horizons)) + ], + } + + +def _execution_attempts( + cells: Sequence[Mapping[str, Any]], +) -> Iterable[Mapping[str, Any]]: + for cell in cells: + yield cell + retries = cell.get("_quality_cost_retry_attempts") or [] + if not isinstance(retries, Sequence) or isinstance(retries, (str, bytes)): + raise QualityCostError( + f"cell {_cell_label(cell)} retry attempts must be an array" + ) + for retry in retries: + if not isinstance(retry, Mapping): + raise QualityCostError( + f"cell {_cell_label(cell)} contains a malformed retry attempt" + ) + normalized = dict(retry) + normalized["model"] = cell["model"] + normalized["_quality_cost_is_retry"] = True + yield normalized + + +def _looks_like_cell(value: Any) -> bool: + required = { + "model", + "instance_id", + "query_id", + "subset_id", + "rep", + "success", + } + return isinstance(value, Mapping) and required <= set(value) + + +def _looks_like_retry_failure(path: Path, value: Any) -> bool: + return ( + isinstance(value, Mapping) + and path.name == "failed_cell.json" + and "failure_retries" in path.parts + and isinstance(value.get("cell_id"), str) + and bool(value.get("cell_id")) + ) + + +def _primary_source_root(roots: Iterable[str]) -> str: + selected = sorted(set(roots), key=lambda root: (len(Path(root).parts), root)) + if not selected: + raise QualityCostError("internal error: candidate JSON has no source root") + return selected[-1] + + +def _required_cell_string(cell: Mapping[str, Any], key: str, index: int) -> str: + value = cell.get(key) + if not isinstance(value, str) or not value.strip(): + source = cell.get("_quality_cost_source") or index + raise QualityCostError(f"cell {source} {key} must be a non-empty string") + return value.strip() + + +def _unit_key(cell: Mapping[str, Any]) -> tuple[str, int]: + return str(cell["query_id"]), int(cell["rep"]) + + +def _format_unit_key(key: tuple[str, int]) -> str: + return f"{key[0]}#rep{key[1]}" + + +def _repository_key(cell: Mapping[str, Any]) -> str: + explicit = cell.get("repository") or cell.get("repo") + if isinstance(explicit, str) and explicit.strip(): + return explicit.strip() + instance_id = str(cell["instance_id"]) + head, separator, tail = instance_id.rpartition("-") + return head if separator and tail.isdigit() else instance_id + + +def _cell_label(cell: Mapping[str, Any]) -> str: + return str(cell.get("cell_id") or _format_unit_key(_unit_key(cell))) + + +def _optional_nonnegative(value: Any, field: str) -> Optional[float]: + if value is None: + return None + return _nonnegative(value, field) + + +def _nonnegative(value: Any, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise QualityCostError(f"{field} must be a non-negative number") + number = float(value) + if number < 0 or number != number or number in {float("inf"), float("-inf")}: + raise QualityCostError(f"{field} must be a finite non-negative number") + return number + + +def _format_number(value: Optional[float], digits: int) -> str: + if value is None: + return "n/a" + return f"{value:,.{digits}f}" + + +def _format_money(value: Optional[float]) -> str: + if value is None: + return "n/a" + return f"${value:.6f}" + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="codenib-quality-cost-report", + description="Report quality-constrained cost per successful localization.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("inputs", nargs="+", type=Path) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--baseline", default=DEFAULT_BASELINE) + parser.add_argument("--quality-scope", default=DEFAULT_QUALITY_SCOPE) + parser.add_argument("--quality-field", default=DEFAULT_QUALITY_FIELD) + parser.add_argument("--quality-k", type=int, default=DEFAULT_QUALITY_K) + parser.add_argument( + "--success-threshold", type=float, default=DEFAULT_SUCCESS_THRESHOLD + ) + parser.add_argument( + "--noninferiority-margin", + type=float, + default=DEFAULT_NONINFERIORITY_MARGIN, + ) + parser.add_argument( + "--bootstrap-samples", type=int, default=DEFAULT_BOOTSTRAP_SAMPLES + ) + parser.add_argument("--bootstrap-seed", type=int, default=0) + parser.add_argument("--allow-unmatched", action="store_true") + parser.add_argument( + "--arm", + dest="included_arms", + action="append", + help="Include only this arm; repeat for each compared arm.", + ) + parser.add_argument("--pricing-snapshot", type=Path) + parser.add_argument( + "--usd-source", choices=("auto", "recorded", "projected"), default="auto" + ) + parser.add_argument("--shared-build-cost-usd", type=float) + parser.add_argument("--query-horizon", type=int, action="append", default=[]) + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> int: + """CLI entry point.""" + + args = _build_parser().parse_args(argv) + try: + cells, load_audit = load_quality_cost_cells(args.inputs) + snapshot = ( + load_pricing_snapshot(args.pricing_snapshot) + if args.pricing_snapshot + else None + ) + report = write_quality_cost_report( + cells=cells, + output_dir=args.output_dir, + load_audit=load_audit, + baseline=args.baseline, + quality_scope=args.quality_scope, + quality_field=args.quality_field, + quality_k=args.quality_k, + success_threshold=args.success_threshold, + noninferiority_margin=args.noninferiority_margin, + bootstrap_samples=args.bootstrap_samples, + bootstrap_seed=args.bootstrap_seed, + strict_denominators=not args.allow_unmatched, + included_arms=args.included_arms, + pricing_snapshot=snapshot, + usd_source=args.usd_source, + shared_build_cost_usd=args.shared_build_cost_usd, + query_horizons=args.query_horizon, + ) + except (QualityCostError, ValueError) as exc: + print(f"codenib-quality-cost-report: {exc}", file=sys.stderr) + return 2 + print(render_quality_cost_markdown(report), end="") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = [ + "DEFAULT_BASELINE", + "DEFAULT_BOOTSTRAP_SAMPLES", + "DEFAULT_NONINFERIORITY_MARGIN", + "DEFAULT_QUALITY_FIELD", + "DEFAULT_QUALITY_K", + "DEFAULT_QUALITY_SCOPE", + "DEFAULT_SUCCESS_THRESHOLD", + "QUALITY_COST_REPORT_SCHEMA", + "QualityCostError", + "analyze_quality_cost", + "load_quality_cost_cells", + "main", + "render_quality_cost_markdown", + "write_quality_cost_report", +] diff --git a/codenib/llm/usage.py b/codenib/llm/usage.py index 2225c59b..60dc5544 100644 --- a/codenib/llm/usage.py +++ b/codenib/llm/usage.py @@ -20,7 +20,9 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, List, Optional +from functools import lru_cache +from importlib.metadata import PackageNotFoundError, version +from typing import Any, Dict, List, Optional from ..log_utils import get_logger @@ -32,9 +34,11 @@ class TokenUsage: """Cumulative token counts (plus optional USD cost). ``cache_read_input_tokens`` / ``cache_creation_input_tokens`` are the - Anthropic prompt-cache breakdown of ``prompt_tokens`` (the cached prefix is - *included* in ``prompt_tokens``): reads are billed ~0.1x input, writes - ~1.25x. Both are 0 when prompt caching is off or unsupported. + provider's prompt-cache classes. Anthropic reports them separately from + uncached ``input_tokens``; total processed input is therefore their sum. + Other providers may normalize ``prompt_tokens`` differently, so offline + repricing must declare the token semantics in its pricing snapshot rather + than infer them here. Cache fields are 0 when caching is off or unsupported. """ prompt_tokens: int = 0 @@ -128,6 +132,30 @@ def totals(self) -> TokenUsage: return total +def runtime_cost_provenance(model: str, cost_usd: Optional[float]) -> Dict[str, Any]: + """Describe a best-effort LiteLLM runtime estimate without pinning a price. + + The calculator version makes historical records auditable, but this is not + an immutable pricing snapshot. Reports that need reproducible repricing + must use raw token classes and a separate catalog. + """ + + return { + "kind": "runtime_estimate" if cost_usd is not None else "unavailable", + "calculator": "litellm.completion_cost", + "calculator_version": _installed_litellm_version(), + "model": str(model), + } + + +@lru_cache(maxsize=1) +def _installed_litellm_version() -> Optional[str]: + try: + return version("litellm") + except PackageNotFoundError: + return None + + def _extract_token_usage(response: Any) -> TokenUsage: """Pull ``prompt_tokens``/``completion_tokens``/``total_tokens`` off a response. diff --git a/docs/evaluation_artifacts.md b/docs/evaluation_artifacts.md index 109ca19e..77393c72 100644 --- a/docs/evaluation_artifacts.md +++ b/docs/evaluation_artifacts.md @@ -127,3 +127,27 @@ whose per-cell JSON already existed and were reused under `resume`), `skipped` (cell- or instance-level failures, each with a reason). Bundle the sweep output directory — summary plus `cells/` — with the manifest flow above to freeze both the protocol and the per-cell results. + +## Quality-constrained cost reports + +Use `codenib-quality-cost-report` to compare agent arms without dropping misses +or selecting a cheap arm that fails the localization-quality guard. Inputs may +be complete sweep directories or individual cell JSON files; sharded `cells/` +directories are discovered recursively. + +```bash +codenib-quality-cost-report /path/to/baseline-and-policy-results \ + --output-dir /path/to/report \ + --arm grep_only \ + --arm preinj_eager \ + --arm preinj_eager_compact \ + --noninferiority-margin 0.05 +``` + +The report emits JSON and Markdown with exact paired denominators, localization +success, token cost per attempt and per success, repository-clustered confidence +intervals, and the least-cost arm that clears the declared quality margin. +Recorded USD estimates remain labeled as unpinned. Offline repricing requires +complete raw token classes and an explicit content-hashed pricing snapshot; +unknown local-model prices are reported as unavailable. These values are not +GitHub or Copilot credit estimates. diff --git a/docs/experiments/quality_cost.md b/docs/experiments/quality_cost.md new file mode 100644 index 00000000..76ab6049 --- /dev/null +++ b/docs/experiments/quality_cost.md @@ -0,0 +1,80 @@ + + +# Quality-constrained localization cost + +## Claim boundary + +`codenib-quality-cost-report` measures cost per **successful localization**. It +does not measure patch correctness, issue resolution, Copilot premium requests, +or self-hosted GPU cost. One attempted query repetition is successful when its +`answer_blocks.recall@5` reaches the declared threshold. Misses and recorded +infrastructure failures stay in the denominator. + +An arm is eligible for cost selection only when the lower bound of its paired, +repository-clustered 95% confidence interval is above the declared recall +margin relative to `grep_only`. The report then minimizes tokens per successful +localization among eligible arms. This ordering prevents a cheap but +quality-regressing arm from winning. + +## Accounting rules + +- Inputs are joined by exact `(model, query_id, rep, arm)` identity. Duplicate + cells and unmatched arm denominators fail by default. +- The compared arms can be pinned with repeated `--arm` options. Unrelated + policy variants in the same experiment roots cannot enter the selection. +- Retry failures are linked to the final cell within the same input root. If a + retry lacks token or cost data, the affected arm's complete cost is marked + unavailable rather than undercounted. +- `total_tokens` is the primary historical measure. New query sweeps also + preserve prompt, completion, cache-read, and cache-write token classes. +- USD is either an explicitly labeled, unpinned runtime estimate or an offline + projection from a content-hashed pricing snapshot. A zero emitted for an + unpriced local model is unavailable, not free. +- Shared index build cost is optional and reported separately at declared query + horizons. It is never silently added to model-call cost. + +## Five-model validation + +The retained 500-query synthesis study uses the same three predeclared arms and +the paper's 5 percentage-point recall margin: + +| model | selected policy | tokens / successful localization | reduction vs. grep/read | USD status | +| --- | --- | ---: | ---: | --- | +| Claude Haiku 4.5 | eager | 114,855 | 49.2% | recorded runtime estimate | +| Gemini 2.5 Flash | compact | 20,240 | 68.1% | recorded runtime estimate | +| Gemma 4 12B | compact | 20,145 | 87.5% | unavailable | +| Qwen3.5 9B | compact | 115,404 | 53.9% | unavailable | +| Qwen3.5 27B | compact | 74,772 | 54.4% | unavailable | + +The Qwen3.5-27B source contains one failed eager retry without usage fields. +The report therefore withholds complete eager cost for that model; the selected +compact arm is unaffected. Cloud USD values are historical LiteLLM estimates, +not immutable repricing and not GitHub or Copilot credits. + +## Reproduce without model calls + +```bash +codenib-quality-cost-report \ + "${CODENIB_RESULTS_DIR}/haiku_compare_py" \ + "${CODENIB_RESULTS_DIR}/haiku_compare_multilang" \ + "${CODENIB_RESULTS_DIR}/haiku_compare_cpp" \ + "${CODENIB_RESULTS_DIR}/haiku_synth_compact" \ + "${CODENIB_RESULTS_DIR}/gemini25_flash_synth_runtime_v1" \ + "${CODENIB_RESULTS_DIR}/gemma4_12b_synth_runtime_v1" \ + "${CODENIB_RESULTS_DIR}/qwen35_9b_synth_runtime_v1" \ + "${CODENIB_RESULTS_DIR}/qwen35_27b_synth_runtime_v1" \ + --output-dir "${CODENIB_RESULTS_DIR}/quality_cost_v1/five_models" \ + --arm grep_only \ + --arm preinj_eager \ + --arm preinj_eager_compact \ + --noninferiority-margin 0.05 \ + --bootstrap-samples 5000 \ + --bootstrap-seed 0 +``` + +The command emits `quality_cost.json` and `quality_cost.md`. The JSON preserves +the metric contract, bootstrap settings, per-input model counts, denominator +audits, retry coverage, arm-level measurements, and the constrained optimum. diff --git a/pyproject.toml b/pyproject.toml index 7de52886..21a7e832 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,6 +66,7 @@ codenib-lsp-agent-study-artifacts = "codenib.eval.agent_runner.lsp_agent_study_a codenib-lsp-agent-study-run = "codenib.eval.agent_runner.lsp_agent_study_runner:main" codenib-artifact-bundle = "codenib.eval.artifact_bundle:main" codenib-swe-explore-benchmark = "codenib.eval.benchmarks.swe_explore_runner:main" +codenib-quality-cost-report = "codenib.eval.reports.quality_cost:main" [project.optional-dependencies] agent = [ @@ -168,6 +169,7 @@ exclude = ["eval*"] # exclude packages matching these glob patterns [tool.setuptools.package-data] "codenib.agent.skills" = ["*/config.yaml", "*/skill.md"] +"codenib.eval" = ["pricing/*.json"] "codenib.scip_interface" = ["*.proto", "*.yml", "*.sh", "README.md"] [tool.pytest.ini_options] diff --git a/test/agent/test_runner_usage.py b/test/agent/test_runner_usage.py index 84997f94..08ad42e7 100644 --- a/test/agent/test_runner_usage.py +++ b/test/agent/test_runner_usage.py @@ -25,6 +25,7 @@ UsageRecord, UsageTracker, _extract_token_usage, + runtime_cost_provenance, ) # --------------------------------------------------------------------------- @@ -205,6 +206,20 @@ def test_completion_cost_failure_is_none(self): assert tracker.totals().total_tokens == 7 +def test_runtime_cost_provenance_labels_unpinned_estimates(): + with patch("codenib.llm.usage._installed_litellm_version", return_value="1.2.3"): + priced = runtime_cost_provenance("provider/model", 0.25) + unpriced = runtime_cost_provenance("provider/local", None) + + assert priced == { + "kind": "runtime_estimate", + "calculator": "litellm.completion_cost", + "calculator_version": "1.2.3", + "model": "provider/model", + } + assert unpriced["kind"] == "unavailable" + + # --------------------------------------------------------------------------- # LiteLLMChat._call_raw usage_tracker wiring # --------------------------------------------------------------------------- diff --git a/test/eval/test_pricing.py b/test/eval/test_pricing.py new file mode 100644 index 00000000..36b47e30 --- /dev/null +++ b/test/eval/test_pricing.py @@ -0,0 +1,127 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for immutable offline pricing snapshots.""" + +from __future__ import annotations + +import json + +import pytest + +from codenib.eval.reports.pricing import ( + PricingSnapshotError, + bundled_pricing_snapshot_path, + load_pricing_snapshot, + parse_pricing_snapshot, + project_cell_cost, +) + + +def _snapshot(*, token_semantics: str = "prompt_excludes_cache", input_rate=1.0): + return { + "schema": "codenib.pricing-snapshot.v1", + "snapshot_id": "fixture-2026-08-04", + "currency": "USD", + "effective_date": "2026-08-04", + "retrieved_date": "2026-08-04", + "models": [ + { + "model_id": "provider/model-v1", + "aliases": ["provider/model"], + "channel": "direct-api", + "token_semantics": token_semantics, + "cache_write_ttl": "5m", + "rates_per_million": { + "uncached_input": input_rate, + "output": 5.0, + "cache_read_input": 0.1, + "cache_write_input": 1.25, + }, + "source_url": "https://example.com/pricing", + } + ], + } + + +def _cell(*, prompt_tokens=1_000_000): + return { + "model": "provider/model", + "prompt_tokens": prompt_tokens, + "completion_tokens": 1_000_000, + "cache_read_input_tokens": 1_000_000, + "cache_creation_input_tokens": 1_000_000, + } + + +def test_projects_each_token_class_under_declared_semantics(): + snapshot = parse_pricing_snapshot(_snapshot()) + + projected = project_cell_cost(_cell(), snapshot) + + assert projected["available"] is True + assert projected["billable_tokens"]["uncached_input"] == 1_000_000 + assert projected["total"] == pytest.approx(7.35) + + +def test_prompt_including_cache_subtracts_cached_classes_once(): + snapshot = parse_pricing_snapshot( + _snapshot(token_semantics="prompt_includes_cache") + ) + + projected = project_cell_cost(_cell(prompt_tokens=3_000_000), snapshot) + + assert projected["billable_tokens"]["uncached_input"] == 1_000_000 + assert projected["total"] == pytest.approx(7.35) + + +def test_projection_refuses_missing_or_inconsistent_token_classes(): + snapshot = parse_pricing_snapshot(_snapshot()) + missing = _cell() + missing["completion_tokens"] = None + + assert project_cell_cost(missing, snapshot)["reason"] == "missing_token_classes" + + inclusive = parse_pricing_snapshot( + _snapshot(token_semantics="prompt_includes_cache") + ) + inconsistent = project_cell_cost(_cell(prompt_tokens=1), inclusive) + assert inconsistent["reason"] == "inconsistent_token_classes" + + +def test_snapshot_rejects_invalid_dates_and_ambiguous_aliases(): + invalid_date = _snapshot() + invalid_date["retrieved_date"] = "2026-02-30" + with pytest.raises(PricingSnapshotError, match="valid calendar date"): + parse_pricing_snapshot(invalid_date) + + ambiguous = _snapshot() + duplicate = dict(ambiguous["models"][0]) + duplicate.update(model_id="provider/other", aliases=["provider/model"]) + ambiguous["models"].append(duplicate) + with pytest.raises(PricingSnapshotError, match="ambiguous"): + parse_pricing_snapshot(ambiguous) + + +def test_bundled_snapshot_has_hashed_portable_identity(): + path = bundled_pricing_snapshot_path() + + snapshot = load_pricing_snapshot(path) + identity = snapshot.public_identity() + + assert json.loads(path.read_text(encoding="utf-8"))["schema"] == identity["schema"] + assert identity["source_file"] == path.name + assert str(path.parent) not in json.dumps(identity) + assert identity["sha256"].startswith("sha256:") + + +def test_price_changes_affect_projection_without_changing_tokens(): + low = parse_pricing_snapshot(_snapshot(input_rate=1.0)) + high = parse_pricing_snapshot(_snapshot(input_rate=2.0)) + + low_cost = project_cell_cost(_cell(), low) + high_cost = project_cell_cost(_cell(), high) + + assert low_cost["billable_tokens"] == high_cost["billable_tokens"] + assert high_cost["total"] - low_cost["total"] == pytest.approx(1.0) diff --git a/test/eval/test_quality_cost_report.py b/test/eval/test_quality_cost_report.py new file mode 100644 index 00000000..4dc6ba11 --- /dev/null +++ b/test/eval/test_quality_cost_report.py @@ -0,0 +1,368 @@ +# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors +# +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for quality-constrained localization cost reports.""" + +from __future__ import annotations + +import json + +import pytest + +from codenib.eval.reports.pricing import parse_pricing_snapshot +from codenib.eval.reports.quality_cost import ( + QualityCostError, + analyze_quality_cost, + load_quality_cost_cells, + main, + write_quality_cost_report, +) + + +def _cell( + query_id: str, + arm: str, + *, + quality: float, + tokens: int, + success: bool = True, + cost_usd: float | None = None, + model: str = "provider/model", +): + instance_id = "org__repo-1" if query_id == "q1" else "other__repo-2" + return { + "cell_id": f"{query_id}__{arm}__rep1", + "model": model, + "instance_id": instance_id, + "query_id": query_id, + "query": f"Locate {query_id}", + "subset_id": arm, + "rep": 1, + "success": success, + "metrics_meaningful": True, + "metrics": ({"answer_blocks": {"5": {"recall": quality}}} if success else {}), + "prompt_tokens": tokens - 2, + "completion_tokens": 2, + "total_tokens": tokens, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "cost_usd": cost_usd, + } + + +def _paired_cells(): + return [ + _cell("q1", "grep_only", quality=1.0, tokens=100, cost_usd=0.10), + _cell("q2", "grep_only", quality=0.0, tokens=100, cost_usd=0.10), + _cell("q1", "compact", quality=1.0, tokens=40, cost_usd=0.04), + _cell("q2", "compact", quality=0.0, tokens=40, cost_usd=0.04), + ] + + +def _pricing_snapshot(): + return parse_pricing_snapshot( + { + "schema": "codenib.pricing-snapshot.v1", + "snapshot_id": "fixture", + "currency": "USD", + "effective_date": "2026-08-04", + "retrieved_date": "2026-08-04", + "models": [ + { + "model_id": "provider/model", + "aliases": [], + "channel": "fixture", + "token_semantics": "prompt_excludes_cache", + "cache_write_ttl": "5m", + "rates_per_million": { + "uncached_input": 1.0, + "output": 5.0, + "cache_read_input": 0.1, + "cache_write_input": 1.25, + }, + "source_url": "https://example.com/pricing", + } + ], + } + ) + + +def test_selects_lowest_cost_quality_qualified_arm(): + report = analyze_quality_cost(_paired_cells(), bootstrap_samples=100) + model = report["models"]["provider/model"] + + assert model["arms"]["grep_only"]["tokens"]["per_success"] == 200 + assert model["arms"]["compact"]["tokens"]["per_success"] == 80 + optimum = model["constrained_optima"]["tokens_per_success"] + assert optimum["selected_arm"] == "compact" + assert optimum["ratio_to_baseline"] == pytest.approx(0.4) + assert optimum["savings_fraction_vs_baseline"] == pytest.approx(0.6) + + +def test_infrastructure_failure_scores_zero_and_remains_charged(): + cells = _paired_cells() + failed = next( + cell + for cell in cells + if cell["query_id"] == "q2" and cell["subset_id"] == "compact" + ) + failed.update(success=False, metrics={}, total_tokens=60) + + report = analyze_quality_cost(cells, bootstrap_samples=50) + compact = report["models"]["provider/model"]["arms"]["compact"] + + assert compact["attempted"] == 2 + assert compact["infrastructure_success_rate"] == 0.5 + assert compact["quality_mean"] == 0.5 + assert compact["tokens"]["total"] == 100 + + +def test_strict_denominator_rejects_unmatched_arms(): + cells = [ + cell + for cell in _paired_cells() + if not (cell["query_id"] == "q2" and cell["subset_id"] == "compact") + ] + + with pytest.raises(QualityCostError, match="unmatched paired denominators"): + analyze_quality_cost(cells, bootstrap_samples=10) + + report = analyze_quality_cost( + cells, bootstrap_samples=10, strict_denominators=False + ) + audit = report["models"]["provider/model"]["denominator_audit"] + assert audit["common_unit_count"] == 1 + assert audit["mismatches"]["compact"]["missing_count"] == 1 + + +def test_quality_regression_cannot_win_on_cost(): + cells = _paired_cells() + for cell in cells: + if cell["subset_id"] == "compact": + cell["metrics"]["answer_blocks"]["5"]["recall"] = 0.0 + + report = analyze_quality_cost(cells, bootstrap_samples=100) + model = report["models"]["provider/model"] + + assert model["arms"]["compact"]["quality_qualified"] is False + assert ( + model["constrained_optima"]["tokens_per_success"]["selected_arm"] == "grep_only" + ) + + +def test_arm_allowlist_excludes_unplanned_policy_variants(): + cells = _paired_cells() + cells.extend( + [ + _cell(query, "unplanned", quality=1.0, tokens=1, cost_usd=0.001) + for query in ("q1", "q2") + ] + ) + + report = analyze_quality_cost( + cells, + bootstrap_samples=10, + included_arms=["grep_only", "compact"], + ) + + model = report["models"]["provider/model"] + assert set(model["arms"]) == {"grep_only", "compact"} + assert report["contract"]["included_arms"] == ["grep_only", "compact"] + assert report["arm_selection_audit"] == { + "input_cell_count": 6, + "analyzed_cell_count": 4, + "excluded_cell_count": 2, + "excluded_arms": {"provider/model": {"unplanned": 2}}, + } + + with pytest.raises(QualityCostError, match="missing requested arms"): + analyze_quality_cost( + cells, + bootstrap_samples=10, + included_arms=["grep_only", "missing"], + ) + + +def test_zero_success_and_zero_only_local_cost_are_unavailable(): + cells = [ + _cell(query, arm, quality=0.0, tokens=10, cost_usd=0.0) + for query in ("q1", "q2") + for arm in ("grep_only", "compact") + ] + + report = analyze_quality_cost(cells, bootstrap_samples=10) + model = report["models"]["provider/model"] + + assert model["arms"]["compact"]["tokens"]["per_success"] is None + assert model["arms"]["compact"]["usd"]["available"] is False + assert ( + model["arms"]["compact"]["usd"]["reason"] == "ambiguous_zero_runtime_estimates" + ) + assert model["constrained_optima"]["tokens_per_success"]["selected_arm"] is None + + +def test_mixed_zero_and_positive_runtime_estimates_are_not_undercounted(): + cells = _paired_cells() + cells[0]["cost_usd"] = 0.0 + + report = analyze_quality_cost(cells, bootstrap_samples=10) + baseline = report["models"]["provider/model"]["arms"]["grep_only"] + + assert baseline["usd"]["available"] is False + assert baseline["usd"]["reason"] == "ambiguous_zero_runtime_estimates" + + +def test_projection_requires_complete_raw_classes_and_uses_snapshot(): + report = analyze_quality_cost( + _paired_cells(), + bootstrap_samples=10, + pricing_snapshot=_pricing_snapshot(), + usd_source="projected", + ) + compact = report["models"]["provider/model"]["arms"]["compact"] + assert compact["usd"]["kind"] == "offline_projection" + assert compact["usd"]["available"] is True + + cells = _paired_cells() + cells[0]["completion_tokens"] = None + unavailable = analyze_quality_cost( + cells, + bootstrap_samples=10, + pricing_snapshot=_pricing_snapshot(), + usd_source="projected", + ) + baseline = unavailable["models"]["provider/model"]["arms"]["grep_only"] + assert baseline["usd"]["available"] is False + assert baseline["usd"]["reason"] == "missing_token_classes" + + +def test_unmetered_retry_attempt_makes_complete_cost_unavailable(): + cells = _paired_cells() + compact = next( + cell + for cell in cells + if cell["query_id"] == "q1" and cell["subset_id"] == "compact" + ) + compact["_quality_cost_retry_attempts"] = [ + {"cell_id": compact["cell_id"], "success": False, "cost_usd": None} + ] + + report = analyze_quality_cost(cells, bootstrap_samples=10) + compact_report = report["models"]["provider/model"]["arms"]["compact"] + + assert compact_report["tokens"]["available"] is False + assert compact_report["tokens"]["reason"] == "unmetered_retry_attempts" + assert compact_report["tokens"]["execution_attempts"] == 3 + assert compact_report["usd"]["reason"] == "unmetered_retry_attempts" + + +def test_recursive_loader_audits_non_cells_and_duplicates(tmp_path): + nested = tmp_path / "shard" / "cells" + nested.mkdir(parents=True) + cell = _cell("q1", "grep_only", quality=1.0, tokens=10) + (nested / "cell.json").write_text(json.dumps(cell), encoding="utf-8") + (tmp_path / "summary.json").write_text( + json.dumps({"schema": "summary"}), encoding="utf-8" + ) + + cells, audit = load_quality_cost_cells([tmp_path]) + assert len(cells) == 1 + assert audit["skipped_non_cell_json"] == 1 + + retry_dir = tmp_path / "protocol" / "failure_retries" / cell["cell_id"] + retry_dir.mkdir(parents=True) + (retry_dir / "failed_cell.json").write_text( + json.dumps( + { + "cell_id": cell["cell_id"], + "instance_id": cell["instance_id"], + "query_id": cell["query_id"], + "subset_id": cell["subset_id"], + "success": False, + "error": "provider failure", + } + ), + encoding="utf-8", + ) + cells, audit = load_quality_cost_cells([tmp_path]) + assert audit["retry_failure_records"] == 1 + assert audit["retry_failures_with_total_tokens"] == 0 + assert len(cells[0]["_quality_cost_retry_attempts"]) == 1 + + duplicate = tmp_path / "duplicate.json" + duplicate.write_text(json.dumps(cell), encoding="utf-8") + with pytest.raises(QualityCostError, match="duplicate cell"): + load_quality_cost_cells([tmp_path]) + + +def test_retry_links_within_input_root_when_models_share_cell_ids(tmp_path): + roots = [tmp_path / "model-a", tmp_path / "model-b"] + for index, root in enumerate(roots): + cells_dir = root / "cells" + cells_dir.mkdir(parents=True) + cell = _cell( + "q1", + "grep_only", + quality=1.0, + tokens=10, + model=f"provider/model-{index}", + ) + (cells_dir / "shared.json").write_text(json.dumps(cell), encoding="utf-8") + + retry_dir = roots[1] / "protocol" / "failure_retries" / "shared" + retry_dir.mkdir(parents=True) + (retry_dir / "failed_cell.json").write_text( + json.dumps({"cell_id": "q1__grep_only__rep1", "success": False}), + encoding="utf-8", + ) + + cells, audit = load_quality_cost_cells(roots) + by_model = {cell["model"]: cell for cell in cells} + + assert "_quality_cost_retry_attempts" not in by_model["provider/model-0"] + assert len(by_model["provider/model-1"]["_quality_cost_retry_attempts"]) == 1 + assert audit["roots"]["input-02"]["name"] == "model-b" + assert audit["roots"]["input-02"]["retry_failure_records"] == 1 + + +def test_report_writer_and_cli_emit_json_and_markdown(tmp_path, capsys): + output = tmp_path / "direct" + report = write_quality_cost_report( + cells=_paired_cells(), output_dir=output, bootstrap_samples=10 + ) + assert ( + json.loads((output / "quality_cost.json").read_text())["schema"] + == report["schema"] + ) + assert "Token optimum" in (output / "quality_cost.md").read_text() + + cells_dir = tmp_path / "cells" + cells_dir.mkdir() + for cell in _paired_cells(): + (cells_dir / f"{cell['cell_id']}.json").write_text( + json.dumps(cell), encoding="utf-8" + ) + cli_output = tmp_path / "cli" + assert ( + main( + [ + str(cells_dir), + "--output-dir", + str(cli_output), + "--bootstrap-samples", + "10", + ] + ) + == 0 + ) + assert "Quality-constrained cost" in capsys.readouterr().out + assert (cli_output / "quality_cost.json").is_file() + + +def test_rejects_queries_without_quality_ground_truth(): + cells = _paired_cells() + cells[0]["metrics_meaningful"] = False + + with pytest.raises(QualityCostError, match="meaningful ground truth"): + analyze_quality_cost(cells, bootstrap_samples=10) diff --git a/test/eval/test_query_sweep.py b/test/eval/test_query_sweep.py index 5fe4417e..818bd5b0 100644 --- a/test/eval/test_query_sweep.py +++ b/test/eval/test_query_sweep.py @@ -6,6 +6,7 @@ from __future__ import annotations +import json from types import SimpleNamespace import pytest @@ -211,9 +212,17 @@ def unexpected_symbol_graph(*_args, **_kwargs): "tool_calls": [], "trace_summary": {}, "total_turns": 1, - "total_tokens": 1, + "prompt_tokens": 7, + "completion_tokens": 2, + "total_tokens": 12, "cost_usd": 0.0, + "cost_provenance": { + "kind": "runtime_estimate", + "calculator": "fixture", + "model": "provider/model", + }, "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 3, }, ) monkeypatch.setattr( @@ -276,6 +285,16 @@ def unexpected_symbol_graph(*_args, **_kwargs): str(output_dir / "cache" / "org__repo-1"), {"vector"}, ) + cell = json.loads( + (output_dir / "cells" / "query-1__vector__rep1.json").read_text( + encoding="utf-8" + ) + ) + assert cell["prompt_tokens"] == 7 + assert cell["completion_tokens"] == 2 + assert cell["cache_read_input_tokens"] == 0 + assert cell["cache_creation_input_tokens"] == 3 + assert cell["cost_provenance"]["calculator"] == "fixture" resumed = run_query_sweep( cfg,