feat(delphi): plumb D11 consensus + D12 priorities through to_dict / …

delphi/docs/PLAN_DISCREPANCY_FIXES.md

@@ -28,10 +28,11 @@ This plan's "PR N" labels map to actual GitHub PRs as follows:
 | PR 7 (D8) | #2522 | Stack 15/17 | Fix D8: finalize comment stats |
 | PR 12 (D15) | #2523 | Stack 16/17 | Fix D15: moderation handling |
 | (K-inv) | #2524 | Stack 17/17 | Fix K-means k divergence: preserve row order |
-| PR 8 (D10) | — (WIP) | — | Fix D10: rep comment selection — **NEEDS REWORK** |
-| PR 9 (D11) | — (WIP) | — | Fix D11: consensus selection — **NEEDS REWORK** |
+| PR 14a (scalar deletion) | #2564 | — | Delete dead scalar paths in `repness.py`; migrate blob injection tests to vectorized |
+| PR 8 (D10) | #2566 | — | Fix D10: rep comment selection — single-pass reduce matching Clojure |
+| PR 9 (D11) | #2567 | — | Fix D11: consensus selection — whole-conv stats + per-side top-5 matching Clojure |
 | PR 10 (D3) | — (WIP) | — | Fix D3: k-smoother buffer — **NEEDS REWORK** |
-| PR 11 (D12) | — (WIP) | — | Fix D12: comment priorities — **NEEDS REWORK** |
+| PR 11 (D12) | — (in flight) | — | Fix D12: comment priorities — Clojure-parity importance/priority metrics + PCA comment projection |
 | PR 13 (D1) | — (WIP) | — | Fix D1: PCA sign flip prevention — **NEEDS REWORK** |
 | PR 15 | — (WIP) | — | Fix load_votes timestamp ordering — **NEEDS REWORK** |
 
@@ -914,3 +915,30 @@ Tagging this as a follow-up. No code changes until we discuss.
 - **`to_dynamo_dict` parallel inline implementations** were refactored to
   route through the same helpers as `to_dict` in PR #2523 follow-up. No
   further action needed.
+- **`ns` includes-PASS-divergence** (DISCOVERED 2026-06-11 during D11). Clojure's
+  `:ns` (via `count-votes` with `filter identity` — repness.clj:56-61) INCLUDES
+  PASS votes. Python's `compute_group_comment_stats_df` and `consensus_stats_df`
+  both compute `ns = na + nd`, excluding PASS. This is a real divergence that
+  affects `pa, pd, pat, pdt, ra, rd, rat, rdt` everywhere — every downstream
+  metric and selection. The D5 PR #2519 journal claim ("PASS NOT included,
+  matching Clojure") was based on a misreading of `count-votes`. Currently
+  causing 3-5% divergence in pat values for tids with non-zero PASS counts;
+  visible at the consensus-selection margins (4/6 overlap on vw cold_start
+  agree, 1/3 on disagree). Needs a dedicated PR — affects:
+    - `compute_group_comment_stats_df` (line ~283: `ns = na + nd`).
+    - `consensus_stats_df` (line ~435: `ns = na + nd`).
+  Fix: `ns = (vote_matrix_df != 0).sum(axis=0)` no, actually we want to
+  count non-NaN: `ns = vote_matrix_df.notna().sum(axis=0)` for wide format;
+  for long-format `votes_long.groupby('comment').size()` after dropna.
+  Re-record goldens afterward.
+- **D10 take-5 eviction edge case** (2026-06-11). The Clojure-parity
+  `select_rep_comments_df` introduced in PR 8 mirrors Clojure exactly:
+  `take(5)` runs AFTER prepending the `best_agree` slot. When `best_agree`
+  was kept by `beats_best_agr?` as a non-significant agree-priority fallback
+  (i.e. it failed `passes_by_test?` but qualified via Branch 4) AND
+  `:sufficient` already has 5 entries, the prepend pushes the total to 6
+  and `take(5)` silently evicts the 5th-highest-metric `sufficient` entry
+  — possibly a strong dissenting view. Mirrored for blob parity; flag for
+  future product review. See `# TODO(parity-eviction)` in
+  `delphi/polismath/pca_kmeans_rep/repness.py::select_rep_comments_df` and
+  the synthetic test `TestD10SelectRepCommentsBoundary::test_take_5_eviction_when_best_agree_outside_sufficient`.

delphi/polismath/conversation/conversation.py

@@ -15,7 +15,11 @@
 from datetime import datetime
 from natsort import natsorted
 
