From 042deef5f66a7975d23d91c4c342134650e5efa6 Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Sat, 18 Jul 2026 02:12:29 +0100 Subject: [PATCH] =?UTF-8?q?python-math=20#29:=20feat(math):=20moderation?= =?UTF-8?q?=20+=20restart=20replay=20parity=20=E2=80=94=20mod=5Fupdate,=20?= =?UTF-8?q?restart-seam=20restore,=20Q13-Q18?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Certification-battery coverage (the battery: recorded real-conversation vote schedules replayed through both engines and compared step by step) for the goal-doc's moderation-heavy, meta-tids (comments flagged `is_meta`) and restart-seam edge cases, plus the quirks they surfaced. All TDD, ledgered in `CLOJURE_QUIRKS.md`; design in `MOD_RESTART_PORT_SPEC.md`. ## Engine ports - `Conversation.mod_update`: parity with Clojure's `mod-update` reducer (`conversation.clj:846-884`) — a comment flagged `is_meta` lands in BOTH mod sets, un-moderation removes via `disj`, application is order-sensitive, and the moderation watermark takes a max floor. - Quirk Q15: legacy `recompute()` drops the moderation watermark, because Clojure's `conv-update` plumbing graph has no `:last-mod-timestamp` node — replicated. - Quirk Q16: rank-1 principal components collapse ALL projections to `[0,0]` (Clojure's `utils/zip` truncates when pc2 is nil) — replicated at both projection helpers; found by review-driven 1xN/Nx1 tiny-shape tests with Clojure-derived reference values. - group-votes tally source: tallies read from `raw-rating-mat` (the raw, pre-moderation-filter rating matrix; `conversation.clj:601-608`) at ALL THREE tally sites — the compute path, the inline `to_dict`, and `to_dynamo_dict`. ## Replay drivers - Mod-event weaving: votes-then-moderation per batch, one result blob (the JSON math result) per step, plus a load-or-init restart seam (kill the engine mid-schedule, then resume from the persisted blob) — both in BOTH drivers. The blob round-trips via `db/load-conv`'s longs-else-keywords key-fn and `restructure-json-conv`, with full-history `raw-rating-mat` and `mod-update`; the Clojure seam was validated bit-identical pre-seam vs cached recordings. - Restart-seam restore (2026-07-24): `Conversation.from_dict` now restores `zid` (blobs emit `'zid'` in both engine modes), base-clusters (unfold + the legacy un-negate — the warm-start lineage input), and group-votes (tid keys re-intified — the recovery tick's previous-tick priorities input, per quirk Q2: comment priorities always read the PREVIOUS tick's stored group-votes) — mirroring Clojure's `restructure-json-conv` (`conv_man.clj:171-186`). Both restart battery entries MATCH at every step. - Restart-driver review fixes: at the seam, replay woven mods only (Clojure `restart-conv`'s `(mapcat :mods steps-so-far)`, never `dataset.mod_events`); `restart_after` range validation; modified-only comments-CSV header (`is-meta` optional); `to_dynamo_dict` tally gate. ## Quirks resolved this round - Q17 PORTED: legacy `cluster_step` resolves k-means distance ties by Clojure's cleared-clusters map iteration order — insertion order for <=8 clusters, PersistentHashMap trie order (via `clj_hash`) for >8. Quirk Q11's cancellation floor (the Clojure distance formula floors near-coincident distances to exactly 0.0) makes exact 0.0 ties common, not measure-zero. - Q18 CARVED out of certification: `uniqify` merge-center exactness is value-dependent ulp luck, amplified by exact-equality comparison over ~1e-16 cross-engine PCA noise (real-vectorz evidence recorded in the quirk row). Consequently `pc-modheavy-01` + `pc-meta-01` are re-scheduled to single-cut-mod (cold tick, full moderation weave — deterministic), and a NEW entry `pc-meta-02` (moderate coincidence density) carries the warm-chain mod/meta coverage. ## Tooling + results - prodclone extractor (pulls datasets from the production-database clone): additive `is-meta`/`modified` comment columns; certify passes `--comments` for moderation schedules. - Battery: 20 entries — 20/20 MATCH on TWO consecutive full passes (2026-07-24); `divergences.json` at 0 open (70 resolved + 11 carved-out). Co-Authored-By: Claude Fable 5 commit-id:9467ac51 --- delphi/polismath/conversation/conversation.py | 187 +++++++++++-- .../polismath/pca_kmeans_rep/legacy_kmeans.py | 28 +- delphi/polismath/pca_kmeans_rep/pca.py | 23 +- delphi/polismath/replay/certify.py | 46 +++- delphi/polismath/replay/driver.py | 141 ++++++++-- delphi/polismath/replay/prodclone.py | 23 +- delphi/polismath/replay/real_data.py | 71 ++++- delphi/polismath/replay/schedule.py | 17 +- delphi/polismath/replay/types.py | 14 +- delphi/scripts/certify_battery.json | 236 +++++++++------- .../schedules/pc-meta-01-single-cut-mod.json | 16 ++ .../schedules/pc-meta-01-uniform6-mod.json | 21 ++ .../schedules/pc-meta-02-uniform6-mod.json | 21 ++ .../pc-midmix-01-uniform6-restart3.json | 22 ++ .../pc-modheavy-01-single-cut-mod.json | 16 ++ .../pc-modheavy-01-uniform6-mod.json | 21 ++ .../schedules/vw-uniform8-restart4.json | 24 ++ delphi/tests/poller/test_load_or_init.py | 27 +- delphi/tests/replay_harness/test_certify.py | 90 +++++++ delphi/tests/replay_harness/test_driver.py | 251 ++++++++++++++++++ .../test_real_data_mod_events.py | 174 ++++++++++++ delphi/tests/replay_harness/test_schedule.py | 95 +++++++ delphi/tests/test_legacy_blob_shape.py | 137 ++++++++++ delphi/tests/test_legacy_kmeans.py | 51 ++++ delphi/tests/test_mod_update_parity.py | 229 ++++++++++++++++ delphi/tests/test_prodclone_extract.py | 45 +++- math/dev/proj_probe.clj | 191 +++++++++++++ math/dev/replay.clj | 225 +++++++++++++--- 28 files changed, 2224 insertions(+), 218 deletions(-) create mode 100644 delphi/scripts/schedules/pc-meta-01-single-cut-mod.json create mode 100644 delphi/scripts/schedules/pc-meta-01-uniform6-mod.json create mode 100644 delphi/scripts/schedules/pc-meta-02-uniform6-mod.json create mode 100644 delphi/scripts/schedules/pc-midmix-01-uniform6-restart3.json create mode 100644 delphi/scripts/schedules/pc-modheavy-01-single-cut-mod.json create mode 100644 delphi/scripts/schedules/pc-modheavy-01-uniform6-mod.json create mode 100644 delphi/scripts/schedules/vw-uniform8-restart4.json create mode 100644 delphi/tests/replay_harness/test_real_data_mod_events.py create mode 100644 delphi/tests/test_mod_update_parity.py diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index 4d51e8d90c..de58092d49 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/delphi/polismath/conversation/conversation.py @@ -207,7 +207,7 @@ def __init__(self, # clojure-legacy emission can distinguish "never moderated" (null) # from "moderated to empty" ([]). See FP-2f5714ce9c / FP-2975bbfb04. self.moderation_applied = False - self.last_mod_timestamp = None + self.last_mod_timestamp: Optional[int] = None # Clojure named-matrix column order = first-vote arrival order per tid # (update-nmat appends unseen colnames in encounter order); python's # internal matrix is natsorted instead (update_votes). Tracked so @@ -661,9 +661,59 @@ def update_moderation(self, # Recompute clustering if requested if recompute: result = result.recompute() - + return result - + + def mod_update(self, mods: List[Dict[str, Any]]) -> 'Conversation': + """Clojure ``mod-update`` parity (conversation.clj:846-884). + + Reduces raw moderation rows ``{tid, is_meta, mod, modified}`` over the + current sets, in row order: mod-out conj when ``is_meta OR mod == -1`` + else disj; mod-in conj when ``is_meta OR mod == 1`` else disj; + meta-tids conj when ``is_meta`` else disj. Consequences pinned by + tests/test_mod_update_parity.py: is_meta rows land in BOTH mod sets, + un-moderation REMOVES (which ``update_moderation`` cannot express), + and the last row per tid wins. Watermark: + ``last_mod_timestamp = max(existing or 0, *modified)``. + + NO math recompute — Clojure's ``:moderation`` message handler runs + ``mod-update`` alone and re-emits the blob with updated sets and + unchanged math (conv_man.clj:274-276 + 328-345); the sets take effect + at the next votes recompute (``_apply_moderation`` runs inside + ``update_votes``). ``moderation_applied`` becomes True even for empty + ``mods``: any mod-update leaves Clojure's sets as real (possibly + empty) sets, which the blob emits as ``[]`` rather than ``null``. + """ + result = deepcopy(self) + mod_out = set(result.mod_out_tids) + mod_in = set(result.mod_in_tids) + meta = set(result.meta_tids) + for row in mods: + tid = row['tid'] + is_meta = bool(row.get('is_meta')) + mod = row.get('mod') + if is_meta or mod == -1: + mod_out.add(tid) + else: + mod_out.discard(tid) + if is_meta or mod == 1: + mod_in.add(tid) + else: + mod_in.discard(tid) + if is_meta: + meta.add(tid) + else: + meta.discard(tid) + result.mod_out_tids = mod_out + result.mod_in_tids = mod_in + result.meta_tids = meta + result.moderation_applied = True + result.last_mod_timestamp = max( + [result.last_mod_timestamp or 0] + + [row['modified'] for row in mods] + ) + return result + def _compute_pca(self, n_components: int = 2, prev_pca: Optional[Dict[str, Any]] = None) -> None: """ @@ -1401,6 +1451,15 @@ def recompute(self) -> 'Conversation': # (conversation.clj:658). Captured here, consumed in legacy mode only. prev_group_votes = getattr(result, 'group_votes', {}) + # Q15: Clojure's conv-update is a plumbing-graph compile whose output + # has ONLY graph-node keys — :last-mod-timestamp is not one + # (conversation.clj:780-820), so every votes recompute DROPS the mod + # watermark; blobs carry lastModTimestamp only when the tick's last + # write was a mod-update. Improved mode keeps the persistent watermark + # (documented divergence). tests/test_mod_update_parity.py. + if resolve_engine_mode() == ENGINE_MODE_LEGACY: + result.last_mod_timestamp = None + # Compute PCA and projections result._compute_pca(prev_pca=prev_pca) @@ -1867,6 +1926,18 @@ def _compute_group_votes(self) -> Dict[str, Any]: # Expand base-cluster IDs to participant IDs (matches Clojure group-votes) unfolded = self._unfolded_group_clusters() + # Clojure's group-votes aggregates votes-base, whose fnk reads + # RAW-rating-mat (conversation.clj:601-608): moderated-out comments + # report the ACTUAL votes cast and true seen-counts, not the + # post-zeroing pass-shaped columns (a zeroed column would tally + # A=0/D=0 with S = every member). Legacy mode mirrors that; improved + # mode keeps the zeroed-matrix tally it was snapshotted with (its + # S-inflation on moderated tids is a known later-fix). + # tests/test_mod_update_parity.py TestGroupVotesTallyRawMatrix. + tally_mat = (self.raw_rating_mat + if resolve_engine_mode() == ENGINE_MODE_LEGACY + else self.rating_mat) + group_votes = {} # Helper to count votes of a specific type for a group @@ -1886,7 +1957,7 @@ def count_votes_for_group(group_id, comment_id, vote_type): row_indices = [] for member in members: try: - member_idx = self.rating_mat.index.get_loc(member) + member_idx = tally_mat.index.get_loc(member) row_indices.append(member_idx) except ValueError: # Skip members not found in matrix @@ -1894,13 +1965,13 @@ def count_votes_for_group(group_id, comment_id, vote_type): # Get the column index for this comment try: - col_idx = self.rating_mat.columns.get_loc(comment_id) + col_idx = tally_mat.columns.get_loc(comment_id) except ValueError: # If comment not found, return 0 return 0 # Count votes of specified type - votes = self.rating_mat.values[row_indices, col_idx] + votes = tally_mat.values[row_indices, col_idx] if vote_type == 'A': # Agree return int(np.sum(np.abs(votes - 1.0) < 0.001)) @@ -2347,8 +2418,16 @@ def numpy_to_list(arr): # Reuse the already-unfolded group clusters (computed above) unfolded_groups = unfolded_gc + # Same tally-source rule as _compute_group_votes: Clojure's + # group-votes aggregates votes-base, which reads RAW-rating-mat + # (conversation.clj:601-608) — moderated-out comments report the + # actual votes cast, not the zeroed pass-shaped columns. + tally_mat = (self.raw_rating_mat + if resolve_engine_mode() == ENGINE_MODE_LEGACY + else self.rating_mat) + # Precompute indices for each participant for faster lookups - ptpt_indices = {ptpt_id: i for i, ptpt_id in enumerate(self.rating_mat.index)} + ptpt_indices = {ptpt_id: i for i, ptpt_id in enumerate(tally_mat.index)} # Process each group for group in unfolded_groups: @@ -2360,19 +2439,19 @@ def numpy_to_list(arr): member_indices = [] for member in group.get('members', []): idx = ptpt_indices.get(member) - if idx is not None and idx < self.rating_mat.values.shape[0]: + if idx is not None and idx < tally_mat.values.shape[0]: member_indices.append(idx) - + # Skip groups with no valid members if not member_indices: continue - + # Get the vote submatrix for this group - group_matrix = self.rating_mat.values[member_indices, :] - + group_matrix = tally_mat.values[member_indices, :] + # Calculate vote stats for each comment using vectorized operations votes = {} - for j, comment_id in enumerate(self.rating_mat.columns): + for j, comment_id in enumerate(tally_mat.columns): if j >= group_matrix.shape[1]: continue @@ -2746,8 +2825,14 @@ def from_dict(cls, data: Dict[str, Any]) -> 'Conversation': Returns: Conversation instance """ - # Create empty conversation - conv = cls(data.get('conversation_id', '')) + # Create empty conversation. to_dict emits the id under 'zid' (both + # modes — it renames conversation_id at emission), matching Clojure + # prep-main blobs; accept either key so a recorded blob round-trips + # with its id intact (restart-seam root, journal 2026-07-24). + # Key-presence check, not truthiness: a legitimately-falsy id (0) + # must not fall through to the other key (#2656 review). + conv = cls(data['conversation_id'] if 'conversation_id' in data + else data.get('zid', '')) # Restore basic attributes conv.last_updated = data.get('last_updated', int(time.time() * 1000)) @@ -2816,6 +2901,55 @@ def from_dict(cls, data: Dict[str, Any]) -> 'Conversation': # Restore cluster data conv.group_clusters = data.get('group_clusters', []) + + # Restore base clusters — the blob emits them in the Clojure folded + # column-store shape ({'id': [...], 'members': [...], 'x': [...], + # 'y': [...], 'count': [...]}); unfold to the internal row shape + # exactly as restructure-json-conv does (conv_man.clj:171-186 → + # clusters.clj:402-414 unfold-clusters: center := [x, y]). Without + # this, a warm restart cold-starts the base-cluster lineage and the + # first post-restart tick re-mints every id (restart-seam root, + # journal 2026-07-24). Legacy blobs carry emission-NEGATED x/y (see + # _apply_legacy_blob_shape) — un-negate back to the internal sign + # convention, mirroring the pca center restore above. + folded_bc = data.get('base-clusters') + if folded_bc: + unfolded_bc = conv._unfold_base_clusters(folded_bc) + if legacy: + for c in unfolded_bc: + c['center'] = [-v for v in c['center']] + conv.base_clusters = unfolded_bc + + # Restore group-votes — restructure-json-conv keeps :group-votes + # (conv_man.clj:174) and the recovery tick's comment-priorities read + # it as the PREVIOUS tick's group-votes (Q2, conversation.clj:658); + # without this a warm restart computes priorities against empty prev + # group-votes (every comment looks unseen → inflated priorities — + # vw-restart4 step-5 divergence, journal 2026-07-24). A JSON + # round-trip stringifies the per-group vote tid keys; re-intify + # them, mirroring parse-blob-json turning numeric-string keys back + # into longs (postgres.clj:419-433). gid keys stay as emitted (the + # priorities reduce only iterates values). Improved mode is + # unaffected in practice: priorities there read the CURRENT tick's + # group-votes, and the recompute overwrites this attribute first. + def _numeric_key(k): + try: + return int(k) + except (ValueError, TypeError): + return k + + blob_gv = data.get('group-votes') + if blob_gv: + conv.group_votes = { + gid: { + **{k: v for k, v in g.items() if k != 'votes'}, + 'votes': { + _numeric_key(t): e + for t, e in (g.get('votes') or {}).items() + }, + } + for gid, g in blob_gv.items() + } # Restore representativeness data. Legacy blobs emit 'repness' in # Clojure per-group shape and park the internal dict under @@ -2959,9 +3093,18 @@ def float_to_decimal(obj): # Expand base-cluster IDs to participant IDs for vote counting unfolded_groups = self._unfolded_group_clusters() + # Same tally-source rule as _compute_group_votes / to_dict: + # Clojure's group-votes aggregates votes-base, which reads + # RAW-rating-mat (conversation.clj:601-608) — moderated-out + # comments report the actual votes cast, not the zeroed + # pass-shaped columns. + tally_mat = (self.raw_rating_mat + if resolve_engine_mode() == ENGINE_MODE_LEGACY + else self.rating_mat) + # Precompute indices for each participant ptpt_indices = {} - for i, ptpt_id in enumerate(self.rating_mat.index): + for i, ptpt_id in enumerate(tally_mat.index): ptpt_indices[ptpt_id] = i # Process each group @@ -2974,19 +3117,19 @@ def float_to_decimal(obj): member_indices = [] for member in group.get('members', []): idx = ptpt_indices.get(member) - if idx is not None and idx < self.rating_mat.values.shape[0]: + if idx is not None and idx < tally_mat.values.shape[0]: member_indices.append(idx) - + # Skip groups with no valid members if not member_indices: continue - + # Get the submatrix for this group - group_matrix = self.rating_mat.values[member_indices, :] - + group_matrix = tally_mat.values[member_indices, :] + # Calculate votes for each comment group_votes = {} - for j, comment_id in enumerate(self.rating_mat.columns): + for j, comment_id in enumerate(tally_mat.columns): if j >= group_matrix.shape[1]: continue diff --git a/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py b/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py index 17ff431f87..804bf37239 100644 --- a/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py +++ b/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py @@ -44,6 +44,7 @@ # seed rows as ``kmeans_sklearn``'s ``use_first_k_init`` branch. Sharing this is # what keeps the base-level cold-start invariant tight (see module tests). from polismath.pca_kmeans_rep.clusters import _get_first_k_distinct_centers +from polismath.utils.clj_hash import clojure_hash_map_key_order # Clojure ``same-clustering?`` default tolerance (clusters.clj:71). SAME_CLUSTERING_THRESHOLD = 0.01 @@ -205,12 +206,31 @@ def cluster_step(data: _NamedData, members: List[List[Any]] = [[] for _ in range(n)] positions: List[List[np.ndarray]] = [[] for _ in range(n)] + # Assignment SCAN order: Clojure's add-to-closest iterates the + # cleared-clusters map — ``(into {})`` of [id cluster] pairs is an + # array-map in insertion (input) order for <=8 clusters but a + # PersistentHashMap for >8, whose seq order is the HAMT trie order of + # the id hashes (clusters.clj:79-86, 149). min-key keeps the LAST + # minimal entry in that order, so the scan order is semantic exactly on + # distance ties — and Q11's cancellation floor makes exact 0.0 ties + # COMMON, not measure-zero (pc-modheavy-01 step 2: 12 seed clusters + # emptied clj-side purely by hash-order ties, recorded 80 vs 92; + # journal 2026-07-24). clojure_hash_map_key_order reproduces the real + # Clojure order (cross-validated against clojure -M for n=9/20). + if n > 8: + hash_pos = {cid: i for i, cid in enumerate( + clojure_hash_map_key_order([c['id'] for c in clusters]))} + scan = sorted(range(n), key=lambda j: hash_pos[clusters[j]['id']]) + else: + scan = list(range(n)) + for name, row in zip(data.row_names, data.matrix): - best_idx = 0 - best_dist = _euclidean(row, centers[0]) - for j in range(1, n): + best_idx = scan[0] + best_dist = _euclidean(row, centers[scan[0]]) + for j in scan[1:]: d = _euclidean(row, centers[j]) - # ``<=`` => ties go to the LATER cluster (Clojure min-key semantics). + # ``<=`` => ties go to the LATER cluster in scan order (Clojure + # min-key semantics over the map's iteration order). if d <= best_dist: best_dist = d best_idx = j diff --git a/delphi/polismath/pca_kmeans_rep/pca.py b/delphi/polismath/pca_kmeans_rep/pca.py index 9605e260c8..def16079d6 100644 --- a/delphi/polismath/pca_kmeans_rep/pca.py +++ b/delphi/polismath/pca_kmeans_rep/pca.py @@ -367,14 +367,16 @@ def pca_project_dataframe(df: pd.DataFrame, projections = ((matrix_data_no_nan - pca_results['center']) @ pca_results['comps'].T) # comps are RANK-CAPPED (min(n_comps, data dim), matching - # Clojure's emitted comps) but projections are always 2-D: - # Clojure's [pc1 pc2] destructure zero-fills a missing second - # component (sparsity-aware-project-ptpt, pca.clj:134-157). + # Clojure's emitted comps) but projections are always 2-D — and + # with fewer than 2 comps rows they are all-ZERO (Q16): Clojure's + # `[pc1 pc2] comps` destructure leaves pc2 nil, and `utils/zip` + # (map vector) truncates to the shortest input — EMPTY — so the + # sparsity-aware reduce (pca.clj:134-157) never runs and EVERY + # projection (both components, participants and comments alike) + # collapses to 0.0. Verified against a 3-ptpt x 1-comment clj + # replay reference, 2026-07-22 s4 (base-clusters x/y = [0.0]). if projections.ndim == 2 and projections.shape[1] < n_comps: - projections = np.pad( - projections, - ((0, 0), (0, n_comps - projections.shape[1])), - ) + projections = np.zeros((projections.shape[0], n_comps)) projections = np.ascontiguousarray(projections) @@ -458,6 +460,13 @@ def pca_project_cmnts(center: np.ndarray, comps: np.ndarray) -> np.ndarray: n_cmnts = len(center) if n_cmnts == 0: return np.zeros((0, comps.shape[0] if comps.ndim == 2 else 0)) + if comps.ndim == 2 and comps.shape[0] < 2: + # Q16: with fewer than 2 comps rows, Clojure's `[pc1 pc2] comps` + # 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])) 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 9e23ce1411..91b85299d6 100644 --- a/delphi/polismath/replay/certify.py +++ b/delphi/polismath/replay/certify.py @@ -224,6 +224,18 @@ def votes_csv_path(dataset: str) -> Path | None: return hits[0] if hits else None +def comments_csv_path(dataset: str) -> Path | None: + """Locate a dataset's comments CSV the same way :func:`votes_csv_path` + locates its votes CSV. ``None`` when the dataset (or its comments CSV) + isn't there — moderation-interleaving datasets have one, but not every + dataset does (MOD_RESTART_PORT_SPEC.md "Python ports" item 5).""" + d = real_data.dataset_dir(dataset) + if d is None: + return None + hits = sorted(d.glob("*-comments.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": @@ -468,11 +480,21 @@ def run_py_driver(spec_path: Path, *, out_root: Path, engine_mode: str) -> subpr raise CertifyError("py-driver-launch", str(exc)) from exc -def run_clj_driver(spec_path: Path, votes_csv: Path, *, out_dir: Path) -> subprocess.CompletedProcess: +def run_clj_driver( + spec_path: Path, votes_csv: Path, *, out_dir: Path, comments_csv: Path | None = None, +) -> subprocess.CompletedProcess: """Runs ``clojure -M:replay --schedule --votes - --out `` in a SUBPROCESS with cwd=math/ (dev/replay.clj:57).""" + --out `` in a SUBPROCESS with cwd=math/ (dev/replay.clj:57). + + ``comments_csv`` adds ``--comments `` — the clj driver's + moderation-interleave source (MOD_RESTART_PORT_SPEC.md "Python ports" + item 5). Omitted (``None``, the default) for every schedule that doesn't + request moderation interleaving, so existing recordings' invocation is + byte-for-byte unchanged.""" cmd = ["clojure", "-M:replay", "--schedule", str(spec_path), "--votes", str(votes_csv), "--out", str(out_dir)] + if comments_csv is not None: + cmd += ["--comments", str(comments_csv)] try: return _run_subprocess(cmd, cwd=_MATH_ROOT, env=dict(os.environ)) except OSError as exc: @@ -541,13 +563,18 @@ def ensure_py_recording( def ensure_clj_recording( entry: BatteryEntry, spec: sched.ScheduleSpec, votes_sha: str, votes_csv: Path, *, - root: Path, refresh: bool = False, + 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.""" + 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.""" rec_dir = st.recording_dir(entry.dataset, entry.schedule_id, root=root) clj_dir = rec_dir / "clj" manifest_path = clj_dir / "cache_manifest.json" @@ -562,7 +589,7 @@ def ensure_clj_recording( 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) + result = run_clj_driver(tmp_schedule, votes_csv, out_dir=rec_dir, comments_csv=comments_csv) if result.returncode != 0: raise CertifyError( "clj-driver", (result.stderr or result.stdout or "non-zero exit").strip()[:1000] @@ -707,8 +734,15 @@ def certify_entry( ds = real_data.load_export_votes(entry.dataset) spec = build_effective_spec(entry, ds) + # --comments only when the schedule actually requests moderation + # (interleaving or an explicit list) AND the dataset has a comments + # CSV to weave from — existing moderation="none" entries never pass + # it, so their recordings/caches are untouched (MOD_RESTART_PORT_ + # SPEC.md "Python ports" item 5). + comments_csv = comments_csv_path(entry.dataset) if spec.moderation != "none" else None + clj_dir, _ = ensure_clj_recording(entry, spec, votes_sha, votes_csv, root=root, - refresh=refresh_clj) + 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, diff --git a/delphi/polismath/replay/driver.py b/delphi/polismath/replay/driver.py index 43f6e03f86..b966815a85 100644 --- a/delphi/polismath/replay/driver.py +++ b/delphi/polismath/replay/driver.py @@ -49,7 +49,8 @@ from polismath.conversation.conversation import Conversation from polismath.replay.schedule import ReplayStep, ScheduleSpec, slice_schedule -from polismath.replay.types import ReplayDataset +from polismath.replay.types import ModEvent, ReplayDataset +from polismath.utils.engine_mode import ENGINE_MODE_LEGACY, resolve_engine_mode # Vote sign convention recorded in provenance; the future Clojure driver flips. VOTE_SIGN_CONVENTION = "delphi" # AGREE=+1 (export convention, no re-flip) @@ -85,6 +86,17 @@ def run_replay( steps = slice_schedule(dataset, spec) total = len(steps) + # replay.clj CLI parity: restart_after must leave at least one step after + # the seam (0 <= r <= n_steps-2), else the "restart" would never be + # observed by any subsequent step — reject rather than silently no-op. + if spec.restart_after is not None and not ( + 0 <= spec.restart_after <= total - 2 + ): + raise ValueError( + "restart_after must be a step index with at least one step after " + f"it; got {spec.restart_after!r} for {total} steps" + ) + # `or 1`: a first vote at t_ms==0 would seed last_updated=0, which # Conversation's `last_updated or now` footgun (conversation.py:205) turns # into wall-clock — breaking determinism. Floor to 1 (nonzero). @@ -102,36 +114,76 @@ def run_replay( # certify-cold-start-pca. conv.pca = {'center': np.zeros(1), 'comps': np.array([[1.0], [1.0]])} - # Cumulative latest-wins moderation value per tid across the whole replay. + # Cumulative latest-wins moderation value per tid across the whole replay + # (improved-mode path only; legacy mode carries its own mod state on + # `conv` via `mod_update` — see the branch below). mod_state: dict[int, int] = {} + legacy = resolve_engine_mode() == ENGINE_MODE_LEGACY + + # The restart seam replays woven mods via mod_update — Clojure's (and + # legacy mode's) reducer semantics. Improved mode moderates through + # update_moderation (truthy-replace lists); silently applying mod_update + # at its restart seam would mix semantics (#2656 review, 2026-07-24). + if spec.restart_after is not None and spec.moderation != "none" and not legacy: + raise NotImplementedError( + "restart_after with a moderation-bearing schedule is only " + "implemented for clojure-legacy engine mode: the restart seam " + "replays woven mods via mod_update (legacy reducer semantics), " + "which does not mirror improved mode's update_moderation." + ) records: list[StepRecord] = [] + # Mods woven into steps so far — the restart seam replays exactly these + # (clj restart-conv: (mapcat :mods steps-so-far)), NEVER dataset.mod_events + # (a new-format comments CSV carries mod events even for schedules that + # weave none of them). + woven_mods: list[ModEvent] = [] for step in steps: if progress is not None: progress(step.index, total) conv = conv.update_votes(_votes_dict(step), recompute=False) - if step.mod_events: - for m in step.mod_events: - mod_state[m.tid] = m.mod - mod = _mod_dict(mod_state) - _guard_moderation_clear(conv, mod) - conv = conv.update_moderation(mod, recompute=True) - else: + if legacy: + # Clojure batch order (:votes :moderation, conv_man.clj:361-371): + # the votes recompute runs FIRST, on the PRIOR step's mod state. + # mod_update then touches only sets/watermark for THIS step's + # blob — NO recompute — so a mod change's effect on the math + # lands at the NEXT votes recompute (module docstring / conv/ + # mod_update docstring). moderation="none" schedules never reach + # the `if step.mod_events` branch below, so this is bit-identical + # to the pre-existing (unconditional) `conv.recompute()` call for + # every schedule that doesn't request moderation. conv = conv.recompute() + if step.mod_events: + conv = conv.mod_update(_mod_rows(step.mod_events)) + else: + if step.mod_events: + for m in step.mod_events: + mod_state[m.tid] = m.mod + mod = _mod_dict(mod_state) + _guard_moderation_clear(conv, mod) + conv = conv.update_moderation(mod, recompute=True) + else: + conv = conv.recompute() + + record = StepRecord( + index=step.index, + prev_slot=step.prev_slot, + cut_slot=step.cut_slot, + batch_size=len(step.vote_events), + cut_time_ms=step.cut_time_ms, + blob=conv.to_dict(), + extras=_step_extras(conv), + ) + records.append(record) + woven_mods.extend(step.mod_events) - records.append( - StepRecord( - index=step.index, - prev_slot=step.prev_slot, - cut_slot=step.cut_slot, - batch_size=len(step.vote_events), - cut_time_ms=step.cut_time_ms, - blob=conv.to_dict(), - extras=_step_extras(conv), + if spec.restart_after is not None and step.index == spec.restart_after: + conv = _restart_conversation( + dataset, cut_slot=step.cut_slot, cut_time_ms=step.cut_time_ms, + blob=record.blob, mod_events=tuple(woven_mods), ) - ) return records @@ -177,6 +229,57 @@ def _guard_moderation_clear(conv: Conversation, mod: dict[str, list[int]]) -> No ) +def _mod_rows(events: tuple[ModEvent, ...]) -> list[dict[str, Any]]: + """Map a batch of ModEvents to ``Conversation.mod_update``'s row shape + (``{tid, is_meta, mod, modified}`` — conversation.clj:846-884 parity).""" + return [ + {"tid": m.tid, "is_meta": m.is_meta, "mod": m.mod, "modified": m.t_ms} + for m in events + ] + + +def _restart_conversation( + dataset: ReplayDataset, *, cut_slot: int, cut_time_ms: int, blob: dict[str, Any], + mod_events: tuple[ModEvent, ...], +) -> Conversation: + """Rebuild a conversation from its OWN just-recorded step blob — the + Python mirror of a Clojure worker restart (conv_man.clj load-or-init / + restructure-json-conv; MOD_RESTART_PORT_SPEC.md "Replay-step semantics"). + + ``Conversation.from_dict`` restores the warm state Clojure's + restructure-json-conv keeps (PCA, moderation sets, repness, tid arrival + order, …) but — like Clojure resetting raw-rating-mat — leaves BOTH + rating matrices empty, and never restores the per-k group-clusterings / + group-k-smoother warm-start state at all (poller/__init__.py's + documented "load-or-init finding": ``from_dict`` does not restore + ``raw_rating_mat``/``rating_mat``/``group_clusterings``/ + ``group_k_smoother``). This rebuilds the matrices from the FULL vote + slice (dataset order, ONE batch, no recompute — mirrors update-nmat over + every vote with slot <= cut_slot) and replays the WOVEN mod history so + far via ``mod_update`` — ``mod_events`` is exactly the mods the schedule + wove into steps up to the seam, in woven order (clj restart-conv: + ``(mapcat :mods steps-so-far)``, dev/replay.clj), NEVER + ``dataset.mod_events`` (which a new-format comments CSV populates even + when the schedule weaves none of them). Empty is fine — still called + unconditionally, mirroring Clojure's conv-mod-poll 0 at restart; + ``mod_update`` always sets ``moderation_applied = True``, matching + Clojure set-ifying mod sets so a post-restart blob emits ``[]`` rather + than ``null``. + """ + restored = Conversation.from_dict(blob) + + all_votes = [ + {"pid": v.pid, "tid": v.tid, "vote": v.sign, "created": v.t_ms} + for v in dataset.votes[:cut_slot] + ] + restored = restored.update_votes( + {"votes": all_votes, "lastVoteTimestamp": cut_time_ms}, recompute=False + ) + + restored = restored.mod_update(_mod_rows(mod_events)) + return restored + + def _mod_dict(mod_state: dict[int, int]) -> dict[str, list[int]]: """Cumulative moderation sets from latest-wins per-tid mod values. diff --git a/delphi/polismath/replay/prodclone.py b/delphi/polismath/replay/prodclone.py index 1a9bfb001f..5a7ab29ea4 100644 --- a/delphi/polismath/replay/prodclone.py +++ b/delphi/polismath/replay/prodclone.py @@ -141,8 +141,12 @@ def sql_votes_export() -> str: def sql_comments_export() -> str: + """``is_meta``/``modified`` are additive (MOD_RESTART_PORT_SPEC.md "Data" + bullet) — the replay harness's moderation-interleave source + (real_data.py's mod-event loader reads them as ``is-meta``/``modified`` + on the exported CSV).""" return """ - SELECT tid, pid, created, mod + SELECT tid, pid, created, mod, is_meta, modified FROM comments WHERE zid = %s ORDER BY tid ASC @@ -370,17 +374,23 @@ def format_comments_rows( raw_rows: Iterable[dict[str, Any]], vote_counts: dict[int, tuple[int, int]], ) -> list[dict[str, str]]: - """``raw_rows``: dicts with keys tid, pid, created, mod. - ``vote_counts``: {tid: (agrees, disagrees)}, counted over ALL vote rows - (see :func:`sql_comment_vote_counts`); missing tids default to (0, 0). + """``raw_rows``: dicts with keys tid, pid, created, mod (is_meta/modified + optional — default to False/empty so this stays usable with rows that + don't carry them yet). ``vote_counts``: {tid: (agrees, disagrees)}, + counted over ALL vote rows (see :func:`sql_comment_vote_counts`); missing + tids default to (0, 0). ``comment-body`` is ALWAYS the empty string — comment text is redacted per the privacy rules; the column is present (mirroring the export - format) but never populated.""" + format) but never populated. ``is-meta``/``modified`` are ADDITIVE + columns (MOD_RESTART_PORT_SPEC.md "Data" bullet) appended after the + pre-existing ones — the replay harness's moderation-interleave source + (real_data.py's mod-event loader).""" out = [] for row in raw_rows: agrees, disagrees = vote_counts.get(row["tid"], (0, 0)) created = row["created"] + modified = row.get("modified") out.append({ "timestamp": str(created // 1000), "datetime": format_export_datetime(created), @@ -390,6 +400,8 @@ def format_comments_rows( "disagrees": str(disagrees), "moderated": str(row["mod"]), "comment-body": "", + "is-meta": str(bool(row.get("is_meta", False))), + "modified": "" if modified is None else str(modified), }) return out @@ -398,6 +410,7 @@ def format_comments_rows( _COMMENTS_FIELDNAMES = [ "timestamp", "datetime", "comment-id", "author-id", "agrees", "disagrees", "moderated", "comment-body", + "is-meta", "modified", ] diff --git a/delphi/polismath/replay/real_data.py b/delphi/polismath/replay/real_data.py index eafbac6b64..551c6dff99 100644 --- a/delphi/polismath/replay/real_data.py +++ b/delphi/polismath/replay/real_data.py @@ -17,10 +17,62 @@ import csv from pathlib import Path -from polismath.replay.types import ReplayDataset +from polismath.replay.types import ModEvent, ReplayDataset REAL_DATA_ROOT = Path(__file__).resolve().parents[2] / "real_data" +# 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" +# and are left alone: no mod events, no error. "is-meta" is optional and +# defaults to False when absent, mirroring the clj reader; "comment-id" and +# "moderated" ARE required — the row loop reads them unconditionally, so a +# header missing either takes the graceful no-events path instead of a +# KeyError mid-row (#2656 review finding 3). +_MOD_EVENT_REQUIRED_COLUMNS = frozenset({"modified", "comment-id", "moderated"}) +_TRUE_STRINGS = frozenset({"1", "true", "t", "yes"}) + + +def _parse_bool(value: str | None) -> bool: + if value is None: + return False + return value.strip().lower() in _TRUE_STRINGS + + +def _load_mod_events(comments_csv: Path) -> tuple[list[ModEvent], int]: + """Build ``ModEvent``s from a comments CSV carrying the moderation-history + columns, alongside the existing ``comment-id``/``moderated`` columns + (modified->t_ms, comment-id->tid, moderated->mod, is-meta->is_meta). + + Returns ``([], 0)`` when the required columns are absent (a header-level + check — this is a format detection, not a per-row guess). Rows with no + ``modified`` value cannot be woven into a replay schedule (nothing to + interleave on) — SKIPPED; the count is returned for provenance (surfaced + via :attr:`~polismath.replay.types.ReplayDataset.mod_events_skipped`). + """ + with open(comments_csv, newline="") as fh: + reader = csv.DictReader(fh) + fieldnames = set(reader.fieldnames or []) + if not _MOD_EVENT_REQUIRED_COLUMNS <= fieldnames: + return [], 0 + + events: list[ModEvent] = [] + skipped = 0 + for row in reader: + modified = (row.get("modified") or "").strip() + if not modified: + skipped += 1 + continue + events.append( + ModEvent( + t_ms=int(modified), + tid=int(row["comment-id"]), + mod=int(row["moderated"]), + is_meta=_parse_bool(row.get("is-meta")), + ) + ) + return events, skipped + def dataset_dir(slug: str) -> Path | None: """Locate a dataset directory by slug — public (``real_data/*-``) @@ -38,6 +90,12 @@ def load_export_votes(slug: str) -> ReplayDataset: Comment creation times are inferred as first-vote times (lower bound on availability; adequate because a comment is unobservable in the mark likelihood before its first vote anyway). + + If a ``*-comments.csv`` sits alongside the votes CSV AND carries the + moderation-history columns (``modified``, ``is-meta``), the dataset's + ``mod_events`` are built from it (see :func:`_load_mod_events`) — older + comments CSVs, or datasets with no comments CSV at all, yield no mod + events (unchanged from before this was wired up). """ d = dataset_dir(slug) if d is None: @@ -56,4 +114,13 @@ def load_export_votes(slug: str) -> ReplayDataset: int(row["vote"]), ) ) - return ReplayDataset.build(raw) + + mod_events: list[ModEvent] = [] + mod_events_skipped = 0 + comments_csvs = sorted(d.glob("*-comments.csv")) + if comments_csvs: + mod_events, mod_events_skipped = _load_mod_events(comments_csvs[0]) + + dataset = ReplayDataset.build(raw, mod_events=mod_events) + dataset.mod_events_skipped = mod_events_skipped + return dataset diff --git a/delphi/polismath/replay/schedule.py b/delphi/polismath/replay/schedule.py index 11a0c901e9..671e16d62d 100644 --- a/delphi/polismath/replay/schedule.py +++ b/delphi/polismath/replay/schedule.py @@ -61,6 +61,12 @@ class ScheduleSpec: moderation: Any = "none" clojure: dict[str, Any] = field(default_factory=lambda: {"warm_start": "chain"}) notes: str = "" + # Restart seam (MOD_RESTART_PORT_SPEC.md "Replay-step semantics" / restart + # plumbing): after recording the step at this index, the driver rebuilds + # the conversation the way a Clojure worker restart would (see driver.py's + # `_restart_conversation`). None (default) means no restart — every + # existing schedule is unaffected. + restart_after: int | None = None # Verbatim mapping this spec was loaded from (None → reconstruct on demand). _raw: dict[str, Any] | None = field(default=None, repr=False, compare=False) @@ -75,6 +81,7 @@ def from_dict(cls, d: dict[str, Any]) -> "ScheduleSpec": moderation=d.get("moderation", "none"), clojure=d.get("clojure", {"warm_start": "chain"}), notes=d.get("notes", ""), + restart_after=d.get("restart_after"), _raw=dict(d), ) @@ -95,6 +102,7 @@ def to_dict(self) -> dict[str, Any]: "moderation": self.moderation, "clojure": self.clojure, "notes": self.notes, + "restart_after": self.restart_after, } def write_json(self, path: str | Path) -> None: @@ -230,9 +238,12 @@ def _resolve_mod_events(dataset: ReplayDataset, spec: ScheduleSpec) -> list[ModE return sorted(dataset.mod_events, key=lambda m: m.t_ms) if isinstance(mode, (list, tuple)): parsed = [ - m if isinstance(m, ModEvent) else ModEvent(t_ms=int(m["t_ms"]), - tid=int(m["tid"]), - mod=int(m["mod"])) + m if isinstance(m, ModEvent) else ModEvent( + t_ms=int(m["t_ms"]), + tid=int(m["tid"]), + mod=int(m["mod"]), + is_meta=bool(m.get("is_meta", False)), + ) for m in mode ] return sorted(parsed, key=lambda m: m.t_ms) diff --git a/delphi/polismath/replay/types.py b/delphi/polismath/replay/types.py index 3f85cc277b..f482e61ce7 100644 --- a/delphi/polismath/replay/types.py +++ b/delphi/polismath/replay/types.py @@ -58,12 +58,17 @@ class ModEvent: """A moderation change at ``t_ms`` setting ``comments.mod`` for ``tid``. ``mod`` uses the production convention: -1 moderated-out, 0 unmoderated, - 1 moderated-in. + 1 moderated-in. ``is_meta`` mirrors ``comments.is_meta`` (MOD_RESTART_PORT_ + SPEC.md "Python ports" item 2) — additive, defaults False so every existing + caller (bare ``ModEvent(t_ms, tid, mod)``) is unaffected. Consumed by + ``Conversation.mod_update`` (conversation.clj:846-884 parity): an is_meta + row lands in BOTH mod-out and mod-in regardless of ``mod``. """ t_ms: int tid: int mod: int + is_meta: bool = False Schedule = tuple[int, ...] @@ -78,6 +83,13 @@ class ReplayDataset: comments: dict[int, CommentMeta] mod_events: list[ModEvent] = field(default_factory=list) strict_moderation: bool = False + # Provenance counter (MOD_RESTART_PORT_SPEC.md "Data" bullet): rows in the + # source comments CSV that carried no ``modified`` timestamp and therefore + # could not be woven into a replay schedule as a ModEvent. Populated by + # :func:`polismath.replay.real_data.load_export_votes`; 0 for datasets with + # no moderation-history columns at all (nothing was skipped — there was + # nothing to parse). + mod_events_skipped: int = 0 @property def n(self) -> int: diff --git a/delphi/scripts/certify_battery.json b/delphi/scripts/certify_battery.json index 0a5643b18c..5ca2817ad8 100644 --- a/delphi/scripts/certify_battery.json +++ b/delphi/scripts/certify_battery.json @@ -1,104 +1,134 @@ [ - { - "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 \u2014 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" - }, - { - "dataset": "FLI", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "smallest private dataset (~91k votes) \u2014 pilot for the private-size regime; calibrates clj/py wall-clock before scheduling bg2018/pakistan/engage/bg2050" - }, - { - "dataset": "bg2018", - "preset": "uniform", - "n_cuts": 8, - "engine_mode": "clojure-legacy", - "notes": "~226k votes; revote-rich production conversation" - }, - { - "dataset": "pakistan", - "preset": "uniform", - "n_cuts": 8, - "engine_mode": "clojure-legacy", - "notes": "~400k votes" - }, - { - "dataset": "engage", - "preset": "uniform", - "n_cuts": 8, - "engine_mode": "clojure-legacy", - "notes": "~443k votes" - }, - { - "dataset": "bg2050", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "largest (~1.03M votes) \u2014 6 cuts to bound wall-clock" - }, - { - "dataset": "vw", - "schedule": "schedules/vw-every-vote-56.json", - "engine_mode": "clojure-legacy", - "notes": "every-vote prefix \u2014 see the schedule file for the Q11 truncation rationale" - }, - { - "dataset": "pc-revote-01", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "prodclone: extreme revote conversation (97% revotes, ~56k votes)" - }, - { - "dataset": "pc-banned-01", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "prodclone: banned participants present (participants.mod=-1; Q1 leak territory)" - }, - { - "dataset": "pc-smallmix-01", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "prodclone: small mixed conversation (~5k votes)" - }, - { - "dataset": "pc-midmix-01", - "preset": "uniform", - "n_cuts": 6, - "engine_mode": "clojure-legacy", - "notes": "prodclone: mid mixed conversation (~49k votes)" - }, - { - "dataset": "pc-zerovote-01", - "preset": "single-cut", - "engine_mode": "clojure-legacy", - "notes": "prodclone: zero-vote conversation \u2014 empty-conversation edge" - } -] \ No newline at end of file + { + "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 \u2014 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" + }, + { + "dataset": "FLI", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "smallest private dataset (~91k votes) \u2014 pilot for the private-size regime; calibrates clj/py wall-clock before scheduling bg2018/pakistan/engage/bg2050" + }, + { + "dataset": "bg2018", + "preset": "uniform", + "n_cuts": 8, + "engine_mode": "clojure-legacy", + "notes": "~226k votes; revote-rich production conversation" + }, + { + "dataset": "pakistan", + "preset": "uniform", + "n_cuts": 8, + "engine_mode": "clojure-legacy", + "notes": "~400k votes" + }, + { + "dataset": "engage", + "preset": "uniform", + "n_cuts": 8, + "engine_mode": "clojure-legacy", + "notes": "~443k votes" + }, + { + "dataset": "bg2050", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "largest (~1.03M votes) \u2014 6 cuts to bound wall-clock" + }, + { + "dataset": "vw", + "schedule": "schedules/vw-every-vote-56.json", + "engine_mode": "clojure-legacy", + "notes": "every-vote prefix \u2014 see the schedule file for the Q11 truncation rationale; also the battery's degenerate-tick coverage: its early steps are 1-participant/1..N-comment ticks (goal-doc edge-case list)" + }, + { + "dataset": "pc-revote-02", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "prodclone: revote-heavy small conversation (~28% revotes, ~8.4k votes, 34 ptpts, 277 comments). Replaces pc-revote-01 (~56k votes, 97% revotes, 205 ptpts): its warm-chain split-loop hits a 4-way knife-edge tie (within-engine gaps <=2.5e-16 vs ~1e-5 cross-engine PCA noise) - extraction order irreducible cross-language; carved out per divergences.json + journal 2026-07-22 session 4." + }, + { + "dataset": "pc-banned-01", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "prodclone: banned participants present (participants.mod=-1; Q1 leak territory)" + }, + { + "dataset": "pc-smallmix-01", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "prodclone: small mixed conversation (~5k votes)" + }, + { + "dataset": "pc-midmix-01", + "preset": "uniform", + "n_cuts": 6, + "engine_mode": "clojure-legacy", + "notes": "prodclone: mid mixed conversation (~49k votes)" + }, + { + "dataset": "pc-zerovote-01", + "preset": "single-cut", + "engine_mode": "clojure-legacy", + "notes": "prodclone: zero-vote conversation \u2014 empty-conversation edge" + }, + { + "dataset": "pc-modheavy-01", + "schedule": "schedules/pc-modheavy-01-single-cut-mod.json", + "engine_mode": "clojure-legacy", + "notes": "prodclone: moderation-heavy (~81% modout, ~12k votes, 215 ptpts); single cold cut, full mod weave (Q18 carve - warm chain knife-edges; warm mod coverage: pc-meta-02)" + }, + { + "dataset": "pc-meta-01", + "schedule": "schedules/pc-meta-01-single-cut-mod.json", + "engine_mode": "clojure-legacy", + "notes": "prodclone: meta-rich (~25% is-meta, ~4.7k votes, 109 ptpts); single cold cut, full mod weave (Q18 carve - warm chain knife-edges; warm meta coverage: pc-meta-02)" + }, + { + "dataset": "vw", + "schedule": "schedules/vw-uniform8-restart4.json", + "engine_mode": "clojure-legacy", + "notes": "restart seam: uniform8 with load-or-init worker-restart replay after step 4" + }, + { + "dataset": "pc-midmix-01", + "schedule": "schedules/pc-midmix-01-uniform6-restart3.json", + "engine_mode": "clojure-legacy", + "notes": "restart seam: uniform6 with worker-restart replay after step 3 (medium prodclone)" + }, + { + "dataset": "pc-meta-02", + "schedule": "schedules/pc-meta-02-uniform6-mod.json", + "engine_mode": "clojure-legacy", + "notes": "prodclone: moderate mod density (20 modout + 15 meta of 146 cmts, 2.4k votes, 33 ptpts; ptpt-per-live 0.26); the Q18-safe warm-chain mod/meta entry" + } +] diff --git a/delphi/scripts/schedules/pc-meta-01-single-cut-mod.json b/delphi/scripts/schedules/pc-meta-01-single-cut-mod.json new file mode 100644 index 0000000000..6d14fcbb5e --- /dev/null +++ b/delphi/scripts/schedules/pc-meta-01-single-cut-mod.json @@ -0,0 +1,16 @@ +{ + "dataset": "pc-meta-01", + "schedule_id": "single-cut-mod", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 4689 + ] + }, + "moderation": "interleave-by-timestamp", + "clojure": { + "warm_start": "chain" + }, + "notes": "single cold cut with the full mod history woven (Q18 carve, 2026-07-24: this dataset's mod-narrowed warm geometry knife-edges in uniqify; warm-chain meta coverage lives in pc-meta-02 uniform6-mod)" +} diff --git a/delphi/scripts/schedules/pc-meta-01-uniform6-mod.json b/delphi/scripts/schedules/pc-meta-01-uniform6-mod.json new file mode 100644 index 0000000000..8155ec3ff4 --- /dev/null +++ b/delphi/scripts/schedules/pc-meta-01-uniform6-mod.json @@ -0,0 +1,21 @@ +{ + "dataset": "pc-meta-01", + "schedule_id": "uniform6-mod", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 782, + 1563, + 2344, + 3126, + 3908, + 4689 + ] + }, + "moderation": "interleave-by-timestamp", + "clojure": { + "warm_start": "chain" + }, + "notes": "6 evenly-spaced recomputes with moderation interleaved (meta-rich prodclone extraction: meta-tids enter via mod-update, the production-reachable route) [NOT IN BATTERY since 2026-07-24: warm-chain steps knife-edge on Q18 (uniqify merge-center ulp chaos, CLOJURE_QUIRKS.md) \u2014 kept only to reproduce the diagnosis; certification uses the single-cut-mod variant]" +} diff --git a/delphi/scripts/schedules/pc-meta-02-uniform6-mod.json b/delphi/scripts/schedules/pc-meta-02-uniform6-mod.json new file mode 100644 index 0000000000..2af7c92c9a --- /dev/null +++ b/delphi/scripts/schedules/pc-meta-02-uniform6-mod.json @@ -0,0 +1,21 @@ +{ + "dataset": "pc-meta-02", + "schedule_id": "uniform6-mod", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 402, + 805, + 1207, + 1609, + 2012, + 2414 + ] + }, + "moderation": "interleave-by-timestamp", + "clojure": { + "warm_start": "chain" + }, + "notes": "6 evenly-spaced recomputes with moderation interleaved (moderate-density prodclone extraction: meta + modout on coincidence-sparse geometry - the Q18-safe warm-chain mod/meta coverage)" +} diff --git a/delphi/scripts/schedules/pc-midmix-01-uniform6-restart3.json b/delphi/scripts/schedules/pc-midmix-01-uniform6-restart3.json new file mode 100644 index 0000000000..cb91b6f397 --- /dev/null +++ b/delphi/scripts/schedules/pc-midmix-01-uniform6-restart3.json @@ -0,0 +1,22 @@ +{ + "dataset": "pc-midmix-01", + "schedule_id": "uniform6-restart3", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 8191, + 16382, + 24573, + 32764, + 40955, + 49146 + ] + }, + "moderation": "none", + "clojure": { + "warm_start": "chain" + }, + "notes": "uniform6 with a worker-restart seam after step 3 (medium-size prodclone extraction)", + "restart_after": 3 +} \ No newline at end of file diff --git a/delphi/scripts/schedules/pc-modheavy-01-single-cut-mod.json b/delphi/scripts/schedules/pc-modheavy-01-single-cut-mod.json new file mode 100644 index 0000000000..1491a6886b --- /dev/null +++ b/delphi/scripts/schedules/pc-modheavy-01-single-cut-mod.json @@ -0,0 +1,16 @@ +{ + "dataset": "pc-modheavy-01", + "schedule_id": "single-cut-mod", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 11712 + ] + }, + "moderation": "interleave-by-timestamp", + "clojure": { + "warm_start": "chain" + }, + "notes": "single cold cut with the full mod history woven (Q18 carve, 2026-07-24: mod-heavy warm chains are certification-hostile - uniqify merge-center exactness knife-edges on coincidence-dense geometry; warm-chain mod/meta coverage lives in pc-meta-02 uniform6-mod)" +} diff --git a/delphi/scripts/schedules/pc-modheavy-01-uniform6-mod.json b/delphi/scripts/schedules/pc-modheavy-01-uniform6-mod.json new file mode 100644 index 0000000000..bceef9accc --- /dev/null +++ b/delphi/scripts/schedules/pc-modheavy-01-uniform6-mod.json @@ -0,0 +1,21 @@ +{ + "dataset": "pc-modheavy-01", + "schedule_id": "uniform6-mod", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 1952, + 3904, + 5856, + 7808, + 9760, + 11712 + ] + }, + "moderation": "interleave-by-timestamp", + "clojure": { + "warm_start": "chain" + }, + "notes": "6 evenly-spaced recomputes with moderation interleaved by modified timestamp (modout-heavy prodclone extraction, meta rows included) [NOT IN BATTERY since 2026-07-24: warm-chain steps knife-edge on Q18 (uniqify merge-center ulp chaos, CLOJURE_QUIRKS.md) \u2014 kept only to reproduce the diagnosis; certification uses the single-cut-mod variant]" +} diff --git a/delphi/scripts/schedules/vw-uniform8-restart4.json b/delphi/scripts/schedules/vw-uniform8-restart4.json new file mode 100644 index 0000000000..b3bc5e9377 --- /dev/null +++ b/delphi/scripts/schedules/vw-uniform8-restart4.json @@ -0,0 +1,24 @@ +{ + "dataset": "vw", + "schedule_id": "uniform8-restart4", + "source": "votes-csv", + "cuts": { + "mode": "vote-count", + "at": [ + 585, + 1171, + 1756, + 2342, + 2927, + 3512, + 4098, + 4683 + ] + }, + "moderation": "none", + "clojure": { + "warm_start": "chain" + }, + "notes": "uniform8 with a worker-restart seam after step 4 (load-or-init replay: blob round-trip + full-history raw-rating-mat + mod-update)", + "restart_after": 4 +} \ No newline at end of file diff --git a/delphi/tests/poller/test_load_or_init.py b/delphi/tests/poller/test_load_or_init.py index 609221d119..6f19eaf083 100644 --- a/delphi/tests/poller/test_load_or_init.py +++ b/delphi/tests/poller/test_load_or_init.py @@ -1,9 +1,11 @@ """load-or-init + the from_dict restoration finding. These tests LOCK the finding documented in polismath/poller/__init__.py: -``Conversation.from_dict`` restores warm state (pca, moderation, counts) but NOT -the rating matrices or base_clusters, so load-or-init must ALWAYS rebuild the -matrices from the full vote history (mirroring conv_man.clj:188-207). +``Conversation.from_dict`` restores warm state (pca, moderation, counts — and, +since the 2026-07-24 restart-seam fix, zid, base_clusters and group_votes, +mirroring what restructure-json-conv keeps, conv_man.clj:171-186) but NOT the +rating matrices or the group-clusterings/smoother memory, so load-or-init must +ALWAYS rebuild the matrices from the full vote history (conv_man.clj:188-207). """ import time @@ -46,16 +48,25 @@ def test_from_dict_restores_pca_and_moderation_but_not_matrices(self): blob = conv.to_dict() restored = Conversation.from_dict(blob) - # RESTORED (warm state): pca, moderation, counts. + # RESTORED (warm state): pca, moderation, counts — and, since the + # restart-seam fix (journal 2026-07-24), zid + base clusters + # (id/members faithful — the warm-start lineage input) + group-votes, + # exactly what restructure-json-conv keeps (conv_man.clj:171-186). assert restored.pca is not None assert set(restored.mod_out_tids) == {"3"} assert restored.participant_count == conv.participant_count - - # NOT RESTORED: the vote matrices and base_clusters — hence a full - # rebuild is mandatory in load-or-init. + assert restored.conversation_id == "42" + assert [c["id"] for c in restored.base_clusters] == \ + [c["id"] for c in conv.base_clusters] + assert [c["members"] for c in restored.base_clusters] == \ + [c["members"] for c in conv.base_clusters] + + # NOT RESTORED: the vote matrices (and the per-k clusterings/smoother + # memory) — hence a full rebuild is mandatory in load-or-init. assert restored.raw_rating_mat.size == 0 assert restored.rating_mat.size == 0 - assert restored.base_clusters == [] + assert restored.group_clusterings == {} + assert restored.group_k_smoother == {} class TestLoadOrInit: diff --git a/delphi/tests/replay_harness/test_certify.py b/delphi/tests/replay_harness/test_certify.py index 9fd0ca4b55..aaefb25b61 100644 --- a/delphi/tests/replay_harness/test_certify.py +++ b/delphi/tests/replay_harness/test_certify.py @@ -453,6 +453,96 @@ def fake_run(cmd, *, cwd, env): assert exc_info.value.stage +# --------------------------------------------------------------------------- +# --comments plumbing (MOD_RESTART_PORT_SPEC.md "Python ports" item 5): the +# clj driver gets --comments only when the schedule requests moderation != +# "none" AND the dataset has a comments CSV. Existing (moderation="none") +# recordings must be unaffected -- no --comments flag, no manifest change. +# --------------------------------------------------------------------------- +def test_comments_csv_path_locates_existing_public_dataset(): + path = cert.comments_csv_path("vw") + assert path is not None + assert path.name.endswith("-comments.csv") + assert path.exists() + + +def test_comments_csv_path_returns_none_for_missing_dataset(): + assert cert.comments_csv_path("no-such-dataset-xyz") is None + + +def test_run_clj_driver_includes_comments_flag_when_given(tmp_path, monkeypatch): + captured = {} + + def fake_run(cmd, *, cwd, env): + captured["cmd"] = cmd + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + comments_csv = tmp_path / "comments.csv" + cert.run_clj_driver(tmp_path / "sched.json", tmp_path / "votes.csv", out_dir=tmp_path, + comments_csv=comments_csv) + cmd = captured["cmd"] + assert "--comments" in cmd + assert cmd[cmd.index("--comments") + 1] == str(comments_csv) + + +def test_run_clj_driver_omits_comments_flag_by_default(tmp_path, monkeypatch): + captured = {} + + def fake_run(cmd, *, cwd, env): + captured["cmd"] = cmd + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + cert.run_clj_driver(tmp_path / "sched.json", tmp_path / "votes.csv", out_dir=tmp_path) + assert "--comments" not in captured["cmd"] + + +@requires_math_tree +def test_certify_entry_passes_comments_when_moderation_requested(tmp_path, monkeypatch): + calls = [] + + def fake_run(cmd, *, cwd, env): + calls.append(cmd) + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + schedule_path = tmp_path / "sched.json" + schedule_path.write_text(json.dumps({ + "dataset": "vw", "schedule_id": "mod-comments-test", + "cuts": {"mode": "vote-count", "at": [10]}, + "moderation": "interleave-by-timestamp", + "clojure": {"warm_start": "chain"}, "notes": "", + })) + entry = cert.parse_battery_entry( + {"dataset": "vw", "schedule": str(schedule_path), "engine_mode": "clojure-legacy"}, + ) + cert.certify_entry(entry, root=tmp_path, ledger={}) + + clj_cmds = [c for c in calls if c and c[0] == "clojure"] + assert len(clj_cmds) == 1 + assert "--comments" in clj_cmds[0] + + +@requires_math_tree +def test_certify_entry_omits_comments_when_moderation_none(tmp_path, monkeypatch): + calls = [] + + def fake_run(cmd, *, cwd, env): + calls.append(cmd) + return _fake_completed() + + monkeypatch.setattr(cert, "_run_subprocess", fake_run) + + entry = _make_entry() # dataset=vw, preset=single-cut -> moderation "none" + cert.certify_entry(entry, root=tmp_path, ledger={}) + + clj_cmds = [c for c in calls if c and c[0] == "clojure"] + assert len(clj_cmds) == 1 + assert "--comments" not in clj_cmds[0] + + 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={}) diff --git a/delphi/tests/replay_harness/test_driver.py b/delphi/tests/replay_harness/test_driver.py index ab39336beb..85499afd8b 100644 --- a/delphi/tests/replay_harness/test_driver.py +++ b/delphi/tests/replay_harness/test_driver.py @@ -16,7 +16,9 @@ import pytest +from polismath.conversation.conversation import Conversation from polismath.replay.real_data import load_export_votes +from polismath.replay import driver from polismath.replay import schedule as sched from polismath.replay.driver import run_replay, VOTE_SIGN_CONVENTION from polismath.replay.types import ModEvent, ReplayDataset @@ -194,3 +196,252 @@ def test_determinism_bit_identical_except_wall_clock(vw_dataset, spec, run1): blob_a = {k: v for k, v in a.blob.items() if k not in WALL_CLOCK_FIELDS} blob_b = {k: v for k, v in b.blob.items() if k not in WALL_CLOCK_FIELDS} assert blob_a == blob_b, f"non-wall-clock blob differs at step {a.index}" + + +# --- legacy-mode moderation: mod_update, votes-then-mods, no mod recompute -- +# MOD_RESTART_PORT_SPEC.md "Python ports" item 4 / "Replay-step semantics": +# in 'clojure-legacy' engine mode, the votes batch recomputes FIRST (using the +# PRIOR step's mod state); mod_update then only touches sets/watermark for +# THIS step's blob — no recompute — mirroring Clojure's :moderation handler +# (mod-update's effect on the math lands at the NEXT votes recompute). +def _run_legacy(ds, spec): + logging.disable(logging.CRITICAL) + try: + return run_replay(ds, spec) + finally: + logging.disable(logging.NOTSET) + + +def test_legacy_mode_applies_mod_events_via_mod_update(monkeypatch): + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + mods = [ModEvent(35, 100, -1), ModEvent(55, 101, 1)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + records = _run_legacy(ds, _mod_spec(mods)) + assert len(records) == 2 + step0 = records[0].blob["moderation"] + assert step0["mod_out_tids"] == [100] + assert step0["mod_in_tids"] == [] + step1 = records[1].blob["moderation"] + assert sorted(step1["mod_out_tids"]) == [100] + assert sorted(step1["mod_in_tids"]) == [101] + + +def test_legacy_mode_un_moderation_disjs_the_set(monkeypatch): + # The un-moderating sequence that DEFEATS update_moderation/_guard in + # improved mode (test_driver_fails_loudly_on_moderation_set_emptying) + # must be representable in legacy mode via mod_update's disj semantics. + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + mods = [ModEvent(35, 100, -1), ModEvent(35, 101, 1), ModEvent(55, 100, 0)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + records = _run_legacy(ds, _mod_spec(mods)) + assert len(records) == 2 + step1 = records[1].blob["moderation"] + assert step1["mod_out_tids"] == [] # tid 100 un-moderated -> disj, not stuck + assert step1["mod_in_tids"] == [101] + + +def test_legacy_mode_none_moderation_never_calls_mod_update(monkeypatch): + """Task-3 exact-preservation rule: zero mod events -> zero mod_update + calls, not even with an empty list, so schedules with moderation="none" + stay bit-identical (mod_update unconditionally flips moderation_applied, + so a stray call would be observable even with nothing in the sets).""" + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + calls = [] + original = Conversation.mod_update + + def _spy(self, mods): + calls.append(list(mods)) + return original(self, mods) + + monkeypatch.setattr(Conversation, "mod_update", _spy) + + raw = [(100 * (i + 1), (i % 3) + 1, (i % 2) + 10, 1) for i in range(6)] + ds = ReplayDataset.build(raw) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "legacy-none", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [3, 6]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", + }) + records = _run_legacy(ds, spec) + assert calls == [] + assert records[-1].blob["moderation"]["mod_out_tids"] == [] + + +def test_improved_mode_still_uses_update_moderation_and_guard(monkeypatch): + # Explicit control: 'improved' (default, no env override) keeps using + # update_moderation + _guard_moderation_clear, never mod_update. + calls = [] + original = Conversation.mod_update + + def _spy(self, mods): + calls.append(list(mods)) + return original(self, mods) + + monkeypatch.setattr(Conversation, "mod_update", _spy) + mods = [ModEvent(35, 100, -1), ModEvent(55, 101, 1)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + records = _run_legacy(ds, _mod_spec(mods)) + assert len(records) == 2 + assert calls == [] # mod_update never called on the improved path + + +# --- restart_after: worker-restart seam ------------------------------------ +# MOD_RESTART_PORT_SPEC.md "Replay-step semantics" / restart plumbing: after +# recording the step at spec.restart_after, the driver rebuilds the +# conversation the way a Clojure worker restart would (parse the just- +# recorded blob, restore via Conversation.from_dict, rebuild BOTH rating +# matrices from the full vote slice, replay the WOVEN mod history so far via +# mod_update — clj restart-conv's (mapcat :mods steps-so-far)) and continues +# the schedule from there. +_RESTART_RAW_VOTES = [ + (100 * (i + 1), (i % 4) + 1, (i % 3) + 10, [1, -1, 1][i % 3]) + for i in range(20) +] + + +def test_restart_after_does_not_change_step_count(monkeypatch): + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + ds = ReplayDataset.build(_RESTART_RAW_VOTES) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "restart-e2e", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [5, 10, 15, 20]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 1, + }) + records = _run_legacy(ds, spec) + assert [r.index for r in records] == [0, 1, 2, 3] + assert [r.cut_slot for r in records] == [5, 10, 15, 20] + + +def test_restart_after_none_is_a_no_op(monkeypatch): + # restart_after absent (None, the default) must not touch the replay at + # all — same step count/content as never having the field. + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + ds = ReplayDataset.build(_RESTART_RAW_VOTES) + spec_no_restart = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "no-restart", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [5, 10, 15, 20]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", + }) + records = _run_legacy(ds, spec_no_restart) + assert len(records) == 4 + + +def test_restart_conversation_rebuilds_matrices_and_drops_smoother_state(monkeypatch): + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + ds = ReplayDataset.build(_RESTART_RAW_VOTES) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "restart-unit", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [10]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", + }) + records = _run_legacy(ds, spec) + blob = records[0].blob + + restored = driver._restart_conversation( + ds, cut_slot=records[0].cut_slot, cut_time_ms=records[0].cut_time_ms, + blob=blob, mod_events=(), + ) + # from_dict never restores these (poller/__init__.py's documented + # "load-or-init finding") -- confirmed dropped on the restart path too. + assert restored.group_clusterings == {} + assert restored.group_k_smoother == {} + # Rating matrices are rebuilt fresh from the full vote slice, not left + # empty (from_dict alone would leave them at the cls() default). + assert restored.raw_rating_mat.shape[0] > 0 + assert restored.raw_rating_mat.shape[1] > 0 + assert restored.rating_mat.shape == restored.raw_rating_mat.shape + # Restart-seam root (journal 2026-07-24): the warm-start lineage input + # must survive the restore — zid and base clusters come back from the + # blob (clj restructure-json-conv keeps :zid and unfolds :base-clusters). + assert restored.conversation_id == "t" + blob_bc = blob["base-clusters"] + assert len(blob_bc["id"]) > 0, "recorded blob unexpectedly has no base clusters" + assert [c["id"] for c in restored.base_clusters] == list(blob_bc["id"]) + assert [c["members"] for c in restored.base_clusters] == list(blob_bc["members"]) + + +def test_restart_conversation_replays_woven_mod_history_not_just_blob_state(): + # A blob with NO moderation recorded (e.g. recorded before the mods were + # applied) — restart must derive the mod state from the WOVEN mod history + # passed in (clj restart-conv: (mapcat :mods steps-so-far)) via + # mod_update, not trust the (here: empty) blob moderation. + raw = [(100, 1, 10, 1), (200, 2, 11, -1)] + woven = (ModEvent(t_ms=50, tid=10, mod=-1), ModEvent(t_ms=150, tid=11, mod=1)) + ds = ReplayDataset.build(raw, mod_events=list(woven)) + blob = { + "conversation_id": "t", "last_updated": 200, "participant_count": 2, + "comment_count": 2, "vote_stats": {}, + "moderation": {"mod_out_tids": [], "mod_in_tids": [], "meta_tids": [], + "mod_out_ptpts": []}, + } + restored = driver._restart_conversation( + ds, cut_slot=ds.n, cut_time_ms=200, blob=blob, mod_events=woven, + ) + assert restored.mod_out_tids == {10} + assert restored.mod_in_tids == {11} + assert restored.moderation_applied is True + + +def test_restart_replays_only_woven_mods_not_dataset_mods(monkeypatch): + # #2656 review finding 1 (the landmine): a NEW-format comments CSV always + # yields dataset.mod_events, but a moderation="none" schedule weaves NONE + # of them into steps. clj restart-conv replays only the woven mods + # ((mapcat :mods steps-so-far), replay.clj) — the py restart must not + # smuggle dataset-level mods the chain never saw into the warm state. + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + mods = [ModEvent(t_ms=150, tid=10, mod=-1)] + ds = ReplayDataset.build(_RESTART_RAW_VOTES, mod_events=mods) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "restart-unwoven", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [10, 20]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 0, + }) + records = _run_legacy(ds, spec) + post = records[1].blob["moderation"] + assert post["mod_out_tids"] == [] # dataset-level mod never woven -> never replayed + + +def test_restart_replays_woven_mods_so_far(monkeypatch): + # Control for the test above: mods that ARE woven into steps up to the + # seam must survive the restart (replayed via mod_update). + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + mods = [ModEvent(35, 100, -1)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "restart-woven", "source": "votes-csv", + "cuts": _MOD_CUTS, "moderation": "interleave-by-timestamp", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 0, + }) + records = _run_legacy(ds, spec) + assert records[1].blob["moderation"]["mod_out_tids"] == [100] + + +def test_restart_with_moderation_requires_legacy_mode(): + # #2656 review (2026-07-24): the restart seam replays woven mods via + # mod_update (legacy reducer semantics); combining restart_after with a + # moderation-bearing schedule in IMPROVED mode would silently apply the + # wrong moderation semantics — fail loudly instead. + mods = [ModEvent(35, 100, -1)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "restart-improved", "source": "votes-csv", + "cuts": _MOD_CUTS, "moderation": "interleave-by-timestamp", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 0, + }) + with pytest.raises(NotImplementedError, match="clojure-legacy"): + _run_legacy(ds, spec) # improved mode: no env override set + + +@pytest.mark.parametrize("bad", [-1, 3, 4]) +def test_restart_after_out_of_range_raises(monkeypatch, bad): + # replay.clj CLI parity: restart_after must be a step index with at least + # one step after it (0 <= r <= n_steps-2); 4 cuts -> valid r in [0, 2]. + monkeypatch.setenv("POLISMATH_ENGINE_MODE", "clojure-legacy") + ds = ReplayDataset.build(_RESTART_RAW_VOTES) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "restart-range", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [5, 10, 15, 20]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": bad, + }) + with pytest.raises(ValueError, match="restart_after"): + _run_legacy(ds, spec) diff --git a/delphi/tests/replay_harness/test_real_data_mod_events.py b/delphi/tests/replay_harness/test_real_data_mod_events.py new file mode 100644 index 0000000000..4cacf44d27 --- /dev/null +++ b/delphi/tests/replay_harness/test_real_data_mod_events.py @@ -0,0 +1,174 @@ +"""``real_data.load_export_votes`` builds ``dataset.mod_events`` from the +comments CSV when it carries the moderation-history columns +(MOD_RESTART_PORT_SPEC.md "Python ports" item 3: modified->t_ms, +comment-id->tid, moderated->mod, is-meta->is_meta). + +Synthetic fixtures only, written under ``tmp_path`` with ``REAL_DATA_ROOT`` +monkeypatched — never touches ``real_data/.local``. +""" + +from __future__ import annotations + +import csv + +import pytest + +from polismath.replay import real_data as rd + +_VOTES_HEADER = ["timestamp", "datetime", "comment-id", "voter-id", "vote"] + +# New-format header: existing columns unchanged, is-meta/modified ADDITIVE +# at the end (mirrors the prodclone extractor's planned column order). +_COMMENTS_HEADER_NEW = [ + "timestamp", "datetime", "comment-id", "author-id", + "agrees", "disagrees", "moderated", "comment-body", + "is-meta", "modified", +] +_COMMENTS_HEADER_LEGACY = [ + "timestamp", "datetime", "comment-id", "author-id", + "agrees", "disagrees", "moderated", "comment-body", +] + + +def _write_csv(path, header, rows) -> None: + with open(path, "w", newline="") as fh: + w = csv.writer(fh) + w.writerow(header) + w.writerows(rows) + + +@pytest.fixture() +def fake_root(tmp_path, monkeypatch): + monkeypatch.setattr(rd, "REAL_DATA_ROOT", tmp_path) + return tmp_path + + +def _seed_votes(d, slug: str) -> None: + _write_csv( + d / f"{slug}-votes.csv", _VOTES_HEADER, + [ + [100, "t1", 10, 1, 1], + [200, "t2", 11, 2, -1], + ], + ) + + +def test_mod_events_built_from_new_columns(fake_root): + slug = "modtest" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + _write_csv( + d / f"{slug}-comments.csv", _COMMENTS_HEADER_NEW, + [ + [100, "t1", 10, 1, 3, 1, -1, "", "False", 150], + [200, "t2", 11, 2, 1, 0, 1, "", "True", 250], + ], + ) + ds = rd.load_export_votes(slug) + assert len(ds.mod_events) == 2 + events = sorted(ds.mod_events, key=lambda m: m.t_ms) + assert (events[0].t_ms, events[0].tid, events[0].mod) == (150, 10, -1) + assert events[0].is_meta is False + assert (events[1].t_ms, events[1].tid, events[1].mod) == (250, 11, 1) + assert events[1].is_meta is True + assert ds.mod_events_skipped == 0 + + +def test_rows_without_modified_are_skipped_and_counted(fake_root): + slug = "modtest2" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + _write_csv( + d / f"{slug}-comments.csv", _COMMENTS_HEADER_NEW, + [ + [100, "t1", 10, 1, 3, 1, -1, "", "False", 150], + [200, "t2", 11, 2, 1, 0, 1, "", "False", ""], # no modified -> skip + ], + ) + ds = rd.load_export_votes(slug) + assert len(ds.mod_events) == 1 + assert ds.mod_events[0].tid == 10 + assert ds.mod_events_skipped == 1 + + +def test_legacy_comments_csv_without_new_columns_yields_no_mod_events(fake_root): + # Pre-existing comments CSVs (moderated but no modified/is-meta columns) + # must not error, and must not fabricate mod events out of the existing + # "moderated" column alone — nothing to interleave on. + slug = "modtest3" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + _write_csv( + d / f"{slug}-comments.csv", _COMMENTS_HEADER_LEGACY, + [[100, "t1", 10, 1, 3, 1, -1, ""]], + ) + ds = rd.load_export_votes(slug) + assert ds.mod_events == [] + assert ds.mod_events_skipped == 0 + + +def test_no_comments_csv_yields_no_mod_events(fake_root): + slug = "modtest4" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + ds = rd.load_export_votes(slug) + assert ds.mod_events == [] + assert ds.mod_events_skipped == 0 + + +def test_mod_events_sorted_by_t_ms_regardless_of_row_order(fake_root): + slug = "modtest5" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + _write_csv( + d / f"{slug}-comments.csv", _COMMENTS_HEADER_NEW, + [ + [200, "t2", 11, 2, 1, 0, 1, "", "False", 999], + [100, "t1", 10, 1, 3, 1, -1, "", "False", 111], + ], + ) + ds = rd.load_export_votes(slug) + assert [m.t_ms for m in ds.mod_events] == [111, 999] + + +def test_modified_without_moderated_column_yields_no_events(fake_root): + # #2656 review finding 3: a malformed CSV carrying "modified" but missing + # the moderated column must take the graceful no-mod-events path the + # docstring promises, not crash with a KeyError mid-row. + slug = "modtest7" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + header = ["timestamp", "datetime", "comment-id", "author-id", "comment-body", "modified"] + _write_csv(d / f"{slug}-comments.csv", header, [[100, "t1", 10, 1, "", 150]]) + ds = rd.load_export_votes(slug) + assert ds.mod_events == [] + assert ds.mod_events_skipped == 0 + + +def test_modified_without_is_meta_column_yields_events_meta_false(fake_root): + # #2656 review finding 3: clj's mod-event reader keys ONLY on "modified" + # (is-meta optional -> false). A comments CSV carrying modified but not + # is-meta must still yield mod events, with is_meta defaulting to False. + slug = "modtest6" + d = fake_root / f"rFAKE-{slug}" + d.mkdir() + _seed_votes(d, slug) + _write_csv( + d / f"{slug}-comments.csv", _COMMENTS_HEADER_LEGACY + ["modified"], + [ + [100, "t1", 10, 1, 3, 1, -1, "", 150], + [200, "t2", 11, 2, 1, 0, 1, "", 250], + ], + ) + ds = rd.load_export_votes(slug) + assert [(m.t_ms, m.tid, m.mod) for m in ds.mod_events] == [ + (150, 10, -1), (250, 11, 1), + ] + assert all(m.is_meta is False for m in ds.mod_events) + assert ds.mod_events_skipped == 0 diff --git a/delphi/tests/replay_harness/test_schedule.py b/delphi/tests/replay_harness/test_schedule.py index dc35a4304b..37970a4bff 100644 --- a/delphi/tests/replay_harness/test_schedule.py +++ b/delphi/tests/replay_harness/test_schedule.py @@ -262,6 +262,101 @@ def test_per_day_preset_from_real_timestamps(): assert [len(s.vote_events) for s in steps] == [2, 2, 2] +# -------------------------------------------------------------------------- +# is_meta plumbing (MOD_RESTART_PORT_SPEC.md "Python ports" item 2). +# -------------------------------------------------------------------------- +def test_mod_event_is_meta_defaults_false(): + m = ModEvent(t_ms=1, tid=2, mod=0) + assert m.is_meta is False + + +def test_mod_event_is_meta_explicit_true(): + m = ModEvent(t_ms=1, tid=2, mod=0, is_meta=True) + assert m.is_meta is True + + +def test_explicit_mod_list_parses_is_meta_key(ds8): + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "s", + "cuts": {"mode": "vote-count", "at": [2, 5, "end"]}, + "moderation": [{"t_ms": 250, "tid": 10, "mod": -1, "is_meta": True}], + }) + steps = sched.slice_schedule(ds8, spec) + mods = [m for s in steps for m in s.mod_events] + assert len(mods) == 1 + assert mods[0].is_meta is True + + +def test_explicit_mod_list_defaults_is_meta_false_when_absent(ds8): + # Backward compat: dict rows written before is_meta existed must still + # parse (missing key -> False, not a KeyError). + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "s", + "cuts": {"mode": "vote-count", "at": [2, 5, "end"]}, + "moderation": [{"t_ms": 250, "tid": 10, "mod": -1}], + }) + steps = sched.slice_schedule(ds8, spec) + mods = [m for s in steps for m in s.mod_events] + assert len(mods) == 1 + assert mods[0].is_meta is False + + +def test_explicit_mod_list_passthrough_of_existing_modevent_keeps_is_meta(ds8): + # A pre-built ModEvent in the list (not a dict) passes through verbatim. + spec = sched.ScheduleSpec.from_dict({ + "dataset": "t", "schedule_id": "s", + "cuts": {"mode": "vote-count", "at": [2, 5, "end"]}, + "moderation": [ModEvent(t_ms=250, tid=10, mod=-1, is_meta=True)], + }) + steps = sched.slice_schedule(ds8, spec) + mods = [m for s in steps for m in s.mod_events] + assert mods[0].is_meta is True + + +# -------------------------------------------------------------------------- +# restart_after plumbing (MOD_RESTART_PORT_SPEC.md restart-seam schedule field). +# -------------------------------------------------------------------------- +def test_restart_after_parses_from_dict(): + spec = sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "s", + "cuts": {"mode": "vote-count", "at": [4]}, + "restart_after": 4, + }) + assert spec.restart_after == 4 + + +def test_restart_after_defaults_to_none_when_absent(): + spec = sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "s", + "cuts": {"mode": "vote-count", "at": [4]}, + }) + assert spec.restart_after is None + + +def test_restart_after_round_trips_verbatim(): + d = { + "dataset": "vw", "schedule_id": "s", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [4]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 3, + } + spec = sched.ScheduleSpec.from_dict(d) + assert spec.to_dict() == d + + +def test_restart_after_included_when_constructed_directly(): + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": [1]}, + restart_after=2, + ) + assert spec.to_dict()["restart_after"] == 2 + + +def test_restart_after_none_by_default_when_constructed_directly(): + spec = sched.ScheduleSpec(dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": [1]}) + assert spec.restart_after is None + assert spec.to_dict()["restart_after"] is None + + # -------------------------------------------------------------------------- # ScheduleSpec JSON round-trip (verbatim). # -------------------------------------------------------------------------- diff --git a/delphi/tests/test_legacy_blob_shape.py b/delphi/tests/test_legacy_blob_shape.py index bdb773bb2b..7210fafda5 100644 --- a/delphi/tests/test_legacy_blob_shape.py +++ b/delphi/tests/test_legacy_blob_shape.py @@ -30,6 +30,8 @@ from __future__ import annotations +import json + import numpy as np import pytest @@ -430,6 +432,66 @@ def test_legacy_from_dict_restores_arrival_order(conv, legacy): assert restored.tid_arrival_order == conv.tid_arrival_order +# --------------------------------------------------------------------------- +# from_dict warm-restart restore: base clusters + zid (restart-seam root, +# journal 2026-07-24: from_dict restored ZERO base_clusters and zid '' from a +# recorded step blob, so the recovery tick cold-started the base-cluster +# lineage and re-minted every id). Mirrors Clojure restructure-json-conv +# (conv_man.clj:171-186): keep :zid, unfold :base-clusters (clusters.clj: +# 402-414, center := [x y]). +# --------------------------------------------------------------------------- +def _assert_base_clusters_round_trip(conv, restored): + assert [c["id"] for c in restored.base_clusters] == \ + [c["id"] for c in conv.base_clusters] + assert [c["members"] for c in restored.base_clusters] == \ + [c["members"] for c in conv.base_clusters] + # Centers come back in the INTERNAL sign convention: legacy emission + # negates x/y at the blob boundary and the restore un-negates (double + # negation is exact in IEEE); improved emission is verbatim. + assert [c["center"] for c in restored.base_clusters] == \ + [list(c["center"][:2]) for c in conv.base_clusters] + + +def test_legacy_from_dict_restores_base_clusters_and_zid(conv, legacy): + restored = Conversation.from_dict(conv.to_dict()) + assert restored.conversation_id == "legacy_blob_shape" + _assert_base_clusters_round_trip(conv, restored) + + +def test_improved_from_dict_restores_base_clusters_and_zid(conv, improved): + restored = Conversation.from_dict(conv.to_dict()) + assert restored.conversation_id == "legacy_blob_shape" + _assert_base_clusters_round_trip(conv, restored) + + +def test_from_dict_preserves_falsy_conversation_id(): + # #2656 review finding 2: `data.get('conversation_id') or data.get('zid')` + # would discard a legitimately-falsy id (e.g. 0) — the key-presence check + # must win, not truthiness. (Real to_dict blobs always carry 'zid'; this + # pins the synthetic/hand-built-blob path.) + restored = Conversation.from_dict({"conversation_id": 0}) + assert restored.conversation_id == 0 + + +def test_legacy_from_dict_restores_group_votes_for_prev_tick_priorities(conv, legacy): + # restructure-json-conv keeps :group-votes (conv_man.clj:174) and the + # recovery tick's comment-priorities read it as the PREVIOUS tick's + # group-votes (Q2, conversation.clj:658) — without the restore, a warm + # restart computes priorities against empty prev group-votes (every + # comment looks unseen → inflated priorities; vw-restart4 step-5 + # divergence, journal 2026-07-24). Round-trip through JSON like a + # recorded blob: tid keys stringify and must come back as ints + # (parse-blob-json numeric-string→long parity). + blob = json.loads(json.dumps(conv.to_dict())) + restored = Conversation.from_dict(blob) + assert restored.group_votes, "group-votes must survive the restore" + assert set(restored.group_votes.keys()) == set(blob["group-votes"].keys()) + for gid, g in blob["group-votes"].items(): + rg = restored.group_votes[gid] + assert rg["n-members"] == g["n-members"] + assert rg["votes"] == {int(t): e for t, e in g["votes"].items()} + + def test_conv_repness_tie_break_follows_tid_order(): """Two comments with IDENTICAL vote patterns tie on every repness stat; Clojure's stable sort keeps them in column (arrival) order. With @@ -571,3 +633,78 @@ def test_improved_from_dict_round_trips_center_sign(conv, improved): np.testing.assert_allclose( np.asarray(restored.pca["center"]), np.asarray(conv.pca["center"]) ) + + +# --------------------------------------------------------------------------- +# Tiny SHAPES beyond 1x1 (review finding on #2653): the relaxed small-dim +# guards cover any `rows < 2 OR cols < 2` matrix. Expectations are REAL +# Clojure outputs (Q14): +# 1xN — vw every-vote-56 clj recording step-002 (public data: pid 1's first +# three AGREEs on tids 24/19/47; recorded with the Q12 pinned start); +# Nx1 — a synthetic 3-ptpt x 1-comment fixture run through the clj replay +# driver 2026-07-22 s4 (votes +1/+1/-1 on tid 0; same pinned start). +# --------------------------------------------------------------------------- +def _pinned_conv(name): + c = Conversation(name) + # The replay drivers' Q12 carve-out: cold-tick PCA start pinned to ones. + c.pca = {"center": np.zeros(1), "comps": np.array([[1.0], [1.0]])} + return c + + +def test_legacy_one_by_n_matches_clojure_recording(legacy): + c = _pinned_conv("tiny_1x3") + c = c.update_votes( + {"votes": [{"pid": 1, "tid": 24, "vote": 1}, + {"pid": 1, "tid": 19, "vote": 1}, + {"pid": 1, "tid": 47, "vote": 1}]}, + recompute=False, + ) + d = c.recompute().to_dict() + assert d["pca"]["center"] == [-1.0, -1.0, -1.0] + assert d["pca"]["comps"] == [[0.0, 0.0, 0.0]] + assert len(d["pca"]["comment-projection"]) == 2 + assert d["pca"]["comment-extremity"] == [0.0, 0.0, 0.0] + rep = d["repness"] + (gid,) = rep.keys() + (entry,) = rep[gid] + assert entry["tid"] == 24 and entry["best-agree"] is True + assert entry["p-success"] == pytest.approx(2 / 3) + agree = d["consensus"]["agree"] + assert [e["tid"] for e in agree] == [24, 19, 47] + for e in agree: + assert e["n-trials"] == 1 + assert e["p-success"] == pytest.approx(2 / 3) + assert e["p-test"] == pytest.approx(1.4142135623730951) + assert d["consensus"]["disagree"] == [] + + +def test_legacy_n_by_one_matches_clojure_reference(legacy): + c = _pinned_conv("tiny_3x1") + c = c.update_votes( + {"votes": [{"pid": 10, "tid": 0, "vote": 1}, + {"pid": 11, "tid": 0, "vote": 1}, + {"pid": 12, "tid": 0, "vote": -1}]}, + recompute=False, + ) + d = c.recompute().to_dict() + assert d["pca"]["center"] == pytest.approx([-1 / 3]) + assert d["pca"]["comps"] == [[1.0]] + assert d["pca"]["comment-projection"] == [[0.0], [0.0]] + assert d["pca"]["comment-extremity"] == [0.0] + rep = d["repness"] + (gid,) = rep.keys() + (entry,) = rep[gid] + assert entry["tid"] == 0 and entry["repful-for"] == "agree" + assert entry["n-success"] == 2 and entry["n-trials"] == 3 + assert entry["p-success"] == pytest.approx(0.6) + assert entry["repness"] == pytest.approx(1.2) + assert entry["best-agree"] is True + assert d["consensus"] == {"agree": [], "disagree": []} + assert set(d["user-vote-counts"]) == {10, 11, 12} or set( + d["user-vote-counts"] + ) == {"10", "11", "12"} + bc = d["base-clusters"] + assert bc["members"] == [[10, 11, 12]] + # Q16 collapse: all-zero projections -> single coincident base cluster + assert bc["x"] == [0.0] and bc["y"] == [0.0] + assert d["comment_priorities"] == {0: 5.0625} diff --git a/delphi/tests/test_legacy_kmeans.py b/delphi/tests/test_legacy_kmeans.py index 917a041f80..51331d0cf0 100644 --- a/delphi/tests/test_legacy_kmeans.py +++ b/delphi/tests/test_legacy_kmeans.py @@ -138,6 +138,57 @@ def test_weighted_recentering(self): np.testing.assert_allclose(stepped[0]['center'], [0.0, 1.5]) +class TestClusterStepHashOrderTieBreak: + """Clojure's cluster-step iterates the cleared-clusters map: ``(into {})`` + of ``[id cluster]`` pairs is an array-map in INSERTION (input) order for + <=8 clusters but a PersistentHashMap for >8, whose seq order is the HAMT + trie order of the id hashes (clusters.clj:79-86, 149). add-to-closest's + min-key keeps the LAST minimal entry in THAT order, so the scan order is + semantic exactly on distance ties — which the Q11 cancellation floor + makes COMMON, not measure-zero (pc-modheavy-01 step 2: 12 seed clusters + emptied clj-side by hash-order ties, 80 vs 92 recorded clusters; journal + 2026-07-24). + + Ground truth from real Clojure (clojure -M eval, 2026-07-24): + (keys (into {} (map (juxt identity identity) (range 9)))) + => (0 7 1 4 6 3 2 5 8) ; ids 1 and 7 INVERT input order + (range 8) stays (0 1 2 3 4 5 6 7) ; array-map, insertion order + polismath.utils.clj_hash.clojure_hash_map_key_order reproduces the n=9 + and n=20 orders bit-for-bit (cross-validated same session).""" + + @staticmethod + def _tie_fixture(n_ids): + # Row 't' ties at distance 0.0 between clusters 1 and 7 (both centers + # exactly its position); every other cluster holds its own coincident + # row so nothing else moves or empties. + names, rows, clusters = [], [], [] + for i in range(n_ids): + if i in (1, 7): + center = [5.0, 5.0] + else: + center = [10.0 * i, -7.0] + names.append(f"p{i}") + rows.append(center) + clusters.append({'id': i, 'members': [], 'center': np.array(center)}) + names.append("t") + rows.append([5.0, 5.0]) + return _nd(names, rows), clusters + + def test_gt8_ties_resolve_by_clojure_hash_map_order(self): + data, clusters = self._tie_fixture(9) + by = _by_id(cluster_step(data, clusters)) + # hash order (0 7 1 4 6 3 2 5 8): id 1 comes AFTER id 7 -> 1 wins. + assert 't' in by[1]['members'] + assert 7 not in by # cluster 7 got no members -> dropped + + def test_le8_ties_resolve_by_input_order(self): + data, clusters = self._tie_fixture(8) + by = _by_id(cluster_step(data, clusters)) + # array-map insertion order == input order: id 7 is later -> 7 wins. + assert 't' in by[7]['members'] + assert 1 not in by + + # --------------------------------------------------------------------------- # safe_recenter_clusters (clusters.clj:171-191) — drop vanished # --------------------------------------------------------------------------- diff --git a/delphi/tests/test_mod_update_parity.py b/delphi/tests/test_mod_update_parity.py new file mode 100644 index 0000000000..d28edc7f7f --- /dev/null +++ b/delphi/tests/test_mod_update_parity.py @@ -0,0 +1,229 @@ +"""Clojure ``mod-update`` parity — ``Conversation.mod_update`` (MOD_RESTART_PORT_SPEC.md). + +Pins the conversation.clj:846-884 reducer semantics that the existing +``update_moderation`` (replace-only-when-truthy) cannot express: + +- un-moderation REMOVES a tid from a set (``disj``); +- ``is_meta`` rows land in BOTH mod-out and mod-in (and meta-tids); +- the reduce is order-sensitive within one batch (last row wins per tid); +- watermark = ``max(existing or 0, *modified)``; +- NO math recompute — sets and watermark only (Clojure's ``:moderation`` + message handler runs ``mod-update`` alone; math changes at the NEXT + votes recompute). + +RED observed 2026-07-22 (session 4): every test fails with AttributeError +(``mod_update`` does not exist); the semantic cases are also inexpressible +via ``update_moderation`` by construction (it never removes set members). +""" + +import numpy as np +import pytest + +from polismath.conversation.conversation import Conversation +from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR + + +@pytest.fixture +def legacy_mode(monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'clojure-legacy') + + +@pytest.fixture +def improved_mode(monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'improved') + + +def _conv(**sets): + conv = Conversation("mod-parity-probe", last_updated=1) + for attr, val in sets.items(): + setattr(conv, attr, set(val)) + return conv + + +def _row(tid, mod: int | None = 0, is_meta=False, modified=100): + return {"tid": tid, "is_meta": is_meta, "mod": mod, "modified": modified} + + +class TestReducerSemantics: + def test_mod_minus_one_conjs_mod_out_and_disjs_mod_in(self): + conv = _conv(mod_in_tids={4}) + result = conv.mod_update([_row(4, mod=-1)]) + assert 4 in result.mod_out_tids + assert 4 not in result.mod_in_tids + assert 4 not in result.meta_tids + + def test_mod_plus_one_conjs_mod_in_and_disjs_mod_out(self): + conv = _conv(mod_out_tids={9}) + result = conv.mod_update([_row(9, mod=1)]) + assert 9 in result.mod_in_tids + assert 9 not in result.mod_out_tids + + def test_unmoderation_removes_from_both_sets(self): + # mod=0 (neither -1 nor 1) disjs from BOTH sets — the removal + # update_moderation cannot express. + conv = _conv(mod_out_tids={5}, mod_in_tids={5}) + result = conv.mod_update([_row(5, mod=0)]) + assert 5 not in result.mod_out_tids + assert 5 not in result.mod_in_tids + + def test_mod_none_behaves_as_disj(self): + # Clojure (= mod -1)/(= mod 1) is false for nil -> disj everywhere. + conv = _conv(mod_out_tids={2}, mod_in_tids={2}) + result = conv.mod_update([_row(2, mod=None)]) + assert 2 not in result.mod_out_tids + assert 2 not in result.mod_in_tids + + def test_is_meta_lands_in_both_mod_sets_and_meta(self): + conv = _conv() + result = conv.mod_update([_row(7, mod=0, is_meta=True)]) + assert 7 in result.mod_out_tids + assert 7 in result.mod_in_tids + assert 7 in result.meta_tids + + def test_meta_unset_disjs_meta_tids(self): + conv = _conv(meta_tids={3}) + result = conv.mod_update([_row(3, mod=1, is_meta=False)]) + assert 3 not in result.meta_tids + assert 3 in result.mod_in_tids + + def test_order_sensitive_last_row_wins(self): + conv = _conv() + fwd = conv.mod_update([_row(3, mod=-1), _row(3, mod=1)]) + assert 3 in fwd.mod_in_tids and 3 not in fwd.mod_out_tids + rev = conv.mod_update([_row(3, mod=1), _row(3, mod=-1)]) + assert 3 in rev.mod_out_tids and 3 not in rev.mod_in_tids + + +class TestWatermark: + def test_watermark_max_of_existing_and_rows(self): + conv = _conv() + conv.last_mod_timestamp = 500 + result = conv.mod_update([_row(1, modified=200), _row(2, modified=900)]) + assert result.last_mod_timestamp == 900 + + def test_watermark_never_regresses(self): + conv = _conv() + conv.last_mod_timestamp = 500 + result = conv.mod_update([_row(1, modified=200)]) + assert result.last_mod_timestamp == 500 + + def test_watermark_from_none_starts_at_zero_floor(self): + conv = _conv() + assert conv.last_mod_timestamp is None + result = conv.mod_update([_row(1, modified=250)]) + assert result.last_mod_timestamp == 250 + + def test_empty_mods_floors_none_watermark_at_zero(self): + # (apply max (or nil 0) '()) = 0 — Clojure's load-or-init calls + # mod-update with the (possibly empty) full mod history. + conv = _conv() + result = conv.mod_update([]) + assert result.last_mod_timestamp == 0 + + def test_empty_mods_preserves_existing_watermark(self): + conv = _conv() + conv.last_mod_timestamp = 42 + result = conv.mod_update([]) + assert result.last_mod_timestamp == 42 + + +class TestWatermarkDroppedByRecompute: + """Clojure's ``conv-update`` is a plumbing-graph compile whose output map + has ONLY graph-node keys — ``:last-mod-timestamp`` is not one + (conversation.clj:780-820), so EVERY votes tick drops the mod watermark; + it is blob-visible only on ticks whose last write was a mod-update. + Observed on the vw restart probe (2026-07-22 s4): restart mod-update set + the watermark, the next conv-update's blob emitted null.""" + + BATCH = { + "votes": [ + {"pid": 0, "tid": 0, "vote": 1, "created": 10}, + {"pid": 1, "tid": 0, "vote": -1, "created": 20}, + ], + "lastVoteTimestamp": 20, + } + + def test_votes_recompute_drops_watermark_in_legacy_mode(self, legacy_mode): + conv = Conversation("wm-drop", last_updated=1) + conv = conv.mod_update([_row(0, mod=-1, modified=777)]) + assert conv.last_mod_timestamp == 777 + conv2 = conv.update_votes(dict(self.BATCH), recompute=True) + assert conv2.last_mod_timestamp is None + + def test_watermark_persists_in_improved_mode(self, improved_mode): + # Documented divergence: improved mode keeps the sane persistent + # watermark instead of Clojure's graph-drop. + conv = Conversation("wm-keep", last_updated=1) + conv = conv.mod_update([_row(0, mod=-1, modified=777)]) + conv2 = conv.update_votes(dict(self.BATCH), recompute=True) + assert conv2.last_mod_timestamp == 777 + + +class TestGroupVotesTallyRawMatrix: + """Clojure's group-votes aggregates votes-base, whose fnk reads + RAW-rating-mat (conversation.clj:601-608): moderated-out comments report + the ACTUAL votes cast (and true seen-counts), not the post-zeroing + pass-shaped columns. Found on pc-meta-01 step 1 (2026-07-22 s4): python + tallied the zeroed rating_mat -> A=0/D=0 with S inflated to every member + ("everyone passed"), where Clojure reports the real A/D/S.""" + + @staticmethod + def _moderated_conv(): + """4 ptpts; tid 0 gets A=3/D=1 then is moderated OUT; the next votes + tick applies the moderation (zeroed rating_mat) and recomputes.""" + conv = Conversation("gv-raw", last_updated=1) + votes = [] + for pid, (v0, v1) in enumerate([(1, 1), (1, -1), (-1, -1), (1, 1)]): + votes.append({"pid": pid, "tid": 0, "vote": v0, "created": 10 + pid}) + votes.append({"pid": pid, "tid": 1, "vote": v1, "created": 20 + pid}) + conv = conv.update_votes({"votes": votes, "lastVoteTimestamp": 30}, + recompute=False) + conv = conv.recompute() + conv = conv.mod_update([_row(0, mod=-1, modified=100)]) + return conv.update_votes( + {"votes": [{"pid": 0, "tid": 1, "vote": 1, "created": 40}], + "lastVoteTimestamp": 40}, + recompute=True, + ) + + def test_legacy_group_votes_report_actual_votes_for_moderated_tid(self, legacy_mode): + # group-votes must still tally tid 0's REAL votes post-moderation. + gv = self._moderated_conv().group_votes + assert gv, "expected at least one group" + tot_a = sum(g["votes"][0]["A"] for g in gv.values()) + tot_d = sum(g["votes"][0]["D"] for g in gv.values()) + tot_s = sum(g["votes"][0]["S"] for g in gv.values()) + assert (tot_a, tot_d) == (3, 1) + assert tot_s == 4 + + def test_legacy_to_dynamo_dict_group_votes_tally_raw_matrix(self, legacy_mode): + # #2656 review finding 4: the THIRD inline group-votes tally + # (to_dynamo_dict) must obey the same raw-matrix rule as + # _compute_group_votes and to_dict — not the zeroed rating_mat. + dyn = self._moderated_conv().to_dynamo_dict() + gv = dyn["group_votes"] + assert gv, "expected at least one group" + tot_a = sum(g["votes"][0]["agree"] for g in gv.values()) + tot_d = sum(g["votes"][0]["disagree"] for g in gv.values()) + tot_s = sum(g["votes"][0]["total"] for g in gv.values()) + assert (tot_a, tot_d) == (3, 1) + assert tot_s == 4 + + +class TestNoRecomputeAndImmutability: + def test_math_state_untouched(self): + conv = _conv() + sentinel_pca = {"center": np.array([0.5]), "comps": np.array([[1.0], [0.0]])} + conv.pca = sentinel_pca + conv.base_clusters = [{"id": 0, "members": [1], "center": [0.0, 0.0]}] + result = conv.mod_update([_row(6, mod=-1)]) + assert result.base_clusters == conv.base_clusters + assert result.pca is not None + np.testing.assert_array_equal(result.pca["center"], sentinel_pca["center"]) + + def test_returns_new_conversation_original_untouched(self): + conv = _conv(mod_out_tids={8}) + result = conv.mod_update([_row(8, mod=0)]) + assert result is not conv + assert 8 in conv.mod_out_tids + assert 8 not in result.mod_out_tids diff --git a/delphi/tests/test_prodclone_extract.py b/delphi/tests/test_prodclone_extract.py index 1484abb27d..fb7d4fc7b0 100644 --- a/delphi/tests/test_prodclone_extract.py +++ b/delphi/tests/test_prodclone_extract.py @@ -343,6 +343,7 @@ def test_format_comments_rows_redacts_text(): assert set(row) == { "timestamp", "datetime", "comment-id", "author-id", "agrees", "disagrees", "moderated", "comment-body", + "is-meta", "modified", } @@ -363,6 +364,37 @@ def test_format_comments_rows_default_zero_votes(): assert row["moderated"] == "-1" +# --------------------------------------------------------------------------- +# is-meta / modified columns (MOD_RESTART_PORT_SPEC.md "Data" bullet): +# additive, after the existing columns; comment-body stays EMPTY regardless. +# --------------------------------------------------------------------------- +def test_format_comments_rows_includes_is_meta_and_modified(): + raw = [{"tid": 5, "pid": 2, "created": 0, "mod": -1, "is_meta": True, "modified": 12345}] + rows = pc.format_comments_rows(raw, vote_counts={}) + row = rows[0] + assert row["is-meta"] == "True" + assert row["modified"] == "12345" + + +def test_format_comments_rows_defaults_is_meta_false_and_modified_empty_when_absent(): + # Tolerates raw rows that don't carry the new keys at all (defensive; + # every SQL-fetched row will, post this port, but the formatter itself + # stays permissive). + raw = [{"tid": 5, "pid": 2, "created": 0, "mod": -1}] + rows = pc.format_comments_rows(raw, vote_counts={}) + row = rows[0] + assert row["is-meta"] == "False" + assert row["modified"] == "" + + +def test_format_comments_rows_modified_none_becomes_empty_string(): + # comments.modified is nullable in the DB (schema permits NULL even + # though it defaults to now_as_millis()) -> empty string, not "None". + raw = [{"tid": 5, "pid": 2, "created": 0, "mod": -1, "is_meta": False, "modified": None}] + rows = pc.format_comments_rows(raw, vote_counts={}) + assert rows[0]["modified"] == "" + + # --------------------------------------------------------------------------- # CSV writers — round-trip through csv.DictReader # --------------------------------------------------------------------------- @@ -386,7 +418,8 @@ def test_write_votes_csv_round_trips(tmp_path): def test_write_comments_csv_round_trips(tmp_path): - raw = [{"tid": 1, "pid": 1, "created": 1_700_000_000_000, "mod": 1}] + raw = [{"tid": 1, "pid": 1, "created": 1_700_000_000_000, "mod": 1, + "is_meta": False, "modified": 1_700_000_000_500}] path = tmp_path / "comments.csv" pc.write_comments_csv(path, pc.format_comments_rows(raw, vote_counts={1: (2, 1)})) with open(path, newline="") as fh: @@ -394,10 +427,13 @@ def test_write_comments_csv_round_trips(tmp_path): assert reader.fieldnames == [ "timestamp", "datetime", "comment-id", "author-id", "agrees", "disagrees", "moderated", "comment-body", + "is-meta", "modified", ] got = list(reader) assert got[0]["comment-body"] == "" assert got[0]["agrees"] == "2" + assert got[0]["is-meta"] == "False" + assert got[0]["modified"] == "1700000000500" # --------------------------------------------------------------------------- @@ -544,6 +580,13 @@ def test_sql_comments_export_has_zid_placeholder(): assert "%s" in sql +def test_sql_comments_export_selects_is_meta_and_modified(): + sql = pc.sql_comments_export() + lowered = sql.lower() + assert "is_meta" in lowered + assert "modified" in lowered + + def test_sql_comment_vote_counts_has_zid_placeholder(): sql = pc.sql_comment_vote_counts() assert "%s" in sql diff --git a/math/dev/proj_probe.clj b/math/dev/proj_probe.clj index 28c1d6d8ef..620259fe3e 100644 --- a/math/dev/proj_probe.clj +++ b/math/dev/proj_probe.clj @@ -156,3 +156,194 @@ (doseq [c (sort-by :id (:base-clusters cur)) :when (> (count (:members c)) 1)] (println " multi-member cluster id=" (:id c) "members=" (pr-str (:members c)))))) + +;; Mod-weaving distinct-rows probe (pc-modheavy-01 step-2 fork, journal +;; 2026-07-22 s4 What's Next #1): replay an N-cut prefix of a mod-interleave +;; schedule WITH woven moderation (replay's own read-mod-events + +;; slice-schedule; meta-tids empty — interleave schedules take meta via +;; mod-update only, as in replay/-main), then report, at the FINAL step, the +;; in-conv projection-row distinct count and every group of pids whose rows +;; are EQUAL in clj at %.17g — to diff against the python side (py: 92 +;; distinct of 105 at step 2; 12 row-pairs collide in clj only). +(defn mod-distinct-probe [votes-csv comments-csv zid & cuts] + (let [votes (->> (replay/read-votes-csv votes-csv) replay/build-dataset) + {mods :events} (replay/read-mod-events comments-csv) + slots (mapv long cuts) + steps (replay/slice-schedule votes slots mods) + results (replay/run-once zid #{} steps) + [_ conv'] (last results) + pnmat (nm/named-matrix (nm/rownames (:rating-mat conv')) ["x" "y"] + (:proj conv')) + inmat (nm/rowname-subset pnmat (:in-conv conv')) + names (nm/rownames inmat) + raw-rows (matrix/rows (nm/get-matrix inmat)) + rows (mapv #(into [] %) raw-rows)] + (println "MODPROBE final-step: in-conv rows=" (count rows) + "distinct(vectorz)=" (count (distinct (into [] raw-rows))) + "distinct(vec)=" (count (distinct rows))) + (doseq [[row prs] (->> (group-by second (map vector names rows)) + (filter (fn [[_ prs]] (> (count prs) 1))) + (sort-by (fn [[_ prs]] (long (ffirst prs)))))] + (println (format "COLLIDE pids=%s row=[%.17g %.17g]" + (pr-str (mapv first prs)) + (double (nth row 0)) (double (nth row 1))))))) + +;; Mod-weaving split-loop walk (pc-modheavy-01 step-2: clj records 80 base +;; clusters vs py 92 while BOTH see 92 distinct in-conv rows — so the clj +;; split loop stops early; this prints WHY). Replays an N-cut prefix with +;; woven mods, then at the FINAL step: runs the REAL clean-start-clusters +;; (count check), then mirrors clusters.clj:250-273 manually printing each +;; iteration's most-distal extraction (id/dist/clst-id at %.20g) up to the +;; stop, plus the remaining multi-member clusters at the stop. +(defn mod-split-probe [votes-csv comments-csv zid & cuts] + (let [votes (->> (replay/read-votes-csv votes-csv) replay/build-dataset) + {mods :events} (replay/read-mod-events comments-csv) + slots (mapv long cuts) + steps (replay/slice-schedule votes slots mods) + results (replay/run-once zid #{} steps) + [_ prev-conv] (nth results (- (count results) 2)) + [_ cur-conv] (last results) + pnmat (nm/named-matrix (nm/rownames (:rating-mat cur-conv)) ["x" "y"] + (:proj cur-conv)) + inmat (nm/rowname-subset pnmat (:in-conv cur-conv)) + prev-bc (:base-clusters prev-conv) + real-cs (clusters/clean-start-clusters inmat prev-bc 100) + rec (clusters/safe-recenter-clusters inmat prev-bc) + uniq (clusters/uniqify-clusters rec) + possible (min 100 (count (distinct (into [] (matrix/rows (nm/get-matrix inmat))))))] + (println "MODSPLIT prev-step clusters:" (count prev-bc) + "safe-recenter:" (count rec) "uniqify:" (count uniq) + "possible:" possible "rows:" (count (nm/rownames inmat)) + "REAL clean-start-clusters:" (count real-cs)) + (loop [clusters uniq, it 0] + (let [clusters (clusters/recenter-clusters inmat clusters)] + (if (> possible (count clusters)) + (let [outlier (clusters/most-distal inmat clusters)] + (println (format "MODSPLIT iter %d: n=%d extract pid=%s d=%.20g clst=%s" + it (count clusters) (str (:id outlier)) + (double (:dist outlier)) (str (:clst-id outlier)))) + (if (> (:dist outlier) 0) + (recur + (-> + (mapv + (fn [clst] + (assoc clst :members + (remove (set [(:id outlier)]) (:members clst)))) + clusters) + (conj {:id (inc (apply max (map :id clusters))) + :members [(:id outlier)] + :center (nm/get-row-by-name inmat (:id outlier))})) + (inc it)) + (do + (println "MODSPLIT STOPPED (zero-dist outlier) at n=" (count clusters)) + (doseq [c clusters + :when (> (count (:members c)) 1)] + (println " multi-member id=" (:id c) "members=" (pr-str (:members c))))))) + (println "MODSPLIT done (possible reached) n=" (count clusters))))))) + +;; Step-1 lineage probe (pc-modheavy-01 {1,3,8,11} id 2-vs-8): replay an +;; N-cut prefix with woven mods, then at the FINAL step print the REAL +;; clean-start seed clusters holding the tracked pids (centers %.17g), the +;; distances of each tracked row to the tracked cluster ids under +;; matrix/distance (the add-to-closest path), and the final kmeans outcome +;; for those pids — to pin WHERE clj's id survives vs the py port. +(defn lineage-probe [votes-csv comments-csv zid track-pids track-ids & cuts] + (let [votes (->> (replay/read-votes-csv votes-csv) replay/build-dataset) + {mods :events} (replay/read-mod-events comments-csv) + slots (mapv long cuts) + steps (replay/slice-schedule votes slots mods) + results (replay/run-once zid #{} steps) + [_ prev-conv] (nth results (- (count results) 2)) + [_ cur-conv] (last results) + pnmat (nm/named-matrix (nm/rownames (:rating-mat cur-conv)) ["x" "y"] + (:proj cur-conv)) + inmat (nm/rowname-subset pnmat (:in-conv cur-conv)) + prev-bc (:base-clusters prev-conv) + track-pids (set track-pids) + track-ids (set track-ids) + seed (clusters/clean-start-clusters inmat prev-bc 100)] + (println "LINEAGE prev-step ids holding tracked pids:") + (doseq [c prev-bc :when (seq (clojure.set/intersection track-pids (set (:members c))))] + (println (format " prev id=%d members=%s center=[%.17g %.17g]" + (long (:id c)) (pr-str (:members c)) + (double (first (:center c))) (double (second (:center c)))))) + (println "LINEAGE seed clusters holding tracked pids or ids:") + (doseq [c seed :when (or (seq (clojure.set/intersection track-pids (set (:members c)))) + (contains? track-ids (:id c)))] + (println (format " seed id=%d members=%s center=[%.17g %.17g]" + (long (:id c)) (pr-str (:members c)) + (double (first (:center c))) (double (second (:center c)))))) + (doseq [p track-pids] + (let [row (nm/get-row-by-name inmat p)] + (doseq [c seed :when (contains? track-ids (:id c))] + (println (format " d(row%s, c%d) = %.20g" + (str p) (long (:id c)) + (double (matrix/distance row (:center c)))))))) + (let [km (clusters/kmeans inmat 100 + :last-clusters prev-bc + :max-iters 100)] + (println "LINEAGE final kmeans clusters holding tracked pids:") + (doseq [c (sort-by :id km) + :when (seq (clojure.set/intersection track-pids (set (:members c))))] + (println (format " final id=%d members=%s" + (long (:id c)) (pr-str (:members c)))))))) + +;; Split-loop probe (pc-revote-01 step-1 extraction tie): replay two vote-count +;; batches like batch-probe, then walk clean-start-clusters' split loop +;; MANUALLY (mirroring clusters.clj:250-273 verbatim) printing, per iteration, +;; the ACTUAL most-distal extraction (id/dist/clst-id) plus the top-3 candidate +;; ranking with runner-up gaps, so the sequence can be diffed against the +;; python probe (delphi/scratch/probe_revote_split.py). +(defn split-probe [csv-path cut1 cut2] + (let [votes (->> (replay/read-votes-csv csv-path) replay/build-dataset) + b1 (subvec votes 0 cut1) + b2 (subvec votes cut1 cut2) + seed (-> (conv/new-conv) + (assoc :zid 99998 :meta-tids #{} + :pca replay/certify-cold-start-pca)) + prev (conv/conv-update seed (replay/->conv-votes b1) + replay/certify-conv-opts) + cur (conv/conv-update prev (replay/->conv-votes b2) + replay/certify-conv-opts) + pnmat (nm/named-matrix (nm/rownames (:rating-mat cur)) ["x" "y"] + (:proj cur)) + inmat (nm/rowname-subset pnmat (:in-conv cur)) + rec (clusters/safe-recenter-clusters inmat (:base-clusters prev)) + uniq (clusters/uniqify-clusters rec) + possible (min 100 (count (distinct (into [] (matrix/rows (nm/get-matrix inmat))))))] + (println "SPLIT start-clusters:" (count uniq) "possible:" possible + "rows:" (count (nm/rownames inmat))) + (loop [clusters uniq, it 0] + (let [clusters (clusters/recenter-clusters inmat clusters)] + (if (> possible (count clusters)) + (let [outlier (clusters/most-distal inmat clusters) + ranks (->> (nm/rownames inmat) + (map (fn [mem] + (let [row (nm/get-row-by-name inmat mem)] + [(apply min (map #(matrix/distance row (:center %)) + clusters)) + mem]))) + (sort-by first) + reverse + (take 3)) + [[d0 m0] [d1 m1] [d2 m2]] ranks] + (println (format "SPLIT iter %d: extract pid=%s d=%.20g clst=%s | top3 %s:%.20g %s:%.20g %s:%.20g | gap01=%.3e" + it (str (:id outlier)) (double (:dist outlier)) + (str (:clst-id outlier)) + (str m0) (double d0) (str m1) (double d1) + (str m2) (double d2) + (double (- d0 d1)))) + (if (> (:dist outlier) 0) + (recur + (-> + (mapv + (fn [clst] + (assoc clst :members + (remove (set [(:id outlier)]) (:members clst)))) + clusters) + (conj {:id (inc (apply max (map :id clusters))) + :members [(:id outlier)] + :center (nm/get-row-by-name inmat (:id outlier))})) + (inc it)) + (println "SPLIT done (zero-dist outlier) after iter" it))) + (println "SPLIT done (possible reached) n=" (count clusters))))))) diff --git a/math/dev/replay.clj b/math/dev/replay.clj index 9b1dd515ca..d9a950f30d 100644 --- a/math/dev/replay.clj +++ b/math/dev/replay.clj @@ -65,6 +65,7 @@ [com.stuartsierra.component :as component] [clojure.core.matrix :as matrix] [polismath.math.conversation :as conv] + [polismath.math.named-matrix :as nm] [polismath.conv-man :as cm] [polismath.components.core-matrix-boot :as cmb]) (:import [java.security MessageDigest] @@ -165,16 +166,27 @@ ;; --------------------------------------------------------------------------- (defn slice-schedule - [votes slots] - (loop [prev 0 [cut & more] slots i 0 acc []] - (if (nil? cut) - acc - (recur cut more (inc i) - (conj acc {:index i - :prev-slot prev - :cut-slot cut - :votes (subvec votes prev cut) ; (prev, cut] 0-based - :cut-time-ms (:t-ms (nth votes (dec cut)))}))))) + "Mods weave per schedule.py:204-210: a mod event attaches to the FIRST cut + whose cut-time reaches its :modified (and which is past the previous cut's + time); events after the last cut are dropped, like tail votes." + ([votes slots] (slice-schedule votes slots [])) + ([votes slots mod-events] + (loop [prev 0 [cut & more] slots i 0 acc []] + (if (nil? cut) + acc + (let [cut-time (:t-ms (nth votes (dec cut))) + prev-time (when (pos? prev) (:t-ms (nth votes (dec prev)))) + mods (filterv #(and (<= (long (:modified %)) (long cut-time)) + (or (nil? prev-time) + (> (long (:modified %)) (long prev-time)))) + mod-events)] + (recur cut more (inc i) + (conj acc {:index i + :prev-slot prev + :cut-slot cut + :votes (subvec votes prev cut) ; (prev, cut] 0-based + :mods mods + :cut-time-ms cut-time}))))))) ;; --------------------------------------------------------------------------- ;; Feeding conv-update: FLIP the export sign to raw-DB (design §5). @@ -215,27 +227,77 @@ (def certify-cold-start-pca {:comps [[1.0] [1.0]]}) +(defn parse-blob-json + "EXACTLY db/load-conv's key-fn (postgres.clj:419-433): numeric-string keys + become longs, everything else keywords — including the keyword/long + hash-map-key mismatches its own docstring warns about (e.g. :repness), + which are part of production restart semantics." + [s] + (json/parse-string s (fn [x] (try (Long/parseLong x) + (catch Exception _ (keyword x)))))) + +(defn restart-conv + "Replicate conv-man's load-or-init restart (conv_man.clj:188-207) + mid-schedule: rebuild the conv from its OWN just-computed math_main blob + (prep-main → JSON round-trip → restructure-json-conv), :recompute :reboot, + raw-rating-mat from the FULL vote log so far ([pid tid raw-db-vote] in + dataset order — conv-poll's created-order equivalent), then mod-update with + the FULL mod history so far (called even when empty, as load-or-init does). + Everything restructure-json-conv drops (rating-mat, per-k + :group-clusterings smoother memory, …) is LOST, exactly as in production." + [conv steps-so-far] + (let [votes-so-far (mapcat :votes steps-so-far) + mods-so-far (mapcat :mods steps-so-far)] + (-> (cm/prep-main conv) + json/generate-string + parse-blob-json + cm/restructure-json-conv + (assoc :recompute :reboot) + (assoc :raw-rating-mat + (nm/update-nmat (nm/named-matrix) + (mapv (fn [{:keys [pid tid sign]}] + [pid tid (- (long sign))]) + votes-so-far))) + (conv/mod-update (vec mods-so-far))))) + (defn run-once "Returns a vector of [step conv-after-update] pairs, one per cut slot. The reduce threading the conv IS the implicit warm-start chain. conv-update runs with certify-conv-opts (Q10 full-PCA carve-out) and the - seed conv carries certify-cold-start-pca (Q12 pinned cold start)." - [zid meta-tids steps] - (binding [*out* *err*] - (println "Q10 carve-out: large-conv mini-batch PCA disabled" - "(ptpt/cmt cutoffs pinned to 10^9; full PCA at every size)") - (println "Q12 carve-out: cold-tick PCA start pinned to ones" - "(production start is unseeded-random)")) - (let [seed (-> (conv/new-conv) - (assoc :zid zid - :meta-tids (set meta-tids) - :pca certify-cold-start-pca))] - (loop [conv seed [s & more] steps acc []] - (if (nil? s) - acc - (let [conv' (conv/conv-update conv (->conv-votes (:votes s)) - certify-conv-opts)] - (recur conv' more (conj acc [s conv']))))))) + seed conv carries certify-cold-start-pca (Q12 pinned cold start). + Step semantics mirror conv-man's per-batch [:votes :moderation] order + (conv_man.clj:361-371): votes → conv-update (recompute), then mods → + conv/mod-update (sets+watermark ONLY, no recompute — the mods take effect + at the NEXT votes recompute); ONE blob per step, recorded post-mods. + After recording step `restart-after`, the chain continues from + `restart-conv` (the production worker-restart seam)." + ([zid meta-tids steps] (run-once zid meta-tids steps nil)) + ([zid meta-tids steps restart-after] + (binding [*out* *err*] + (println "Q10 carve-out: large-conv mini-batch PCA disabled" + "(ptpt/cmt cutoffs pinned to 10^9; full PCA at every size)") + (println "Q12 carve-out: cold-tick PCA start pinned to ones" + "(production start is unseeded-random)")) + (let [seed (-> (conv/new-conv) + (assoc :zid zid + :meta-tids (set meta-tids) + :pca certify-cold-start-pca))] + (loop [conv seed [s & more] steps acc []] + (if (nil? s) + acc + (let [conv' (conv/conv-update conv (->conv-votes (:votes s)) + certify-conv-opts) + conv' (if (seq (:mods s)) + (conv/mod-update conv' (vec (:mods s))) + conv') + acc' (conj acc [s conv']) + conv'' (if (and restart-after (= (long (:index s)) (long restart-after))) + (do (binding [*out* *err*] + (println (format "restart seam after step %d (load-or-init replay)" + (long (:index s))))) + (restart-conv conv' (map first acc'))) + conv')] + (recur conv'' more acc'))))))) ;; --------------------------------------------------------------------------- ;; Recording. @@ -292,8 +354,13 @@ (defn build-provenance [{:keys [schedule schedule-id source votes-path comments-path zid meta-tids - meta-tids-source warm-start repeats n-steps edn?]}] + meta-tids-source warm-start repeats n-steps edn? + moderation n-mod-events n-mod-skipped restart-after]}] {:engine "clj" + :moderation (or moderation "none") + :n_mod_events (or n-mod-events 0) + :n_mod_skipped_no_modified (or n-mod-skipped 0) + :restart_after restart-after :mode "A" :schedule_id schedule-id :source source @@ -348,6 +415,51 @@ (into #{})) "comments CSV is-meta column"]))))) +;; --------------------------------------------------------------------------- +;; Moderation rows from the comments CSV (interleave-by-timestamp schedules). +;; --------------------------------------------------------------------------- + +(defn read-mod-events + "Raw moderation rows {:tid :is_meta :mod :modified} from the comments CSV, + sorted by (modified, file order) — the conv-mod-poll stream equivalent. + `modified` is the DB value in MILLISECONDS, compared directly against vote + :t-ms at weave time (the py loader reads the same column identically). + Rows with an empty `modified` cannot be woven and are SKIPPED (counted in + :n-skipped for provenance). Columns: comment-id/tid, is-meta/is_meta, + mod/moderated, modified." + [comments-path] + (with-open [rdr (io/reader comments-path)] + (let [rows (doall (csv/read-csv rdr)) + header (first rows) + idx (zipmap header (range)) + ci (or (idx "comment-id") (idx "tid")) + mi (or (idx "is-meta") (idx "is_meta")) + modi (or (idx "mod") (idx "moderated")) + tsi (idx "modified")] + (when (some nil? [ci modi tsi]) + (throw (ex-info (str "comments CSV lacks moderation columns " + "(need comment-id, mod/moderated, modified); header=" + (vec header)) + {:header header}))) + (let [parsed (->> (rest rows) + (keep-indexed + (fn [i r] + (let [modified-raw (str/trim (str (nth r tsi "")))] + (when (seq modified-raw) + {:tid (Long/parseLong (str/trim (nth r ci))) + :is_meta (boolean + (when mi + (#{"1" "true" "t" "yes"} + (str/lower-case (str/trim (str (nth r mi ""))))))) + :mod (Long/parseLong (str/trim (nth r modi))) + :modified (Long/parseLong modified-raw) + :file-idx i}))))) + events (->> parsed + (sort-by (juxt :modified :file-idx)) + (mapv #(dissoc % :file-idx)))] + {:events events + :n-skipped (- (count (rest rows)) (count events))})))) + ;; --------------------------------------------------------------------------- ;; CLI. ;; --------------------------------------------------------------------------- @@ -395,12 +507,16 @@ out (io/file (:out options)) clj-dir (io/file out "clj")] - (when-not (contains? #{"none" nil} moderation) + (when-not (contains? #{"none" "interleave-by-timestamp" nil} moderation) (throw (ex-info - (str "Moderation interleaving is NOT implemented in the Mode A " - "driver (the vw dataset has none). Got moderation=" - (pr-str moderation) ". Use \"none\" or add mod-update interleaving.") + (str "Unknown moderation mode " (pr-str moderation) + ". Use \"none\" or \"interleave-by-timestamp\" " + "(mod rows from --comments, woven by modified timestamp).") {:moderation moderation}))) + (when (and (= moderation "interleave-by-timestamp") + (nil? (:comments options))) + (throw (ex-info "moderation=interleave-by-timestamp requires --comments" + {:moderation moderation}))) ;; Only "chain" warm-start is implemented (it is IMPLICIT: the reduce ;; threads the conv, whose :pca :comps seed the next step's start-vectors, @@ -422,14 +538,45 @@ (let [raw (read-votes-csv (:votes options)) votes (build-dataset raw) slots (resolve-cut-slots votes cuts) - steps (slice-schedule votes slots) - [meta-tids meta-src] (read-meta-tids (:comments options))] + restart-after (get schedule "restart_after") + {mod-events :events n-mod-skipped :n-skipped} + (if (= moderation "interleave-by-timestamp") + (read-mod-events (:comments options)) + {:events [] :n-skipped 0}) + steps (slice-schedule votes slots mod-events) + ;; Under interleave moderation, meta-tids enter EXCLUSIVELY via + ;; the woven mod-update rows (the production-reachable route) — + ;; seeding them at conv creation as well would front-load every + ;; is-meta comment into step 0's compute, which no production + ;; state can produce (found on pc-meta-01 step 0, 2026-07-22 s4: + ;; clj meta-tids = seed ∪ woven vs py's woven-only). The + ;; creation-time seed remains for moderation="none" runs with + ;; --comments (the original vw-compat path). + [meta-tids meta-src] + (if (= moderation "interleave-by-timestamp") + [#{} "empty (interleave moderation: meta-tids via mod-update only)"] + (read-meta-tids (:comments options)))] + + (when restart-after + (when-not (and (integer? restart-after) + (<= 0 (long restart-after) (- (count steps) 2))) + (throw (ex-info (str "restart_after must be a step index with at " + "least one step after it; got " + (pr-str restart-after) " for " (count steps) + " steps") + {:restart_after restart-after :n-steps (count steps)})))) (binding [*out* *err*] (println (format "dataset=%s n_votes=%d schedule=%s cuts=%s" dataset (count votes) schedule-id (pr-str slots))) (println (format "steps=%d repeats=%d edn=%s zid=%s meta-tids=%d" - (count steps) repeats edn? (pr-str zid) (count meta-tids)))) + (count steps) repeats edn? (pr-str zid) (count meta-tids))) + (when (= moderation "interleave-by-timestamp") + (println (format "moderation=interleave-by-timestamp mod-events=%d skipped-no-modified=%d woven=%d" + (count mod-events) (long n-mod-skipped) + (reduce + (map (comp count :mods) steps))))) + (when restart-after + (println (format "restart_after=%d (load-or-init seam)" (long restart-after))))) (.mkdirs clj-dir) ;; schedule.json verbatim (byte-faithful copy of the §4 input). @@ -438,7 +585,7 @@ ;; Run repeats. rep 0 is also written flat to clj/ (the canonical ;; cross-language surface); rep i>0 (and rep 0) go to clj/rep-i/. (dotimes [rep repeats] - (let [results (run-once zid meta-tids steps) + (let [results (run-once zid meta-tids steps restart-after) rep-dir (if (> repeats 1) (io/file clj-dir (str "rep-" rep)) clj-dir)] (write-results! rep-dir results edn?) (when (and (> repeats 1) (zero? rep)) @@ -453,7 +600,11 @@ :votes-path (:votes options) :comments-path (:comments options) :zid zid :meta-tids meta-tids :meta-tids-source meta-src :warm-start warm-start :repeats repeats - :n-steps (count steps) :edn? edn?}) + :n-steps (count steps) :edn? edn? + :moderation moderation + :n-mod-events (count mod-events) + :n-mod-skipped n-mod-skipped + :restart-after restart-after}) prov-json (json/generate-string prov {:pretty true})] (spit (io/file out "provenance.json") prov-json) (spit (io/file clj-dir "provenance.json") prov-json))