| 1 | """Prevent aggregation of observably incompatible measurement populations.
|
| 2 |
|
| 3 | Legacy rows with uniformly absent metadata remain usable for diagnostics.
|
| 4 | Missing and present metadata are distinct: absence never establishes equality
|
| 5 | with a known environment. This check is not a public-provenance attestation.
|
| 6 | """
|
| 7 |
|
| 8 | from __future__ import annotations
|
| 9 |
|
| 10 | import json
|
| 11 | from typing import Any
|
| 12 |
|
| 13 |
|
| 14 | DIMENSIONS = (
|
| 15 | "lane", "benchmark_lane", "measurement_lane", "profile", "benchmark_track", "command_semantics_version",
|
| 16 | "fixture_version", "scenario_version", "oracle_version", "runner_class", "runner_id", "cache_state",
|
| 17 | "remote_transport", "remote_server", "netshape_profile", "env_isolation_version",
|
| 18 | "calibration_version", "platform_semantics_version", "binary_sha256", "load_tier",
|
| 19 | "instruction_level", "agent.name", "agent.model", "agent.adapter",
|
| 20 | "agent.provider", "agent.version", "agent.temperature", "agent.max_output_tokens",
|
| 21 | "agent.environment", "agent.instruction_level", "fixture.version", "fixture.fixture_version",
|
| 22 | "oracle.version", "task_prompt_id",
|
| 23 | )
|
| 24 |
|
| 25 |
|
| 26 | def population_conflicts(rows: list[dict[str, Any]]) -> list[str]:
|
| 27 | """Return differing identity fields within ONE subject/scenario/op cell."""
|
| 28 | fields = {name: set() for name in DIMENSIONS}
|
| 29 | fields.update({"subject_details.version": set(), "subject_details.binary_sha256": set()})
|
| 30 | for row in rows:
|
| 31 | for name in DIMENSIONS:
|
| 32 | value: Any = row
|
| 33 | for part in name.split("."):
|
| 34 | value = value.get(part) if isinstance(value, dict) else None
|
| 35 | fields[name].add(json.dumps(value, sort_keys=True))
|
| 36 | details = row.get("subject_details") or {}
|
| 37 | if isinstance(details, dict):
|
| 38 | details = details.get(str(row.get("subject")), details)
|
| 39 | for name in ("version", "binary_sha256"):
|
| 40 | value = details.get(name) if isinstance(details, dict) else None
|
| 41 | fields[f"subject_details.{name}"].add(json.dumps(value, sort_keys=True))
|
| 42 | return sorted(name for name, values in fields.items() if len(values) > 1)
|