-from polismath.pca_kmeans_rep.pca import pca_project_dataframe
+from polismath.pca_kmeans_rep.pca import (
+    pca_project_dataframe,
+    pca_project_cmnts,
+    compute_comment_extremity,
+)
 from polismath.pca_kmeans_rep.clusters import (
     kmeans_sklearn,
     calculate_silhouette_sklearn
@@ -37,6 +41,88 @@
     logger.setLevel(logging.INFO)
 
 
+# =============================================================================
+# D12: Comment-priority metrics (Clojure parity)
+# =============================================================================
+#
+# Ports of `importance-metric` and `priority-metric` from Clojure
+# (math/src/polismath/math/conversation.clj:311-330). Public so they can be
+# unit-tested in isolation.
+
+META_PRIORITY = 7  # Clojure: meta-priority (conversation.clj:319). "TODO TUNE."
+
+
+def importance_metric(A: float, P: float, S: float, E: float) -> float:
+    """
+    Clojure importance-metric (conversation.clj:311-315).
+
+        (defn importance-metric
+          [A P S E]
+          (let [p (/ (+ P 1) (+ S 2))
+                a (/ (+ A 1) (+ S 2))]
+            (* (- 1 p) (+ E 1) a)))
+
+    Smoothed (Beta(2,2)) probability of pass `p`, smoothed agree `a`, with
+    extremity boost `(E + 1)`. Higher when fewer passes, more agrees, more
+    extreme (higher PCA extremity).
+
+    Args:
+        A: agree count (across all groups).
+        P: pass count = S - (A + D) across all groups.
+        S: seen count (total votes seen — agree + disagree + pass).
+        E: comment extremity (L2 norm of PCA projection).
+    """
+    p = (P + 1) / (S + 2)
+    a = (A + 1) / (S + 2)
+    return (1 - p) * (E + 1) * a
+
+
+def priority_metric(is_meta: bool,
+                    A: float, P: float, S: float, E: float) -> float:
+    """
+    Clojure priority-metric (conversation.clj:321-330).
+
+        (defn priority-metric
+          [is-meta A P S E]
+          (matrix/pow
+            (if is-meta
+              meta-priority
+              (* (importance-metric A P S E)
+                 (+ 1 (* 8 (matrix/pow 2 (/ S -5))))))
+            2))
+
+    Squared to deepen bias (toward extremes). Meta comments get a constant
+    `META_PRIORITY^2 = 49`. Non-meta comments get `importance * decay`, where
+    the decay factor `1 + 8 * 2^(-S/5)` lets new (low-S) comments bubble up
+    and fades as more votes accumulate.
+
+    Args:
+        is_meta: True for meta comments (treated as constant priority).
+        A, P, S, E: see `importance_metric`.
+
+    Returns:
+        Squared priority value.
+    """
+    # TODO(clojure-parity-bug): Clojure (conversation.clj:325) treats meta-tid
+    # value 0 as TRUTHY in (if is-meta ...), so every tid takes the meta branch.
+    # We mirror this bug for byte-for-byte Clojure parity. Switch back to
+    # honoring `is_meta` once the GitHub issue resolves:
+    # https://github.com/compdemocracy/polis/issues/2571
+    # Original semantic-correct code preserved below for reference and future
+    # restoration.
+    #
+    # Clojure-parity-bug-mirror: ALWAYS take the meta branch, ignoring is_meta.
+    return META_PRIORITY ** 2
+
+    # Original semantically-correct logic, restore when Clojure bug is fixed:
+    # if is_meta:
+    #     inner = META_PRIORITY
+    # else:
+    #     decay_factor = 1 + 8 * (2 ** (-S / 5))
+    #     inner = importance_metric(A, P, S, E) * decay_factor
+    # return inner ** 2
+
+
 class Conversation:
     """
     Manages the state and computation for a Pol.is conversation.
@@ -82,6 +168,7 @@ def __init__(self,
         self.participant_info = {}
         self.vote_stats = {}
         self.group_votes = {}  # Initialize group_votes to avoid attribute errors
+        self.comment_priorities: Dict[Any, float] = {}  # D12 (PR 11)
 
         # Initialize with votes if provided
         if votes:
@@ -753,16 +840,22 @@ def _compute_repness(self) -> None:
 
         # Check if we have groups
         if not self.group_clusters:
+            # B1 fix (D11 sub-agent review): consensus_comments must always be
+            # `{'agree': [], 'disagree': []}` (dict) post-D11, never `[]` (list).
             self.repness = {
                 'comment_ids': list(self.rating_mat.columns),
                 'group_repness': {},
-                'consensus_comments': []
+                'consensus_comments': {'agree': [], 'disagree': []}
             }
             logger.info(f"Representativeness completed in {time.time() - start_time:.2f}s (no groups)")
             return
 
-        # Compute representativeness (needs participant IDs, not base-cluster IDs)
-        self.repness = conv_repness(self.rating_mat, self._unfolded_group_clusters())
+        # 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).
+        self.repness = conv_repness(self.rating_mat,
+                                    self._unfolded_group_clusters(),
+                                    mod_out=self.mod_out_tids)
         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]:
@@ -1001,11 +1094,78 @@ def recompute(self) -> 'Conversation':
 
         # Compute representativeness
         result._compute_repness()
-
+
+        # Compute comment priorities (D12 / PR 11). Needs PCA + group_votes.
+        result._compute_comment_priorities()
+
         # Compute participant info
         result._compute_participant_info()
-        
+
         return result
+
+    def _compute_comment_priorities(self) -> Dict[Any, float]:
+        """
+        Compute per-tid comment priorities matching Clojure
+        `:comment-priorities` (conversation.clj:648-679).
+
+        Per-tid: sum A/D/S across all groups → P = S - (A + D) → call
+        `priority_metric(is_meta, A, P, S, E)` where E is the comment
+        extremity computed from PCA.
+
+        Stores the result on `self.comment_priorities` and also returns it.
+        TS server `nextComment.ts::getNextPrioritizedComment` consumes this
+        for weighted comment routing — pre-D12 Python emitted nothing, so
+        the server fell back to uniform random selection.
+        """
+        if self.pca is None or self.rating_mat is None or self.rating_mat.empty:
+            self.comment_priorities = {}
+            return self.comment_priorities
+
+        center = np.asarray(self.pca.get('center'))
+        comps = np.asarray(self.pca.get('comps'))
+        if center.size == 0 or comps.size == 0:
+            self.comment_priorities = {}
+            return self.comment_priorities
+
+        # Comment projection + extremity (Clojure with-proj-and-extremtiy,
+        # conversation.clj:341-352).
+        cmnt_proj = pca_project_cmnts(center, comps)
+        extremity_arr = compute_comment_extremity(cmnt_proj)
+
+        # Column order of `center`/`comps`/`extremity_arr` matches
+        # `self.rating_mat.columns` (PCA is computed on rating_mat).
+        tid_extremity = dict(zip(self.rating_mat.columns, extremity_arr))
+
+        # Per-group A/D/S aggregation. `_compute_group_votes` returns
+        # {str(gid): {'n-members': N, 'votes': {tid: {A, D, S}}}}. S includes
+        # PASS (line ~1222: `np.sum(~np.isnan(votes))`), matching Clojure.
+        group_votes = self._compute_group_votes()
+
+        priorities: Dict[Any, float] = {}
+        for tid in self.rating_mat.columns:
+            A_total = 0
+            D_total = 0
+            S_total = 0
+            for gv_data in group_votes.values():
+                votes_for_tid = gv_data.get('votes', {}).get(
+                    tid, {'A': 0, 'D': 0, 'S': 0})
+                A_total += votes_for_tid.get('A', 0)
+                D_total += votes_for_tid.get('D', 0)
+                S_total += votes_for_tid.get('S', 0)
+            # Clojure: P = S - (A + D)  (conversation.clj:661).
+            P_total = S_total - (A_total + D_total)
+            E = float(tid_extremity.get(tid, 0))
+            is_meta = tid in self.meta_tids
+            # Match key type with the rest of the codebase (int when possible).
+            try:
+                tid_key = int(tid)
+            except (ValueError, TypeError):
+                tid_key = tid
+            priorities[tid_key] = float(priority_metric(
+                is_meta, A_total, P_total, S_total, E))
+
+        self.comment_priorities = priorities
+        return priorities
 
     def get_summary(self) -> Dict[str, Any]:
         """
@@ -1731,12 +1891,15 @@ def numpy_to_list(arr):
         # a list-of-dicts format that would break server/src/report.ts,
         # server/src/utils/pca.ts, and client-participation-alpha consumers.
 
-        # Add empty consensus structure for compatibility
-        result['consensus'] = {
-            'agree': [],
-            'disagree': [],
-            'comment-stats': {}
-        }
+        # Surface D11 consensus comments (Clojure parity: client-report's Majority
+        # view consumes result['consensus']). Pre-Investigation-B this block was
+        # hardcoded empty, which silently zeroed the Majority view regardless of
+        # the D11 selection. Falls back to the empty shape when repness is missing
+        # or did not produce a consensus_comments dict (older blobs, no-group convs).
+        result['consensus'] = (
+            self.repness.get('consensus_comments', {'agree': [], 'disagree': []})
+            if self.repness else {'agree': [], 'disagree': []}
+        )
 
         # Add math_tick value
         current_time = int(time.time())
@@ -2272,12 +2435,14 @@ def float_to_decimal(obj):
             }
             result['pca'] = float_to_decimal(pca_data)
 
