From 5e139c14f7ce091bb0141e43d03b343a24809078 Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Wed, 22 Jul 2026 02:27:23 +0100 Subject: [PATCH] =?UTF-8?q?feat(math):=20clojure-legacy=20blob=20emission?= =?UTF-8?q?=20+=20selection=20parity=20=E2=80=94=20Clojure-exact=20math=5F?= =?UTF-8?q?main=20at=20step=200?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aligns clojure-legacy emission and selection semantics with Clojure's prep-main blob (conv_man.clj:45-74), per the step-0 battery diagnosis (journal 2026-07-22 session 3; fingerprints in docs/divergences.json). Battery effect: vw:uniform8 (8 steps), vw:single-cut and biodiversity:uniform8 (8 steps) went from DIVERGENCE to full MATCH; vw:front-loaded6 remains open (base-cluster count edge at small N). Emission (Conversation.to_dict, legacy-gated, improved unchanged): - group-clusters: members emitted as BASE-CLUSTER ids (Clojure folded form); unfolded pids stay under the snake alias group_clusters. - votes-base: per-base-cluster A/D/S bucket lists over sort-by-id clustered members (agg-bucket-votes-for-tid, conversation.clj:601-608) instead of whole-matrix int totals. - pca: comment-projection + comment-extremity emitted (existing D12 fns). - Sign parity: pca.center, base-clusters.x/y, group-clusters centers, comment-projection negated at emission — Delphi's matrix is the NEGATION of Clojure's (AGREE=+1 vs -1); comps are covariance-derived and already equal (verified max|clj+py| = 1e-16 on center). - repness: {gid: [finalize-cmt-stats-shaped entries]} (repness.clj:173), internal dict preserved under repness_full for from_dict round-trip. - mod-in/mod-out/lastModTimestamp: null until moderation applied. - tids emitted in Clojure column (first-vote arrival) order with pca arrays permuted alongside; from_dict inverts the center sign and restores the arrival tracker in legacy mode. Selection/stat parity: - group-aware-consensus multiplies (A+1)/(S+2) over EVERY group — a zero-S group contributes 1/2, not skipped (conversation.clj:639-641). - repness rest-stats sum over the OTHER GROUPS only (repness.clj:125), excluding unclustered voters from the comparison domain. - Exact-score ties in repness/consensus selection resolve by Clojure's stable sort over named-matrix column order: update_votes tracks first-vote arrival order, conv_repness accepts tid_order and the selectors iterate as-given in legacy mode. Also: skip the three math-tree-hashing harness tests in delphi-only CI images (no /math in the container; run 29881478306's 3 failures). TDD: tests/test_legacy_blob_shape.py (25 tests, RED observed per fix). commit-id:0add28f3 --- delphi/polismath/conversation/conversation.py | 280 +++++++++- delphi/polismath/pca_kmeans_rep/repness.py | 110 +++- delphi/tests/replay_harness/test_certify.py | 9 + .../tests/replay_harness/test_timing_probe.py | 4 + delphi/tests/test_engine_mode.py | 42 +- delphi/tests/test_legacy_blob_shape.py | 518 ++++++++++++++++++ 6 files changed, 912 insertions(+), 51 deletions(-) create mode 100644 delphi/tests/test_legacy_blob_shape.py diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index 405c2d7be..9f80df692 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/delphi/polismath/conversation/conversation.py @@ -201,6 +201,18 @@ def __init__(self, self.mod_in_tids = set() # Featured comments self.meta_tids = set() # Meta comments self.mod_out_ptpts = set() # Excluded participants + # Clojure conv state carries no :mod-in/:mod-out (nil in the blob) + # until the poller delivers moderation; these two track that seam so + # 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 + # 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 + # clojure-legacy tie-breaking (stable sorts over column order) and + # blob tid emission can replicate Clojure exactly. Append-only. + self.tid_arrival_order = [] # Clustering and projection state self.pca = None @@ -350,6 +362,15 @@ def update_votes(self, # Log validation results logger.info(f"[{time.time() - start_time:.2f}s] Vote processing summary: {len(vote_updates)} valid, {invalid_count} invalid, {null_count} null") + # Record first-vote arrival order for unseen tids (valid votes only, + # in payload order — the same order Clojure's update-nmat encounters + # them). See tid_arrival_order in __init__. + seen_tids = set(result.tid_arrival_order) + for _, comment_id, _ in vote_updates: + if comment_id not in seen_tids: + seen_tids.add(comment_id) + result.tid_arrival_order.append(comment_id) + # Get existing row and column indices existing_rows = self.raw_rating_mat.index existing_cols = self.raw_rating_mat.columns @@ -611,6 +632,11 @@ def update_moderation(self, mod_in_tids = moderation.get('mod_in_tids', []) meta_tids = moderation.get('meta_tids', []) mod_out_ptpts = moderation.get('mod_out_ptpts', []) + + result.moderation_applied = True + result.last_mod_timestamp = moderation.get( + 'lastModTimestamp', result.last_mod_timestamp + ) # Update moderation sets if mod_out_tids: @@ -1109,9 +1135,18 @@ def _compute_repness(self) -> None: # Compute representativeness (needs participant IDs, not base-cluster IDs). # `mod_out=self.mod_out_tids` forwards moderated-out tids to the rep + consensus # selectors (Clojure parity per D11 / PR 9; matches repness.clj:222 and :296). + # In clojure-legacy mode, tid_order carries the first-vote arrival + # order so exact-score ties resolve like Clojure's stable sorts over + # named-matrix column order (FP-eaea8c1b7f / FP-0d73f006f4). + tid_order = ( + self.tid_arrival_order + if resolve_engine_mode() == ENGINE_MODE_LEGACY + else None + ) self.repness = conv_repness(self.rating_mat, self._unfolded_group_clusters(), - mod_out=self.mod_out_tids) + mod_out=self.mod_out_tids, + tid_order=tid_order) logger.info(f"Representativeness completed in {time.time() - start_time:.2f}s") def _compute_participant_info_optimized(self, vote_matrix: pd.DataFrame, group_clusters: List[Dict[str, Any]]) -> Dict[str, Any]: @@ -1638,7 +1673,170 @@ def _compute_votes_base(self) -> Dict[str, Any]: votes_base[tid] = entry return votes_base - + + def _compute_votes_base_buckets(self) -> Dict[str, Any]: + """ + Clojure-exact votes-base: per-tid A/D/S vectors indexed by base-cluster + bucket, matching `agg-bucket-votes-for-tid` over `bid-to-pid` + (conversation.clj:593-608). Buckets are the base clusters SORTED BY + :id; the aggregation domain is each bucket's member pids only — votes + from unclustered participants never appear (this is why the improved + int totals run up to +1 higher on some tids; FP-81fda13ef6). + + Values come from raw_rating_mat (D15 parity: the actual votes cast, + not post-moderation zeros), same as `_compute_votes_base`. + + Returns: + {tid: {'A': [per-bucket count], 'D': [...], 'S': [...]}}. + """ + mat = self.raw_rating_mat + values = mat.values + row_pos = {pid: i for i, pid in enumerate(mat.index)} + bucket_rows = [ + np.asarray( + [row_pos[p] for p in c['members'] if p in row_pos], dtype=int + ) + for c in sorted(self.base_clusters or [], key=lambda c: c['id']) + ] + + agree_mask = np.abs(values - 1.0) < 0.001 + disagree_mask = np.abs(values + 1.0) < 0.001 + valid_mask = ~np.isnan(values) + per_bucket = [ + ( + agree_mask[rows].sum(axis=0), + disagree_mask[rows].sum(axis=0), + valid_mask[rows].sum(axis=0), + ) + for rows in bucket_rows + ] + + votes_base = {} + for j, tid in enumerate(mat.columns): + entry = { + 'A': [int(a[j]) for a, _, _ in per_bucket], + 'D': [int(d[j]) for _, d, _ in per_bucket], + 'S': [int(s[j]) for _, _, s in per_bucket], + } + try: + votes_base[int(tid)] = entry + except (ValueError, TypeError): + votes_base[tid] = entry + return votes_base + + @staticmethod + def _legacy_repness_entry(e: Dict[str, Any]) -> Dict[str, Any]: + """ + One repness entry in Clojure `finalize-cmt-stats` shape + (repness.clj:173-188): direction chosen by rat > rdt, stat fields + renamed to the blob spellings, repness-test cast through float32 + (Clojure applies a literal `(float ...)`). Best-agree entries carry + the two extra keys per repness.clj:262-264. + """ + repful = e.get('repful') or ('agree' if e['rat'] > e['rdt'] else 'disagree') + agree = repful == 'agree' + out = { + 'tid': e['comment_id'], + 'n-success': e['na'] if agree else e['nd'], + 'n-trials': e['ns'], + 'p-success': e['pa'] if agree else e['pd'], + 'p-test': e['pat'] if agree else e['pdt'], + 'repness': e['ra'] if agree else e['rd'], + 'repness-test': float(np.float32(e['rat'] if agree else e['rdt'])), + 'repful-for': repful, + } + if e.get('best_agree'): + out['best-agree'] = True + out['n-agree'] = e['n_agree'] + return out + + def _apply_legacy_blob_shape(self, result: Dict[str, Any]) -> None: + """ + Clojure-exact emission overrides for clojure-legacy mode (mutates + ``result``). Improved-mode emission is untouched. Diagnosis and + fingerprints: docs/divergences.json + journal session 2026-07-22-3. + + - Sign parity (FP-80ca42344a/FP-3781b5768f/FP-af281c8386/ + FP-194ee5ad04): Delphi's rating matrix is the NEGATION of Clojure's + (AGREE=+1 vs AGREE=-1), so every mean/projection-derived float + (pca.center, base-clusters x/y, group-cluster centers, + comment-projection) negates at this boundary; comps are + covariance-derived and already equal — emitted unchanged. + - pca comment-projection/comment-extremity (FP-2393072de1): Clojure's + with-proj-and-extremtiy (conversation.clj:341-352); projection is + emitted TRANSPOSED (component-major), extremity is a norm and + therefore sign-invariant. + - group-clusters (FP-55e290562e): members are BASE-CLUSTER ids + (Clojure folded form); the unfolded-pids view stays available under + the snake alias ``group_clusters``. + - repness (FP-c3cee15f8b): {gid: [finalize-cmt-stats entries]}; the + internal dict moves to ``repness_full`` (python-only key) so + ``from_dict`` round-trips losslessly. + - moderation seam (FP-2f5714ce9c/FP-872f81716c/FP-2975bbfb04): + mod-in/mod-out/lastModTimestamp are null until moderation has been + applied. + """ + # tids in Clojure column (first-vote arrival) order; tid-aligned pca + # arrays are permuted with them so the blob stays internally + # consistent. Falls back to emitted order for tids that predate the + # tracker (e.g. conversations restored from pre-tracker blobs). + emitted_tids = result.get('tids') or [] + emitted_set = set(emitted_tids) + arrival = [t for t in self.tid_arrival_order if t in emitted_set] + arrival_set = set(arrival) + arrival += [t for t in emitted_tids if t not in arrival_set] + tid_pos = {t: i for i, t in enumerate(emitted_tids)} + perm = [tid_pos[t] for t in arrival] + + if self.pca: + center = np.asarray(self.pca['center'], dtype=float) + comps = np.asarray(self.pca['comps'], dtype=float) + cmnt_proj = pca_project_cmnts(center, comps) # Delphi sign, (n_cmts, n_comps) + extremity = compute_comment_extremity(cmnt_proj) + pca_out = dict(result.get('pca', {})) + if len(perm) == center.shape[0]: + # Permute tids and every tid-aligned pca array TOGETHER so + # the blob stays internally consistent. + result['tids'] = arrival + center = center[perm] + comps = comps[:, perm] + cmnt_proj = cmnt_proj[perm, :] + extremity = extremity[perm] + pca_out['center'] = (-center).tolist() + pca_out['comps'] = comps.tolist() + pca_out['comment-projection'] = (-cmnt_proj.T).tolist() + pca_out['comment-extremity'] = extremity.tolist() + result['pca'] = pca_out + else: + # No tid-aligned arrays to keep in sync — reorder tids alone. + result['tids'] = arrival + + bc = result.get('base-clusters') + if bc: + bc['x'] = [-v for v in bc['x']] + bc['y'] = [-v for v in bc['y']] + + result['group-clusters'] = [ + { + 'id': g['id'], + 'members': list(g['members']), + 'center': [-c for c in g['center']], + } + for g in (self.group_clusters or []) + ] + + if self.repness and self.repness.get('group_repness') is not None: + result['repness_full'] = self.repness + result['repness'] = { + gid: [self._legacy_repness_entry(e) for e in entries] + for gid, entries in self.repness['group_repness'].items() + } + + if not self.moderation_applied: + result['mod-in'] = None + result['mod-out'] = None + result['lastModTimestamp'] = self.last_mod_timestamp + def _compute_group_votes(self) -> Dict[str, Any]: """ Compute group votes structure which maps group IDs to vote statistics by comment. @@ -2109,7 +2307,12 @@ def numpy_to_list(arr): # raw_rating_mat so that moderated-out columns report the actual votes cast, # not the post-D15 zeros (which would inflate every column's 'S' count). votes_base_start = time.time() - result['votes-base'] = self._compute_votes_base() + if resolve_engine_mode() == ENGINE_MODE_LEGACY: + # Clojure-exact per-base-cluster bucket vectors (agg-bucket-votes- + # for-tid parity); improved mode keeps the int totals. + result['votes-base'] = self._compute_votes_base_buckets() + else: + result['votes-base'] = self._compute_votes_base() logger.info(f"Votes base: {time.time() - votes_base_start:.4f}s") # Compute group votes with optimized approach @@ -2190,6 +2393,7 @@ def numpy_to_list(arr): # Compute in one pass using existing structure if 'group-votes' in result: + gac_legacy = resolve_engine_mode() == ENGINE_MODE_LEGACY # Store consensus values per comment ID for tid in self.rating_mat.columns: # Try converting to integer for consistent keys @@ -2197,22 +2401,29 @@ def numpy_to_list(arr): tid_key = int(tid) except (ValueError, TypeError): tid_key = tid - + # Start with consensus value of 1 consensus_value = 1.0 has_data = False - + # Multiply probabilities from all groups (same as reduce * in Clojure) for gid, gid_data in result['group-votes'].items(): votes_data = gid_data.get('votes', {}) - + if tid_key in votes_data: vote_stats = votes_data[tid_key] agree_count = vote_stats.get('A', 0) total_count = vote_stats.get('S', 0) - + + if gac_legacy: + # Clojure parity (conversation.clj:639-641, + # FP-b3670cb052): every group's factor multiplies + # in, `:or {A 0 S 0}` — a zero-S group contributes + # (0+1)/(0+2) = 1/2, it is NOT skipped. + consensus_value *= (agree_count + 1.0) / (total_count + 2.0) + has_data = True # Calculate probability with Laplace smoothing - if total_count > 0: + elif total_count > 0: prob = (agree_count + 1.0) / (total_count + 2.0) consensus_value *= prob has_data = True @@ -2290,6 +2501,10 @@ def numpy_to_list(arr): # Add math_tick value and return result['math_tick'] = math_tick_value + + if resolve_engine_mode() == ENGINE_MODE_LEGACY: + self._apply_legacy_blob_shape(result) + logger.info(f"Total to_dict time: {time.time() - overall_start_time:.4f}s") return result @@ -2526,13 +2741,50 @@ def from_dict(cls, data: Dict[str, Any]) -> 'Conversation': conv.mod_in_tids = set(moderation.get('mod_in_tids', [])) conv.meta_tids = set(moderation.get('meta_tids', [])) conv.mod_out_ptpts = set(moderation.get('mod_out_ptpts', [])) - + + legacy = resolve_engine_mode() == ENGINE_MODE_LEGACY + # Best-effort inference (the blob carries no explicit flag): any + # restored moderation set implies moderation was applied. A + # moderated-then-emptied conversation restores as not-applied — the + # same information loss Clojure has on a cold restore. + conv.moderation_applied = bool( + conv.mod_out_tids or conv.mod_in_tids + or conv.meta_tids or conv.mod_out_ptpts + ) + if legacy: + # Legacy blobs emit the real (possibly null) mod watermark; + # improved blobs reuse last_updated there, which is NOT a mod + # timestamp — leave the attribute at its None default for those. + conv.last_mod_timestamp = data.get('lastModTimestamp') + # Legacy blobs emit tids in Clojure column (arrival) order — + # restore the tracker so tie-breaking survives a warm restart. + conv.tid_arrival_order = list(data.get('tids', [])) + # Restore PCA data pca_data = data.get('pca') if pca_data: + center = np.array(pca_data['center']) + comps = np.array(pca_data['comps']) + if legacy: + # Inverse of the legacy emission sign parity: blobs carry the + # Clojure-convention (negated) center; internal state stays in + # Delphi convention (see _apply_legacy_blob_shape). + center = -center + # Inverse of the legacy emission ORDER parity: blobs emit tids + # (and every tid-aligned pca array) in Clojure ARRIVAL order, + # while internal state aligns with the natsorted matrix + # columns. Without un-permuting, a warm restore would seed the + # next PCA with column-misaligned center/comps (#2649 review). + blob_tids = data.get('tids') or [] + if len(blob_tids) == center.shape[0]: + pos = {t: i for i, t in enumerate(blob_tids)} + perm = [pos[t] for t in natsorted(blob_tids)] + center = center[perm] + if comps.ndim == 2 and comps.shape[1] == len(perm): + comps = comps[:, perm] conv.pca = { - 'center': np.array(pca_data['center']), - 'comps': np.array(pca_data['comps']) + 'center': center, + 'comps': comps } # Restore projection data @@ -2543,8 +2795,10 @@ def from_dict(cls, data: Dict[str, Any]) -> 'Conversation': # Restore cluster data conv.group_clusters = data.get('group_clusters', []) - # Restore representativeness data - conv.repness = data.get('repness') + # Restore representativeness data. Legacy blobs emit 'repness' in + # Clojure per-group shape and park the internal dict under + # 'repness_full' — prefer the lossless internal copy when present. + conv.repness = data.get('repness_full', data.get('repness')) # Restore participant info conv.participant_info = data.get('participant_info', {}) diff --git a/delphi/polismath/pca_kmeans_rep/repness.py b/delphi/polismath/pca_kmeans_rep/repness.py index 56808eb7f..9deeddaf9 100644 --- a/delphi/polismath/pca_kmeans_rep/repness.py +++ b/delphi/polismath/pca_kmeans_rep/repness.py @@ -9,6 +9,7 @@ import pandas as pd from typing import Any, Dict, Iterable, List, Optional, Tuple +from polismath.utils.engine_mode import ENGINE_MODE_LEGACY, resolve_engine_mode from polismath.utils.general import AGREE, DISAGREE @@ -175,7 +176,8 @@ def two_prop_test(succ_in, succ_out, pop_in, pop_out): def compute_group_comment_stats_df(votes_long: pd.DataFrame, - group_clusters: List[Dict[str, Any]]) -> pd.DataFrame: + group_clusters: List[Dict[str, Any]], + tid_order: Optional[List[Any]] = None) -> pd.DataFrame: """ Compute vote counts and probabilities for all (group, comment) pairs. @@ -221,34 +223,60 @@ def compute_group_comment_stats_df(votes_long: pd.DataFrame, # Return empty DataFrame with correct schema return pd.DataFrame(columns=['na', 'nd', 'ns', 'pa', 'pd', 'pat', 'pdt']) - # Compute total counts per comment BEFORE filtering to group members - # This matches the old behavior where "other" included ALL participants - # not in the current group (even those not in any cluster) + # Add group column and identify votes from clustered participants + votes_with_group = votes_only.copy() + votes_with_group['group_id'] = votes_with_group['participant'].map(ptpt_to_group) + + # Keep only votes from participants in some group (for group-specific counts) + votes_in_groups = votes_with_group.dropna(subset=['group_id']) + + # Totals feed the "other" (rest) side of the comparison below. + # + # clojure-legacy: Clojure's rest-stats sum per-group comment-stats over + # the OTHER GROUPS only (utils/mapv-rest, repness.clj:125-131), and group + # membership is unfolded through base clusters — so votes from + # participants in NO cluster never enter the comparison. Totals must + # therefore come from clustered voters only (FP-69c7a13580/FP-faac8c6125). + # + # improved: keeps the historical behavior where "other" included ALL + # participants not in the current group (even those not in any cluster). # # total_votes counts agree + disagree + PASS, matching Clojure's # `count-votes` (math/src/polismath/math/repness.clj:56-61, :70). # `count-votes` called with no `vote` arg uses `identity` as the filter # predicate; in Clojure 0 is truthy, so PASS (0) votes are kept. NaN # entries are already dropped above. Use size() to count non-NaN rows. - total_counts = votes_only.groupby('comment').agg( + total_source = ( + votes_in_groups + if resolve_engine_mode() == ENGINE_MODE_LEGACY + else votes_only + ) + total_counts = total_source.groupby('comment').agg( total_agree=('vote', lambda x: (x == AGREE).sum()), total_disagree=('vote', lambda x: (x == DISAGREE).sum()), total_votes=('vote', 'size'), ) - - # Now add group column and filter to only group members - votes_with_group = votes_only.copy() - votes_with_group['group_id'] = votes_with_group['participant'].map(ptpt_to_group) - - # Keep only votes from participants in some group (for group-specific counts) - votes_in_groups = votes_with_group.dropna(subset=['group_id']) + # The comment universe stays votes_only-based in BOTH modes (Clojure + # iterates every matrix column; a comment voted on only by unclustered + # participants still gets an all-zero stats row). + all_voted_comments = votes_only['comment'].unique() + total_counts = total_counts.reindex(all_voted_comments, fill_value=0) if votes_in_groups.empty: # Return empty DataFrame with correct schema return pd.DataFrame(columns=['na', 'nd', 'ns', 'pa', 'pd', 'pat', 'pdt']) - # Get all unique comments that have at least one vote (from anyone) + # Get all unique comments that have at least one vote (from anyone). + # With tid_order (clojure-legacy), rows follow Clojure's named-matrix + # column order (first-vote arrival) so downstream stable sorts break + # exact-score ties identically; unknown comments keep their default + # position at the tail (defensive — tid_order normally covers all). all_comments = total_counts.index.tolist() + if tid_order is not None: + known = set(all_comments) + ordered = [t for t in tid_order if t in known] + ordered_set = set(ordered) + all_comments = ordered + [t for t in all_comments if t not in ordered_set] # Get all group IDs all_group_ids = [group['id'] for group in group_clusters] @@ -474,7 +502,8 @@ def _finalize_row_for_output(row: Dict[str, Any], *, def select_rep_comments_df(stats_df: pd.DataFrame, - mod_out: Optional[Iterable[int]] = None + mod_out: Optional[Iterable[int]] = None, + preserve_order: bool = False ) -> Tuple[pd.DataFrame, Optional[Dict[str, Any]]]: """ Select representative comments for a single group (Clojure parity). @@ -532,15 +561,24 @@ def select_rep_comments_df(stats_df: pd.DataFrame, best_max_rt: Optional[float] = None best_agree: Optional[Dict[str, Any]] = None - # Sort by `comment` (tid) ascending BEFORE iterating, so ties in - # `beats_best_by_test` (max(rat, rdt) tied) and in `beats_best_agr` - # (Branch 2/3 product tied) resolve deterministically. The chosen order - # matches Clojure's named-matrix column iteration: after normalization - # the columns are insertion-ordered, and for cold-start that's tid - # ascending (see Clojure named_matrix.clj:130-131 — insertion order). - # All Clojure beats-*? predicates use strict `>` so the FIRST row at a - # tied score wins; sorting ascending here mirrors that (decision D10.8.1). - iter_df = stats_df.sort_values('comment', kind='mergesort') + # Iteration order decides ties: all Clojure beats-*? predicates use + # strict `>` so the FIRST row at a tied score wins, and repness-sort is + # a stable sort over the iteration order (repness.clj:196-200). + # + # preserve_order=True (clojure-legacy via conv_repness's tid_order): rows + # already follow Clojure's named-matrix column order — first-vote ARRIVAL + # order, which is NOT tid-ascending in general (verified on the vw replay: + # clj tids open [24, 19, 47, …]) — so iterate as-given. + # + # preserve_order=False (improved / direct callers): sort by `comment` + # ascending for deterministic ties (decision D10.8.1; its "insertion + # order == ascending" cold-start assumption holds only for tid-ordered + # vote streams, hence the legacy path above). + iter_df = ( + stats_df + if preserve_order + else stats_df.sort_values('comment', kind='mergesort') + ) for row in iter_df.to_dict('records'): if row['comment'] in mod_out_set: @@ -605,7 +643,8 @@ def _sort_key(s: Dict[str, Any]) -> float: def _assemble_rep_comments(stats_df: pd.DataFrame, - mod_out: Optional[Iterable[int]] = None + mod_out: Optional[Iterable[int]] = None, + preserve_order: bool = False ) -> List[Dict[str, Any]]: """Thin wrapper around `select_rep_comments_df` that returns the flat output list (best-agree slot prepended, then the DataFrame's rows, @@ -618,7 +657,8 @@ def _assemble_rep_comments(stats_df: pd.DataFrame, the flat List[Dict] form, so we keep one place that does the prepend and the final agrees-before-disagrees stable partition. """ - rep_df, best_agree_dict = select_rep_comments_df(stats_df, mod_out=mod_out) + rep_df, best_agree_dict = select_rep_comments_df( + stats_df, mod_out=mod_out, preserve_order=preserve_order) head: List[Dict[str, Any]] = [best_agree_dict] if best_agree_dict is not None else [] tail: List[Dict[str, Any]] = ( rep_df.to_dict('records') if not rep_df.empty else [] @@ -772,6 +812,7 @@ def _disagree_entry(tid: Any, row: pd.Series) -> Dict[str, Any]: def conv_repness(vote_matrix_df: pd.DataFrame, group_clusters: List[Dict[str, Any]], mod_out: Optional[Iterable[int]] = None, + tid_order: Optional[List[Any]] = None, ) -> Dict[str, Any]: """ Calculate representativeness for all comments and groups. @@ -785,6 +826,11 @@ def conv_repness(vote_matrix_df: pd.DataFrame, mod_out: Optional iterable of tids to exclude (moderated-out comments). Forwarded to `select_rep_comments_df` and `consensus_stats_df`. See `Conversation.mod_out_tids`. + tid_order: Optional comment order for tie-breaking (clojure-legacy: + first-vote arrival order == Clojure's named-matrix column order). + When given, stats rows and consensus stats follow it and the + selectors iterate as-given instead of tid-ascending, so + exact-score ties resolve like Clojure's stable sorts. Returns: Dictionary with representativeness data for each group: @@ -819,7 +865,8 @@ def conv_repness(vote_matrix_df: pd.DataFrame, votes_long['vote'] = pd.to_numeric(votes_long['vote'], errors='coerce') # Compute all stats using vectorized function - stats_df = compute_group_comment_stats_df(votes_long, group_clusters) + stats_df = compute_group_comment_stats_df(votes_long, group_clusters, + tid_order=tid_order) if stats_df.empty: return empty_result @@ -870,7 +917,8 @@ def conv_repness(vote_matrix_df: pd.DataFrame, # partition applied). Forward `mod_out` from conv_repness (D11 # added this kwarg). result['group_repness'][group_id] = _assemble_rep_comments( - group_stats, mod_out=mod_out) + group_stats, mod_out=mod_out, + preserve_order=tid_order is not None) except Exception as e: print(f"Error selecting representative comments for group {group_id}: {e}") result['group_repness'][group_id] = [] @@ -880,6 +928,14 @@ def conv_repness(vote_matrix_df: pd.DataFrame, # `len(group_clusters) > 1` guard. try: cons_stats = consensus_stats_df(vote_matrix_df, mod_out=mod_out) + if tid_order is not None and not cons_stats.empty: + # Rank ties resolve by row order (nlargest keep='first'): follow + # Clojure's column (arrival) order, unknown tids at the tail. + known = set(cons_stats.index) + ordered = [t for t in tid_order if t in known] + ordered_set = set(ordered) + ordered += [t for t in cons_stats.index if t not in ordered_set] + cons_stats = cons_stats.reindex(ordered) result['consensus_comments'] = select_consensus_comments_df(cons_stats) except Exception as e: print(f"Error selecting consensus comments: {e}") diff --git a/delphi/tests/replay_harness/test_certify.py b/delphi/tests/replay_harness/test_certify.py index 8c30b3e47..868271092 100644 --- a/delphi/tests/replay_harness/test_certify.py +++ b/delphi/tests/replay_harness/test_certify.py @@ -401,6 +401,14 @@ def fake_run(cmd, *, cwd, env): assert exc_info.value.stage +requires_math_tree = pytest.mark.skipif( + not (cert._MATH_ROOT / "dev" / "replay.clj").exists(), + reason="math/ tree not present (delphi-only CI image runs from /app; " + "the clj cache manifest hashes real math/ files)", +) + + +@requires_math_tree def test_ensure_clj_recording_cache_hit_then_miss_on_change(tmp_path, monkeypatch): calls = {"n": 0} @@ -425,6 +433,7 @@ def fake_run(cmd, *, cwd, env): assert cached3 is False and calls["n"] == 2 +@requires_math_tree def test_ensure_clj_recording_raises_certify_error_on_nonzero_exit(tmp_path, monkeypatch): def fake_run(cmd, *, cwd, env): return _fake_completed(returncode=1, stderr="clojure blew up") diff --git a/delphi/tests/replay_harness/test_timing_probe.py b/delphi/tests/replay_harness/test_timing_probe.py index 7865457a8..2abad1046 100644 --- a/delphi/tests/replay_harness/test_timing_probe.py +++ b/delphi/tests/replay_harness/test_timing_probe.py @@ -374,6 +374,10 @@ def test_build_report_shape(mod): # CLI end-to-end (mocked subprocess). # --------------------------------------------------------------------------- +@pytest.mark.skipif( + not (Path(__file__).resolve().parents[3] / "math" / "dev" / "replay.clj").exists(), + reason="math/ tree not present (delphi-only CI image runs from /app)", +) def test_cli_probe_end_to_end_mocked(mod, tmp_path, monkeypatch): votes_src = tmp_path / "votes.csv" _write_votes_csv(votes_src, 200) diff --git a/delphi/tests/test_engine_mode.py b/delphi/tests/test_engine_mode.py index fac7319f3..87ca2d001 100644 --- a/delphi/tests/test_engine_mode.py +++ b/delphi/tests/test_engine_mode.py @@ -8,9 +8,13 @@ - 'clojure-legacy' : threads warm-start state across ticks, matching Clojure (PCA :start-vectors, group-k-smoother). -On the FIRST tick (cold start) the two modes MUST coincide, because Clojure's -warm-start state is empty on the first tick (no previous comps, no smoother -state). This module asserts: +On the FIRST tick (cold start) the two modes MUST coincide STRUCTURALLY, +because Clojure's warm-start state is empty on the first tick (no previous +comps, no smoother state). Since the 2026-07-22 session-3 parity port, the +modes legitimately differ at cold start in a DOCUMENTED set of stat/selection +semantics and in the clojure-legacy blob emission surface (see +_recompute_to_dict for the list); the invariance gate covers everything else — +memberships, cluster structure, pca, vote aggregates. This module asserts: 1. Flag resolution semantics (default, valid, invalid, case/whitespace, read-at-call-time) — mirrors tests/test_powerit_pca.py::TestPcaImplFlag. 2. Cold-start invariance: a single-shot vw pipeline run is identical under @@ -158,15 +162,31 @@ def _recompute_to_dict(self, monkeypatch, mode): # Pin last_updated so the two runs share a deterministic value. conv.last_updated = 0 result = conv.recompute() + # Serialize BOTH runs under IMPROVED emission: since session-3 + # (2026-07-22) clojure-legacy has its own blob SURFACE (negated + # center/x/y, bids-vs-pids group members, bucketed votes-base, + # arrival-order tids, finalize-cmt-stats repness shape, null + # moderation seam) — pinned bidirectionally in + # test_legacy_blob_shape.py. Forcing improved emission here makes the + # comparison test what the legacy PLUMBING COMPUTED, not how legacy + # serializes it. + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ENGINE_MODE_IMPROVED) d = _strip_volatile(result.to_dict()) - # ONE documented first-tick exception (Q2, 2026-07-22): Clojure's own - # :comment-priorities reads the PREVIOUS tick's group-votes - # (conversation.clj:658), which is nil on the first tick — so - # Clojure-faithful legacy tick-1 priorities come from zero counts and - # CANNOT equal improved's current-tick-based values. Every other key - # keeps the cold-start invariance guarantee. Legacy tick-1 zero - # semantics are pinned in test_priority_unmirror.py. - d.pop('comment_priorities', None) + # Documented first-tick mode differences (each pinned elsewhere): + # - comment_priorities (Q2): Clojure reads the PREVIOUS tick's + # group-votes (conversation.clj:658) — nil on tick 1, so legacy + # tick-1 priorities come from zero counts (test_priority_unmirror). + # - repness + consensus (session 3): legacy rest-stats sum over the + # OTHER GROUPS only (repness.clj:125 — unclustered voters excluded, + # changing ra/rat values) and exact-score ties resolve by first-vote + # ARRIVAL order instead of tid-ascending (test_legacy_blob_shape). + # - group-aware-consensus (session 3): legacy multiplies the 1/2 + # factor of zero-S groups instead of skipping (conversation.clj:639). + # Everything else — memberships, in-conv, base/group clusters, pca, + # votes-base, vote aggregates — keeps the cold-start invariance gate. + for key in ('comment_priorities', 'repness', 'consensus', + 'group-aware-consensus'): + d.pop(key, None) return d def test_vw_cold_run_identical_across_modes(self, monkeypatch): diff --git a/delphi/tests/test_legacy_blob_shape.py b/delphi/tests/test_legacy_blob_shape.py new file mode 100644 index 000000000..ff2b3b9b3 --- /dev/null +++ b/delphi/tests/test_legacy_blob_shape.py @@ -0,0 +1,518 @@ +"""Legacy blob-shape alignment — Clojure-exact math_main emission (to_dict). + +Pins the clojure-legacy emission surface against Clojure's prep-main blob +(conv_man.clj:45-74), per the step-0 battery diagnosis (journal 2026-07-22 +session 3; fingerprints FP-55e290562e, FP-81fda13ef6, FP-2393072de1, +FP-80ca42344a, FP-c3cee15f8b, FP-2f5714ce9c, FP-2975bbfb04 in +docs/divergences.json): + +- group-clusters members are BASE-CLUSTER ids (Clojure folded form), not + unfolded participant ids. +- votes-base is per-base-cluster A/D/S bucket lists over the sort-by-id + clustered members (agg-bucket-votes-for-tid, conversation.clj:601-608), + not whole-matrix int totals. +- pca carries comment-projection + comment-extremity + (with-proj-and-extremtiy, conversation.clj:341-352). +- Mean/projection-derived floats are negated at emission: Delphi's matrix is + the NEGATION of Clojure's (AGREE=+1 vs AGREE=-1), comps are + covariance-derived and already equal, so pca.center, base-clusters.x/y, + group-clusters centers, comment-projection negate (verified empirically: + max|clj+py| = 1e-16 on center, <=1.4e-7 on x/y, vw single-cut). +- repness is {gid: [selected entries]} with finalize-cmt-stats key names + (repness.clj:173-188), direction by rat > rdt. +- mod-in / mod-out / lastModTimestamp are None until moderation is applied + (Clojure conv state holds no :mod-in until the poller delivers moderation). + +Improved-mode emission stays byte-for-byte as before (regression-guarded +here; the unfolded-pids contract is separately pinned by +test_serialization_unfolding.py, which runs in the default improved mode). +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from polismath.conversation.conversation import Conversation +from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR + + +# --------------------------------------------------------------------------- +# Fixture: polarized conversation + one under-threshold voter. +# --------------------------------------------------------------------------- +def _make_votes(): + """20 clustered participants (two polarized groups) + p20, who casts only + 3 votes — below the min(7, n_cmts) inclusion threshold — so their votes + exist in the matrix but they are NOT clustered. Distinguishes the + clustered-members aggregation domain (Clojure votes-base) from the + whole-matrix domain (improved totals).""" + votes = [] + for i in range(20): + pid = i + sign = 1 if i < 10 else -1 + for j in range(10): + votes.append({"pid": pid, "tid": j, "vote": sign if j < 5 else -sign}) + for j in range(3): + votes.append({"pid": 20, "tid": j, "vote": 1}) + # tid 10: voted ONLY by group A (p0-p9, all agree) — group B has S=0 on + # it, exercising Clojure's unconditional (A+1)/(S+2) factor for zero-S + # groups in group-aware-consensus (conversation.clj:633-653). + for i in range(10): + votes.append({"pid": i, "tid": 10, "vote": 1}) + return {"votes": votes, "lastVoteTimestamp": 1700000000000} + + +@pytest.fixture(scope="module") +def conv(): + c = Conversation("legacy_blob_shape") + c = c.update_votes(_make_votes(), recompute=False) + c = c.recompute() + assert len(c.base_clusters) > 0 + assert len(c.group_clusters) >= 2 + assert c.repness and c.repness.get("group_repness") + return c + + +@pytest.fixture() +def legacy(monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, "clojure-legacy") + + +@pytest.fixture() +def improved(monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, "improved") + + +def _sorted_base_clusters(conv): + return sorted(conv.base_clusters, key=lambda c: c["id"]) + + +# --------------------------------------------------------------------------- +# group-clusters: members are bids in legacy mode. +# --------------------------------------------------------------------------- +def test_legacy_group_clusters_members_are_bids(conv, legacy): + result = conv.to_dict() + bc_ids = {c["id"] for c in conv.base_clusters} + for gc in result["group-clusters"]: + assert set(gc["members"]) <= bc_ids, ( + f"legacy group-clusters[{gc['id']}].members must be base-cluster " + f"ids, got {gc['members']}" + ) + # partition of base clusters: union of members covers every bid exactly once + all_members = [m for gc in result["group-clusters"] for m in gc["members"]] + assert sorted(all_members) == sorted(bc_ids) + + +def test_improved_group_clusters_members_stay_pids(conv, improved): + result = conv.to_dict() + pids = set(conv.rating_mat.index) + for gc in result["group-clusters"]: + assert set(gc["members"]) <= pids + + +# --------------------------------------------------------------------------- +# votes-base: per-base-cluster bucket lists in legacy mode. +# --------------------------------------------------------------------------- +def test_legacy_votes_base_is_bucketed_lists(conv, legacy): + result = conv.to_dict() + n_buckets = len(conv.base_clusters) + vb = result["votes-base"] + assert len(vb) == conv.rating_mat.shape[1] + for tid, entry in vb.items(): + for k in ("A", "D", "S"): + assert isinstance(entry[k], list), f"votes-base[{tid}][{k}] must be a list" + assert len(entry[k]) == n_buckets + + # Buckets align to sort-by-id base clusters: recompute one tid by hand. + buckets = [c["members"] for c in _sorted_base_clusters(conv)] + mat = conv.raw_rating_mat + tid0 = list(mat.columns)[0] + expect_A = [int(sum(1 for p in b if mat.at[p, tid0] == 1)) for b in buckets] + expect_D = [int(sum(1 for p in b if mat.at[p, tid0] == -1)) for b in buckets] + expect_S = [ + int(sum(1 for p in b if not np.isnan(mat.at[p, tid0]))) for b in buckets + ] + key = tid0 if tid0 in vb else int(tid0) + assert vb[key]["A"] == expect_A + assert vb[key]["D"] == expect_D + assert vb[key]["S"] == expect_S + + +def test_legacy_votes_base_excludes_unclustered_votes(conv, legacy): + """p20 voted agree on tids 0-2 but is not clustered — Clojure's votes-base + never sees those votes (aggregation runs over bid-to-pid members only).""" + result = conv.to_dict() + vb = result["votes-base"] + clustered = {p for c in conv.base_clusters for p in c["members"]} + assert 20 not in clustered, "fixture broken: p20 must stay unclustered" + for tid in (0, 1, 2): + entry = vb[tid if tid in vb else str(tid)] + # 10 group-A members agreed on tids 0-2; p20's agree must NOT appear. + assert sum(entry["A"]) == 10 + assert sum(entry["S"]) == 20 + + +def test_improved_votes_base_stays_int_totals(conv, improved): + result = conv.to_dict() + entry = next(iter(result["votes-base"].values())) + assert isinstance(entry["A"], int) + assert isinstance(entry["D"], int) + assert isinstance(entry["S"], int) + + +# --------------------------------------------------------------------------- +# pca: comment-projection / comment-extremity emitted + sign parity. +# --------------------------------------------------------------------------- +def test_legacy_pca_emits_comment_projection_and_extremity(conv, legacy): + result = conv.to_dict() + pca = result["pca"] + n_cmts = conv.rating_mat.shape[1] + assert "comment-projection" in pca and "comment-extremity" in pca + # Clojure transposes cmnt-proj: rows are components, tid-aligned. + cp = pca["comment-projection"] + assert len(cp) == len(pca["comps"]) + assert all(len(row) == n_cmts for row in cp) + assert len(pca["comment-extremity"]) == n_cmts + # extremity is the column norm of the (sign-invariant) projection + norms = np.linalg.norm(np.asarray(cp), axis=0) + np.testing.assert_allclose(pca["comment-extremity"], norms, rtol=1e-9) + + +def test_legacy_sign_negation_of_center_and_projections(conv, legacy): + result = conv.to_dict() + # pca.center emits the NEGATION of the internal (Delphi-convention) center + np.testing.assert_allclose( + result["pca"]["center"], -np.asarray(conv.pca["center"]), rtol=0, atol=0 + ) + # comps emit unchanged (covariance-derived) + np.testing.assert_allclose(result["pca"]["comps"], np.asarray(conv.pca["comps"])) + # base-clusters x/y emit the negation of the internal cluster centers + bc = result["base-clusters"] + by_id = {c["id"]: c for c in conv.base_clusters} + for i, bid in enumerate(bc["id"]): + assert bc["x"][i] == pytest.approx(-by_id[bid]["center"][0]) + assert bc["y"][i] == pytest.approx(-by_id[bid]["center"][1]) + # group-clusters centers negate too + gby_id = {g["id"]: g for g in conv.group_clusters} + for gc in result["group-clusters"]: + np.testing.assert_allclose( + gc["center"], -np.asarray(gby_id[gc["id"]]["center"]) + ) + # comment-projection is the negation of the internal D12 projection + from polismath.pca_kmeans_rep.pca import pca_project_cmnts + + internal = pca_project_cmnts( + np.asarray(conv.pca["center"]), np.asarray(conv.pca["comps"]) + ) + np.testing.assert_allclose( + result["pca"]["comment-projection"], -internal.T, rtol=1e-12 + ) + + +def test_improved_pca_emission_unchanged(conv, improved): + result = conv.to_dict() + np.testing.assert_allclose(result["pca"]["center"], np.asarray(conv.pca["center"])) + assert "comment-projection" not in result["pca"] + bc = result["base-clusters"] + by_id = {c["id"]: c for c in conv.base_clusters} + for i, bid in enumerate(bc["id"]): + assert bc["x"][i] == pytest.approx(by_id[bid]["center"][0]) + + +# --------------------------------------------------------------------------- +# repness: Clojure finalize-cmt-stats shape in legacy mode. +# --------------------------------------------------------------------------- +def test_legacy_repness_shape_and_direction_mapping(conv, legacy): + result = conv.to_dict() + rep = result["repness"] + internal = conv.repness["group_repness"] + assert set(rep.keys()) == set(internal.keys()) + for gid, entries in rep.items(): + assert entries, f"group {gid} has no selected repness entries" + for got, src in zip(entries, internal[gid]): + repful = "agree" if src["rat"] > src["rdt"] else "disagree" + assert got["tid"] == src["comment_id"] + assert got["repful-for"] == repful + assert got["n-trials"] == src["ns"] + if repful == "agree": + assert got["n-success"] == src["na"] + assert got["p-success"] == pytest.approx(src["pa"]) + assert got["p-test"] == pytest.approx(src["pat"]) + assert got["repness"] == pytest.approx(src["ra"]) + assert got["repness-test"] == pytest.approx(src["rat"], rel=1e-6) + else: + assert got["n-success"] == src["nd"] + assert got["p-success"] == pytest.approx(src["pd"]) + assert got["p-test"] == pytest.approx(src["pdt"]) + assert got["repness"] == pytest.approx(src["rd"]) + assert got["repness-test"] == pytest.approx(src["rdt"], rel=1e-6) + # best-agree entries carry the two extra Clojure keys + if src.get("best_agree"): + assert got["best-agree"] is True + assert got["n-agree"] == src["n_agree"] + else: + assert "best-agree" not in got and "n-agree" not in got + # no internal spellings leak into the legacy blob + assert "comment_id" not in got and "na" not in got and "rat" not in got + + +def test_improved_repness_stays_internal_shape(conv, improved): + result = conv.to_dict() + assert set(result["repness"].keys()) == { + "comment_ids", "group_repness", "comment_repness", "consensus_comments", + } + + +# --------------------------------------------------------------------------- +# repness rest-domain: "other" = the OTHER GROUPS only in legacy mode. +# --------------------------------------------------------------------------- +def _rest_domain_fixture(): + import pandas as pd + + votes_long = pd.DataFrame( + [ + {"participant": 1, "comment": 0, "vote": 1}, + {"participant": 2, "comment": 0, "vote": 1}, + {"participant": 3, "comment": 0, "vote": -1}, + {"participant": 4, "comment": 0, "vote": -1}, + # unclustered voter — in the matrix, in no group + {"participant": 99, "comment": 0, "vote": 1}, + ] + ) + groups = [{"id": 0, "members": [1, 2]}, {"id": 1, "members": [3, 4]}] + return votes_long, groups + + +def test_legacy_repness_rest_domain_excludes_unclustered(legacy): + """Clojure's rest-stats sum ONLY over the other groups' (clustered) + members (utils/mapv-rest over per-group comment-stats, repness.clj:125-131 + + group-members unfolding) — an unclustered participant's votes never + enter the comparison (FP-69c7a13580 / FP-faac8c6125 root).""" + from polismath.pca_kmeans_rep.repness import compute_group_comment_stats_df + + votes_long, groups = _rest_domain_fixture() + df = compute_group_comment_stats_df(votes_long, groups) + row = df.loc[(0, 0)] + # group 0: na=2 ns=2 → pa = 3/4. rest = group 1 only: na=0 ns=2 → + # other_pa = (0+1)/(2+2) = 1/4. ra = 3. + assert row["pa"] == pytest.approx(0.75) + assert row["ra"] == pytest.approx(3.0) + + +def test_improved_repness_rest_domain_includes_all_voters(improved): + from polismath.pca_kmeans_rep.repness import compute_group_comment_stats_df + + votes_long, groups = _rest_domain_fixture() + df = compute_group_comment_stats_df(votes_long, groups) + row = df.loc[(0, 0)] + # rest = group 1 + p99: na=1 ns=3 → other_pa = (1+1)/(3+2) = 0.4; ra = 1.875 + assert row["ra"] == pytest.approx(0.75 / 0.4) + + +# --------------------------------------------------------------------------- +# group-aware-consensus: zero-S groups contribute (A+1)/(S+2) = 1/2 in legacy. +# --------------------------------------------------------------------------- +def _gac_group_stats(result, tid): + stats = {} + for gid, gdata in result["group-votes"].items(): + vs = gdata["votes"].get(tid, gdata["votes"].get(str(tid), {})) + stats[gid] = (vs.get("A", 0), vs.get("S", 0)) + return stats + + +def test_legacy_gac_multiplies_zero_s_groups(conv, legacy): + """Clojure multiplies (A+1)/(S+2) over EVERY group — a zero-S group + contributes 1/2 (conversation.clj:639-641, `:or {A 0 S 0}`), it is not + skipped. tid 10 was voted on only by group A members.""" + result = conv.to_dict() + stats = _gac_group_stats(result, 10) + assert any(s == 0 for _, s in stats.values()), ( + f"fixture broken: expected a zero-S group on tid 10, stats={stats}" + ) + expected = 1.0 + for a, s in stats.values(): + expected *= (a + 1.0) / (s + 2.0) + assert result["group-aware-consensus"][10] == pytest.approx(expected) + + +def test_improved_gac_skips_zero_s_groups(conv, improved): + result = conv.to_dict() + stats = _gac_group_stats(result, 10) + expected = 1.0 + for a, s in stats.values(): + if s > 0: + expected *= (a + 1.0) / (s + 2.0) + assert result["group-aware-consensus"][10] == pytest.approx(expected) + + +# --------------------------------------------------------------------------- +# moderation-state semantics: None until moderation applied (legacy). +# --------------------------------------------------------------------------- +def test_legacy_mod_keys_none_until_moderation(conv, legacy): + result = conv.to_dict() + assert result["mod-in"] is None + assert result["mod-out"] is None + assert result["lastModTimestamp"] is None + + +def test_legacy_mod_keys_populated_after_moderation(conv, legacy): + moderated = conv.update_moderation({"mod_out_tids": [3]}, recompute=False) + result = moderated.to_dict() + assert result["mod-out"] == [3] + assert result["mod-in"] == [] + # no moderation timestamp was supplied — stays None (vote-only replays + # match Clojure's null; real poller feeds will carry one) + assert result["lastModTimestamp"] is None + + +def test_improved_mod_keys_stay_lists(conv, improved): + result = conv.to_dict() + assert result["mod-in"] == [] + assert result["mod-out"] == [] + assert result["lastModTimestamp"] == conv.last_updated + + +# --------------------------------------------------------------------------- +# Arrival-order parity: Clojure's named-matrix column order is first-vote +# arrival order (update-nmat appends unseen colnames in encounter order); +# python's internal matrix is natsorted (conversation.py:414). Ties in +# repness/consensus selection resolve by stable sort over COLUMN order, so +# legacy mode tracks arrival order and uses it for tie-breaking + emission. +# --------------------------------------------------------------------------- +def test_tid_arrival_order_tracked_across_updates(): + c = Conversation("arrival_tracking") + c = c.update_votes( + {"votes": [ + {"pid": 1, "tid": 5, "vote": 1}, + {"pid": 1, "tid": 2, "vote": 1}, + {"pid": 2, "tid": 9, "vote": -1}, + {"pid": 2, "tid": 2, "vote": 1}, + {"pid": 3, "tid": 5, "vote": 0}, + ]}, + recompute=False, + ) + assert c.tid_arrival_order == [5, 2, 9] + c = c.update_votes( + {"votes": [{"pid": 3, "tid": 1, "vote": 1}, {"pid": 3, "tid": 9, "vote": 1}]}, + recompute=False, + ) + assert c.tid_arrival_order == [5, 2, 9, 1] + + +def test_legacy_tids_emitted_in_arrival_order_with_aligned_pca(conv, legacy): + result = conv.to_dict() + assert result["tids"] == conv.tid_arrival_order + # pca arrays must be re-aligned to the emitted tid order: emitted center[i] + # is the (negated) internal center entry for tids[i]. + internal_center = dict(zip(conv.rating_mat.columns, conv.pca["center"])) + for i, tid in enumerate(result["tids"]): + assert result["pca"]["center"][i] == pytest.approx(-internal_center[tid]) + from polismath.pca_kmeans_rep.pca import pca_project_cmnts + + internal_ext = np.linalg.norm( + pca_project_cmnts( + np.asarray(conv.pca["center"]), np.asarray(conv.pca["comps"]) + ), + axis=1, + ) + ext = dict(zip(conv.rating_mat.columns, internal_ext)) + for i, tid in enumerate(result["tids"]): + assert result["pca"]["comment-extremity"][i] == pytest.approx(ext[tid]) + + +def test_improved_tids_stay_natsorted(conv, improved): + result = conv.to_dict() + assert result["tids"] == list(conv.rating_mat.columns) + + +def test_legacy_from_dict_restores_arrival_order(conv, legacy): + restored = Conversation.from_dict(conv.to_dict()) + assert restored.tid_arrival_order == conv.tid_arrival_order + + +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 + tid_order=[7, 3], 7 must precede 3; without, ascending order wins.""" + import pandas as pd + + from polismath.pca_kmeans_rep.repness import conv_repness + + mat = pd.DataFrame( + {3: [1, 1, -1, -1], 7: [1, 1, -1, -1]}, index=[1, 2, 3, 4] + ) + groups = [{"id": 0, "members": [1, 2]}, {"id": 1, "members": [3, 4]}] + ordered = conv_repness(mat, groups, tid_order=[7, 3]) + tids = [e["comment_id"] for e in ordered["group_repness"][0]] + assert tids == [7, 3] + default = conv_repness(mat, groups) + tids = [e["comment_id"] for e in default["group_repness"][0]] + assert tids == [3, 7] + + +def test_conv_repness_consensus_tie_break_follows_tid_order(): + """Universal-agree clones tie on the consensus agree metric; tid_order + decides their relative rank (Clojure stable sort over column order).""" + import pandas as pd + + from polismath.pca_kmeans_rep.repness import conv_repness + + mat = pd.DataFrame( + { + 3: [1, 1, 1, 1], + 7: [1, 1, 1, 1], + 0: [1, -1, 1, -1], + }, + index=[1, 2, 3, 4], + ) + groups = [{"id": 0, "members": [1, 2]}, {"id": 1, "members": [3, 4]}] + ordered = conv_repness(mat, groups, tid_order=[7, 0, 3]) + agree_tids = [e["tid"] for e in ordered["consensus_comments"]["agree"]] + assert agree_tids[:2] == [7, 3] + default = conv_repness(mat, groups) + agree_tids = [e["tid"] for e in default["consensus_comments"]["agree"]] + assert agree_tids[:2] == [3, 7] + + +# --------------------------------------------------------------------------- +# from_dict inverse: legacy round-trip restores the internal convention. +# --------------------------------------------------------------------------- +def test_legacy_from_dict_unpermutes_pca_alignment(legacy): + """Legacy blobs emit tids (and pca arrays) in ARRIVAL order; internal + state is natsorted-aligned. from_dict must invert the permutation as well + as the sign, or a warm restore seeds PCA with column-misaligned + center/comps (review finding on #2649 — the plain round-trip fixture + below can't catch it because its arrival order is ascending).""" + arrival = [5, 2, 9, 0, 7, 1, 3, 4, 6, 8] + votes = [] + for pid in range(20): + sign = 1 if pid < 10 else -1 + for j, tid in enumerate(arrival): + votes.append( + {"pid": pid, "tid": tid, "vote": sign if j < 5 else -sign} + ) + c = Conversation("roundtrip_perm") + c = c.update_votes({"votes": votes}, recompute=False) + c = c.recompute() + assert c.tid_arrival_order != sorted(c.tid_arrival_order) + restored = Conversation.from_dict(c.to_dict()) + np.testing.assert_allclose( + np.asarray(restored.pca["center"]), np.asarray(c.pca["center"]) + ) + np.testing.assert_allclose( + np.asarray(restored.pca["comps"]), np.asarray(c.pca["comps"]) + ) + + +def test_legacy_from_dict_round_trips_center_sign(conv, legacy): + restored = Conversation.from_dict(conv.to_dict()) + np.testing.assert_allclose( + np.asarray(restored.pca["center"]), np.asarray(conv.pca["center"]) + ) + + +def test_improved_from_dict_round_trips_center_sign(conv, improved): + restored = Conversation.from_dict(conv.to_dict()) + np.testing.assert_allclose( + np.asarray(restored.pca["center"]), np.asarray(conv.pca["center"]) + )