diff --git a/delphi/docs/CLJ-PARITY-FIXES-JOURNAL.md b/delphi/docs/CLJ-PARITY-FIXES-JOURNAL.md index 151861d3d..1a785b18c 100644 --- a/delphi/docs/CLJ-PARITY-FIXES-JOURNAL.md +++ b/delphi/docs/CLJ-PARITY-FIXES-JOURNAL.md @@ -4045,3 +4045,117 @@ Category-1 nondeterminism issues opened: #2660 (Q10), #2661 (Q12), #2662 (Q13/Q18) — all "fixed by the Python push", determinism pinned by test_driver.py::test_determinism_bit_identical_except_wall_clock + the battery's pass-pair bit-comparisons. + +## Session 7 (2026-07-27): Phase 0 battery speedup + Phase 1 engine_mode inventory + +Goal: GOAL_CUTOVER_READY.md. Orientation: s6 battery verdicts CONFIRMED +(fresh cached run on the triaged tree: 20/20 MATCH in 22.3s); all 21 stack +PRs (#2638-#2658) show 0 unresolved review threads (GraphQL sweep). + +### Phase 0 — battery tooling speedup (certify.py) + +TDD (RED: 7 failing tests → GREEN; replay_harness 464 passed / 5 skipped +vs 456/5 baseline — delta is exactly the 8 new tests): + +- **0a hash scoping**: py recording cache key is now the ENGINE-scoped tree + hash — `sha256_tree(..., exclude=_ENGINE_TREE_EXCLUDE)` with + `poller/`, `replay/certify.py`, `replay/poller_equiv.py`, + `replay/prodclone.py`, `replay/shard_bench.py` excluded (none is in the + replay subprocess import graph: scripts/replay_driver.py → driver/ + schedule/real_data/store/stepcompare/types → engine; verified by import + audit). Manifest key renamed `py_tree_sha256` → `engine_tree_sha256`; + the one-time invalidation of all 20 py recordings was DELIBERATE — it + doubled as the timed parallel first-pass A/B run. +- **0b parallelism**: `run_battery(..., workers=N)`; per-entry heavy work + (`_certify_entry_heavy`: drivers + hash-first compare, NO ledger) fans + out on a ThreadPoolExecutor; `_fold_entry_into_ledger` stays strictly + serial in battery order (annotate-before-update preserved), so report + + ledger are bit-identical to workers=1 (pinned by test). Step-verdict + cache writes made atomic (tmp + os.replace). CLI `--workers` default 6. +- **0c A/B**: serial baseline 36 min (journal s6); parallel first-pass + timing recorded below when the run (launched this session) completes. + Cached-pass integrity + harness-edit cache retention proven after. + +### Phase 1 — engine_mode inventory (grep gate: delphi/polismath/, 0 hits) + +Branch sites and classification (DELETE = remove outright; PARK = extract +VERBATIM to improvements/* side commit first; KEEP = legacy side becomes +the only path): + +| Site | What branches | Classification | +|---|---|---| +| conversation.py:514 | Q1 ban filtering (improved drops banned rows) | DELETE (Julien: bans dropped as a feature; mod_out_ptpts stays inert) | +| conversation.py:744 | <2 PCA guard (improved short-circuits tiny) | PARK item 2; KEEP empty-only short-circuit | +| conversation.py:769 | PCA warm start (legacy powerit+start vectors) | KEEP legacy; PARK improved cold+solver-choice with item 8 | +| conversation.py:905 | degenerate-tick clustering guard | PARK item 2; KEEP legacy fall-through | +| conversation.py:1213 | repness tid_order arrival-order tie-break | KEEP legacy (always pass tid_order); improved None side trivial, no park | +| conversation.py:1473 | Q15 watermark drop on votes tick | KEEP legacy drop; PARK item 4 (persistent watermark) | +| conversation.py:1569 | Q2 prev-tick group-votes for priorities | KEEP legacy prev; PARK item 5 (current-tick) | +| conversation.py:1951, 2439, 3115 | tally from raw vs zeroed mat | KEEP legacy raw (improved side has known S-inflation bug — DELETE, no park) | +| conversation.py:2134 | in-conv carry + greedy floor (PR-E) | KEEP legacy; improved threshold-only DELETE (not queued) | +| conversation.py:2416 | votes-base bucket vectors vs int totals | KEEP legacy buckets; improved DELETE (cleanup item 11 territory) | +| conversation.py:2510 | group-aware-consensus legacy formula | KEEP legacy; improved DELETE | +| conversation.py:2555 | in-conv serialization of persisted set | KEEP legacy | +| conversation.py:2619 | _apply_legacy_blob_shape | KEEP (unconditional) | +| conversation.py:2865 | from_dict restore seam (legacy flag) | KEEP legacy side | +| repness.py:251 | total_source votes_in_groups vs votes_only | KEEP legacy; improved DELETE (not queued) | +| repness.py:858 | <2 repness guard | PARK item 2; KEEP legacy proceed | +| replay/driver.py:121+ | mod semantics: mod_update (legacy) vs update_moderation; improved+restart NotImplementedError | KEEP legacy path; DELETE improved branch + guard | + +Flag machinery (delete last, after all callers): utils/engine_mode.py +(whole file), poller/service.py engine_mode config/apply_engine_mode/log +fields, poller/__init__.py + env_flags.py docstring mentions. + +**Discovery — the gate covers harness files too**: certify.py (63 refs: +BatteryEntry.engine_mode, battery JSON key, run_py_driver env, manifest +key, fingerprint component), poller_equiv.py (12), store.py env recording. +All identifier references must go. BUT: schedule-id STRINGS +("uniform8-clojure-legacy") and ledger fingerprint keys do NOT match the +grep gate — keep them verbatim so recordings dirs and the historical +divergence ledger stay valid. Fingerprints: hardcode the literal +"clojure-legacy" as the mode component to preserve keys. + +**Other impl flag**: POLISMATH_PCA_IMPL (pca.py, recorded by store.py). +Legacy requires powerit, so after collapse the sklearn path is dead code — +park it with item 8 and delete the flag (goal spirit: ONE code path), +even though the grep gate doesn't name it. + +**No park needed for queue items 3/6/7** (Q3 kmeans iters, Q11 euclidean, +Q16 rank-1): they have no improved-mode branches today — they're future +fixes, not extractions. Park commits needed only for items 2, 4, 5, 8. + +Phase 2 chunk order (battery per chunk): (1) conversation.py + +repness.py engine branches, (2) driver.py mod semantics + restart guard, +(3) flag machinery deletion, (4) harness identifier purge (one commit — +manifest "engine_mode" key drop pays its single py re-record together +with the Phase 4 goldens re-record). + +### Phase 0c A/B results (2026-07-27, this session) + +- **First pass, all 20 py recordings invalidated** (manifest key change), + `--workers 6` (10-core host): **19m17s wall / 37m36s user** vs ~36 min + serial (journal s6). Speedup capped by the long-pole entry — + pakistan:uniform8 alone ran ~18 of the 19 minutes; everything else + finished by ~minute 7. The <8 min target is unreachable without + intra-entry parallelism (out of scope); the practical win is that the + OTHER 19 entries certify in ~7 min and harness edits no longer trigger + re-replay at all. +- **Cached pass**: 20/20 MATCH in 22s (unchanged). +- **Harness-only edit live proof**: appended a comment line to certify.py + → battery 20/20 MATCH in 2m9s with ZERO re-replays (recordings stayed + cached — the hash scoping works). The 2m is step-verdict re-derivation: + certify.py is deliberately part of `_comparer_code_hash` (stale MATCH + is the worst failure mode), so tolerant-family mismatch steps re-run + the comparer. Cost table now: harness edit ≈2m, engine edit ≈19m, + no edit ≈22s. + +### Gotcha (hit + recovered this session): `git checkout --` in the colocated repo + +Reverting the probe line with `git checkout -- ` clobbered the +working-copy file back to the PARENT commit's version (git HEAD sits at +@- in jj-colocated repos) — wiping the session's uncommitted certify.py +changes. Recovered from jj's last auto-snapshot (`git show +:`), verified byte-identical (@ id unchanged), tests +green. Rule: in this repo, undo scratch edits with a targeted edit (sed/ +editor), NEVER `git checkout --`/`git restore` (index = parent, not @), +and NEVER `jj restore --from @-` for a file carrying uncommitted work. diff --git a/delphi/polismath/replay/certify.py b/delphi/polismath/replay/certify.py index 0f6582a23..9f3ca7dce 100644 --- a/delphi/polismath/replay/certify.py +++ b/delphi/polismath/replay/certify.py @@ -48,6 +48,8 @@ import os import re import subprocess +import threading +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -285,16 +287,22 @@ def sha256_file(path: str | Path) -> str: return h.hexdigest() -def sha256_tree(root: str | Path, pattern: str = "**/*") -> str: +def sha256_tree(root: str | Path, pattern: str = "**/*", *, + exclude: tuple[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.""" + iteration order, sensitive to both a file's path and its content. + + ``exclude`` entries are posix relpaths under ``root``: a trailing ``/`` + excludes that whole subtree, otherwise the exact file is excluded.""" 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() + if any(rel == e or (e.endswith("/") and rel.startswith(e)) for e in exclude): + continue h.update(rel.encode()) h.update(b"\0") h.update(p.read_bytes()) @@ -537,9 +545,35 @@ def _write_manifest(manifest_path: Path, manifest: dict[str, Any]) -> None: json.dump(manifest, fh, indent=2, sort_keys=True) +#: Pure-harness paths (relative to ``polismath/``) excluded from the py +#: recording cache key: none of them is reachable from the replay subprocess +#: import graph (``scripts/replay_driver.py`` → driver/schedule/real_data/ +#: store/stepcompare/types → the engine), so editing them cannot change +#: replay outputs (Julien ruling 2026-07-27, GOAL_CUTOVER_READY.md Phase 0a). +#: Trailing ``/`` = whole subtree. driver.py/schedule.py/real_data.py DO +#: shape replays and deliberately stay in the hash. +_ENGINE_TREE_EXCLUDE: tuple[str, ...] = ( + "poller/", + "replay/certify.py", + "replay/poller_equiv.py", + "replay/prodclone.py", + "replay/shard_bench.py", +) + + +def engine_tree_hash(polismath_root: str | Path | None = None) -> str: + """Tree hash of the ENGINE surface: every ``polismath/**/*.py`` except + :data:`_ENGINE_TREE_EXCLUDE` — the py recording cache key. Harness-only + edits therefore keep recordings cached (the ~36-min full py re-replay is + reserved for actual engine changes). Uncached because tests mutate trees; + the battery hot path goes through :func:`_engine_tree_hash_cached`.""" + root = Path(polismath_root) if polismath_root is not None else _DELPHI_ROOT / "polismath" + return sha256_tree(root, "**/*.py", exclude=_ENGINE_TREE_EXCLUDE) + + @functools.lru_cache(maxsize=1) -def _py_tree_hash() -> str: - return sha256_tree(_DELPHI_ROOT / "polismath", "**/*.py") +def _engine_tree_hash_cached() -> str: + return engine_tree_hash() @functools.lru_cache(maxsize=1) @@ -558,8 +592,13 @@ def ensure_py_recording( 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)``.""" + sha256, schedule hash, engine_mode, ENGINE-scoped tree hash); else (re)run + the Python driver in a subprocess. Returns ``(py_dir, was_cached)``. + + The 2026-07-27 switch from the full-``polismath`` tree hash to the + engine-scoped one (key renamed ``py_tree_sha256`` → ``engine_tree_sha256``) + deliberately invalidated every existing py recording ONCE — that forced + re-replay doubled as the timed A/B run for the parallel battery.""" rec_dir = st.recording_dir(entry.dataset, entry.schedule_id, root=root) py_dir = rec_dir / "py" manifest_path = py_dir / "cache_manifest.json" @@ -567,7 +606,7 @@ def ensure_py_recording( "votes_sha256": votes_sha, "schedule_hash": canonical_schedule_hash(spec), "engine_mode": entry.engine_mode, - "py_tree_sha256": _py_tree_hash(), + "engine_tree_sha256": _engine_tree_hash_cached(), } if not refresh and _manifest_matches(manifest_path, expected): return py_dir, True @@ -675,8 +714,15 @@ def compare_recording_pair( 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: + # Atomic write (tmp + rename): parallel battery workers may reach + # the same hash-pair key concurrently; a reader must never see a + # torn file served as a cached verdict. + tmp_path = cache_path.with_suffix( + f".tmp-{os.getpid()}-{threading.get_ident()}" + ) + with open(tmp_path, "w") as fh: json.dump(report, fh, indent=2, sort_keys=True, default=str) + os.replace(tmp_path, cache_path) report = dict(report) report["hash_match"] = False per_step.append(report) @@ -736,21 +782,18 @@ def _summarize_divergences(cmp_result: dict[str, Any], *, engine_mode: str) -> d # --------------------------------------------------------------------------- # Per-entry certification. # --------------------------------------------------------------------------- -def certify_entry( +def _certify_entry_heavy( 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() - +) -> dict[str, Any]: + """The parallel-safe part of certifying one entry: ensure both recordings + and run the hash-first compare — NO ledger access, so a whole battery can + fan these out across workers. Terminal verdicts (SKIPPED/ERROR/MATCH) come + back complete; a divergence carries its summary under ``"_summary"`` for + the strictly-serial ledger fold (:func:`_fold_entry_into_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) + return {"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "SKIPPED", + "reason": "dataset-unavailable"} try: votes_csv = votes_csv_path(entry.dataset) @@ -771,50 +814,77 @@ def certify_entry( refresh=refresh_clj, comments_csv=comments_csv) 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) + return {"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "ERROR", + "stage": exc.stage, "reason": str(exc)} 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) + return {"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "ERROR", + "stage": "setup", "reason": str(exc)} 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) + 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"} 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) + return {"dataset": entry.dataset, "schedule_id": entry.schedule_id, + "engine_mode": entry.engine_mode, "verdict": "MATCH", + "n_steps": cmp_result["aligned_steps"]} summary = _summarize_divergences(cmp_result, engine_mode=entry.engine_mode) + return {"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"], "_summary": summary} + + +def _fold_entry_into_ledger( + result: dict[str, Any], ledger: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + """Serial half of certifying an entry: annotate a divergence's top paths + against the (accumulating) ledger, then record its observations. Annotate + BEFORE update — a fingerprint first seen in THIS entry reads as new, not + known — exactly matching the pre-parallel serial semantics.""" + summary = result.pop("_summary", None) + if summary is None: + return result, ledger + 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"]} + {"path_pattern": o["path_pattern"], "family": o["family"], + "engine_mode": result["engine_mode"], "dataset": result["dataset"], + "schedule_id": result["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"], - } + result["top_paths"] = summary["top_paths"] return result, ledger +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() + heavy = _certify_entry_heavy(entry, root=root, refresh_clj=refresh_clj, + refresh_py=refresh_py) + return _fold_entry_into_ledger(heavy, ledger) + + # --------------------------------------------------------------------------- # Battery-level orchestration. # --------------------------------------------------------------------------- @@ -828,10 +898,17 @@ def _filter_only(entries: list[BatteryEntry], only: str) -> list[BatteryEntry]: 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, + workers: int = 1, ) -> 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.""" + :func:`render_run_lines` for the stdout rendering. + + ``workers`` > 1 fans the per-entry heavy work (driver subprocesses + + hash-first compare) across threads — entries are independent by + construction (disjoint recording dirs, atomic verdict-cache writes). The + ledger fold stays strictly serial and in battery order, so the report and + ledger are identical to a ``workers=1`` run.""" root = root or st.replays_root() ledger_path = ledger_path or default_ledger_path() ledger = load_ledger(ledger_path) @@ -839,10 +916,19 @@ def run_battery( if only: entries = _filter_only(entries, only) + def _heavy(entry: BatteryEntry) -> dict[str, Any]: + return _certify_entry_heavy(entry, root=root, refresh_clj=refresh_clj, + refresh_py=refresh_py) + + if workers > 1 and len(entries) > 1: + with ThreadPoolExecutor(max_workers=min(workers, len(entries))) as pool: + heavies = list(pool.map(_heavy, entries)) + else: + heavies = [_heavy(e) for e in entries] + results = [] - for entry in entries: - result, ledger = certify_entry(entry, root=root, refresh_clj=refresh_clj, - refresh_py=refresh_py, ledger=ledger) + for heavy in heavies: + result, ledger = _fold_entry_into_ledger(heavy, ledger) results.append(result) save_ledger(ledger, ledger_path) diff --git a/delphi/scripts/certify.py b/delphi/scripts/certify.py index df0318d99..54e33833a 100644 --- a/delphi/scripts/certify.py +++ b/delphi/scripts/certify.py @@ -53,11 +53,14 @@ def cli() -> None: 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): +@click.option("--workers", type=int, default=6, show_default=True, + help="Parallel battery entries (drivers + compare); the ledger " + "fold stays serial, so results match --workers 1 exactly.") +def run(battery_path, only, refresh_clj, refresh_py, strict, root, workers): """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) + refresh_py=refresh_py, only=only, workers=workers) for line in cert.render_run_lines(report): click.echo(line) sys.exit(cert.battery_exit_code(report["battery"], strict=strict)) diff --git a/delphi/tests/replay_harness/test_certify.py b/delphi/tests/replay_harness/test_certify.py index fa044dd0a..645cf5257 100644 --- a/delphi/tests/replay_harness/test_certify.py +++ b/delphi/tests/replay_harness/test_certify.py @@ -184,6 +184,104 @@ def test_sha256_tree_sensitive_to_relpath_not_just_content(tmp_path): assert cert.sha256_tree(d1, "**/*.py") != cert.sha256_tree(d2, "**/*.py") +def test_sha256_tree_exclude_file_and_dir_prefix(tmp_path): + """``exclude`` drops exact file relpaths and (trailing-slash) dir subtrees + from the digest — an excluded file's content no longer moves the hash.""" + d = tmp_path / "pkg" + d.mkdir() + (d / "engine.py").write_text("e = 1\n") + (d / "harness.py").write_text("h = 1\n") + (d / "tools").mkdir() + (d / "tools" / "aux.py").write_text("t = 1\n") + + exclude = ("harness.py", "tools/") + h_all = cert.sha256_tree(d, "**/*.py") + h1 = cert.sha256_tree(d, "**/*.py", exclude=exclude) + assert h1 != h_all # exclusion actually removes content from the digest + + (d / "harness.py").write_text("h = 2\n") + (d / "tools" / "aux.py").write_text("t = 2\n") + assert cert.sha256_tree(d, "**/*.py", exclude=exclude) == h1 + + (d / "engine.py").write_text("e = 2\n") + assert cert.sha256_tree(d, "**/*.py", exclude=exclude) != h1 + + +def test_engine_tree_exclude_entries_exist_and_keep_replay_shapers(): + """Every exclusion names a real path under polismath/ (a rename must not + turn it into a silent no-op), and the replay-shaping files stay hashed.""" + pm = cert._DELPHI_ROOT / "polismath" + for e in cert._ENGINE_TREE_EXCLUDE: + p = pm / e.rstrip("/") + if e.endswith("/"): + assert p.is_dir(), e + else: + assert p.is_file(), e + kept = {"replay/driver.py", "replay/schedule.py", "replay/real_data.py", + "replay/store.py", "replay/stepcompare.py", "replay/types.py"} + assert not kept & set(cert._ENGINE_TREE_EXCLUDE) + + +def test_engine_tree_hash_ignores_harness_edits_sees_engine_edits(tmp_path): + pm = tmp_path / "polismath" + for rel in ("replay/certify.py", "replay/prodclone.py", "replay/driver.py", + "poller/service.py", "conversation/conversation.py"): + p = pm / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text("v = 1\n") + + h0 = cert.engine_tree_hash(pm) + (pm / "replay" / "certify.py").write_text("v = 2\n") + (pm / "poller" / "service.py").write_text("v = 2\n") + assert cert.engine_tree_hash(pm) == h0 + + (pm / "replay" / "driver.py").write_text("v = 2\n") + h1 = cert.engine_tree_hash(pm) + assert h1 != h0 + (pm / "conversation" / "conversation.py").write_text("v = 2\n") + assert cert.engine_tree_hash(pm) != h1 + + +def test_ensure_py_recording_cache_survives_harness_only_edit(tmp_path, monkeypatch): + """The py cache manifest is keyed on the ENGINE-scoped tree hash: editing + an excluded harness file must NOT invalidate a recording; editing an + engine file must.""" + calls = {"n": 0} + + def fake_run(cmd, *, cwd, env): + calls["n"] += 1 + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + fake_delphi = tmp_path / "delphi" + pm = fake_delphi / "polismath" + (pm / "replay").mkdir(parents=True) + (pm / "replay" / "certify.py").write_text("h = 1\n") + (pm / "replay" / "driver.py").write_text("d = 1\n") + monkeypatch.setattr(cert, "_DELPHI_ROOT", fake_delphi) + cert._engine_tree_hash_cached.cache_clear() + try: + entry = _make_entry() + spec = sched.preset_single_cut("vw", 100, schedule_id=entry.schedule_id) + root = tmp_path / "root" + + _, cached1 = cert.ensure_py_recording(entry, spec, "sha", root=root) + assert cached1 is False and calls["n"] == 1 + + (pm / "replay" / "certify.py").write_text("h = 2\n") # harness-only edit + cert._engine_tree_hash_cached.cache_clear() + _, cached2 = cert.ensure_py_recording(entry, spec, "sha", root=root) + assert cached2 is True and calls["n"] == 1 + + (pm / "replay" / "driver.py").write_text("d = 2\n") # engine edit + cert._engine_tree_hash_cached.cache_clear() + _, cached3 = cert.ensure_py_recording(entry, spec, "sha", root=root) + assert cached3 is False and calls["n"] == 2 + finally: + cert._engine_tree_hash_cached.cache_clear() + + 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") @@ -711,3 +809,76 @@ def spy(cmd, *, cwd, env): result2, _ = cert.certify_entry(entry, root=tmp_path, ledger=ledger) assert calls["n"] == 0 assert result2["verdict"] == result["verdict"] + + +# --------------------------------------------------------------------------- +# Parallel battery (Phase 0b): workers>1 must produce an identical report and +# ledger to the serial path, results in battery order. +# --------------------------------------------------------------------------- +def _seed_cached_pair(root: Path, ds: str, sid: str, *, divergent: bool) -> None: + """Pre-write a one-step clj/py recording pair under ``///`` + so certify_entry takes the fully-cached path (manifest check mocked).""" + rec = root / ds / sid + blob = _acceptance_blob() + _write_clj_step(rec / "clj", 0, blob) + py_blob = dict(blob, n=blob["n"] + 5) if divergent else blob + _write_py_step(rec / "py", 0, py_blob) + + +def test_run_battery_parallel_matches_serial_report_and_ledger(tmp_path, monkeypatch): + monkeypatch.setattr(cert, "_manifest_matches", lambda mp, exp: True) + monkeypatch.setattr(cert, "_clj_source_hashes", lambda: ("x", "y")) + + entries = [ + _make_entry(schedule_id="p0-match"), + _make_entry(schedule_id="p0-div"), + _make_entry(schedule_id="p0-match2"), + ] + root_a = tmp_path / "root_a" + root_b = tmp_path / "root_b" + for e, div in zip(entries, (False, True, False)): + _seed_cached_pair(root_a, "vw", e.schedule_id, divergent=div) + _seed_cached_pair(root_b, "vw", e.schedule_id, divergent=div) + + ledger_a = tmp_path / "ledger_a.json" + ledger_b = tmp_path / "ledger_b.json" + rep_a = cert.run_battery(entries, root=root_a, ledger_path=ledger_a) + rep_b = cert.run_battery(entries, root=root_b, ledger_path=ledger_b, workers=3) + + assert rep_a["battery"] == rep_b["battery"] + assert [(r["dataset"], r["schedule_id"]) for r in rep_b["battery"]] == \ + [("vw", e.schedule_id) for e in entries] + assert rep_b["battery"][1]["verdict"] == "DIVERGENCE" + assert json.loads(ledger_a.read_text()) == json.loads(ledger_b.read_text()) + assert json.loads(ledger_b.read_text()) # non-vacuous: divergence reached it + + +def test_run_battery_workers_one_is_default_and_identical(tmp_path, monkeypatch): + monkeypatch.setattr(cert, "_manifest_matches", lambda mp, exp: True) + monkeypatch.setattr(cert, "_clj_source_hashes", lambda: ("x", "y")) + + entries = [_make_entry(schedule_id="w1-only")] + root_a = tmp_path / "root_a" + root_b = tmp_path / "root_b" + _seed_cached_pair(root_a, "vw", "w1-only", divergent=False) + _seed_cached_pair(root_b, "vw", "w1-only", divergent=False) + + rep_default = cert.run_battery(entries, root=root_a, + ledger_path=tmp_path / "la.json") + rep_w1 = cert.run_battery(entries, root=root_b, + ledger_path=tmp_path / "lb.json", workers=1) + assert rep_default["battery"] == rep_w1["battery"] + + +def test_step_verdict_cache_write_leaves_no_tmp_files(tmp_path): + clj_dir = tmp_path / "clj" + py_dir = tmp_path / "py" + blob = _acceptance_blob() + _write_clj_step(clj_dir, 0, blob) + _write_py_step(py_dir, 0, dict(blob, n=99)) + cert.compare_recording_pair(clj_dir, py_dir, engine_mode="clojure-legacy", + cache_root=tmp_path) + cache_dir = tmp_path / ".certify_cache" / "stepverdicts" + files = list(cache_dir.iterdir()) + assert files + assert all(f.suffix == ".json" for f in files) diff --git a/delphi/tests/replay_harness/test_certify_cli.py b/delphi/tests/replay_harness/test_certify_cli.py index cbdd963db..c3d97f2ef 100644 --- a/delphi/tests/replay_harness/test_certify_cli.py +++ b/delphi/tests/replay_harness/test_certify_cli.py @@ -104,6 +104,27 @@ def fake_run_battery(entries, **kw): assert captured["refresh_py"] is True +def test_run_passes_workers_through_and_defaults_to_six(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", "--workers", "3"]) + assert res.exit_code == 0, res.output + assert captured["workers"] == 3 + + captured.clear() + res = CliRunner().invoke(mod.cli, ["run"]) + assert res.exit_code == 0, res.output + assert captured["workers"] == 6 + + def test_run_stdout_budget_with_large_mocked_battery(monkeypatch): mod = _module() report = {