-        # Add consensus structure
-        result['consensus'] = {
-            'agree': [],
-            'disagree': [],
-            'comment_stats': {}
-        }
+        # Surface D11 consensus comments (Clojure parity). Pre-Investigation-B
+        # this block was hardcoded empty, so the DynamoDB blob never carried the
+        # D11 dict even when repness produced one. Falls back to the empty shape
+        # when repness is missing or didn't produce consensus_comments.
+        result['consensus'] = (
+            self.repness.get('consensus_comments', {'agree': [], 'disagree': []})
+            if self.repness else {'agree': [], 'disagree': []}
+        )
 
         # Add math_tick value
         current_time = int(time.time())

delphi/polismath/database/dynamodb.py

@@ -300,6 +300,13 @@ def write_conversation(self, conv) -> bool:
             if analysis_table:
                 if dynamo_data:
                     # Use pre-formatted data
+                    # D11 cascade fix (Investigation B, Site 1): source the FULL
+                    # consensus dict from repness.consensus_comments, preserving
+                    # both `agree` and `disagree` lists. The old code dropped
+                    # `disagree` entirely (`.get('consensus', {}).get('agree', [])`).
+                    consensus_comments = dynamo_data.get('repness', {}).get(
+                        'consensus_comments', {'agree': [], 'disagree': []}
+                    )
                     analysis_table.put_item(Item={
                         'zid': zid,
                         'math_tick': math_tick,
@@ -308,7 +315,7 @@ def write_conversation(self, conv) -> bool:
                         'comment_count': dynamo_data.get('comment_count', 0),
                         'group_count': dynamo_data.get('group_count', 0),
                         'pca': dynamo_data.get('pca', {}),
-                        'consensus_comments': dynamo_data.get('consensus', {}).get('agree', [])
+                        'consensus_comments': consensus_comments
                     })
                 else:
                     # Legacy format
