From 8d6fafb6affcf9febceaeb4466474316f9afd0ea Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Sat, 18 Jul 2026 02:12:29 +0100 Subject: [PATCH] feat(delphi): certify.py battery runner + focuser + fingerprint ledger (Spec A, Phase 0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sonnet-agent build from SPEC_A — the R1 certification backbone: - polismath/replay/certify.py + scripts/certify.py (click CLI: run / focus) over scripts/certify_battery.json (starter battery: vw uniform8 / front-loaded6 / single-cut + biodiversity uniform8, all clojure-legacy). - Clojure recording cache: reuse iff manifest matches (votes sha256, canonical schedule hash, replay.clj + math/src tree hashes); Python recording cache keyed on polismath tree hash + engine_mode. Both drivers invoked as subprocesses via --schedule (the only way to bake the collision-free schedule_id into the recording path without touching replay_driver.py). - Hash-first compare (canonical-JSON step hashes after acceptance projection) + content-addressed step-verdict cache; a same-input re-run is a pure cache hit (verified: zero subprocess calls on repeat). - Acceptance projection = prep-main whitelist MINUS subgroup-* trio (CLOJURE_QUIRKS.md Q7); the exclusion is printed on EVERY run, never silent. - First-divergence focuser (earliest divergent step, divergent key-paths only, ≤40-line stdout, full detail to focus-report.json) + fingerprint ledger docs/divergences.json (normalized path × family × engine_mode; ships empty — diagnoses accrue as certification proceeds). - Crosslang bridge PROMOTED tests/replay_harness/clj_crosslang.py → polismath/replay/crosslang.py (now a production API); behavior-identical (84 pre-existing replay-harness tests green before and after), import sites updated (test_clj_crosslang.py, replay_smoke.sh heredoc). Tests: 132 passed + 1 gated integration (ran for real: actual clojure + python drivers on vw single-cut, ~11s, then cache-hit re-run). Agent's real smoke surfaced ~19 undiagnosed divergent fingerprints on the PRE-port base — the certification loop's starting worklist (ledger deliberately left empty; diagnosis is top-level work). commit-id:4a87a6ae --- delphi/docs/divergences.json | 1 + delphi/polismath/replay/certify.py | 944 ++++++++++++++++++ .../replay/crosslang.py} | 11 +- delphi/scripts/certify.py | 81 ++ delphi/scripts/certify_battery.json | 29 + delphi/tests/replay_harness/test_certify.py | 607 +++++++++++ .../tests/replay_harness/test_certify_cli.py | 158 +++ .../replay_harness/test_clj_crosslang.py | 2 +- math/dev/replay_smoke.sh | 3 +- 9 files changed, 1830 insertions(+), 6 deletions(-) create mode 100644 delphi/docs/divergences.json create mode 100644 delphi/polismath/replay/certify.py rename delphi/{tests/replay_harness/clj_crosslang.py => polismath/replay/crosslang.py} (92%) create mode 100644 delphi/scripts/certify.py create mode 100644 delphi/scripts/certify_battery.json create mode 100644 delphi/tests/replay_harness/test_certify.py create mode 100644 delphi/tests/replay_harness/test_certify_cli.py diff --git a/delphi/docs/divergences.json b/delphi/docs/divergences.json new file mode 100644 index 000000000..0967ef424 --- /dev/null +++ b/delphi/docs/divergences.json @@ -0,0 +1 @@ +{} diff --git a/delphi/polismath/replay/certify.py b/delphi/polismath/replay/certify.py new file mode 100644 index 000000000..d36db97a0 --- /dev/null +++ b/delphi/polismath/replay/certify.py @@ -0,0 +1,944 @@ +"""Certification battery runner — Clojure<->Python math parity (SPEC A). + +The replay harness (schedule.py/driver.py/store.py/stepcompare.py, Phase H-A) +and the Clojure cross-language bridge (crosslang.py, Phase H-B) already let a +human run ONE (dataset, schedule) replay through both engines and diff the +result. This module turns that into a repeatable, cheap-to-re-run BATTERY: + +- A committed battery config (``scripts/certify_battery.json``) declares which + (dataset, schedule) pairs to certify, and under which + :mod:`polismath.utils.engine_mode` the Python side runs. ``clojure-legacy`` + is the parity target (Clojure's warm-start behavior); ``improved`` is + Python's production default. certify sets the mode EXPLICITLY per entry + (never inherited from the ambient environment) by launching the Python + driver in a subprocess with the env var freshly set — see :func:`run_py_driver`. +- Both engines' recordings are CACHED on disk, keyed by content hashes (votes + CSV, resolved schedule, and — for Python — the ``polismath`` source tree, or + — for Clojure — ``dev/replay.clj`` + the ``math/src`` tree). Re-running + certify after an unrelated code change should cost near-zero: cache hits + short-circuit the (slow) driver subprocess entirely. +- Comparison is HASH-FIRST: each step's post-acceptance-projection blob is + content-hashed per engine; equal hashes mean an exact MATCH with zero + diffing. Only a hash MISMATCH falls back to the (cached, by hash-pair) full + :class:`~polismath.replay.stepcompare.StepComparer` run. +- Acceptance projection is the prep-main 23-key whitelist (crosslang.py) MINUS + the dead ``subgroup-*`` trio (subgroup-clusters/subgroup-votes/subgroup-repness + — see CLOJURE_QUIRKS.md Q7). This is never silent: every certify run prints + :data:`ACCEPTANCE_NOTICE`. +- Every divergence is FINGERPRINTED (index/step-stripped path pattern + family + + engine_mode -> 10 hex chars) and tracked in a committed ledger + (``docs/divergences.json``) so recurring, already-diagnosed divergences are + annotated instead of re-discovered cold every run. + +Design note — the clj cache is scoped PER (dataset, schedule_id) directory (as +literally specified), not globally content-addressed across engine-mode +siblings of the same underlying schedule. Two battery entries that share a +preset+cuts but differ only in ``engine_mode`` will each get their own +``/clj/`` (and therefore each pay for one clojure driver run) even +though the clj recording would be byte-identical — engine_mode has no effect +on the Clojure reference. The starter battery (all ``clojure-legacy``) never +hits this; flagged here for whoever adds a second engine_mode to the battery. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import subprocess +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from polismath.replay import real_data +from polismath.replay import schedule as sched +from polismath.replay import store as st +from polismath.replay.crosslang import PREP_MAIN_KEYS, _kebab, load_clj_blobs, project_prep_main +from polismath.replay.stepcompare import DEFAULT_TOLERANT_STAT_KEYS, StepComparer +from polismath.replay.types import ReplayDataset +from polismath.utils.engine_mode import ( + ENGINE_MODE_CHOICES, + ENGINE_MODE_DEFAULT, + ENGINE_MODE_ENV_VAR, + ENGINE_MODE_LEGACY, +) + +# --------------------------------------------------------------------------- +# Paths. +# --------------------------------------------------------------------------- +# certify.py -> replay -> polismath -> delphi -> repo root (mirrors store.py). +_DELPHI_ROOT = Path(__file__).resolve().parents[2] +_REPO_ROOT = _DELPHI_ROOT.parents[0] +_MATH_ROOT = _REPO_ROOT / "math" + +DEFAULT_BATTERY_PATH = _DELPHI_ROOT / "scripts" / "certify_battery.json" + + +def default_ledger_path() -> Path: + return _DELPHI_ROOT / "docs" / "divergences.json" + + +# --------------------------------------------------------------------------- +# Acceptance projection: prep-main whitelist MINUS the dead subgroup-* trio. +# --------------------------------------------------------------------------- +ACCEPTANCE_EXCLUDED_KEYS = frozenset({"subgroup-clusters", "subgroup-votes", "subgroup-repness"}) +ACCEPTANCE_KEYS = PREP_MAIN_KEYS - ACCEPTANCE_EXCLUDED_KEYS +ACCEPTANCE_NOTICE = "subgroup-* keys excluded from acceptance (CLOJURE_QUIRKS.md Q7)" + + +def project_acceptance(blob: dict[str, Any]) -> dict[str, Any]: + """Project a math_main blob onto :data:`ACCEPTANCE_KEYS` (kebab-canonical). + + Reuses :func:`polismath.replay.crosslang.project_prep_main` for the + snake/kebab canonicalisation, then drops the dead subgroup-* trio. + """ + proj = project_prep_main(blob) + return {k: v for k, v in proj.items() if k not in ACCEPTANCE_EXCLUDED_KEYS} + + +def _acceptance_projecting_comparer(**kwargs: Any) -> StepComparer: + """A :class:`StepComparer` that projects both blobs onto acceptance keys + before diffing — the comparer used for the (cached) hash-mismatch path.""" + tolerant = frozenset(_kebab(k) for k in DEFAULT_TOLERANT_STAT_KEYS) - ACCEPTANCE_EXCLUDED_KEYS + + class _AcceptanceProjectingComparer(StepComparer): + def compare_step(self, blob_a: dict, blob_b: dict, index: int) -> dict[str, Any]: + return super().compare_step( + project_acceptance(blob_a), project_acceptance(blob_b), index + ) + + return _AcceptanceProjectingComparer(tolerant_stat_keys=tolerant, **kwargs) + + +# --------------------------------------------------------------------------- +# Battery config: parsing + collision-free schedule-id derivation. +# --------------------------------------------------------------------------- +_NCUTS_PRESETS = frozenset({"uniform", "front-loaded", "back-loaded"}) +_VALID_PRESETS = _NCUTS_PRESETS | frozenset({"single-cut", "every-vote", "per-day"}) + + +@dataclass(frozen=True) +class BatteryEntry: + """One parsed ``certify_battery.json`` entry (either preset- or + schedule-file-based), with its collision-free ``schedule_id`` already + resolved (bakes in ``engine_mode`` — see :func:`derive_schedule_id`).""" + + dataset: str + engine_mode: str + schedule_id: str + preset: str | None = None + n_cuts: int | None = None + schedule_path: Path | None = None + notes: str = "" + + +def derive_schedule_id( + *, engine_mode: str, preset: str | None = None, n_cuts: int | None = None, + base_schedule_id: str | None = None, +) -> str: + """Collision-free schedule id: ``{base}-{engine_mode}``. + + ``base`` is either an explicit ``base_schedule_id`` (schedule-file-based + entries — the id the file itself declares) or ``{preset}{n_cuts}`` for + presets that take a cut count (``uniform8``, ``front-loaded6``, …) or bare + ``preset`` for those that don't (``single-cut``, ``every-vote``, ``per-day``). + Distinct (preset, n_cuts, engine_mode) triples always yield distinct ids + because the preset name is embedded verbatim in ``base``. + """ + if base_schedule_id is not None: + base = base_schedule_id + elif preset in _NCUTS_PRESETS: + if n_cuts is None: + raise ValueError(f"preset {preset!r} requires n_cuts to derive a schedule_id") + base = f"{preset}{n_cuts}" + else: + base = preset + return f"{base}-{engine_mode}" + + +def parse_battery_entry(e: dict[str, Any], *, battery_dir: Path | None = None) -> BatteryEntry: + """Parse one battery entry — either ``{"schedule": ""}`` (base id + read verbatim from the referenced schedule.json) or ``{"preset": ..., + "n_cuts": ...}``. ``engine_mode`` defaults to ``clojure-legacy`` (the + parity target) when absent — the starter battery spells it out anyway.""" + dataset = e["dataset"] + engine_mode = e.get("engine_mode", ENGINE_MODE_LEGACY) + + if "schedule" in e: + schedule_path = Path(e["schedule"]) + if battery_dir is not None and not schedule_path.is_absolute(): + schedule_path = battery_dir / schedule_path + base_id = json.loads(schedule_path.read_text())["schedule_id"] + schedule_id = derive_schedule_id(engine_mode=engine_mode, base_schedule_id=base_id) + return BatteryEntry(dataset=dataset, engine_mode=engine_mode, schedule_id=schedule_id, + schedule_path=schedule_path, notes=e.get("notes", "")) + + preset = e.get("preset") + if preset not in _VALID_PRESETS: + raise ValueError( + f"unknown preset {preset!r} in battery entry {e!r}; expected one of " + f"{sorted(_VALID_PRESETS)}" + ) + n_cuts = e.get("n_cuts") + if preset in _NCUTS_PRESETS and n_cuts is None: + raise ValueError(f"preset {preset!r} requires n_cuts in battery entry {e!r}") + schedule_id = derive_schedule_id(engine_mode=engine_mode, preset=preset, n_cuts=n_cuts) + return BatteryEntry(dataset=dataset, engine_mode=engine_mode, schedule_id=schedule_id, + preset=preset, n_cuts=n_cuts, notes=e.get("notes", "")) + + +def load_battery(path: str | Path = DEFAULT_BATTERY_PATH) -> list[BatteryEntry]: + path = Path(path) + data = json.loads(path.read_text()) + return [parse_battery_entry(e, battery_dir=path.parent) for e in data] + + +# --------------------------------------------------------------------------- +# Dataset availability + effective schedule construction. +# --------------------------------------------------------------------------- +def dataset_available(dataset: str) -> bool: + return real_data.dataset_dir(dataset) is not None + + +def votes_csv_path(dataset: str) -> Path | None: + d = real_data.dataset_dir(dataset) + if d is None: + return None + hits = sorted(d.glob("*-votes.csv")) + return hits[0] if hits else None + + +def _spec_from_preset(entry: BatteryEntry, ds: ReplayDataset) -> sched.ScheduleSpec: + n = ds.n + if entry.preset == "uniform": + return sched.preset_uniform(entry.dataset, n, n_cuts=entry.n_cuts) + if entry.preset == "front-loaded": + return sched.preset_front_loaded(entry.dataset, n, n_cuts=entry.n_cuts) + if entry.preset == "back-loaded": + return sched.preset_back_loaded(entry.dataset, n, n_cuts=entry.n_cuts) + if entry.preset == "every-vote": + return sched.preset_every_vote(entry.dataset, n) + if entry.preset == "single-cut": + return sched.preset_single_cut(entry.dataset, n) + if entry.preset == "per-day": + return sched.preset_per_day(entry.dataset, ds) + raise ValueError(f"unknown preset {entry.preset!r}") + + +def build_effective_spec(entry: BatteryEntry, ds: ReplayDataset) -> sched.ScheduleSpec: + """The :class:`ScheduleSpec` actually run, with ``schedule_id`` overridden + to ``entry.schedule_id`` (the collision-free, engine_mode-baked id) so the + recording lands in the right directory regardless of preset or file origin. + """ + base = (sched.ScheduleSpec.from_json_file(entry.schedule_path) if entry.schedule_path + else _spec_from_preset(entry, ds)) + d = base.to_dict() + d["schedule_id"] = entry.schedule_id + return sched.ScheduleSpec.from_dict(d) + + +# --------------------------------------------------------------------------- +# Hashing helpers. +# --------------------------------------------------------------------------- +def sha256_file(path: str | Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 16), b""): + h.update(chunk) + return h.hexdigest() + + +def sha256_tree(root: str | Path, pattern: str = "**/*") -> str: + """sha256 over sorted (relpath, content) pairs of every FILE matching + ``pattern`` under ``root`` — deterministic regardless of filesystem + iteration order, sensitive to both a file's path and its content.""" + root = Path(root) + h = hashlib.sha256() + for p in sorted(root.glob(pattern)): + if not p.is_file(): + continue + rel = p.relative_to(root).as_posix() + h.update(rel.encode()) + h.update(b"\0") + h.update(p.read_bytes()) + h.update(b"\0") + return h.hexdigest() + + +def _canonical_json(obj: Any) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str) + + +def _canonical_hash(obj: Any) -> str: + return hashlib.sha256(_canonical_json(obj).encode()).hexdigest() + + +def canonical_schedule_hash(spec: sched.ScheduleSpec) -> str: + """Hash of the parts of a schedule that affect the REPLAY (cuts, + moderation, source) — deliberately excludes ``schedule_id``/``notes`` so + two differently-named but content-identical schedules hash equal.""" + payload = {"cuts": spec.cuts, "moderation": spec.moderation, "source": spec.source} + return _canonical_hash(payload) + + +def _comparer_code_hash() -> str: + """Hash of the comparison-logic SOURCE files. Folded into the verdict + cache key so a bugfix to the comparer (with unchanged tolerances) busts + cached step verdicts instead of silently serving stale MATCH/DIVERGENCE + results — for a certification tool a stale MATCH is the worst failure + mode. (Review finding, 2026-07-22.)""" + import polismath.regression.comparer as _comparer_mod + from polismath.replay import crosslang as _crosslang_mod + from polismath.replay import stepcompare as _stepcompare_mod + + h = hashlib.sha256() + for mod in (_stepcompare_mod, _crosslang_mod, _comparer_mod): + h.update(Path(mod.__file__).read_bytes()) + h.update(Path(__file__).read_bytes()) + return h.hexdigest() + + +def _comparer_cfg_hash(cmp: StepComparer) -> str: + cfg = { + "abs_tol": cmp._cmp.abs_tol, + "rel_tol": cmp._cmp.rel_tol, + "ignore_pca_sign_flip": cmp._cmp.ignore_pca_sign_flip, + "outlier_fraction": cmp._cmp.outlier_fraction, + "tolerant_keys": sorted(cmp._tolerant_keys), + "code": _comparer_code_hash(), + } + return _canonical_hash(cfg) + + +# --------------------------------------------------------------------------- +# Fingerprints. +# --------------------------------------------------------------------------- +_STEP_PREFIX_RE = re.compile(r"^step_\d+\.") +_BRACKET_IDX_RE = re.compile(r"\[\d+\]") + + +def normalize_path(path: str) -> str: + """Strip the ``step_N.`` prefix, collapse bracket indices to ``[]``, and + collapse purely-numeric dotted segments (dict keys, e.g. a tid) to ``N`` — + so two divergences at the same structural location (different step, + different list index, different dict key) fingerprint identically. + + ``step_3.pca.comps[0][1]`` -> ``pca.comps[][]`` (spec example, verbatim). + """ + p = _STEP_PREFIX_RE.sub("", path or "") + p = _BRACKET_IDX_RE.sub("[]", p) + parts = ["N" if part.isdigit() else part for part in p.split(".")] + return ".".join(parts) + + +def _fp_from_normalized(norm_path: str, family: str, engine_mode: str) -> str: + digest = hashlib.sha1(f"{norm_path}|{family}|{engine_mode}".encode()).hexdigest() + return digest[:10] + + +def compute_fingerprint(path: str, family: str, engine_mode: str) -> str: + return _fp_from_normalized(normalize_path(path), family, engine_mode) + + +def fingerprint_key_for(path_pattern: str, family: str, engine_mode: str) -> str: + """Ledger key for an ALREADY-normalized path pattern.""" + return f"FP-{_fp_from_normalized(path_pattern, family, engine_mode)}" + + +def fingerprint_key(path: str, family: str, engine_mode: str) -> str: + return fingerprint_key_for(normalize_path(path), family, engine_mode) + + +def _abbrev(value: Any) -> Any: + """Abbreviate a float to a short string; pass through everything else.""" + if isinstance(value, float): + return f"{value:.6g}" + return value + + +# --------------------------------------------------------------------------- +# Ledger (docs/divergences.json). +# --------------------------------------------------------------------------- +def load_ledger(path: str | Path | None = None) -> dict[str, Any]: + path = Path(path) if path is not None else default_ledger_path() + if not path.exists(): + return {} + return json.loads(path.read_text()) + + +def save_ledger(ledger: dict[str, Any], path: str | Path | None = None) -> None: + path = Path(path) if path is not None else default_ledger_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as fh: + json.dump(ledger, fh, indent=2, sort_keys=True) + fh.write("\n") + + +def update_ledger(ledger: dict[str, Any], observations: list[dict[str, Any]]) -> dict[str, Any]: + """Append fingerprints newly observed in ``observations`` as + ``status=open``. NEVER overwrites an existing entry — a human-entered + ``diagnosis``/``status`` on a known fingerprint is always preserved. + + Each observation: ``{"path_pattern", "family", "engine_mode", "dataset", + "schedule_id", "step"}`` (``path_pattern`` already normalized). + """ + updated = dict(ledger) + for obs in observations: + key = fingerprint_key_for(obs["path_pattern"], obs["family"], obs["engine_mode"]) + if key in updated: + continue + updated[key] = { + "path_pattern": obs["path_pattern"], + "family": obs["family"], + "engine_mode": obs["engine_mode"], + "first_seen": {"dataset": obs["dataset"], "schedule": obs["schedule_id"], + "step": obs["step"]}, + "status": "open", + "diagnosis": None, + } + return updated + + +def annotate_by_key(ledger: dict[str, Any], key: str) -> str | None: + entry = ledger.get(key) + if entry is None: + return None + diagnosis = entry.get("diagnosis") + if diagnosis: + return f"[known {key}: {diagnosis[:60]}]" + return f"[known {key}: status={entry.get('status', 'open')}]" + + +# --------------------------------------------------------------------------- +# Subprocess drivers — `_run_subprocess` is the single mockable seam; tests +# NEVER invoke real clojure or the real py driver (monkeypatch this). +# --------------------------------------------------------------------------- +class CertifyError(RuntimeError): + """A battery entry failed at a specific stage (driver subprocess, setup).""" + + def __init__(self, stage: str, message: str): + super().__init__(message) + self.stage = stage + + +# Generous ceilings — driver runs are ~10s (clj, JVM-startup-bound) to a few +# minutes (py warm-start chains); these exist so a hung JVM or Python driver +# fails the ENTRY instead of blocking an unattended battery run forever. +# (Review finding, 2026-07-22.) +DRIVER_TIMEOUT_SEC = 3600.0 + + +def _run_subprocess(cmd: list[str], *, cwd: Path, env: dict[str, str], + timeout: float = DRIVER_TIMEOUT_SEC) -> subprocess.CompletedProcess: + return subprocess.run(cmd, cwd=str(cwd), env=env, capture_output=True, + text=True, timeout=timeout) + + +def run_py_driver(spec_path: Path, *, out_root: Path, engine_mode: str) -> subprocess.CompletedProcess: + """Runs ``scripts/replay_driver.py run --schedule --out + `` in a SUBPROCESS (cwd=delphi/) with ``OMP_NUM_THREADS`` / + ``OPENBLAS_NUM_THREADS`` pinned to 1 and ``POLISMATH_ENGINE_MODE`` set + explicitly — so the mode is picked up fresh per entry, never inherited + from whatever happens to be in the calling shell's environment.""" + env = dict(os.environ) + env["OMP_NUM_THREADS"] = "1" + env["OPENBLAS_NUM_THREADS"] = "1" + env[ENGINE_MODE_ENV_VAR] = engine_mode + cmd = ["uv", "run", "python", "scripts/replay_driver.py", "run", + "--schedule", str(spec_path), "--out", str(out_root)] + try: + return _run_subprocess(cmd, cwd=_DELPHI_ROOT, env=env) + except OSError as exc: + raise CertifyError("py-driver-launch", str(exc)) from exc + + +def run_clj_driver(spec_path: Path, votes_csv: Path, *, out_dir: Path) -> subprocess.CompletedProcess: + """Runs ``clojure -M:replay --schedule --votes + --out `` in a SUBPROCESS with cwd=math/ (dev/replay.clj:57).""" + cmd = ["clojure", "-M:replay", "--schedule", str(spec_path), "--votes", str(votes_csv), + "--out", str(out_dir)] + try: + return _run_subprocess(cmd, cwd=_MATH_ROOT, env=dict(os.environ)) + except OSError as exc: + raise CertifyError("clj-driver-launch", str(exc)) from exc + + +def _write_temp_schedule(spec: sched.ScheduleSpec, root: Path) -> Path: + """Write ``spec`` (with its final, collision-free schedule_id already + baked in) to a scratch file used purely as the driver CLI's ``--schedule`` + input — overwritten deterministically per (dataset, schedule_id).""" + tmp_dir = root / ".certify_cache" / "tmp_schedules" + tmp_dir.mkdir(parents=True, exist_ok=True) + p = tmp_dir / f"{spec.dataset}__{spec.schedule_id}.json" + spec.write_json(p) + return p + + +def _manifest_matches(manifest_path: Path, expected: dict[str, Any]) -> bool: + if not manifest_path.exists(): + return False + try: + existing = json.loads(manifest_path.read_text()) + except (OSError, json.JSONDecodeError): + return False + return existing == expected + + +def _write_manifest(manifest_path: Path, manifest: dict[str, Any]) -> None: + manifest_path.parent.mkdir(parents=True, exist_ok=True) + with open(manifest_path, "w") as fh: + json.dump(manifest, fh, indent=2, sort_keys=True) + + +def _py_tree_hash() -> str: + return sha256_tree(_DELPHI_ROOT / "polismath", "**/*.py") + + +def ensure_py_recording( + entry: BatteryEntry, spec: sched.ScheduleSpec, votes_sha: str, *, root: Path, + refresh: bool = False, +) -> tuple[Path, bool]: + """Reuse ``///py/`` iff its cache manifest matches (votes + sha256, schedule hash, engine_mode, py tree hash); else (re)run the Python + driver in a subprocess. Returns ``(py_dir, was_cached)``.""" + rec_dir = st.recording_dir(entry.dataset, entry.schedule_id, root=root) + py_dir = rec_dir / "py" + manifest_path = py_dir / "cache_manifest.json" + expected = { + "votes_sha256": votes_sha, + "schedule_hash": canonical_schedule_hash(spec), + "engine_mode": entry.engine_mode, + "py_tree_sha256": _py_tree_hash(), + } + if not refresh and _manifest_matches(manifest_path, expected): + return py_dir, True + + tmp_schedule = _write_temp_schedule(spec, root) + result = run_py_driver(tmp_schedule, out_root=root, engine_mode=entry.engine_mode) + if result.returncode != 0: + raise CertifyError( + "py-driver", (result.stderr or result.stdout or "non-zero exit").strip()[:1000] + ) + _write_manifest(manifest_path, expected) + return py_dir, False + + +def ensure_clj_recording( + entry: BatteryEntry, spec: sched.ScheduleSpec, votes_sha: str, votes_csv: Path, *, + root: Path, refresh: bool = False, +) -> tuple[Path, bool]: + """Reuse ``///clj/`` iff its cache manifest matches (votes + sha256, schedule hash, sha256 of dev/replay.clj, sha256 of math/src); else + (re)run the Clojure driver in a subprocess (cwd=math/). Returns + ``(clj_dir, was_cached)``. Engine_mode plays no part in the Clojure + reference, so it is deliberately NOT one of the cache keys.""" + rec_dir = st.recording_dir(entry.dataset, entry.schedule_id, root=root) + clj_dir = rec_dir / "clj" + manifest_path = clj_dir / "cache_manifest.json" + expected = { + "votes_sha256": votes_sha, + "schedule_hash": canonical_schedule_hash(spec), + "replay_clj_sha256": sha256_file(_MATH_ROOT / "dev" / "replay.clj"), + "math_src_sha256": sha256_tree(_MATH_ROOT / "src", "**/*"), + } + if not refresh and _manifest_matches(manifest_path, expected): + return clj_dir, True + + tmp_schedule = _write_temp_schedule(spec, root) + rec_dir.mkdir(parents=True, exist_ok=True) + result = run_clj_driver(tmp_schedule, votes_csv, out_dir=rec_dir) + if result.returncode != 0: + raise CertifyError( + "clj-driver", (result.stderr or result.stdout or "non-zero exit").strip()[:1000] + ) + _write_manifest(manifest_path, expected) + return clj_dir, False + + +# --------------------------------------------------------------------------- +# Hash-first compare + step-verdict cache. +# --------------------------------------------------------------------------- +def _step_verdict_cache_path(cache_root: Path, clj_hash: str, py_hash: str, cfg_hash: str) -> Path: + key = hashlib.sha256(f"{clj_hash}{py_hash}{cfg_hash}".encode()).hexdigest() + return cache_root / ".certify_cache" / "stepverdicts" / f"{key}.json" + + +def compare_recording_pair( + clj_dir: str | Path, py_dir: str | Path, *, engine_mode: str, cache_root: str | Path, + comparer: StepComparer | None = None, +) -> dict[str, Any]: + """Hash-first, cached comparison of one clj/py recording pair. + + Each aligned step is projected onto :data:`ACCEPTANCE_KEYS` and hashed + PER ENGINE; equal hashes short-circuit to a zero-cost MATCH. A mismatch + consults the on-disk step-verdict cache (keyed on the hash pair + comparer + config) before running the (acceptance-projecting) :class:`StepComparer`. + """ + clj_blobs = load_clj_blobs(Path(clj_dir)) + py_blobs = st.load_step_blobs(Path(py_dir)) + aligned = min(len(clj_blobs), len(py_blobs)) + cmp = comparer if comparer is not None else _acceptance_projecting_comparer() + cfg_hash = _comparer_cfg_hash(cmp) + cache_root = Path(cache_root) + + per_step: list[dict[str, Any]] = [] + for i in range(aligned): + clj_proj = project_acceptance(clj_blobs[i]) + py_proj = project_acceptance(py_blobs[i]) + clj_hash = _canonical_hash(clj_proj) + py_hash = _canonical_hash(py_proj) + + if clj_hash == py_hash: + per_step.append({ + "step": i, "match": True, "n_divergences": 0, + "families": {"exact": [], "tolerant": []}, "sign_flips": [], + "hash_match": True, + }) + continue + + cache_path = _step_verdict_cache_path(cache_root, clj_hash, py_hash, cfg_hash) + report = None + if cache_path.exists(): + try: + report = json.loads(cache_path.read_text()) + except (OSError, json.JSONDecodeError): + report = None + if report is None: + report = cmp.compare_step(clj_proj, py_proj, i) + cache_path.parent.mkdir(parents=True, exist_ok=True) + with open(cache_path, "w") as fh: + json.dump(report, fh, indent=2, sort_keys=True, default=str) + report = dict(report) + report["hash_match"] = False + per_step.append(report) + + return { + "n_steps_clj": len(clj_blobs), + "n_steps_py": len(py_blobs), + "aligned_steps": aligned, + "step_count_mismatch": len(clj_blobs) != len(py_blobs), + "per_step": per_step, + } + + +def _summarize_divergences(cmp_result: dict[str, Any], *, engine_mode: str) -> dict[str, Any]: + """Aggregate ALL divergences across every divergent step into distinct + (normalized path, family) patterns, ranked by frequency (ties broken + alphabetically for determinism) — top ≤3 for display, full set for the + ledger.""" + per_step = cmp_result["per_step"] + div_steps = [s for s in per_step if not s["match"]] + first_div_step = div_steps[0]["step"] if div_steps else None + + counts: dict[tuple[str, str], int] = {} + examples: dict[tuple[str, str], tuple[Any, Any]] = {} + first_step_seen: dict[tuple[str, str], int] = {} + for step in div_steps: + for fam in ("exact", "tolerant"): + for d in step["families"][fam]: + key = (normalize_path(d.get("path") or ""), fam) + counts[key] = counts.get(key, 0) + 1 + if key not in examples: + examples[key] = (d.get("a"), d.get("b")) + first_step_seen[key] = step["step"] + + ranked = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])) + top_paths = [ + { + "path_pattern": path_pattern, "family": fam, "count": count, + "fingerprint": fingerprint_key_for(path_pattern, fam, engine_mode), + "a": _abbrev(examples[(path_pattern, fam)][0]), + "b": _abbrev(examples[(path_pattern, fam)][1]), + } + for (path_pattern, fam), count in ranked[:3] + ] + all_observed = [ + {"path_pattern": path_pattern, "family": fam, "step": first_step_seen[(path_pattern, fam)]} + for (path_pattern, fam) in counts + ] + return { + "first_div_step": first_div_step, + "n_div_steps": len(div_steps), + "top_paths": top_paths, + "all_observed": all_observed, + } + + +# --------------------------------------------------------------------------- +# Per-entry certification. +# --------------------------------------------------------------------------- +def certify_entry( + entry: BatteryEntry, *, root: Path, refresh_clj: bool = False, refresh_py: bool = False, + ledger: dict[str, Any] | None = None, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Certify one battery entry: ensure both recordings, hash-first compare, + fingerprint + ledger any divergences. Returns ``(result, updated_ledger)`` + — the ledger is threaded explicitly (not saved here) so a whole-battery + run persists it exactly once. + """ + ledger = dict(ledger) if ledger is not None else load_ledger() + + if not dataset_available(entry.dataset): + return ({"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "SKIPPED", + "reason": "dataset-unavailable"}, ledger) + + try: + votes_csv = votes_csv_path(entry.dataset) + if votes_csv is None: + raise CertifyError("setup", f"no *-votes.csv found for dataset {entry.dataset!r}") + votes_sha = sha256_file(votes_csv) + ds = real_data.load_export_votes(entry.dataset) + spec = build_effective_spec(entry, ds) + + clj_dir, _ = ensure_clj_recording(entry, spec, votes_sha, votes_csv, root=root, + refresh=refresh_clj) + py_dir, _ = ensure_py_recording(entry, spec, votes_sha, root=root, refresh=refresh_py) + except CertifyError as exc: + return ({"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "ERROR", + "stage": exc.stage, "reason": str(exc)}, ledger) + except Exception as exc: # noqa: BLE001 - one bad entry must not crash the battery + return ({"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "ERROR", + "stage": "setup", "reason": str(exc)}, ledger) + + cmp_result = compare_recording_pair(clj_dir, py_dir, engine_mode=entry.engine_mode, + cache_root=root) + + if cmp_result["step_count_mismatch"]: + return ({"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "ERROR", + "stage": "step-count-mismatch", + "reason": f"clj={cmp_result['n_steps_clj']} steps, " + f"py={cmp_result['n_steps_py']} steps"}, ledger) + + div_steps = [s for s in cmp_result["per_step"] if not s["match"]] + if not div_steps: + return ({"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "MATCH", + "n_steps": cmp_result["aligned_steps"]}, ledger) + + summary = _summarize_divergences(cmp_result, engine_mode=entry.engine_mode) + for p in summary["top_paths"]: + p["known"] = annotate_by_key(ledger, p["fingerprint"]) + + observations = [ + {"path_pattern": o["path_pattern"], "family": o["family"], "engine_mode": entry.engine_mode, + "dataset": entry.dataset, "schedule_id": entry.schedule_id, "step": o["step"]} + for o in summary["all_observed"] + ] + ledger = update_ledger(ledger, observations) + + result = { + "dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "DIVERGENCE", + "first_div_step": summary["first_div_step"], "n_div_steps": summary["n_div_steps"], + "top_paths": summary["top_paths"], + } + return result, ledger + + +# --------------------------------------------------------------------------- +# Battery-level orchestration. +# --------------------------------------------------------------------------- +def _filter_only(entries: list[BatteryEntry], only: str) -> list[BatteryEntry]: + if ":" in only: + ds, sid = only.split(":", 1) + return [e for e in entries if e.dataset == ds and e.schedule_id == sid] + return [e for e in entries if e.dataset == only] + + +def run_battery( + entries: list[BatteryEntry], *, root: Path | None = None, refresh_clj: bool = False, + refresh_py: bool = False, ledger_path: str | Path | None = None, only: str | None = None, +) -> dict[str, Any]: + """Certify every (filtered) entry, persist the ledger once, and write the + machine report to ``/certify_report.json``. Does NOT print — see + :func:`render_run_lines` for the stdout rendering.""" + root = root or st.replays_root() + ledger_path = ledger_path or default_ledger_path() + ledger = load_ledger(ledger_path) + + if only: + entries = _filter_only(entries, only) + + results = [] + for entry in entries: + result, ledger = certify_entry(entry, root=root, refresh_clj=refresh_clj, + refresh_py=refresh_py, ledger=ledger) + results.append(result) + + save_ledger(ledger, ledger_path) + report = {"battery": results, "root": str(root)} + _write_json(root / "certify_report.json", report) + return report + + +def battery_exit_code(results: list[dict[str, Any]], *, strict: bool) -> int: + bad = any(r["verdict"] in ("DIVERGENCE", "ERROR") for r in results) + if strict: + bad = bad or any(r["verdict"] == "SKIPPED" for r in results) + return 1 if bad else 0 + + +def _write_json(path: Path, data: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as fh: + json.dump(data, fh, indent=2, sort_keys=True, default=str) + fh.write("\n") + + +# --------------------------------------------------------------------------- +# Rendering (pure — CLI just echoes the returned lines). +# --------------------------------------------------------------------------- +def _format_entry_line(result: dict[str, Any]) -> str: + tag = f"{result['dataset']}:{result['schedule_id']}" + verdict = result["verdict"] + if verdict == "MATCH": + return f" {tag} MATCH ({result['n_steps']} steps)" + if verdict == "SKIPPED": + return f" {tag} SKIPPED {result['reason']}" + if verdict == "ERROR": + return f" {tag} ERROR [{result['stage']}] {result['reason']}" + # DIVERGENCE + paths = ", ".join( + f"{p['path_pattern']}({p['family']})" + (f" {p['known']}" if p.get("known") else "") + for p in result["top_paths"] + ) + return (f" {tag} DIVERGENCE first_div_step={result['first_div_step']} " + f"n_div_steps={result['n_div_steps']} top=[{paths}]") + + +def _format_footer(results: list[dict[str, Any]]) -> str: + from collections import Counter + + counts = Counter(r["verdict"] for r in results) + return (f"certify: {len(results)} entries — MATCH={counts.get('MATCH', 0)} " + f"DIVERGENCE={counts.get('DIVERGENCE', 0)} SKIPPED={counts.get('SKIPPED', 0)} " + f"ERROR={counts.get('ERROR', 0)}") + + +def render_run_lines(report: dict[str, Any], *, max_lines: int = 40) -> list[str]: + """Render ``run_battery``'s report to ≤``max_lines`` stdout lines: the + acceptance notice, a header, one line per entry (truncated with a + '+N more' line if the battery is too large to fit), and a footer.""" + header = [ACCEPTANCE_NOTICE, f"certify: {len(report['battery'])} entries root={report['root']}"] + footer = [_format_footer(report["battery"])] + budget = max_lines - len(header) - len(footer) + entries = report["battery"] + if len(entries) <= budget: + body = [_format_entry_line(r) for r in entries] + else: + shown = entries[: max(budget - 1, 0)] + body = [_format_entry_line(r) for r in shown] + body.append(f" … +{len(entries) - len(shown)} more entries — see certify_report.json") + return header + body + footer + + +# --------------------------------------------------------------------------- +# Focuser: first-divergence-only inspection of an EXISTING recording pair. +# --------------------------------------------------------------------------- +def _engine_mode_from_schedule_id(schedule_id: str) -> str: + """certify's own ``derive_schedule_id`` always suffixes ``-{engine_mode}``; + recover it from known suffixes (best-effort fallback to the default).""" + for mode in ENGINE_MODE_CHOICES: + if schedule_id.endswith(f"-{mode}"): + return mode + return ENGINE_MODE_DEFAULT + + +def run_focus( + dataset: str, schedule_id: str, *, root: Path | None = None, + ledger_path: str | Path | None = None, +) -> dict[str, Any]: + """Inspect the EARLIEST divergent step of an existing (dataset, + schedule_id) recording pair. Does NOT run the drivers — `certify run` + (or a manual replay) must have produced ``clj/`` and ``py/`` already. + Writes the full per-step detail to ``///focus-report.json`` + and returns a result dict for :func:`render_focus_lines`. ``ledger_path`` + defaults to the committed ``docs/divergences.json`` — override for tests. + """ + root = root or st.replays_root() + rec_dir = st.recording_dir(dataset, schedule_id, root=root) + clj_dir, py_dir = rec_dir / "clj", rec_dir / "py" + if not clj_dir.is_dir() or not py_dir.is_dir(): + return {"dataset": dataset, "schedule_id": schedule_id, "verdict": "ERROR", + "stage": "recording-missing", + "reason": f"expected clj/ and py/ both present under {rec_dir}"} + + engine_mode = _engine_mode_from_schedule_id(schedule_id) + ledger = load_ledger(ledger_path) + cmp_result = compare_recording_pair(clj_dir, py_dir, engine_mode=engine_mode, cache_root=root) + div_steps = [s for s in cmp_result["per_step"] if not s["match"]] + + _write_json(rec_dir / "focus-report.json", { + "dataset": dataset, "schedule_id": schedule_id, "engine_mode": engine_mode, + "n_steps_clj": cmp_result["n_steps_clj"], "n_steps_py": cmp_result["n_steps_py"], + "step_count_mismatch": cmp_result["step_count_mismatch"], + "first_divergent_step": div_steps[0]["step"] if div_steps else None, + "per_step": cmp_result["per_step"], + }) + + if not div_steps: + return {"dataset": dataset, "schedule_id": schedule_id, "engine_mode": engine_mode, + "verdict": "MATCH", "n_steps": cmp_result["aligned_steps"], + "focus_report_path": str(rec_dir / "focus-report.json")} + + step = div_steps[0] + families: dict[str, list[dict[str, Any]]] = {"exact": [], "tolerant": []} + observations = [] + for fam in ("exact", "tolerant"): + for d in step["families"][fam]: + path = d.get("path") or "" + norm = normalize_path(path) + key = fingerprint_key_for(norm, fam, engine_mode) + families[fam].append({ + "path": path, "path_pattern": norm, "a": _abbrev(d.get("a")), + "b": _abbrev(d.get("b")), "fingerprint": key, + "known": annotate_by_key(ledger, key), + }) + observations.append({"path_pattern": norm, "family": fam, "engine_mode": engine_mode, + "dataset": dataset, "schedule_id": schedule_id, + "step": step["step"]}) + + ledger = update_ledger(ledger, observations) + save_ledger(ledger, ledger_path) + + return {"dataset": dataset, "schedule_id": schedule_id, "engine_mode": engine_mode, + "verdict": "DIVERGENCE", "step": step["step"], "families": families, + "focus_report_path": str(rec_dir / "focus-report.json")} + + +def render_focus_lines( + result: dict[str, Any], *, max_per_family: int = 5, max_lines: int = 40, +) -> list[str]: + """Render :func:`run_focus`'s result to ≤``max_lines`` stdout lines: + divergent key-paths ONLY, grouped by family, with a/b values shown for up + to ``max_per_family`` divergences per family (floats abbreviated).""" + lines = [ACCEPTANCE_NOTICE] + tag = f"{result['dataset']}:{result['schedule_id']}" + if result["verdict"] == "ERROR": + lines.append(f"focus: {tag} — ERROR [{result['stage']}] {result['reason']}") + return lines + if result["verdict"] == "MATCH": + lines.append(f"focus: {tag} — no divergence in {result['n_steps']} steps (MATCH)") + return lines + + lines.append(f"focus: {tag} — earliest divergence at step {result['step']}") + for fam in ("exact", "tolerant"): + diffs = result["families"][fam] + if not diffs: + continue + lines.append(f" [{fam}] {len(diffs)} divergence(s)") + shown = diffs[:max_per_family] + for d in shown: + suffix = f" {d['known']}" if d.get("known") else "" + lines.append(f" {d['path']}: a={d['a']} b={d['b']}{suffix}") + if len(diffs) > len(shown): + lines.append(f" … +{len(diffs) - len(shown)} more (see focus-report.json)") + + if len(lines) > max_lines: + lines = lines[: max_lines - 1] + [f"… output truncated at {max_lines} lines — see focus-report.json"] + return lines diff --git a/delphi/tests/replay_harness/clj_crosslang.py b/delphi/polismath/replay/crosslang.py similarity index 92% rename from delphi/tests/replay_harness/clj_crosslang.py rename to delphi/polismath/replay/crosslang.py index e43d12e25..7583e80c5 100644 --- a/delphi/tests/replay_harness/clj_crosslang.py +++ b/delphi/polismath/replay/crosslang.py @@ -11,9 +11,14 @@ blob into the py-store payload shape under a throwaway ``py/`` directory, after which ``compare_recordings(shim_dir, py_dir, engine="py")`` runs verbatim. -This is deliberately kept in ``tests/`` (not in ``polismath/replay/``) — it is -harness glue for the H-B cross-language smoke and its regression test, not a -production API. +Promoted from ``tests/replay_harness/`` (Phase H-B) into ``polismath/replay/`` +(SPEC A — certify.py): it started as harness glue for the H-B cross-language +smoke and its regression test, but ``polismath.replay.certify`` now depends on +it directly (the crosslang shim + prep-main projection are load-bearing for +certification, not just a manual smoke), so it lives alongside the rest of the +replay package as a production API. No behavior change from the move itself — +see ``tests/replay_harness/test_clj_crosslang.py`` (also updated to import from +the new location) for the regression coverage that pins it. """ from __future__ import annotations diff --git a/delphi/scripts/certify.py b/delphi/scripts/certify.py new file mode 100644 index 000000000..df0318d99 --- /dev/null +++ b/delphi/scripts/certify.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Certification CLI (SPEC A) — battery runner + first-divergence focuser. + +Thin click wrapper over ``polismath.replay.certify``: this script owns ALL +printing (``click.echo``) and process exit codes; the library itself stays +pure / side-effect-scoped so it is directly unit-testable without a CliRunner +(see ``tests/replay_harness/test_certify.py`` for the library, and +``tests/replay_harness/test_certify_cli.py`` for this CLI). + +Usage (from delphi/):: + + # Certify the whole starter battery (scripts/certify_battery.json): + uv run python scripts/certify.py run + + # Restrict to one dataset, or one (dataset, schedule_id) pair: + uv run python scripts/certify.py run --only vw + uv run python scripts/certify.py run --only vw:uniform8-clojure-legacy + + # Force re-running a driver (bypass the content-hash cache): + uv run python scripts/certify.py run --refresh-clj --refresh-py + + # SKIPPED (dataset-unavailable) entries also fail the run: + uv run python scripts/certify.py run --strict + + # Inspect the earliest divergent step of an EXISTING recording pair + # (produced by a prior `run`, or a manual replay): + uv run python scripts/certify.py focus vw uniform8-clojure-legacy +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import click + +from polismath.replay import certify as cert + + +@click.group() +def cli() -> None: + """Clojure<->Python math parity certification.""" + + +@cli.command() +@click.option("--battery", "battery_path", type=click.Path(exists=True, path_type=Path), + default=cert.DEFAULT_BATTERY_PATH, show_default=True, + help="Battery config JSON.") +@click.option("--only", default=None, help="Restrict to dataset[:schedule_id].") +@click.option("--refresh-clj", is_flag=True, help="Force re-run the Clojure driver.") +@click.option("--refresh-py", is_flag=True, help="Force re-run the Python driver.") +@click.option("--strict", is_flag=True, + help="SKIPPED (dataset-unavailable) entries also fail the run.") +@click.option("--root", type=click.Path(path_type=Path), default=None, + help="Recording store root (default: real_data/.local/replays).") +def run(battery_path, only, refresh_clj, refresh_py, strict, root): + """Certify every entry in the battery (or a filtered subset).""" + entries = cert.load_battery(battery_path) + report = cert.run_battery(entries, root=root, refresh_clj=refresh_clj, + refresh_py=refresh_py, only=only) + for line in cert.render_run_lines(report): + click.echo(line) + sys.exit(cert.battery_exit_code(report["battery"], strict=strict)) + + +@cli.command() +@click.argument("dataset") +@click.argument("schedule_id") +@click.option("--root", type=click.Path(path_type=Path), default=None, + help="Recording store root (default: real_data/.local/replays).") +def focus(dataset, schedule_id, root): + """First-divergence focuser: earliest divergent step of an EXISTING + (dataset, schedule_id) recording pair (produced by a prior `run`).""" + result = cert.run_focus(dataset, schedule_id, root=root) + for line in cert.render_focus_lines(result): + click.echo(line) + sys.exit(0 if result["verdict"] == "MATCH" else 1) + + +if __name__ == "__main__": + cli() diff --git a/delphi/scripts/certify_battery.json b/delphi/scripts/certify_battery.json new file mode 100644 index 000000000..eb5345268 --- /dev/null +++ b/delphi/scripts/certify_battery.json @@ -0,0 +1,29 @@ +[ + { + "dataset": "vw", + "preset": "uniform", + "n_cuts": 8, + "engine_mode": "clojure-legacy", + "notes": "8 evenly-spaced recomputes over the full vw conversation" + }, + { + "dataset": "vw", + "preset": "front-loaded", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "6 front-loaded recomputes — early-conversation warm-start stress" + }, + { + "dataset": "vw", + "preset": "single-cut", + "engine_mode": "clojure-legacy", + "notes": "single cold-start recompute over all votes" + }, + { + "dataset": "biodiversity", + "preset": "uniform", + "n_cuts": 8, + "engine_mode": "clojure-legacy", + "notes": "8 evenly-spaced recomputes over the full biodiversity conversation" + } +] diff --git a/delphi/tests/replay_harness/test_certify.py b/delphi/tests/replay_harness/test_certify.py new file mode 100644 index 000000000..8c30b3e47 --- /dev/null +++ b/delphi/tests/replay_harness/test_certify.py @@ -0,0 +1,607 @@ +"""Certification battery runner — SPEC A unit tests. + +Covers the ``polismath.replay.certify`` library: battery parsing + collision- +free schedule-id derivation, clj/py recording caches (subprocess calls always +MOCKED — never invoke real clojure or the real py driver here), the hash-first +step-compare shortcut + step-verdict cache, the acceptance projection +(subgroup-* dropped, CLOJURE_QUIRKS.md Q7), fingerprint normalization + the +divergences.json ledger, the first-divergence focuser, and the stdout +line-budget for both `run` and `focus`. + +CLI-level (click) tests live in test_certify_cli.py. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path + +import pytest + +from polismath.replay import certify as cert +from polismath.replay import schedule as sched +from polismath.replay.crosslang import PREP_MAIN_KEYS + +CERTIFY_BATTERY_PATH = Path(__file__).resolve().parents[2] / "scripts" / "certify_battery.json" + + +# --------------------------------------------------------------------------- +# Fixtures / helpers. +# --------------------------------------------------------------------------- +def _acceptance_blob(n: int = 3, first_comp: float = 1.0) -> dict: + """A minimal math_main-shaped blob (prep-main key spelling).""" + return { + "zid": "t", + "n": n, + "n-cmts": 2, + "in-conv": [1, 2, 3], + "tids": [0, 1], + "pca": {"center": [0.1, 0.2], "comps": [[first_comp, 0.0], [0.0, 1.0]]}, + "base-clusters": {"id": [0, 1], "x": [0.1, -0.1], "y": [0.2, -0.2], + "count": [1, 2], "members": [[1], [2, 3]]}, + "repness": {}, + } + + +def _write_clj_step(clj_dir: Path, index: int, blob: dict) -> None: + clj_dir.mkdir(parents=True, exist_ok=True) + (clj_dir / f"step-{index:03d}.blob.json").write_text(json.dumps(blob)) + + +def _write_py_step(py_dir: Path, index: int, blob: dict) -> None: + py_dir.mkdir(parents=True, exist_ok=True) + payload = {"index": index, "prev_slot": None, "cut_slot": None, + "batch_size": None, "cut_time_ms": None, "blob": blob, "extras": {}} + (py_dir / f"step-{index:03d}.json").write_text(json.dumps(payload)) + + +def _fake_completed(returncode: int = 0, stderr: str = "") -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(args=[], returncode=returncode, stdout="", stderr=stderr) + + +def _make_entry(**overrides) -> "cert.BatteryEntry": + defaults = dict(dataset="vw", engine_mode="clojure-legacy", + schedule_id="single-cut-clojure-legacy", preset="single-cut", + n_cuts=None, schedule_path=None, notes="") + defaults.update(overrides) + return cert.BatteryEntry(**defaults) + + +# --------------------------------------------------------------------------- +# Schedule-id derivation. +# --------------------------------------------------------------------------- +def test_derive_schedule_id_ncuts_preset(): + assert cert.derive_schedule_id(engine_mode="clojure-legacy", preset="uniform", + n_cuts=8) == "uniform8-clojure-legacy" + + +def test_derive_schedule_id_no_ncuts_preset(): + assert cert.derive_schedule_id(engine_mode="clojure-legacy", + preset="single-cut") == "single-cut-clojure-legacy" + + +def test_derive_schedule_id_from_explicit_base_id(): + assert cert.derive_schedule_id(engine_mode="improved", + base_schedule_id="hb-3cut") == "hb-3cut-improved" + + +def test_derive_schedule_id_collision_free_across_preset_ncuts_engine_mode(): + ids = { + cert.derive_schedule_id(engine_mode="clojure-legacy", preset="uniform", n_cuts=8), + cert.derive_schedule_id(engine_mode="improved", preset="uniform", n_cuts=8), + cert.derive_schedule_id(engine_mode="clojure-legacy", preset="front-loaded", n_cuts=8), + cert.derive_schedule_id(engine_mode="clojure-legacy", preset="uniform", n_cuts=6), + } + assert len(ids) == 4 + + +# --------------------------------------------------------------------------- +# Battery parsing. +# --------------------------------------------------------------------------- +def test_parse_battery_entry_preset_form(): + e = cert.parse_battery_entry( + {"dataset": "vw", "preset": "uniform", "n_cuts": 8, "engine_mode": "clojure-legacy"} + ) + assert e.dataset == "vw" + assert e.engine_mode == "clojure-legacy" + assert e.preset == "uniform" and e.n_cuts == 8 + assert e.schedule_id == "uniform8-clojure-legacy" + assert e.schedule_path is None + + +def test_parse_battery_entry_defaults_engine_mode_to_clojure_legacy(): + e = cert.parse_battery_entry({"dataset": "vw", "preset": "single-cut"}) + assert e.engine_mode == "clojure-legacy" + + +def test_parse_battery_entry_ncuts_preset_requires_n_cuts(): + with pytest.raises(ValueError, match="n_cuts"): + cert.parse_battery_entry({"dataset": "vw", "preset": "uniform"}) + + +def test_parse_battery_entry_unknown_preset_rejected(): + with pytest.raises(ValueError, match="preset"): + cert.parse_battery_entry({"dataset": "vw", "preset": "bogus"}) + + +def test_parse_battery_entry_schedule_form_reads_base_id_from_file(tmp_path): + schedule_json = {"dataset": "vw", "schedule_id": "hb-3cut", + "cuts": {"mode": "vote-count", "at": ["end"]}} + p = tmp_path / "sched.json" + p.write_text(json.dumps(schedule_json)) + e = cert.parse_battery_entry({"dataset": "vw", "schedule": "sched.json"}, battery_dir=tmp_path) + assert e.schedule_path == p + assert e.schedule_id == "hb-3cut-clojure-legacy" + + +def test_load_battery_starter_file_shape(): + entries = cert.load_battery(CERTIFY_BATTERY_PATH) + ids = {(e.dataset, e.schedule_id) for e in entries} + assert ("vw", "uniform8-clojure-legacy") in ids + assert ("vw", "front-loaded6-clojure-legacy") in ids + assert ("vw", "single-cut-clojure-legacy") in ids + assert ("biodiversity", "uniform8-clojure-legacy") in ids + assert len(entries) == 4 + assert all(e.engine_mode == "clojure-legacy" for e in entries) + + +# --------------------------------------------------------------------------- +# Tree / file hashing. +# --------------------------------------------------------------------------- +def test_sha256_tree_stable_and_sensitive_to_content(tmp_path): + d = tmp_path / "pkg" + d.mkdir() + (d / "a.py").write_text("x = 1\n") + (d / "sub").mkdir() + (d / "sub" / "b.py").write_text("y = 2\n") + + h1 = cert.sha256_tree(d, "**/*.py") + h2 = cert.sha256_tree(d, "**/*.py") + assert h1 == h2 + + (d / "a.py").write_text("x = 2\n") + h3 = cert.sha256_tree(d, "**/*.py") + assert h3 != h1 + + +def test_sha256_tree_sensitive_to_relpath_not_just_content(tmp_path): + d1 = tmp_path / "d1" + d1.mkdir() + (d1 / "a.py").write_text("x=1\n") + d2 = tmp_path / "d2" + d2.mkdir() + (d2 / "b.py").write_text("x=1\n") + assert cert.sha256_tree(d1, "**/*.py") != cert.sha256_tree(d2, "**/*.py") + + +def test_canonical_schedule_hash_ignores_id_but_sees_cuts(): + s1 = sched.preset_uniform("vw", 100, n_cuts=8, schedule_id="a") + s2 = sched.preset_uniform("vw", 100, n_cuts=8, schedule_id="b") + s3 = sched.preset_uniform("vw", 100, n_cuts=6, schedule_id="a") + assert cert.canonical_schedule_hash(s1) == cert.canonical_schedule_hash(s2) + assert cert.canonical_schedule_hash(s1) != cert.canonical_schedule_hash(s3) + + +# --------------------------------------------------------------------------- +# Acceptance projection (CLOJURE_QUIRKS.md Q7 — subgroup-* dead feature). +# --------------------------------------------------------------------------- +def test_project_acceptance_drops_subgroup_keys(): + blob = {k: f"v-{k}" for k in PREP_MAIN_KEYS} + proj = cert.project_acceptance(blob) + assert "subgroup-clusters" not in proj + assert "subgroup-votes" not in proj + assert "subgroup-repness" not in proj + assert set(proj) == PREP_MAIN_KEYS - cert.ACCEPTANCE_EXCLUDED_KEYS + assert proj["pca"] == "v-pca" + + +# --------------------------------------------------------------------------- +# Hash-first compare + step-verdict cache. +# --------------------------------------------------------------------------- +def test_hash_first_shortcut_skips_comparer_entirely(tmp_path, monkeypatch): + blob = _acceptance_blob(n=3) + _write_clj_step(tmp_path / "clj", 0, blob) + _write_py_step(tmp_path / "py", 0, blob) + + comparer = cert._acceptance_projecting_comparer() + calls = {"n": 0} + orig = comparer.compare_step + + def spy(a, b, i): + calls["n"] += 1 + return orig(a, b, i) + + monkeypatch.setattr(comparer, "compare_step", spy) + + result = cert.compare_recording_pair( + tmp_path / "clj", tmp_path / "py", engine_mode="clojure-legacy", + cache_root=tmp_path, comparer=comparer, + ) + assert calls["n"] == 0 + assert result["per_step"][0]["match"] is True + assert result["per_step"][0]["hash_match"] is True + + +def test_hash_mismatch_falls_back_to_comparer(tmp_path, monkeypatch): + _write_clj_step(tmp_path / "clj", 0, _acceptance_blob(n=3)) + _write_py_step(tmp_path / "py", 0, _acceptance_blob(n=99)) + + comparer = cert._acceptance_projecting_comparer() + calls = {"n": 0} + orig = comparer.compare_step + + def spy(a, b, i): + calls["n"] += 1 + return orig(a, b, i) + + monkeypatch.setattr(comparer, "compare_step", spy) + + result = cert.compare_recording_pair( + tmp_path / "clj", tmp_path / "py", engine_mode="clojure-legacy", + cache_root=tmp_path, comparer=comparer, + ) + assert calls["n"] == 1 + assert result["per_step"][0]["match"] is False + assert result["per_step"][0]["hash_match"] is False + + +def test_step_verdict_cache_round_trip(tmp_path, monkeypatch): + _write_clj_step(tmp_path / "clj", 0, _acceptance_blob(n=3)) + _write_py_step(tmp_path / "py", 0, _acceptance_blob(n=99)) + + comparer1 = cert._acceptance_projecting_comparer() + calls1 = {"n": 0} + orig1 = comparer1.compare_step + + def spy1(a, b, i): + calls1["n"] += 1 + return orig1(a, b, i) + + monkeypatch.setattr(comparer1, "compare_step", spy1) + r1 = cert.compare_recording_pair(tmp_path / "clj", tmp_path / "py", + engine_mode="clojure-legacy", cache_root=tmp_path, + comparer=comparer1) + assert calls1["n"] == 1 + + # Fresh comparer instance, same config -> must hit the on-disk step-verdict + # cache and NOT call compare_step again. + comparer2 = cert._acceptance_projecting_comparer() + calls2 = {"n": 0} + orig2 = comparer2.compare_step + + def spy2(a, b, i): + calls2["n"] += 1 + return orig2(a, b, i) + + monkeypatch.setattr(comparer2, "compare_step", spy2) + r2 = cert.compare_recording_pair(tmp_path / "clj", tmp_path / "py", + engine_mode="clojure-legacy", cache_root=tmp_path, + comparer=comparer2) + assert calls2["n"] == 0 + assert r2["per_step"][0]["families"] == r1["per_step"][0]["families"] + + +# --------------------------------------------------------------------------- +# Fingerprint normalization. +# --------------------------------------------------------------------------- +def test_normalize_path_strips_step_prefix_and_bracket_indices(): + assert cert.normalize_path("step_3.pca.comps[0][1]") == "pca.comps[][]" + + +def test_normalize_path_strips_dict_numeric_segments(): + assert cert.normalize_path( + "step_0.comment-priorities.123.priority" + ) == "comment-priorities.N.priority" + + +def test_compute_fingerprint_stable_across_indices(): + fp1 = cert.compute_fingerprint("step_1.pca.comps[0][1]", "tolerant", "clojure-legacy") + fp2 = cert.compute_fingerprint("step_9.pca.comps[3][7]", "tolerant", "clojure-legacy") + assert fp1 == fp2 + assert len(fp1) == 10 + + +def test_compute_fingerprint_differs_by_family_and_engine_mode(): + a = cert.compute_fingerprint("step_1.pca.comps[0][1]", "tolerant", "clojure-legacy") + b = cert.compute_fingerprint("step_1.pca.comps[0][1]", "exact", "clojure-legacy") + c = cert.compute_fingerprint("step_1.pca.comps[0][1]", "tolerant", "improved") + assert len({a, b, c}) == 3 + + +# --------------------------------------------------------------------------- +# Ledger. +# --------------------------------------------------------------------------- +def test_update_ledger_appends_new_as_open(): + obs = [{"path_pattern": "pca.comps[][]", "family": "tolerant", "engine_mode": "clojure-legacy", + "dataset": "vw", "schedule_id": "uniform8-clojure-legacy", "step": 3}] + updated = cert.update_ledger({}, obs) + key = cert.fingerprint_key_for("pca.comps[][]", "tolerant", "clojure-legacy") + assert key in updated + assert updated[key]["status"] == "open" + assert updated[key]["diagnosis"] is None + assert updated[key]["first_seen"] == {"dataset": "vw", "schedule": "uniform8-clojure-legacy", "step": 3} + + +def test_update_ledger_preserves_existing_diagnosis(): + key = cert.fingerprint_key_for("pca.comps[][]", "tolerant", "clojure-legacy") + ledger = {key: {"path_pattern": "pca.comps[][]", "family": "tolerant", + "engine_mode": "clojure-legacy", + "first_seen": {"dataset": "vw", "schedule": "uniform8-clojure-legacy", "step": 1}, + "status": "diagnosed", "diagnosis": "PCA power-iteration seed differs (see #123)"}} + obs = [{"path_pattern": "pca.comps[][]", "family": "tolerant", "engine_mode": "clojure-legacy", + "dataset": "biodiversity", "schedule_id": "uniform8-clojure-legacy", "step": 7}] + updated = cert.update_ledger(ledger, obs) + assert updated[key]["status"] == "diagnosed" + assert updated[key]["diagnosis"] == "PCA power-iteration seed differs (see #123)" + assert updated[key]["first_seen"]["dataset"] == "vw" + + +def test_save_and_load_ledger_round_trip_sorted(tmp_path): + path = tmp_path / "divergences.json" + ledger = {"FP-b000000000": {"status": "open"}, "FP-a000000000": {"status": "open"}} + cert.save_ledger(ledger, path) + text = path.read_text() + assert text.index('"FP-a000000000"') < text.index('"FP-b000000000"') + assert cert.load_ledger(path) == ledger + + +def test_load_ledger_missing_file_returns_empty_dict(tmp_path): + assert cert.load_ledger(tmp_path / "does-not-exist.json") == {} + + +def test_annotate_by_key_with_and_without_diagnosis(): + ledger = {"FP-x": {"status": "open", "diagnosis": None}, + "FP-y": {"status": "diagnosed", "diagnosis": "A" * 100}} + assert cert.annotate_by_key(ledger, "FP-x") == "[known FP-x: status=open]" + assert cert.annotate_by_key(ledger, "FP-y") == f"[known FP-y: {'A' * 60}]" + assert cert.annotate_by_key(ledger, "FP-missing") is None + + +# --------------------------------------------------------------------------- +# Recording caches (subprocess calls MOCKED). +# --------------------------------------------------------------------------- +def test_ensure_py_recording_cache_hit_then_miss_on_change(tmp_path, monkeypatch): + calls = {"n": 0} + + def fake_run(cmd, *, cwd, env): + calls["n"] += 1 + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + entry = _make_entry() + spec = sched.preset_single_cut("vw", 100, schedule_id=entry.schedule_id) + + _, cached1 = cert.ensure_py_recording(entry, spec, "votes-sha-1", root=tmp_path) + assert cached1 is False and calls["n"] == 1 + + _, cached2 = cert.ensure_py_recording(entry, spec, "votes-sha-1", root=tmp_path) + assert cached2 is True and calls["n"] == 1 + + _, cached3 = cert.ensure_py_recording(entry, spec, "votes-sha-2", root=tmp_path) + assert cached3 is False and calls["n"] == 2 + + _, cached4 = cert.ensure_py_recording(entry, spec, "votes-sha-2", root=tmp_path, refresh=True) + assert cached4 is False and calls["n"] == 3 + + +def test_ensure_py_recording_raises_certify_error_on_nonzero_exit(tmp_path, monkeypatch): + def fake_run(cmd, *, cwd, env): + return _fake_completed(returncode=1, stderr="boom") + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + entry = _make_entry() + spec = sched.preset_single_cut("vw", 10, schedule_id=entry.schedule_id) + with pytest.raises(cert.CertifyError) as exc_info: + cert.ensure_py_recording(entry, spec, "sha", root=tmp_path) + assert exc_info.value.stage + + +def test_ensure_clj_recording_cache_hit_then_miss_on_change(tmp_path, monkeypatch): + calls = {"n": 0} + + def fake_run(cmd, *, cwd, env): + calls["n"] += 1 + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + entry = _make_entry() + spec = sched.preset_single_cut("vw", 100, schedule_id=entry.schedule_id) + votes_csv = tmp_path / "vw-votes.csv" + + _, cached1 = cert.ensure_clj_recording(entry, spec, "votes-sha-1", votes_csv, root=tmp_path) + assert cached1 is False and calls["n"] == 1 + + _, cached2 = cert.ensure_clj_recording(entry, spec, "votes-sha-1", votes_csv, root=tmp_path) + assert cached2 is True and calls["n"] == 1 + + _, cached3 = cert.ensure_clj_recording(entry, spec, "votes-sha-1", votes_csv, root=tmp_path, + refresh=True) + assert cached3 is False and calls["n"] == 2 + + +def test_ensure_clj_recording_raises_certify_error_on_nonzero_exit(tmp_path, monkeypatch): + def fake_run(cmd, *, cwd, env): + return _fake_completed(returncode=1, stderr="clojure blew up") + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + entry = _make_entry() + spec = sched.preset_single_cut("vw", 10, schedule_id=entry.schedule_id) + with pytest.raises(cert.CertifyError) as exc_info: + cert.ensure_clj_recording(entry, spec, "sha", tmp_path / "votes.csv", root=tmp_path) + assert exc_info.value.stage + + +def test_certify_entry_skipped_for_missing_dataset(tmp_path): + entry = _make_entry(dataset="no-such-dataset-xyz") + result, ledger = cert.certify_entry(entry, root=tmp_path, ledger={}) + assert result["verdict"] == "SKIPPED" + assert result["reason"] == "dataset-unavailable" + assert ledger == {} + + +# --------------------------------------------------------------------------- +# Focuser. +# --------------------------------------------------------------------------- +def test_run_focus_picks_earliest_divergent_step(tmp_path): + ds, sid = "vw", "uniform8-clojure-legacy" + rec_dir = tmp_path / ds / sid + clj_blobs = [_acceptance_blob(n=3), _acceptance_blob(n=4), _acceptance_blob(n=5)] + py_blobs = [_acceptance_blob(n=3), _acceptance_blob(n=44), _acceptance_blob(n=55)] + for i, b in enumerate(clj_blobs): + _write_clj_step(rec_dir / "clj", i, b) + for i, b in enumerate(py_blobs): + _write_py_step(rec_dir / "py", i, b) + + ledger_path = tmp_path / "divergences.json" + result = cert.run_focus(ds, sid, root=tmp_path, ledger_path=ledger_path) + assert result["verdict"] == "DIVERGENCE" + assert result["step"] == 1 + assert (rec_dir / "focus-report.json").exists() + # The ledger is written to the INJECTED path, never the real committed one. + assert ledger_path.exists() + + +def test_run_focus_match_when_no_divergence(tmp_path): + ds, sid = "vw", "uniform8-clojure-legacy" + rec_dir = tmp_path / ds / sid + blobs = [_acceptance_blob(n=3), _acceptance_blob(n=4)] + for i, b in enumerate(blobs): + _write_clj_step(rec_dir / "clj", i, b) + _write_py_step(rec_dir / "py", i, b) + + result = cert.run_focus(ds, sid, root=tmp_path, ledger_path=tmp_path / "divergences.json") + assert result["verdict"] == "MATCH" + assert result["n_steps"] == 2 + + +def test_run_focus_missing_recording_is_error(tmp_path): + result = cert.run_focus("vw", "no-such-schedule", root=tmp_path, + ledger_path=tmp_path / "divergences.json") + assert result["verdict"] == "ERROR" + + +# --------------------------------------------------------------------------- +# stdout line budget. +# --------------------------------------------------------------------------- +def test_render_run_lines_within_budget_for_many_entries(): + results = [ + {"dataset": "vw", "schedule_id": f"s{i}-clojure-legacy", "engine_mode": "clojure-legacy", + "verdict": "MATCH", "n_steps": 5} + for i in range(100) + ] + report = {"battery": results, "root": "/tmp/x"} + lines = cert.render_run_lines(report) + assert len(lines) <= 40 + assert cert.ACCEPTANCE_NOTICE in lines + + +def test_render_run_lines_small_battery_one_line_per_entry(): + results = [ + {"dataset": "vw", "schedule_id": "single-cut-clojure-legacy", "engine_mode": "clojure-legacy", + "verdict": "MATCH", "n_steps": 3}, + {"dataset": "vw", "schedule_id": "no-dataset", "engine_mode": "clojure-legacy", + "verdict": "SKIPPED", "reason": "dataset-unavailable"}, + ] + report = {"battery": results, "root": "/tmp/x"} + lines = cert.render_run_lines(report) + assert len(lines) <= 40 + assert any("MATCH" in line for line in lines) + assert any("SKIPPED" in line for line in lines) + + +def test_render_focus_lines_within_budget_many_divergences(): + result = { + "dataset": "vw", "schedule_id": "uniform8-clojure-legacy", "engine_mode": "clojure-legacy", + "verdict": "DIVERGENCE", "step": 2, + "families": { + "exact": [{"path": f"step_2.n.{i}", "path_pattern": f"n.{i}", "a": i, "b": i + 1, + "fingerprint": f"FP-{i:010d}", "known": None} for i in range(20)], + "tolerant": [{"path": f"step_2.pca.comps[{i}][0]", "path_pattern": "pca.comps[][]", + "a": 0.1 * i, "b": 0.2 * i, "fingerprint": "FP-aaaa", "known": None} + for i in range(20)], + }, + } + lines = cert.render_focus_lines(result) + assert len(lines) <= 40 + + +def test_render_focus_lines_caps_exact_divergences_shown(): + result = { + "dataset": "vw", "schedule_id": "uniform8-clojure-legacy", "engine_mode": "clojure-legacy", + "verdict": "DIVERGENCE", "step": 0, + "families": { + "exact": [{"path": f"step_0.n.{i}", "path_pattern": f"n.{i}", "a": i, "b": i + 1, + "fingerprint": f"FP-{i:010d}", "known": None} for i in range(8)], + "tolerant": [], + }, + } + lines = cert.render_focus_lines(result, max_per_family=5) + shown = [l for l in lines if l.strip().startswith("step_0.n.")] + assert len(shown) == 5 + assert any("more" in l for l in lines) + + +# --------------------------------------------------------------------------- +# Battery-level exit code. +# --------------------------------------------------------------------------- +def test_battery_exit_code(): + results = [{"verdict": "MATCH"}, {"verdict": "SKIPPED"}] + assert cert.battery_exit_code(results, strict=False) == 0 + assert cert.battery_exit_code(results, strict=True) == 1 + + results2 = [{"verdict": "MATCH"}, {"verdict": "DIVERGENCE"}] + assert cert.battery_exit_code(results2, strict=False) == 1 + + +# --------------------------------------------------------------------------- +# Optional integration test: REAL clojure + REAL py drivers on vw single-cut. +# Opt-in only (slow: JVM startup + a full PCA/clustering pass on both engines). +# --------------------------------------------------------------------------- +@pytest.mark.skipif( + shutil.which("clojure") is None or os.environ.get("RUN_CLJ_INTEGRATION") != "1", + reason="opt-in: needs the clojure CLI on PATH and RUN_CLJ_INTEGRATION=1 " + "(runs the REAL clojure + Python drivers, not mocked)", +) +def test_certify_entry_real_drivers_vw_single_cut(tmp_path): + """End-to-end smoke: real clojure driver + real py driver on vw single-cut. + + Locks the whole pipeline (cache manifests -> subprocess invocation -> + hash-first compare) against the ACTUAL drivers, not mocks. Verdict is + intentionally not asserted to be MATCH — Python's clojure-legacy engine + mode is a parity APPROXIMATION, not a guarantee of bit-identical output; + this test only proves both drivers ran to completion and certify could + compare their results end to end. + """ + entry = cert.parse_battery_entry( + {"dataset": "vw", "preset": "single-cut", "engine_mode": "clojure-legacy"} + ) + result, ledger = cert.certify_entry(entry, root=tmp_path, ledger={}) + assert result["verdict"] in ("MATCH", "DIVERGENCE"), result + + rec_dir = tmp_path / "vw" / entry.schedule_id + assert (rec_dir / "clj" / "step-000.blob.json").exists() + assert (rec_dir / "py" / "step-000.json").exists() + assert (rec_dir / "clj" / "cache_manifest.json").exists() + assert (rec_dir / "py" / "cache_manifest.json").exists() + + # Re-running must be a pure cache hit — no new driver invocation needed. + calls = {"n": 0} + real_run_subprocess = cert._run_subprocess + + def spy(cmd, *, cwd, env): + calls["n"] += 1 + return real_run_subprocess(cmd, cwd=cwd, env=env) + + import unittest.mock + + with unittest.mock.patch.object(cert, "_run_subprocess", spy): + result2, _ = cert.certify_entry(entry, root=tmp_path, ledger=ledger) + assert calls["n"] == 0 + assert result2["verdict"] == result["verdict"] diff --git a/delphi/tests/replay_harness/test_certify_cli.py b/delphi/tests/replay_harness/test_certify_cli.py new file mode 100644 index 000000000..cbdd963db --- /dev/null +++ b/delphi/tests/replay_harness/test_certify_cli.py @@ -0,0 +1,158 @@ +"""CLI-level tests for scripts/certify.py — SPEC A. + +Drives the real click CLI via CliRunner, mocking the polismath.replay.certify +library calls that would otherwise touch real datasets, subprocesses, or the +committed ledger — this file never lets a real clojure/py driver run. Verifies +exit codes, flag plumbing, and the end-to-end ≤40-line stdout budget through +the actual CLI entry point. Unit coverage of the pure `render_*_lines` +functions themselves lives in test_certify.py. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +from click.testing import CliRunner + +_CLI_PATH = Path(__file__).resolve().parents[2] / "scripts" / "certify.py" + + +def _module(): + spec = importlib.util.spec_from_file_location("certify_cli", _CLI_PATH) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def test_run_exits_zero_on_all_match(monkeypatch): + mod = _module() + report = {"battery": [ + {"dataset": "vw", "schedule_id": "single-cut-clojure-legacy", + "engine_mode": "clojure-legacy", "verdict": "MATCH", "n_steps": 3}, + ], "root": "/tmp/x"} + monkeypatch.setattr(mod.cert, "load_battery", lambda path: ["entry"]) + monkeypatch.setattr(mod.cert, "run_battery", lambda entries, **kw: report) + + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 0, res.output + assert mod.cert.ACCEPTANCE_NOTICE in res.output + assert len(res.output.strip().splitlines()) <= 40 + + +def test_run_exits_nonzero_on_divergence(monkeypatch): + mod = _module() + report = {"battery": [ + {"dataset": "vw", "schedule_id": "uniform8-clojure-legacy", + "engine_mode": "clojure-legacy", "verdict": "DIVERGENCE", + "first_div_step": 2, "n_div_steps": 1, "top_paths": []}, + ], "root": "/tmp/x"} + monkeypatch.setattr(mod.cert, "load_battery", lambda path: ["entry"]) + monkeypatch.setattr(mod.cert, "run_battery", lambda entries, **kw: report) + + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 1, res.output + + +def test_run_exits_nonzero_on_error(monkeypatch): + mod = _module() + report = {"battery": [ + {"dataset": "vw", "schedule_id": "x", "engine_mode": "clojure-legacy", + "verdict": "ERROR", "stage": "py-driver", "reason": "boom"}, + ], "root": "/tmp/x"} + monkeypatch.setattr(mod.cert, "load_battery", lambda path: ["entry"]) + monkeypatch.setattr(mod.cert, "run_battery", lambda entries, **kw: report) + + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 1, res.output + + +def test_run_skipped_ok_by_default_but_fails_with_strict(monkeypatch): + mod = _module() + report = {"battery": [ + {"dataset": "vw", "schedule_id": "x", "engine_mode": "clojure-legacy", + "verdict": "SKIPPED", "reason": "dataset-unavailable"}, + ], "root": "/tmp/x"} + monkeypatch.setattr(mod.cert, "load_battery", lambda path: ["entry"]) + monkeypatch.setattr(mod.cert, "run_battery", lambda entries, **kw: report) + + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 0, res.output + + res_strict = CliRunner().invoke(mod.cli, ["run", "--strict"]) + assert res_strict.exit_code == 1, res_strict.output + + +def test_run_passes_cli_flags_through_to_library(monkeypatch): + mod = _module() + captured: dict = {} + + def fake_run_battery(entries, **kw): + captured.update(kw) + return {"battery": [], "root": "/tmp/x"} + + monkeypatch.setattr(mod.cert, "load_battery", lambda path: []) + monkeypatch.setattr(mod.cert, "run_battery", fake_run_battery) + + res = CliRunner().invoke( + mod.cli, + ["run", "--only", "vw:uniform8-clojure-legacy", "--refresh-clj", "--refresh-py"], + ) + assert res.exit_code == 0, res.output + assert captured["only"] == "vw:uniform8-clojure-legacy" + assert captured["refresh_clj"] is True + assert captured["refresh_py"] is True + + +def test_run_stdout_budget_with_large_mocked_battery(monkeypatch): + mod = _module() + report = { + "battery": [ + {"dataset": "vw", "schedule_id": f"s{i}-clojure-legacy", + "engine_mode": "clojure-legacy", "verdict": "MATCH", "n_steps": 3} + for i in range(200) + ], + "root": "/tmp/x", + } + monkeypatch.setattr(mod.cert, "load_battery", lambda path: ["entry"] * 200) + monkeypatch.setattr(mod.cert, "run_battery", lambda entries, **kw: report) + + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 0, res.output + assert len(res.output.strip().splitlines()) <= 40 + + +def test_focus_exits_zero_on_match(monkeypatch): + mod = _module() + monkeypatch.setattr( + mod.cert, "run_focus", + lambda ds, sid, root=None: {"dataset": ds, "schedule_id": sid, + "verdict": "MATCH", "n_steps": 4}, + ) + res = CliRunner().invoke(mod.cli, ["focus", "vw", "uniform8-clojure-legacy"]) + assert res.exit_code == 0, res.output + assert mod.cert.ACCEPTANCE_NOTICE in res.output + + +def test_focus_exits_nonzero_on_divergence(monkeypatch): + mod = _module() + monkeypatch.setattr(mod.cert, "run_focus", lambda ds, sid, root=None: { + "dataset": ds, "schedule_id": sid, "verdict": "DIVERGENCE", "step": 1, + "families": { + "exact": [{"path": "step_1.n", "a": 3, "b": 4, "known": None}], + "tolerant": [], + }, + }) + res = CliRunner().invoke(mod.cli, ["focus", "vw", "uniform8-clojure-legacy"]) + assert res.exit_code == 1, res.output + assert len(res.output.strip().splitlines()) <= 40 + + +def test_focus_exits_nonzero_on_error(monkeypatch): + mod = _module() + monkeypatch.setattr(mod.cert, "run_focus", lambda ds, sid, root=None: { + "dataset": ds, "schedule_id": sid, "verdict": "ERROR", + "stage": "recording-missing", "reason": "nope", + }) + res = CliRunner().invoke(mod.cli, ["focus", "vw", "nope"]) + assert res.exit_code == 1, res.output diff --git a/delphi/tests/replay_harness/test_clj_crosslang.py b/delphi/tests/replay_harness/test_clj_crosslang.py index 4f0e890c4..b1f526a14 100644 --- a/delphi/tests/replay_harness/test_clj_crosslang.py +++ b/delphi/tests/replay_harness/test_clj_crosslang.py @@ -16,7 +16,7 @@ import pytest -from replay_harness.clj_crosslang import ( +from polismath.replay.crosslang import ( PREP_MAIN_KEYS, clj_blob_files, clj_recording_to_py_store, diff --git a/math/dev/replay_smoke.sh b/math/dev/replay_smoke.sh index 7ebac8264..2d72aa616 100755 --- a/math/dev/replay_smoke.sh +++ b/math/dev/replay_smoke.sh @@ -60,8 +60,7 @@ if [[ "${1:-}" == "--xlang" ]]; then echo ">> Cross-language compare (clj vs py)" ( cd "$DELPHI" && OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 uv run python - "$OUT" <<'PY' import sys, tempfile -sys.path.insert(0, "tests/replay_harness") -from clj_crosslang import compare_clj_vs_py +from polismath.replay.crosslang import compare_clj_vs_py from polismath.replay import stepcompare as sc with tempfile.TemporaryDirectory() as shim: print(sc.format_report(compare_clj_vs_py(sys.argv[1], shim_root=shim)))