python-math #8: feat(math): replay harness H-B — Clojure Mode A driver (dev/replay.clj) + cross-language comparer shim

delphi/tests/replay_harness/clj_crosslang.py

@@ -0,0 +1,160 @@
+"""Cross-language store reading — Phase H-B (Clojure Mode A) support.
+
+The Clojure driver (``math/dev/replay.clj``) writes its per-step ``math_main``
+view as ``clj/step-NNN.blob.json`` — the file *is* the raw ``prep-main`` blob
+(design §7), NOT the ``{index, …, blob}`` payload wrapper the Python store uses
+under ``py/step-NNN.json``. This module bridges the two so the EXISTING
+:func:`polismath.replay.stepcompare.compare_recordings` can diff a Clojure
+recording against a Python one with zero changes to production code.
+
+The bridge is a shim: :func:`clj_recording_to_py_store` re-wraps each raw clj
+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.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# prep-main key whitelist + projection.
+# ---------------------------------------------------------------------------
+# The 23 keys Clojure's prep-main emits into math_main (conv_man.clj:43-74),
+# taken VERBATIM from the committed vw cold-start blob
+# (delphi/real_data/*-vw/*math_blob_cold_start.json — all kebab-case). Python's
+# to_dict() spells several of these snake_case (comment_priorities,
+# group_clusters, …) and adds Python-only keys (comment_count, participant_info,
+# vote_stats, proj, moderation, …). Without canonicalising, the comparer's
+# normalize_key (bare str(), comparer.py:611) treats kebab vs snake as DIFFERENT
+# keys, so the values (e.g. comment-priorities) are flagged once as a key-name
+# mismatch and then silently DROPPED from the numeric diff — exactly the routing
+# surface we care about. Projecting BOTH blobs onto this whitelist with canonical
+# kebab spelling before diffing puts them on the same numeric compare path.
+PREP_MAIN_KEYS = frozenset({
+    "base-clusters", "comment-priorities", "consensus", "group-aware-consensus",
+    "group-clusters", "group-votes", "in-conv", "lastModTimestamp",
+    "lastVoteTimestamp", "meta-tids", "mod-in", "mod-out", "n", "n-cmts", "pca",
+    "repness", "subgroup-clusters", "subgroup-repness", "subgroup-votes", "tids",
+    "user-vote-counts", "votes-base", "zid",
+})
+
+
+def _kebab(key: Any) -> Any:
+    """snake_case -> kebab-case for string keys (comment_priorities ->
+    comment-priorities); non-string keys pass through unchanged."""
+    return key.replace("_", "-") if isinstance(key, str) else key
+
+
+def project_prep_main(blob: dict[str, Any]) -> dict[str, Any]:
+    """Project a math_main blob onto the prep-main 23-key whitelist, keys spelled
+    in canonical kebab-case, pulling each key's value from whichever spelling the
+    blob carries. An exact kebab key always wins over a snake alias (Python's
+    to_dict() carries BOTH ``group-clusters`` and ``group_clusters``). Keys absent
+    from the whitelist are dropped on both sides."""
+    if not isinstance(blob, dict):
+        raise TypeError(
+            f"prep-main projection expects a dict blob, got {type(blob).__name__}"
+        )
+    canon: dict[Any, Any] = {}
+    for k, v in blob.items():
+        ck = _kebab(k)
+        # First-writer-wins, but let an exact kebab spelling override a prior
+        # snake alias so the canonical prep-main value is the one compared.
+        if ck not in canon or k == ck:
+            canon[ck] = v
+    return {k: canon[k] for k in canon if k in PREP_MAIN_KEYS}
+
+
+def clj_blob_files(clj_dir: str | Path) -> list[Path]:
+    """The ``step-NNN.blob.json`` files in a clj recording dir, in step order.
+
+    Only the flat top-level blobs are returned (repeat 0 / the canonical
+    cross-language surface); ``rep-*/`` subdirs and ``.edn`` files are ignored.
+    """
+    return sorted(Path(clj_dir).glob("step-*.blob.json"))
+
+
+def load_clj_blobs(clj_dir: str | Path) -> list[dict[str, Any]]:
+    """Load the raw ``prep-main`` blobs from a clj recording dir, in step order."""
+    return [json.loads(p.read_text()) for p in clj_blob_files(clj_dir)]
+
+
+def clj_recording_to_py_store(clj_dir: str | Path, dest_dir: str | Path) -> Path:
+    """Re-wrap a clj recording's blobs into the py-store layout under ``dest_dir``.
+
+    Writes ``dest_dir/py/step-NNN.json`` with the ``{index, …, blob}`` payload
+    :func:`polismath.replay.store.load_step_blobs` expects, so the shimmed dir is
+    a drop-in first argument to ``compare_recordings(..., engine="py")``.
+    Returns ``dest_dir``.
+    """
+    dest_dir = Path(dest_dir)
+    py_dir = dest_dir / "py"
+    py_dir.mkdir(parents=True, exist_ok=True)
+    # Clear stale step payloads from a previous shim run: if the new recording
+    # has fewer steps, leftovers would read back as phantom extra steps.
+    for stale in py_dir.glob("step-*.json"):
+        stale.unlink()
+    for i, p in enumerate(clj_blob_files(clj_dir)):
+        blob = json.loads(p.read_text())
+        payload = {
+            "index": i,
+            "prev_slot": None,
+            "cut_slot": None,
+            "batch_size": None,
+            "cut_time_ms": blob.get("lastVoteTimestamp"),
+            "blob": blob,
+            "extras": {},
+        }
+        (py_dir / f"step-{i:03d}.json").write_text(json.dumps(payload, indent=2))
+    return dest_dir
+
+
+def _prep_main_projecting_comparer():
+    """A StepComparer that projects BOTH step blobs onto the prep-main whitelist
+    (canonical kebab spelling) before diffing, so Python's snake_case keys align
+    with Clojure's kebab and their VALUES land on the numeric compare path. The
+    tolerant-stat keys are kebab-canonicalised too so the float stats that
+    survive projection (repness, comment-priorities, group-aware-consensus) keep
+    their tolerant classification."""
+    from polismath.replay.stepcompare import DEFAULT_TOLERANT_STAT_KEYS, StepComparer
+
+    tolerant = frozenset(_kebab(k) for k in DEFAULT_TOLERANT_STAT_KEYS)
+
+    class PrepMainProjectingComparer(StepComparer):
+        def compare_step(self, blob_a, blob_b, index):
+            return super().compare_step(
+                project_prep_main(blob_a), project_prep_main(blob_b), index
+            )
+
+    return PrepMainProjectingComparer(tolerant_stat_keys=tolerant)
+
+
+def compare_clj_vs_py(
+    recording_dir: str | Path,
+    *,
+    shim_root: str | Path,
+    comparer=None,
+) -> dict[str, Any]:
+    """Compare the ``clj/`` and ``py/`` recordings inside one recording dir.
+
+    Shims ``recording_dir/clj`` into ``shim_root`` (py-store shape), then reuses
+    :func:`polismath.replay.stepcompare.compare_recordings` verbatim. The Clojure
+    blob is the ``a`` (golden) side, Python the ``b`` (current) side.
+
+    By default both blobs are projected onto the prep-main 23-key whitelist
+    (kebab-canonicalised) so snake vs kebab key spellings compare their VALUES
+    rather than being dropped as key-name mismatches (see :func:`project_prep_main`).
+    Pass an explicit ``comparer`` to bypass the projection.
+    """
+    from polismath.replay import stepcompare as sc
+
+    recording_dir = Path(recording_dir)
+    shim = clj_recording_to_py_store(recording_dir / "clj", shim_root)
+    cmp = comparer if comparer is not None else _prep_main_projecting_comparer()
+    return sc.compare_recordings(shim, recording_dir, engine="py", comparer=cmp)

delphi/tests/replay_harness/test_clj_crosslang.py

@@ -0,0 +1,214 @@
+"""Cross-language store-reading tests — Phase H-B (Clojure Mode A).
+
+Verifies the shim that lets the EXISTING
+:func:`polismath.replay.stepcompare.compare_recordings` diff a Clojure
+recording (``clj/step-NNN.blob.json`` — the raw ``prep-main`` blob) against a
+Python one, without touching production code. Uses synthetic blobs only — no
+Clojure toolchain, so it stays in the fast delphi suite.
+
+The real cross-language gap measurement (Clojure driver vs Python driver on the
+vw dataset) is a manual smoke (see math/dev/replay_smoke.sh); this test locks
+the store-bridging logic those smokes rely on.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from replay_harness.clj_crosslang import (
+    PREP_MAIN_KEYS,
+    clj_blob_files,
+    clj_recording_to_py_store,
+    compare_clj_vs_py,
+    load_clj_blobs,
+    project_prep_main,
+)
+from polismath.conversation.conversation import Conversation
+from polismath.replay import stepcompare as sc
+from polismath.replay.store import load_step_blobs
+
+
+def _blob(n=3, first_comp=1.0):
+    """A minimal math_main-shaped blob (prep-main key spelling)."""
+    return {
+        "zid": "t",
+        "math_tick": 30000,  # ignored by the comparer
+        "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_recording(root: Path, blobs) -> Path:
+    """Write raw prep-main blobs as ``root/clj/step-NNN.blob.json``."""
+    clj = root / "clj"
+    clj.mkdir(parents=True, exist_ok=True)
+    for i, b in enumerate(blobs):
+        (clj / f"step-{i:03d}.blob.json").write_text(json.dumps(b))
+    return clj
+
+
+def _write_py_recording(root: Path, blobs) -> None:
+    """Write blobs in the py-store payload shape under ``root/py``."""
+    py = root / "py"
+    py.mkdir(parents=True, exist_ok=True)
+    for i, b in enumerate(blobs):
+        payload = {"index": i, "cut_slot": None, "blob": b, "extras": {}}
+        (py / f"step-{i:03d}.json").write_text(json.dumps(payload))
+
+
+def test_clj_blob_files_ordered_and_scoped(tmp_path):
+    """Only flat step-*.blob.json in order; rep-*/ and .edn are ignored."""
+    clj = _write_clj_recording(tmp_path, [_blob(), _blob(), _blob()])
+    # decoys that must NOT be picked up:
+    (clj / "step-000.edn").write_text("{}")
+    (clj / "rep-1").mkdir()
+    (clj / "rep-1" / "step-000.blob.json").write_text(json.dumps(_blob()))
+
+    files = clj_blob_files(clj)
+    assert [p.name for p in files] == [
+        "step-000.blob.json", "step-001.blob.json", "step-002.blob.json"
+    ]
+    assert len(load_clj_blobs(clj)) == 3
+
+
+def test_shim_produces_loadable_py_store(tmp_path):
+    """The shimmed dir is a drop-in for the py store's load_step_blobs."""
+    blobs = [_blob(n=3), _blob(n=4)]
+    _write_clj_recording(tmp_path, blobs)
+    shim = clj_recording_to_py_store(tmp_path / "clj", tmp_path / "shim")
+
+    loaded = load_step_blobs(shim / "py")
+    assert len(loaded) == 2
+    assert loaded[0]["n"] == 3 and loaded[1]["n"] == 4
+    # cut_time_ms falls back to lastVoteTimestamp (absent here → None), tolerated.
+    assert (shim / "py" / "step-000.json").exists()
+
+
+def test_identical_clj_and_py_recordings_match(tmp_path):
+    """clj vs py with identical content → overall match, zero divergence."""
+    blobs = [_blob(n=3), _blob(n=5)]
+    _write_clj_recording(tmp_path, blobs)
+    _write_py_recording(tmp_path, blobs)
+
+    report = compare_clj_vs_py(tmp_path, shim_root=tmp_path / "shim")
+    assert report["overall_match"] is True
+    assert report["summary"]["diverging_steps"] == 0
+
+
+def test_exact_field_divergence_is_reported(tmp_path):
+    """A count difference surfaces as an EXACT-family divergence."""
+    clj_blobs = [_blob(n=3)]
+    py_blobs = [_blob(n=99)]  # participant count differs
+    _write_clj_recording(tmp_path, clj_blobs)
+    _write_py_recording(tmp_path, py_blobs)
+
+    report = compare_clj_vs_py(tmp_path, shim_root=tmp_path / "shim")
+    assert report["overall_match"] is False
+    step0 = report["per_step"][0]
+    paths = [d["path"] for d in step0["families"]["exact"]]
+    assert any(p.endswith(".n") for p in paths), paths
+
+
+def test_pca_sign_flip_absorbed_as_no_divergence(tmp_path):
+    """A pure PCA comps sign flip is absorbed (ignore_pca_sign_flip=True)."""
+    clj_blobs = [_blob(first_comp=1.0)]
+    py_blobs = [_blob(first_comp=-1.0)]  # sign-flipped first component
+    _write_clj_recording(tmp_path, clj_blobs)
+    _write_py_recording(tmp_path, py_blobs)
+
+    report = compare_clj_vs_py(tmp_path, shim_root=tmp_path / "shim")
+    # The comps sign flip itself must not count as a divergence.
+    step0 = report["per_step"][0]
+    comp_divs = [
+        d for d in step0["families"]["tolerant"] + step0["families"]["exact"]
+        if ".pca.comps" in (d["path"] or "")
+    ]
+    assert comp_divs == [], comp_divs
+
+
+# --- T6: prep-main whitelist projection (kebab vs snake key alignment) --------
+def _real_py_blob():
+    """A REAL Conversation.to_dict() (2 opposing camps -> 2 groups + repness +
+    a populated comment_priorities dict), spelled SNAKE_case as Python emits it."""
+    votes = []
+    n_per, n_cmts = 8, 8
+    for p in range(n_per * 2):
+        camp = 0 if p < n_per else 1
+        for t in range(n_cmts):
+            v = (1.0 if t % 2 == 0 else -1.0) if camp == 0 else (-1.0 if t % 2 == 0 else 1.0)
+            votes.append({"pid": f"p{p}", "tid": f"c{t}", "vote": v})
+    return Conversation("t6").update_votes({"votes": votes}).to_dict()
+
+
+def test_projection_canonicalises_and_whitelists():
+    blob = _real_py_blob()
+    assert "comment_priorities" in blob and "comment-priorities" not in blob
+    proj = project_prep_main(blob)
+    # snake -> kebab, restricted to the whitelist; Python-only keys dropped.
+    assert "comment-priorities" in proj and "comment_priorities" not in proj
+    assert set(proj) <= PREP_MAIN_KEYS
+    assert "comment_count" not in proj and "participant_info" not in proj
+    # value preserved through the projection.
+    assert proj["comment-priorities"] == blob["comment_priorities"]
+
+
+def test_comment_priorities_lands_on_numeric_compare_path(tmp_path):
+    """Feed a REAL snake_case to_dict() on the PY side and a kebab prep-main blob
+    on the CLJ side whose comment-priorities differs numerically. With the
+    whitelist projection the values are COMPARED (Numeric mismatch surfaces);
+    without it they are silently dropped as a key-name mismatch (fails today)."""
+    py_blob = _real_py_blob()
+    tid = next(iter(py_blob["comment_priorities"]))
+
+    # Clojure (golden) side: the projected (kebab) blob with one priority bumped
+    # far beyond tolerance, so a faithful numeric compare MUST flag it.
+    clj_blob = project_prep_main(py_blob)
+    clj_blob["comment-priorities"] = dict(clj_blob["comment-priorities"])
+    clj_blob["comment-priorities"][tid] = clj_blob["comment-priorities"][tid] + 1000.0
+
+    _write_clj_recording(tmp_path, [clj_blob])
+    _write_py_recording(tmp_path, [py_blob])
+
+    report = compare_clj_vs_py(tmp_path, shim_root=tmp_path / "shim")
+    step0 = report["per_step"][0]
+    all_divs = step0["families"]["exact"] + step0["families"]["tolerant"]
+    cp_numeric = [
+        d for d in all_divs
+        if "comment-priorities" in (d["path"] or "")
+        and (d["reason"] or "").startswith("Numeric mismatch")
+    ]
+    assert cp_numeric, (
+        "comment-priorities value must land on the numeric compare path; "
+        f"divergences seen: {[(d['path'], d['reason']) for d in all_divs]}"
+    )
+
+
+def test_shim_clears_stale_step_files_on_rerun(tmp_path):
+    """Re-running the shim into the same dest_dir with FEWER steps must not
+    leave stale step files behind (they would show up as phantom extra steps
+    or divergences in load_step_blobs / compare_recordings)."""
+    _write_clj_recording(tmp_path, [_blob(n=3), _blob(n=4), _blob(n=5)])
+    shim = clj_recording_to_py_store(tmp_path / "clj", tmp_path / "shim")
+    assert len(load_step_blobs(shim / "py")) == 3
+
+    # Shrink the clj recording to 1 step and re-shim into the SAME dest.
+    for stale in sorted((tmp_path / "clj").glob("step-*.blob.json"))[1:]:
+        stale.unlink()
+    shim = clj_recording_to_py_store(tmp_path / "clj", tmp_path / "shim")
+    loaded = load_step_blobs(shim / "py")
+    assert len(loaded) == 1
+    assert loaded[0]["n"] == 3
+
+
+def test_projection_rejects_non_dict_blob():
+    """A non-dict blob is a malformed recording — fail fast rather than leak a
+    non-dict through the dict-typed comparer path."""
+    with pytest.raises(TypeError, match="prep-main"):
+        project_prep_main(["not", "a", "dict"])  # type: ignore[arg-type]

math/deps.edn

@@ -70,7 +70,13 @@
             :exec-fn user/run-with-repl}
            :test
            {:extra-paths ["test"]
-            :main-opts ["-m" "test-runner"]}}
+            :main-opts ["-m" "test-runner"]}
+           ;; Replay harness Phase H-B — Clojure Mode A driver (dev/replay.clj).
+           ;; Loaded via -i (file path) rather than :extra-paths ["dev"] so the
+           ;; dev/user.clj namespace (which requires oz/cider) is NOT auto-loaded
+           ;; at startup — keeps the alias self-contained on the base :deps.
+           :replay
+           {:main-opts ["-i" "dev/replay.clj" "-m" "replay"]}}
  ;:jvm-opts ^:replace []
  :jvm-opts ["-Xmx4g"]}
  ;:mvn/repos {"twitter4j" {:url "https://twitter4j.org/maven2"}}}