@@ -321,11 +328,21 @@ def write_conversation(self, conv) -> bool:
                         }
                         # Replace floats with Decimal for DynamoDB
                         pca_data = self._replace_floats_with_decimals(pca_data)
-
-                    # Create the analysis record with Decimal conversion
-                    consensus_comments = self._numpy_to_list(conv.consensus) if hasattr(conv, 'consensus') else []
+
+                    # D11 cascade fix (Investigation B, Site 2): the old code
+                    # sourced from `conv.consensus`, which is always `[]` post-D11
+                    # (the attribute was deprecated). Source from
+                    # `conv.repness['consensus_comments']` instead — the new shape
+                    # is `{'agree': [...], 'disagree': [...]}`.
+                    if hasattr(conv, 'repness') and conv.repness:
+                        consensus_comments = conv.repness.get(
+                            'consensus_comments', {'agree': [], 'disagree': []}
+                        )
+                    else:
+                        consensus_comments = {'agree': [], 'disagree': []}
+                    consensus_comments = self._numpy_to_list(consensus_comments)
                     consensus_comments = self._replace_floats_with_decimals(consensus_comments)
-                    
+
                     analysis_table.put_item(Item={
                         'zid': zid,
                         'math_tick': math_tick,
@@ -840,7 +857,13 @@ def read_math_by_tick(self, zid: str, math_tick: int) -> Dict[str, Any]:
                         }
 
                     # Set consensus
-                    result['consensus'] = analysis.get('consensus_comments', [])
+                    # D11 cascade fix (Investigation B, Site 3): default to the
+                    # new dict shape `{'agree': [], 'disagree': []}` rather than
+                    # the obsolete empty list `[]`, so downstream consumers
+                    # always receive a uniformly-shaped value.
+                    result['consensus'] = analysis.get(
+                        'consensus_comments', {'agree': [], 'disagree': []}
+                    )
 
             # 2. Get groups data
             groups_table = self.tables.get('Delphi_KMeansClusters')