python-math #34: fix(delphi): Copilot-review triage batch — 16 findings across the stack

delphi/polismath/conversation/conversation.py

@@ -909,15 +909,23 @@ def _compute_clusters(self,
         # in-conv participant its graph still runs the full base->group chain
         # (one base cluster, k=2 group clustering of one point). Legacy mode
         # falls through and replicates that; improved mode keeps the guard.
-        # The 0-participant early return stays in BOTH modes: it is unreachable
-        # past the `not self.proj` short-circuit while the in-conv greedy floor
-        # guarantees >=1 participant (PR-E), and Clojure's kmeans on an empty
-        # matrix has nothing to warm-start either.
+        # The 0-participant early return stays in BOTH modes, but the
+        # greedy-floor unreachability argument (in-conv can't drop below 1) is
+        # LEGACY-only — the greedy floor itself only runs in 'clojure-legacy'
+        # mode (_get_in_conv_participants). In 'improved' mode there is no
+        # floor, so 0 in-conv participants is a REAL, load-bearing case this
+        # guard must handle, not just dead code.
         if len(in_conv_pids_list) == 0 or (not legacy_mode and len(in_conv_pids_list) < 2):
             logger.warning(f"Not enough participants meeting threshold ({len(in_conv_pids_list)})")
             self.base_clusters = []
             self.group_clusters = []
             self.subgroup_clusters = {}
+            if not legacy_mode:
+                # Improved mode has no warm-start use for this state, so a
+                # stale value from a prior tick must not leak forward
+                # (#2642 review finding).
+                self.group_clusterings = {}
+                self.group_k_smoother = {}
             return
 
         logger.info(f"Using {len(in_conv_pids_list)}/{len(self.proj)} participants for clustering")
@@ -1007,6 +1015,11 @@ def _compute_clusters(self,
             else:
                 self.group_clusters = []
             self.subgroup_clusters = {}
+            # Improved-only path (already gated by `not legacy_mode` above): no
+            # warm-start use for this state, so a stale value from a prior
+            # tick must not leak forward (#2642 review finding).
+            self.group_clusterings = {}
+            self.group_k_smoother = {}
             return
 
         # Prepare base cluster centers and weights

delphi/polismath/pca_kmeans_rep/legacy_kmeans.py

@@ -107,7 +107,12 @@ def _euclidean(a: np.ndarray, b: np.ndarray) -> float:
     av = np.asarray(a, dtype=float)
     bv = np.asarray(b, dtype=float)
     d2 = float(np.dot(av, av)) + float(np.dot(bv, bv)) - 2.0 * float(np.dot(av, bv))
-    return float(np.sqrt(max(0.0, d2)))
+    # NaN must propagate, not silently become 0: python's max(0.0, nan) returns
+    # 0.0 (nan compares false against 0.0, so max just returns its first arg),
+    # but real vectorz does no such clamp and would NaN instead.
+    if d2 < 0.0:
+        d2 = 0.0
+    return float(np.sqrt(d2))
 
 
 def weighted_mean(rows: Sequence[np.ndarray],

delphi/polismath/pca_kmeans_rep/pca.py

@@ -318,7 +318,7 @@ def pca_project_dataframe(df: pd.DataFrame,
     # "Determinism verification" entry (2026-07-04/05) in
     # docs/CLJ-PARITY-FIXES-JOURNAL.md.
 
-    # Solver switch (read at call time — see utils.env_flags.resolve_impl_flag):
+    # Solver switch (read at call time — see polismath.utils.env_flags.resolve_impl_flag):
     #   POLISMATH_PCA_IMPL=powerit  (default) legacy/Clojure-parity power iteration
     #   POLISMATH_PCA_IMPL=sklearn  improved exact-SVD path
     # The imputation above and sparsity scaling below are IDENTICAL for both;
@@ -465,8 +465,11 @@ def pca_project_cmnts(center: np.ndarray, comps: np.ndarray) -> np.ndarray:
         # destructure leaves pc2 nil and `utils/zip` truncates the
         # sparsity-aware reduce to EMPTY — every comment projects to 0.0 on
         # BOTH components (pca.clj:134-157; verified on a 3x1 clj replay
-        # reference, 2026-07-22 s4).
-        return np.zeros((n_cmnts, comps.shape[0]))
+        # reference, 2026-07-22 s4). Always 2-wide here — matching
+        # pca_project_dataframe's always-2-wide guarantee — not
+        # comps.shape[0]; the conversation.py:1866 defensive pad becomes a
+        # no-op given this, but is left in place.
+        return np.zeros((n_cmnts, 2))
     scale = np.sqrt(n_cmnts)
     coefs = scale * (AGREE - center)              # shape (n_cmnts,); AGREE = +1 (Delphi)
     return coefs[:, None] * comps.T               # shape (n_cmnts, n_components)

delphi/polismath/replay/certify.py

@@ -42,6 +42,7 @@
 
 from __future__ import annotations
 
+import functools
 import hashlib
 import json
 import os
@@ -184,7 +185,15 @@ def parse_battery_entry(e: dict[str, Any], *, battery_dir: Path | None = None) -
         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_json = json.loads(schedule_path.read_text())
+        schedule_dataset = schedule_json.get("dataset")
+        if schedule_dataset is not None and schedule_dataset != dataset:
+            raise ValueError(
+                f"battery entry dataset {dataset!r} does not match schedule file "
+                f"{schedule_path}'s dataset {schedule_dataset!r} — drivers and certify "
+                f"would disagree on which dataset's votes to replay/cache"
+            )
+        base_id = schedule_json["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", ""))
@@ -528,10 +537,22 @@ def _write_manifest(manifest_path: Path, manifest: dict[str, Any]) -> None:
         json.dump(manifest, fh, indent=2, sort_keys=True)
 
 
+@functools.lru_cache(maxsize=1)
 def _py_tree_hash() -> str:
     return sha256_tree(_DELPHI_ROOT / "polismath", "**/*.py")
 
 
+@functools.lru_cache(maxsize=1)
+def _clj_source_hashes() -> tuple[str, str]:
+    """(sha256 of dev/replay.clj, sha256 of the math/src tree) — cached since
+    both are read-only per process and re-hashing the whole math/src tree on
+    every battery entry is wasted work."""
+    return (
+        sha256_file(_MATH_ROOT / "dev" / "replay.clj"),
+        sha256_tree(_MATH_ROOT / "src", "**/*"),
+    )
+
+
 def ensure_py_recording(
     entry: BatteryEntry, spec: sched.ScheduleSpec, votes_sha: str, *, root: Path,
     refresh: bool = False,
@@ -566,24 +587,29 @@ def ensure_clj_recording(
     root: Path, refresh: bool = False, comments_csv: Path | None = None,
 ) -> tuple[Path, bool]:
     """Reuse ``<root>/<ds>/<sid>/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.
+    sha256, schedule hash, sha256 of dev/replay.clj, sha256 of math/src, and
+    — when ``comments_csv`` is given — its sha256 too); 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.
 
     ``comments_csv`` (when given) is forwarded to :func:`run_clj_driver` as
-    ``--comments`` — deliberately NOT part of the cache manifest, so entries
-    that never pass it (moderation="none") keep their existing cache key and
-    are never invalidated by this parameter's introduction."""
+    ``--comments`` AND its sha256 is added to the cache manifest (STRICT —
+    this deliberately invalidates existing mod-entry clj recordings once; the
+    nightly battery re-records). Entries that never pass it (moderation="none")
+    keep their existing cache key and are unaffected by this parameter."""
     rec_dir = st.recording_dir(entry.dataset, entry.schedule_id, root=root)
     clj_dir = rec_dir / "clj"
     manifest_path = clj_dir / "cache_manifest.json"
+    replay_clj_sha256, math_src_sha256 = _clj_source_hashes()
     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", "**/*"),
+        "replay_clj_sha256": replay_clj_sha256,
+        "math_src_sha256": math_src_sha256,
     }
+    if comments_csv is not None:
+        expected["comments_csv_sha256"] = sha256_file(comments_csv)
     if not refresh and _manifest_matches(manifest_path, expected):
         return clj_dir, True
 

delphi/polismath/replay/prodclone.py

@@ -12,7 +12,8 @@
 CRITICAL privacy rules (see delphi/tests/test_prodclone_extract.py and the
 project CLAUDE.md for the full policy):
 
-- Output goes ONLY under ``<real_data_root>/.local/`` — :func:`assert_under_local`
+- Output goes ONLY under ``<out_root>/.local/`` (the caller-supplied root —
+  ``REAL_DATA_ROOT`` by default, overridable for tests) — :func:`assert_under_local`
   is the hard guard; every write path routes through it.
 - Minted slugs are neutral (``pc-<feature>-<NN>``); the on-disk directory
   prefix is a salted hash of the zid (:func:`fake_report_prefix`), never the
@@ -524,6 +525,8 @@ def run_extract(
     """
     if feature not in FEATURES:
         raise ValueError(f"unknown feature {feature!r}; must be one of {FEATURES}")
+    if out_root.exists() and not out_root.is_dir():
+        raise NotADirectoryError(f"out_root must be a directory: {out_root}")
     if map_path is None:
         map_path = out_root / ".local" / "prodclone_map.json"
 

delphi/polismath/replay/real_data.py

@@ -15,12 +15,20 @@
 """
 
 import csv
+import re
 from pathlib import Path
 
 from polismath.replay.types import ModEvent, ReplayDataset
 
 REAL_DATA_ROOT = Path(__file__).resolve().parents[2] / "real_data"
 
+# Slug allow-list — same precedent as prodclone.py's minted-slug regex
+# (``_SLUG_RE_TEMPLATE``): a slug flows unsanitized into a ``Path.glob()``
+# pattern below, so without this guard a slug containing glob metacharacters
+# (``*``, ``?``, ``[...]``) or path separators (``../``) could escape the
+# intended directory or match unintended files.
+_SLUG_RE = re.compile(r"^[A-Za-z0-9_-]+$")
+
 # Comments-CSV columns a moderation-history-carrying export must have before
 # we attempt to weave mod events out of it — MOD_RESTART_PORT_SPEC.md "Python
 # ports" item 3. Older comments CSVs (pre-dating this port) lack "modified"
@@ -78,6 +86,8 @@ def dataset_dir(slug: str) -> Path | None:
     """Locate a dataset directory by slug — public (``real_data/*-<slug>``)
     first, then private (``real_data/.local/*-<slug>``, gitignored). A public
     match wins a slug collision."""
+    if not _SLUG_RE.match(slug):
+        return None
     hits = sorted(REAL_DATA_ROOT.glob(f"*-{slug}"))
     if not hits:
         hits = sorted(REAL_DATA_ROOT.glob(f".local/*-{slug}"))

delphi/polismath/replay/shard_bench.py

@@ -371,24 +371,34 @@ def run_arm(
         )
         procs.append((idx, proc, ready, out_path, err_path, out_fh, err_fh))
 
+    def _kill_and_close_all() -> None:
+        """Kill every still-alive shard, reap it, and close BOTH of its output
+        handles -- for every proc, not just the one that triggered the abort.
+        Used on both barrier-wait failure paths below so a dead/hung shard
+        never leaves its siblings running unreaped or their fhs leaked."""
+        for _, p, _, _, _, out_fh, err_fh in procs:
+            if p.poll() is None:
+                p.kill()
+                p.wait(timeout=CHILD_TIMEOUT_SEC)
+            out_fh.close()
+            err_fh.close()
+
     # Wait for every child to finish its setup, then release them together.
     deadline = time.monotonic() + CHILD_TIMEOUT_SEC
     while not all(r.exists() for _, _, r, _, _, _, _ in procs):
         dead = [
             (i, p, ep) for i, p, _, _, ep, _, _ in procs if p.poll() is not None
         ]
         if dead:
-            for _, fh in [(p, fh) for _, p, _, _, _, fh, _ in procs]:
-                fh.close()
             i, p, ep = dead[0]
             err = ep.read_text(encoding="utf-8", errors="replace") if ep.exists() else ""
+            _kill_and_close_all()
             raise RuntimeError(
                 f"shard {i} died before the barrier (rc={p.returncode}):\n"
                 f"{err.strip()[-2000:]}"
             )
         if time.monotonic() > deadline:
-            for _, p, _, _, _, _, _ in procs:
-                p.kill()
+            _kill_and_close_all()
             raise RuntimeError("timed out waiting for shards to become ready")
         time.sleep(0.01)
     go.write_text("go", encoding="utf-8")

delphi/polismath/utils/clj_hash.py

@@ -16,9 +16,11 @@
 map in iteration order): n=18, n=30 and n=98 integer-pid maps, all exact.
 
 Caveats, deliberate and documented:
-- Integer keys only. Other key types hash differently (e.g. String hasheq is
-  Murmur3 over ``String.hashCode``); :func:`clojure_hash_map_key_order` falls
-  back to the given order for them rather than guessing.
+- Integer keys, or numeric-string keys that normalize to one (`_as_long`) —
+  both hash as the equivalent Clojure Long. Other key types (e.g. plain
+  strings, whose Clojure hasheq is Murmur3 over ``String.hashCode``, not
+  ``hashLong``) hash differently; :func:`clojure_hash_map_key_order` falls
+  back to the given order for them.
 - Full-hash collisions land in a HashCollisionNode (insertion order). For
   distinct realistic pid ranges Murmur3-32 collisions are vanishingly rare;
   the sort is stable, so colliding keys keep their given relative order —

delphi/polismath/utils/engine_mode.py

@@ -17,7 +17,7 @@
   - 'clojure-legacy'         : threads the warm-start state described above.
 
 The flag is resolved AT CALL TIME (never cached at import) by the shared
-`utils.env_flags.resolve_impl_flag`: unknown values fall back to the default
+`polismath.utils.env_flags.resolve_impl_flag`: unknown values fall back to the default
 with a warning so a typo in a deployment env cannot crash the math worker.
 This lives in a shared spot (polismath.utils) because the mode cross-cuts both
 PCA (conversation._compute_pca) and clustering (conversation._compute_clusters).
@@ -38,7 +38,7 @@ def resolve_engine_mode() -> str:
     """
     Resolve `POLISMATH_ENGINE_MODE` from the environment, at call time.
 
-    Reuses `utils.env_flags.resolve_impl_flag` so the resolution rules
+    Reuses `polismath.utils.env_flags.resolve_impl_flag` so the resolution rules
     (strip + lowercase, unknown -> default with a warning) are identical to
     the PCA-solver switch.
 

delphi/polismath/utils/env_flags.py

@@ -10,8 +10,8 @@
 crash the math worker).
 
 This lives in polismath.utils (not pca.py, where it originated) so that
-lightweight consumers — e.g. `utils.engine_mode`, read on every conv-update
-tick — do not drag in the numpy/pandas pca import chain, and resolution
+lightweight consumers — e.g. `polismath.utils.engine_mode`, read on every
+conv-update tick — do not drag in the numpy/pandas pca import chain, and resolution
 warnings are logged under this module's logger rather than pca's.
 """
 

delphi/scripts/clj_timing_probe.py

@@ -418,6 +418,8 @@ def probe(votes_path, sizes, out_path, budget_min, timeout_sec, dataset) -> None
     out_path = Path(out_path)
 
     n_max = count_data_rows(votes_path)
+    if n_max <= 0:
+        raise click.UsageError(f"votes file has no data rows: {votes_path}")
     size_list = parse_sizes(sizes, n_max)
     if not size_list:
         raise click.UsageError("no sizes to probe")

delphi/tests/replay_harness/test_certify.py

@@ -24,6 +24,7 @@
 from polismath.replay import certify as cert
 from polismath.replay import schedule as sched
 from polismath.replay.crosslang import PREP_MAIN_KEYS
+from polismath.utils.engine_mode import ENGINE_MODE_CHOICES
 
 CERTIFY_BATTERY_PATH = Path(__file__).resolve().parents[2] / "scripts" / "certify_battery.json"
 
@@ -151,7 +152,7 @@ def test_load_battery_starter_file_shape():
     for private_ds in ("FLI", "bg2018", "pakistan", "engage", "bg2050"):
         assert any(e.dataset == private_ds for e in entries), private_ds
     assert len(ids) == len(entries), "duplicate (dataset, schedule) entries"
-    assert all(e.engine_mode == "clojure-legacy" for e in entries)
+    assert all(e.engine_mode in ENGINE_MODE_CHOICES for e in entries)
 
 
 # ---------------------------------------------------------------------------

delphi/tests/replay_harness/test_certify_canonicalization.py

@@ -3,17 +3,22 @@
 The first full battery run (journal 2026-07-22) showed 4/4 entries diverging at
 step 0 with ~800+ "exact" divergences — nearly all of them ORDERING artifacts:
 Clojure emits ``tids``/``in-conv`` (and everything positionally aligned to
-them: pca.center, pca.comps rows, base-clusters columns, votes-base lists) in
-hash/insertion order, while Python emits sorted order. Each blob is internally
-consistent, so cross-engine array order is not a semantic divergence — the
-acceptance criterion (GOAL_R1_PARITY.md) is MEMBERSHIP and value parity.
+them: pca.center, pca.comps rows, base-clusters columns) in hash/insertion
+order, while Python emits sorted order. Each blob is internally consistent, so
+cross-engine array order is not a semantic divergence — the acceptance
+criterion (GOAL_R1_PARITY.md) is MEMBERSHIP and value parity.
 
 ``project_acceptance`` therefore canonicalizes both sides before hashing and
 diffing: id-sets sorted, tid-aligned pca arrays re-indexed by sorted tid,
-base-clusters columns re-indexed by sorted id (votes-base per-cluster lists
-following the same permutation), group-clusters sorted by id with sorted
-members. Real divergences (a differing pid, a differing center value for the
-SAME tid) must still be reported — canonicalization must never mask them.
+base-clusters columns re-indexed by sorted id, group-clusters sorted by id
+with sorted members. votes-base A/D/S per-cluster lists are NOT part of this
+permutation — they are already aligned to sort-by-:id bucket order on BOTH
+engines (bid-to-pid = (mapv :members (sort-by :id base-clusters)),
+conversation.clj:593), so they are identical across the two orderings and the
+canonicalizer leaves them untouched (see lines ~30-35 below and
+crosslang.py:88-90). Real divergences (a differing pid, a differing center
+value for the SAME tid) must still be reported — canonicalization must never
+mask them.
 """
 
 from __future__ import annotations

delphi/tests/test_base_cluster_lineage.py

@@ -86,6 +86,9 @@ def test_legacy_base_is_clojure_faithful_up_to_q11_merges(self, monkeypatch):
         assert all(c['members'] for c in leg.base_clusters)  # no empty clusters
         all_members = [m for c in leg.base_clusters for m in c['members']]
         assert sorted(all_members) == sorted(f'p{i}' for i in range(18))
+        assert any(len(c['members']) > 1 for c in leg.base_clusters), (
+            "fixture must exercise at least one Q11 merge, or this test passes vacuously"
+        )
         pos = {pid: np.asarray(proj) for pid, proj in leg.proj.items()}
         for c in leg.base_clusters:
             for m1 in c['members']:

delphi/tests/test_clj_hash_order.py

@@ -9,8 +9,6 @@
 
 from __future__ import annotations
 
-import pytest
-
 from polismath.conversation.conversation import Conversation
 from polismath.utils.clj_hash import (
     clojure_hash_map_key_order,

delphi/tests/test_degenerate_tick_parity.py

@@ -188,5 +188,25 @@ def test_degenerate_tick_keeps_synthesized_cluster(self, improved_mode):
         assert conv.group_k_smoother == {}
 
 
+class TestImprovedStaleStateReset:
+    """#2642 review finding: the <2-in-conv-participants early return resets
+    base_clusters/group_clusters/subgroup_clusters but used to leave
+    group_clusterings/group_k_smoother untouched. In 'improved' mode there is
+    no warm-start use for that state (unlike 'clojure-legacy'), so a stale
+    value set before a guarded tick would otherwise leak forward into the
+    result unchanged instead of being reset to {}."""
+
+    def test_single_participant_resets_stale_group_state(self, improved_mode):
+        conv = Conversation('solo')
+        conv.group_clusterings = {2: "SENTINEL"}
+        conv.group_k_smoother = {"k": 1}
+
+        result = conv.update_votes(_single_ptpt_votes())
+
+        assert result.base_clusters == [], "sanity: must hit the early-return path"
+        assert result.group_clusterings == {}
+        assert result.group_k_smoother == {}
+
+
 if __name__ == '__main__':
     pytest.main([__file__, '-v'])