diff --git a/tests/canonical/test_run_aggregate_models_snapshot.py b/tests/canonical/test_run_aggregate_models_snapshot.py index 9696f1db..20cbd635 100644 --- a/tests/canonical/test_run_aggregate_models_snapshot.py +++ b/tests/canonical/test_run_aggregate_models_snapshot.py @@ -23,6 +23,7 @@ from __future__ import annotations +import json from datetime import UTC, datetime from typing import Any @@ -345,3 +346,213 @@ def test_failure_attribution_envelope_round_trip() -> None: } _round_trip(FailureAttribution, payload) + + +# --------------------------------------------------------------------------- +# Wire-format invariants (closes #152, #153) +# --------------------------------------------------------------------------- + + +def test_schema_version_survives_exclude_unset() -> None: + """``RunAggregate.schema_version`` must appear in the dumped output + even when the model is constructed without an explicit value AND + dumped with ``exclude_unset=True`` (the mode the migrated writer + will use to preserve wire-format parity with the current dict path). + + The regression this catches: once the writer migrates from + ``json.dump(dict)`` to ``model.model_dump_json(exclude_unset=True)``, + a ``RunAggregate`` whose ``schema_version`` was left at the model + default would silently drop the envelope field — every downstream + dashboard that dispatches on ``schema_version`` would break. The + model's ``@model_serializer`` forces the field into the dump + regardless. Closes #152. + """ + # Payload deliberately omits ``schema_version`` — the model must + # still emit it via the default. + payload = calculate_aggregate_metrics(_sample_task_metrics_list(), weighted=True) + assert "schema_version" not in payload, "guard: payload must not stamp schema_version" + + model = RunAggregate.model_validate(payload) + dumped = model.model_dump(by_alias=True, mode="json", exclude_unset=True) + + assert "schema_version" in dumped, ( + "schema_version must survive exclude_unset=True on RunAggregate — " + f"got dumped keys: {sorted(dumped.keys())}" + ) + assert ( + dumped["schema_version"] == 1 + ), f"schema_version default drifted; expected 1, got {dumped['schema_version']!r}" + + # And it survives a JSON round-trip too. + reloaded = RunAggregate.model_validate_json( + model.model_dump_json(by_alias=True, exclude_unset=True) + ) + assert reloaded.schema_version == 1 + + +def test_int_valued_numeric_fields_preserve_int_type() -> None: + """The ``int | float`` unions on token-count / latency-percentile / + cost-sum fields must preserve ``int`` inputs verbatim through + validation AND JSON dump. A regression that narrows any of these + fields back to ``float`` would coerce ``int 42`` → ``42.0`` on + validation and emit ``42.0`` in the JSON — a byte-level wire drift + invisible to Python-dict comparison (``42 == 42.0``) but real + when downstream tooling diffs the JSON files byte-for-byte. + + Today's producers in ``metrics.py`` happen to always emit + ``float`` for these fields (division short-circuits on empty + input), so no live regression exists. This test guards against a + future producer refactor introducing ``int`` output — the model + must handle it correctly without a coordinated widening. Closes #153. + """ + # Hand-construct a payload with int-valued token counts + latency + # percentiles. The producer path today never emits these as int, + # so this is the only way to exercise the int branch of the union. + task_payload: dict[str, Any] = { + "task_id": "int-preserve-guard", + "benchmark_type": "airline", + "complexity": "simple", + "tags": [], + "expected_failure_modes": [], + "total_trials": 1, + "successful_trials": 1, + "success_rate": 1.0, # rate — stays float + "avg_score": 1.0, # rate — stays float + "avg_latency_s": 5, # int — union widened field + "avg_turns": 3, # int — union widened field + "avg_tool_calls": 2, # int — union widened field + # Token counts — natural integers. + "avg_prompt_tokens": 100, + "avg_completion_tokens": 40, + "avg_reasoning_tokens": 10, + "avg_cached_tokens": 5, + "avg_cache_creation_input_tokens": 3, + "avg_cache_read_input_tokens": 2, + # Latency percentiles — could be int seconds. + "latency_p50_s": 4, + "latency_p90_s": 6, + "latency_p99_s": 8, + "api_call_latency_p50_s": 1, + "api_call_latency_p90_s": 2, + "api_call_latency_p99_s": 3, + "stuck_rate": 0, + } + model = PerTaskMetrics.model_validate(task_payload) + + # Type preservation at the Python level — union picks int on validation. + for field in ( + "avg_latency_s", + "avg_turns", + "avg_tool_calls", + "avg_prompt_tokens", + "avg_completion_tokens", + "avg_reasoning_tokens", + "avg_cached_tokens", + "avg_cache_creation_input_tokens", + "avg_cache_read_input_tokens", + "latency_p50_s", + "latency_p90_s", + "latency_p99_s", + "api_call_latency_p50_s", + "api_call_latency_p90_s", + "api_call_latency_p99_s", + "stuck_rate", + ): + value = getattr(model, field) + assert isinstance(value, int) and not isinstance(value, bool), ( + f"PerTaskMetrics.{field}: int input coerced to {type(value).__name__} — " + f"the int|float union is broken. Got {value!r}." + ) + + # Type preservation on the JSON wire — dump keys must be int, not float. + dumped = model.model_dump(by_alias=True, mode="json") + dumped_json = json.dumps(dumped, sort_keys=True) + # An int field dumped as int has no trailing ``.0`` in the JSON string. + for field in ("avg_prompt_tokens", "latency_p50_s", "stuck_rate"): + assert isinstance(dumped[field], int) and not isinstance(dumped[field], bool), ( + f"PerTaskMetrics.{field} dumped as {type(dumped[field]).__name__}, " + f"expected int. Full dump: {dumped_json}" + ) + + +def test_int_valued_aggregate_fields_preserve_int_type() -> None: + """Same invariant as + :func:`test_int_valued_numeric_fields_preserve_int_type` but for the + ``AggregateMetrics`` shape. Divison-produced rate fields on + ``AggregateMetrics`` (``success_rate_micro/macro``, ``avg_score_*``, + ``pass_at_*_macro``) stay narrow ``float`` — a caller passing ``int`` + there is a producer bug, not a wire invariant to preserve, so we + do NOT assert int preservation for those fields.""" + agg_payload: dict[str, Any] = { + "total_tasks": 2, + "total_trials": 5, + "success_rate_micro": 1.0, # rate — stays float + "avg_score_micro": 1.0, # rate — stays float + "avg_latency_s": 5, # int — widened + "avg_turns": 3, # int — widened + "avg_tool_calls": 2, # int — widened + "stuck_rate": 0, # int — widened + "total_prompt_tokens": 500, + "total_completion_tokens": 200, + "total_reasoning_tokens": 50, + "total_cached_tokens": 25, + "total_cache_creation_input_tokens": 15, + "total_cache_read_input_tokens": 10, + "avg_prompt_tokens": 100, + "avg_completion_tokens": 40, + "avg_reasoning_tokens": 10, + "avg_cached_tokens": 5, + "avg_cache_creation_input_tokens": 3, + "avg_cache_read_input_tokens": 2, + "latency_p50_s_macro": 4, + "latency_p90_s_macro": 6, + "latency_p99_s_macro": 8, + } + model = AggregateMetrics.model_validate(agg_payload) + for field in ( + "avg_latency_s", + "avg_turns", + "avg_tool_calls", + "stuck_rate", + "total_prompt_tokens", + "avg_prompt_tokens", + "latency_p50_s_macro", + ): + value = getattr(model, field) + assert isinstance(value, int) and not isinstance(value, bool), ( + f"AggregateMetrics.{field}: int input coerced to {type(value).__name__}. " + f"Got {value!r}." + ) + + # And rate fields DID stay narrow float — a regression that widens them + # would be caught by mypy at consumer sites, but a runtime guard here + # documents the invariant. + assert isinstance(model.success_rate_micro, float), ( + "AggregateMetrics.success_rate_micro widened to int|float unexpectedly — " + "rate-shaped fields must stay narrow float for consumer type contracts." + ) + + +def test_current_producer_output_matches_model_dump_byte_for_byte() -> None: + """Round-trip guard against the current live producer output. + + Even though the ``int | float`` union covers a case the producer + doesn't emit today, the by-product must still hold: every dict the + production metric-calc functions actually produce today survives a + model round-trip byte-for-byte in the JSON representation. This + catches any accidental type coercion in the round trip for the + fields we DID narrow (``success_rate_*``, ``avg_score_*``, + ``pass_at_*_macro``).""" + + def _canonical(payload: dict[str, Any]) -> str: + return json.dumps(payload, sort_keys=True, default=str) + + task_metrics_list = _sample_task_metrics_list() + for weighted in (True, False): + agg_payload = calculate_aggregate_metrics(task_metrics_list, weighted=weighted) + model = AggregateMetrics.model_validate(agg_payload) + dumped = model.model_dump(by_alias=True, mode="json", exclude_unset=True) + assert _canonical(agg_payload) == _canonical(dumped), ( + f"AggregateMetrics(weighted={weighted}) JSON drift between dict and " + f"model path.\nsource: {_canonical(agg_payload)}\nmodel: {_canonical(dumped)}" + ) diff --git a/tolokaforge/core/output/aggregate_models.py b/tolokaforge/core/output/aggregate_models.py index 876d39f8..055943d5 100644 --- a/tolokaforge/core/output/aggregate_models.py +++ b/tolokaforge/core/output/aggregate_models.py @@ -24,13 +24,38 @@ The ``pass@k`` / ``pass_hat@k`` keys aren't valid Python identifiers, so they carry :class:`~pydantic.Field` ``alias`` values. Dump with ``model_dump(by_alias=True, mode="json")`` to produce the wire shape. + +Two wire-format invariants pinned by the canonical tests: + +1. ``schema_version`` on :class:`RunAggregate` is **always emitted** on + dump, regardless of ``exclude_unset``. See the model-serializer on + the class for details. +2. **Type-preserving numeric fields.** Fields that can naturally be + ``int`` at the producer (token counts, latency percentiles, + stuck-rate counts, cost sums) carry ``int | float`` unions so the + wire type matches whatever the producer emitted. Pydantic v2's + smart-union picks the more specific type on validation, so a + source ``int 42`` round-trips to ``42`` (not ``42.0``) in the + JSON dump. Rate-shaped fields (``success_rate``, ``pass@k``, + ``avg_score`` and their macro/micro aggregates) stay narrow + ``float`` because they're always ``sum(...) / n`` divisions — + never ``int`` at the producer. + + The current producers in ``metrics.py`` and + ``failure_attribution.py`` happen to emit ``float`` for every + widened field today (both metric-calc functions short-circuit on + empty input rather than dividing by zero). The union is + defensive future-proofing: if a future refactor changes a + producer to emit ``int`` (e.g. a counting aggregator that returns + ``0`` for empty), the model preserves the wire type without a + coordinated model change. """ from __future__ import annotations from typing import Any -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_serializer __all__ = [ "AggregateMetrics", @@ -76,42 +101,48 @@ class PerTaskMetrics(BaseModel): pass_hat_at_5: float | None = Field(default=None, alias="pass_hat@5") pass_hat_at_10: float | None = Field(default=None, alias="pass_hat@10") - # Averages across trials. - avg_score: float - avg_latency_s: float - avg_turns: float - avg_tool_calls: float + # Averages across trials. Typed as ``int | float`` unions so the + # model preserves the numeric type the producer emitted: e.g. + # ``sum([]) == 0`` (``int``) for empty aggregates, ``float`` + # otherwise. Pydantic's smart-union picks the more specific type on + # validation, so a source ``0`` round-trips to ``0`` (not ``0.0``) + # in the JSON dump — pinning the wire format against the current + # dict-based writer path. See :class:`RunAggregateWriter` docstring. + avg_score: int | float + avg_latency_s: int | float + avg_turns: int | float + avg_tool_calls: int | float # Per-usage-field averages (Metrics.usage — six fields). - avg_prompt_tokens: float = 0.0 - avg_completion_tokens: float = 0.0 - avg_reasoning_tokens: float = 0.0 - avg_cached_tokens: float = 0.0 - avg_cache_creation_input_tokens: float = 0.0 - avg_cache_read_input_tokens: float = 0.0 + avg_prompt_tokens: int | float = 0 + avg_completion_tokens: int | float = 0 + avg_reasoning_tokens: int | float = 0 + avg_cached_tokens: int | float = 0 + avg_cache_creation_input_tokens: int | float = 0 + avg_cache_read_input_tokens: int | float = 0 # Cost is ``None`` when no trial reported a cost (unknown-provider case). - total_cost_usd: float | None = None - avg_cost_usd: float | None = None + total_cost_usd: int | float | None = None + avg_cost_usd: int | float | None = None # Judge-cost split — ``judge_cost_usd`` is the LLM-judge grader's # spend, tracked separately so the agent-vs-judge cost breakdown # survives round-trips. ``total_cost_incl_judge_usd`` = agent + # judge; ``None`` when neither is known. - judge_cost_usd: float | None = None - total_cost_incl_judge_usd: float | None = None + judge_cost_usd: int | float | None = None + total_cost_incl_judge_usd: int | float | None = None # Per-trial wall-time percentiles. - latency_p50_s: float = 0.0 - latency_p90_s: float = 0.0 - latency_p99_s: float = 0.0 + latency_p50_s: int | float = 0 + latency_p90_s: int | float = 0 + latency_p99_s: int | float = 0 # Per-API-call percentiles aggregated across every call in every trial. - api_call_latency_p50_s: float = 0.0 - api_call_latency_p90_s: float = 0.0 - api_call_latency_p99_s: float = 0.0 + api_call_latency_p50_s: int | float = 0 + api_call_latency_p90_s: int | float = 0 + api_call_latency_p99_s: int | float = 0 - stuck_rate: float = 0.0 + stuck_rate: int | float = 0 class AggregateMetrics(BaseModel): @@ -130,15 +161,20 @@ class AggregateMetrics(BaseModel): total_tasks: int total_trials: int - # Weighted (micro) averages — set when ``weighted=True``. + # Weighted (micro) averages — set when ``weighted=True``. Rate-shaped + # ([0, 1] scalar from ``sum(...) / n``): always ``float`` at the + # producer. Stays narrow — no ``int`` widening — matching the + # ``PerTaskMetrics.success_rate`` / ``avg_score`` siblings. success_rate_micro: float | None = None avg_score_micro: float | None = None - # Unweighted (macro) averages — set when ``weighted=False``. + # Unweighted (macro) averages — set when ``weighted=False``. Same + # rate-shaped rationale. success_rate_macro: float | None = None avg_score_macro: float | None = None # pass@k macro averages across tasks, plus their pass_hat aliases. + # Also always ``float`` (macro-average of ``pass@k`` rates). pass_at_1_macro: float | None = Field(default=None, alias="pass@1_macro") pass_at_5_macro: float | None = Field(default=None, alias="pass@5_macro") pass_at_10_macro: float | None = Field(default=None, alias="pass@10_macro") @@ -147,49 +183,75 @@ class AggregateMetrics(BaseModel): pass_hat_at_10_macro: float | None = Field(default=None, alias="pass_hat@10_macro") # Simple cross-task averages. - avg_latency_s: float = 0.0 - avg_turns: float = 0.0 - avg_tool_calls: float = 0.0 - stuck_rate: float = 0.0 + avg_latency_s: int | float = 0 + avg_turns: int | float = 0 + avg_tool_calls: int | float = 0 + stuck_rate: int | float = 0 # Per-usage-field totals + macro averages. - total_prompt_tokens: float = 0.0 - total_completion_tokens: float = 0.0 - total_reasoning_tokens: float = 0.0 - total_cached_tokens: float = 0.0 - total_cache_creation_input_tokens: float = 0.0 - total_cache_read_input_tokens: float = 0.0 - avg_prompt_tokens: float = 0.0 - avg_completion_tokens: float = 0.0 - avg_reasoning_tokens: float = 0.0 - avg_cached_tokens: float = 0.0 - avg_cache_creation_input_tokens: float = 0.0 - avg_cache_read_input_tokens: float = 0.0 - - total_cost_usd: float | None = None - avg_cost_usd: float | None = None + total_prompt_tokens: int | float = 0 + total_completion_tokens: int | float = 0 + total_reasoning_tokens: int | float = 0 + total_cached_tokens: int | float = 0 + total_cache_creation_input_tokens: int | float = 0 + total_cache_read_input_tokens: int | float = 0 + avg_prompt_tokens: int | float = 0 + avg_completion_tokens: int | float = 0 + avg_reasoning_tokens: int | float = 0 + avg_cached_tokens: int | float = 0 + avg_cache_creation_input_tokens: int | float = 0 + avg_cache_read_input_tokens: int | float = 0 + + total_cost_usd: int | float | None = None + avg_cost_usd: int | float | None = None # Judge-cost split at the run/slice level — same shape as # ``PerTaskMetrics`` but aggregated across the tasks/slice. - judge_cost_usd: float | None = None - total_cost_incl_judge_usd: float | None = None + judge_cost_usd: int | float | None = None + total_cost_incl_judge_usd: int | float | None = None - latency_p50_s_macro: float = 0.0 - latency_p90_s_macro: float = 0.0 - latency_p99_s_macro: float = 0.0 + latency_p50_s_macro: int | float = 0 + latency_p90_s_macro: int | float = 0 + latency_p99_s_macro: int | float = 0 class RunAggregate(AggregateMetrics): """The top-level ``aggregate.json`` shape — :class:`AggregateMetrics` - plus the ``schema_version`` envelope field the orchestrator stamps - before writing. + plus the ``schema_version`` envelope field every downstream consumer + (dashboards, metric collectors, historical-run readers) reads to + dispatch between wire-format generations. Slice values inside ``metadata_slices.json`` are :class:`AggregateMetrics` directly; only the top-level artifact carries the envelope. + + ``schema_version`` is **always emitted** on dump, regardless of + whether the caller constructs the model with an explicit value or + lets the default apply. Pydantic's ``exclude_unset=True`` normally + drops fields left at their default — for this envelope that would + silently ship an aggregate.json without a version, which would + break every version-dispatching downstream consumer. The + :func:`_always_include_schema_version` model-serializer below + guarantees the field is present regardless of dump options. """ schema_version: int = 1 + @model_serializer(mode="wrap") + def _always_include_schema_version(self, handler: Any) -> dict[str, Any]: + """Wrap the default serializer so ``schema_version`` is always + present in the dumped dict, even under ``exclude_unset=True``. + + The wrapped handler produces whatever dict the caller's dump + options normally would; this method injects ``schema_version`` + only when the handler's output omitted it (which happens under + ``exclude_unset=True`` when the field was left at its default). + No forced overwrite of a caller-set value. + """ + data = handler(self) + if isinstance(data, dict) and "schema_version" not in data: + data["schema_version"] = self.schema_version + return data + class MetadataSlices(BaseModel): """The ``metadata_slices.json`` envelope — four ``by_*`` dictionaries. diff --git a/tolokaforge/core/output/aggregates.py b/tolokaforge/core/output/aggregates.py index e4d0bd20..5bb7d81d 100644 --- a/tolokaforge/core/output/aggregates.py +++ b/tolokaforge/core/output/aggregates.py @@ -66,8 +66,20 @@ def write_aggregate( ) -> None: """Persist ``output_dir/aggregate.json`` from *aggregate*. - Callers are responsible for injecting any envelope fields the - artifact carries (e.g. ``schema_version``) before calling. + Wire-format invariants (also pinned by the canonical tests on + :class:`~tolokaforge.core.output.aggregate_models.RunAggregate`): + + * **``schema_version`` is always present** on the written JSON. + Callers passing a dict must set it before calling. Callers + passing a :class:`~tolokaforge.core.output.aggregate_models.RunAggregate` + model get this for free — the model's serializer forces the + field into the dump regardless of ``exclude_unset``. + * **Numeric fields preserve source type.** Values the producer + emitted as ``int`` (e.g. ``sum([]) == 0`` for empty + aggregates) stay ``int`` on the wire; values emitted as + ``float`` stay ``float``. The model achieves this via + ``int | float`` unions; the direct dict path preserves it + naturally. """ ...