diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index 865441a18..56d414996 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/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 diff --git a/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py b/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py index 804bf3723..b58831427 100644 --- a/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py +++ b/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], diff --git a/delphi/polismath/pca_kmeans_rep/pca.py b/delphi/polismath/pca_kmeans_rep/pca.py index def16079d..ba84aab26 100644 --- a/delphi/polismath/pca_kmeans_rep/pca.py +++ b/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) diff --git a/delphi/polismath/replay/certify.py b/delphi/polismath/replay/certify.py index 91b85299d..0f6582a23 100644 --- a/delphi/polismath/replay/certify.py +++ b/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 ``///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 diff --git a/delphi/polismath/replay/prodclone.py b/delphi/polismath/replay/prodclone.py index 5a7ab29ea..94606fa62 100644 --- a/delphi/polismath/replay/prodclone.py +++ b/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 ``/.local/`` — :func:`assert_under_local` +- Output goes ONLY under ``/.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--``); 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" diff --git a/delphi/polismath/replay/real_data.py b/delphi/polismath/replay/real_data.py index 551c6dff9..cece77c72 100644 --- a/delphi/polismath/replay/real_data.py +++ b/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/*-``) first, then private (``real_data/.local/*-``, 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}")) diff --git a/delphi/polismath/replay/shard_bench.py b/delphi/polismath/replay/shard_bench.py index 3f86e3eb9..ca2fe89f4 100644 --- a/delphi/polismath/replay/shard_bench.py +++ b/delphi/polismath/replay/shard_bench.py @@ -371,6 +371,18 @@ 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): @@ -378,17 +390,15 @@ def run_arm( (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") diff --git a/delphi/polismath/utils/clj_hash.py b/delphi/polismath/utils/clj_hash.py index 83736378c..d02f9f729 100644 --- a/delphi/polismath/utils/clj_hash.py +++ b/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 — diff --git a/delphi/polismath/utils/engine_mode.py b/delphi/polismath/utils/engine_mode.py index e68a23f8f..788600970 100644 --- a/delphi/polismath/utils/engine_mode.py +++ b/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. diff --git a/delphi/polismath/utils/env_flags.py b/delphi/polismath/utils/env_flags.py index 3e6cc6bbb..3e0e2c911 100644 --- a/delphi/polismath/utils/env_flags.py +++ b/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. """ diff --git a/delphi/scripts/clj_timing_probe.py b/delphi/scripts/clj_timing_probe.py index 877809f0d..58da468ab 100644 --- a/delphi/scripts/clj_timing_probe.py +++ b/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") diff --git a/delphi/tests/replay_harness/test_certify.py b/delphi/tests/replay_harness/test_certify.py index aaefb25b6..fa044dd0a 100644 --- a/delphi/tests/replay_harness/test_certify.py +++ b/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) # --------------------------------------------------------------------------- diff --git a/delphi/tests/replay_harness/test_certify_canonicalization.py b/delphi/tests/replay_harness/test_certify_canonicalization.py index 29276e63f..ad6997451 100644 --- a/delphi/tests/replay_harness/test_certify_canonicalization.py +++ b/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 diff --git a/delphi/tests/test_base_cluster_lineage.py b/delphi/tests/test_base_cluster_lineage.py index e398f0302..6aa5dcd63 100644 --- a/delphi/tests/test_base_cluster_lineage.py +++ b/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']: diff --git a/delphi/tests/test_clj_hash_order.py b/delphi/tests/test_clj_hash_order.py index 47c47d42e..cb8012a24 100644 --- a/delphi/tests/test_clj_hash_order.py +++ b/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, diff --git a/delphi/tests/test_degenerate_tick_parity.py b/delphi/tests/test_degenerate_tick_parity.py index c2e210c75..c102d241a 100644 --- a/delphi/tests/test_degenerate_tick_parity.py +++ b/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'])