python-math #10: feat(math): in-conv greedy floor + persistent carry (PR-E)

delphi/polismath/conversation/conversation.py

@@ -233,6 +233,15 @@ def __init__(self,
         # them (conv_man.clj:52-74). Unused in the default 'improved' mode.
         self.group_clusterings: Dict[Any, Any] = {}  # k -> (labels, centers, member_lists, silhouette)
         self.group_k_smoother: Dict[str, Any] = {}   # {last_k, last_k_count, smoothed_k}
+        # Persistent in-conv set for 'clojure-legacy' mode (PR-E). Clojure keeps
+        # in-conv on the conv and UNIONS into it every tick, so greedily-admitted
+        # participants never leave (conversation.clj:243-269). Empty on the first
+        # tick; threaded in-memory across update_votes (deepcopy in recompute),
+        # NOT persisted to dynamo — same lifetime as the other warm-start state.
+        # Unused/ignored in the default 'improved' mode (threshold-only
+        # selection) — it can hold carry state from an earlier clojure-legacy
+        # tick if the mode is switched mid-process.
+        self.in_conv: Set[Any] = set()
         self.proj = {}
         self.repness = None
         self.consensus = []
@@ -1746,11 +1755,15 @@ def _compute_user_vote_counts(self) -> Dict[str, int]:
 
         return vote_counts
 
-    def _get_in_conv_participants(self) -> Set[str]:
-        """
-        Get participants who have voted enough to be included in clustering.
+    # Clojure greedy in-conv floor: if fewer than this many participants clear
+    # the vote threshold, greedily admit the top voters up to this count
+    # (conversation.clj:259 `greedy-n 15`).
+    IN_CONV_GREEDY_N = 15
 
-        Matches Clojure's in-conv logic from conversation.clj lines 239-266.
+    def _get_in_conv_participants(self) -> Set[Any]:
+        """
+        Get participants to include in clustering (Clojure :in-conv,
+        conversation.clj:243-269).
 
         Threshold: participant must have voted on at least min(7, n_comments)
         comments (Clojure parity fix D2).
@@ -1765,20 +1778,68 @@ def _get_in_conv_participants(self) -> Set[str]:
         MUST be persisted to DynamoDB. See compdemocracy/polis#2358 and
         Clojure's approach in conv_man.clj:55, conversation.clj:244.
 
+        In the default 'improved' mode this is exactly the threshold set (no
+        carry, no greedy floor) — today's behavior, unchanged. In
+        'clojure-legacy' mode it additionally ports the two Clojure steps the
+        Python pipeline was missing (conversation.clj:243-269):
+
+          1. CARRY: union into the PERSISTENT in-conv set carried on the conv
+             (`(or (:in-conv conv) #{})`, conversation.clj:247) so a participant,
+             once in, stays in — including greedy admits.
+          2. GREEDY FLOOR: if fewer than 15 participants are in, greedily admit
+             the top `15 - n_in` remaining participants by vote count descending
+             (conversation.clj:259-268), and PERSIST them in the carried set.
+
         Returns:
-            Set of participant IDs that meet the threshold
+            Set of participant IDs to feed base clustering.
         """
         n_cmts = len(self.raw_rating_mat.columns) if hasattr(self.raw_rating_mat, 'columns') else 0
         threshold = min(7, n_cmts)
 
-        # Get vote counts for all participants
+        # Get vote counts for all participants (raw_rating_mat, insertion/row
+        # order preserved — the deterministic greedy tie-break below relies on it).
         vote_counts = self._compute_user_vote_counts()
 
-        # Filter participants meeting threshold
-        in_conv = {pid for pid, count in vote_counts.items() if count >= threshold}
-
-        logger.info(f"Filtered {len(in_conv)}/{len(vote_counts)} participants meeting vote threshold {threshold:.1f}")
-
+        # Participants meeting the vote threshold (Clojure conversation.clj:249-256).
+        threshold_set = {pid for pid, count in vote_counts.items() if count >= threshold}
+
+        if resolve_engine_mode() != ENGINE_MODE_LEGACY:
+            # Improved (default): threshold set only — no carry, no greedy floor.
+            logger.info(f"Filtered {len(threshold_set)}/{len(vote_counts)} participants "
+                        f"meeting vote threshold {threshold:.1f}")
+            return threshold_set
+
+        # Legacy: carry forward the persisted in-conv set, then union the
+        # threshold set into it (Clojure `(into in-conv ...)`, conversation.clj:247-256).
+        # PRUNE the carry to participants still present in vote_counts first:
+        # vote_counts is keyed off rating_mat.index, which drops mod_out_ptpts
+        # (banned participants — a Python-only feature Clojure lacks). Without the
+        # intersection a participant banned AFTER being carried would linger in the
+        # set forever, inflating the size check so the greedy floor never re-fires
+        # to top the actually-clustered pool (proj ∩ in_conv) back up, and growing
+        # the carry unboundedly. Clojure-parity is unaffected (no ban feature there).
+        in_conv = (set(self.in_conv) & set(vote_counts.keys())) | threshold_set
+
+        # Greedy floor (conversation.clj:259-268): if under 15, admit the top
+        # remaining voters by count descending. Clojure sorts a hash-map with
+        # `(sort-by (comp - second))`, whose tie order among equal vote counts is
+        # hash-map iteration order (non-deterministic). We instead break ties by
+        # matrix ROW ORDER (vote_counts insertion order) via a STABLE sort — a
+        # deterministic, reproducible surrogate for an inherently underspecified
+        # Clojure tie case. Below-threshold participants ARE eligible here (the
+        # floor guarantees clustering has enough rows in tiny/early conversations).
+        greedy_n = self.IN_CONV_GREEDY_N
+        if len(in_conv) < greedy_n:
+            candidates = [pid for pid in vote_counts if pid not in in_conv]
+            candidates.sort(key=lambda pid: -vote_counts[pid])  # stable -> row-order ties
+            in_conv.update(candidates[:greedy_n - len(in_conv)])
+
+        # Persist for the next tick (Clojure returns this as the conv's new
+        # :in-conv; deepcopy in recompute threads it forward).
+        self.in_conv = set(in_conv)
+
+        logger.info(f"Legacy in-conv: {len(threshold_set)} over threshold "
+                    f"{threshold:.1f}, {len(in_conv)} after carry+greedy floor")
         return in_conv
 
     def _fold_base_clusters(self, clusters: List[Dict]) -> Dict:
@@ -2134,15 +2195,22 @@ def numpy_to_list(arr):
 
         # Calculate in-conv participants
         in_conv_start = time.time()
-
-        # Use pre-calculated vote counts to avoid recalculation
-        in_conv = []
-        min_votes = min(7, self.comment_count)
-
-        for pid, count in result['user-vote-counts'].items():
-            if count >= min_votes:
-                in_conv.append(pid)  # pid is already converted to int where possible
-
+
+        if resolve_engine_mode() == ENGINE_MODE_LEGACY and self.in_conv:
+            # Legacy (PR-E): serialize the PERSISTED carry+greedy set — exactly
+            # the participants that fed base clustering — so the blob's :in-conv
+            # matches the clustered rows (Clojure serializes its carried
+            # in-conv). Keyed off user-vote-counts (same source as self.in_conv)
+            # to preserve pid types and row order.
+            in_conv = [pid for pid in result['user-vote-counts'] if pid in self.in_conv]
+        else:
+            # Improved (default): threshold set only — unchanged.
+            in_conv = []
+            min_votes = min(7, self.comment_count)
+            for pid, count in result['user-vote-counts'].items():
+                if count >= min_votes:
+                    in_conv.append(pid)  # pid is already converted to int where possible
+
         result['in-conv'] = in_conv
         logger.info(f"In-conv: {time.time() - in_conv_start:.4f}s")
 

delphi/tests/test_in_conv_greedy_carry.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+"""
+Tests for the in-conv greedy floor + persistent carry in 'clojure-legacy' mode
+(PR-E, Clojure :in-conv, conversation.clj:243-269).
+
+Clojure keeps a PERSISTENT in-conv set on the conv and, every tick, (1) unions
+the threshold-qualifiers into it and (2) if fewer than 15 are in, greedily
+admits the top voters up to 15 — then carries the whole set forward, so admits
+never leave. The pre-PR Python pipeline had NEITHER the greedy floor NOR the
+carry (only the threshold set). This module verifies:
+
+  1. Improved mode (default) is unchanged: threshold set only, no greedy floor,
+     no carry, and self.in_conv is never populated.
+  2. Legacy mode admits the top (15 - n) voters when under 15, with ties broken
+     by matrix row order (deterministic surrogate for Clojure's hash-order tie).
+  3. Legacy greedy admits PERSIST across ticks even once the conversation grows
+     past 15 threshold-qualifiers (the carry) — improved mode drops them.
+  4. Threshold-qualifiers stay in across ticks in BOTH modes (monotonicity).
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+
+from polismath.conversation.conversation import Conversation
+from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR
+from polismath.pca_kmeans_rep.pca import PCA_IMPL_ENV_VAR
+
+
+TOTAL_CMNTS = 8  # threshold = min(7, 8) = 7
+
+
+def _votes(specs):
+    """specs: list of (pid, n_votes). Participant idx votes on its first
+    n_votes comments (of TOTAL_CMNTS) with a per-idx sign pattern (so rows are
+    distinct). A participant with n_votes >= 7 clears the threshold."""
+    votes = []
+    for idx, (pid, nv) in enumerate(specs):
+        for j in range(nv):
+            v = 1.0 if ((idx + j) % 2 == 0) else -1.0
+            votes.append({'pid': pid, 'tid': f'c{j}', 'vote': v})
+    return {'votes': votes}
+
+
+def _clustered_pids(conv):
+    """The participants that actually fed base clustering = the effective
+    in-conv set (cluster-step assigns every in-conv row to a base cluster)."""
+    return {str(m) for c in conv.base_clusters for m in c['members']}
+
+
+# 2 high voters (qualify) + 20 low voters (6 votes each, below threshold).
+_HIGHS = [(f'H{i}', TOTAL_CMNTS) for i in range(2)]
+_LOWS = [(f'L{i}', 6) for i in range(20)]
+_TICK1_SPECS = _HIGHS + _LOWS  # row order: H0,H1,L0,L1,...,L19
+
+
+def _mode(monkeypatch, mode):
+    monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+    monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode)
+
+
+class TestGreedyFloor:
+
+    def test_improved_has_no_greedy_floor(self, monkeypatch):
+        # Current/improved behavior: only the 2 threshold-qualifiers cluster.
+        _mode(monkeypatch, 'improved')
+        conv = Conversation('g').update_votes(_votes(_TICK1_SPECS))
+        assert _clustered_pids(conv) == {'H0', 'H1'}
+        assert conv.in_conv == set()  # improved never populates the carry set
+
+    def test_legacy_greedy_fills_to_fifteen(self, monkeypatch):
+        _mode(monkeypatch, 'clojure-legacy')
+        conv = Conversation('g').update_votes(_votes(_TICK1_SPECS))
+        clustered = _clustered_pids(conv)
+        assert len(clustered) == 15  # 2 highs + 13 greedy admits
+        assert {'H0', 'H1'}.issubset(clustered)
+        assert conv.in_conv == clustered  # persisted
+
+    def test_legacy_greedy_tie_break_is_row_order(self, monkeypatch):
+        # All 20 lows tie at 6 votes; greedy admits the FIRST 13 by row order
+        # (L0..L12), not L13..L19 (Clojure sort-by is stable; we key ties on
+        # matrix row order deterministically).
+        _mode(monkeypatch, 'clojure-legacy')
+        conv = Conversation('g').update_votes(_votes(_TICK1_SPECS))
+        clustered = _clustered_pids(conv)
+        assert {f'L{i}' for i in range(13)}.issubset(clustered)   # L0..L12 in
+        assert not any(f'L{i}' in clustered for i in range(13, 20))  # L13..L19 out
+
+
+class TestPersistentCarry:
+
+    def _tick2_new_qualifiers(self):
+        # 20 brand-new participants that each clear the threshold.
+        return _votes([(f'Q{i}', TOTAL_CMNTS) for i in range(20)])
+
+    def test_legacy_greedy_admits_persist_after_growth(self, monkeypatch):
+        _mode(monkeypatch, 'clojure-legacy')
+        conv = Conversation('carry').update_votes(_votes(_TICK1_SPECS))
+        admitted_lows = {f'L{i}' for i in range(13)}
+        assert admitted_lows.issubset(conv.in_conv)
+
+        # Grow well past 15 threshold-qualifiers; greedy floor no longer fires.
+        conv = conv.update_votes(self._tick2_new_qualifiers())
+        clustered = _clustered_pids(conv)
+        # The tick-1 greedy admits are STILL in, purely via the carry.
+        assert admitted_lows.issubset(conv.in_conv)
+        assert admitted_lows.issubset(clustered)
+        # And the new qualifiers are in too.
+        assert {f'Q{i}' for i in range(20)}.issubset(clustered)
+
+    def test_improved_drops_non_qualifiers_after_growth(self, monkeypatch):
+        _mode(monkeypatch, 'improved')
+        conv = Conversation('carry').update_votes(_votes(_TICK1_SPECS))
+        conv = conv.update_votes(self._tick2_new_qualifiers())
+        clustered = _clustered_pids(conv)
+        # No carry, no greedy: the below-threshold lows are NOT clustered.
+        assert not any(f'L{i}' in clustered for i in range(20))
+        # Only the threshold-qualifiers (H0,H1 + Q0..Q19) cluster.
+        assert clustered == {'H0', 'H1'} | {f'Q{i}' for i in range(20)}
+
+
+class TestSerializedInConv:
+
+    def test_legacy_blob_in_conv_includes_greedy_admits(self, monkeypatch):
+        _mode(monkeypatch, 'clojure-legacy')
+        conv = Conversation('blob').update_votes(_votes(_TICK1_SPECS))
+        blob_in_conv = {str(p) for p in conv.to_dict()['in-conv']}
+        assert len(blob_in_conv) == 15
+        assert {'H0', 'H1'}.issubset(blob_in_conv)
+        assert {f'L{i}' for i in range(13)}.issubset(blob_in_conv)  # greedy admits
+
+    def test_improved_blob_in_conv_is_threshold_only(self, monkeypatch):
+        _mode(monkeypatch, 'improved')
+        conv = Conversation('blob').update_votes(_votes(_TICK1_SPECS))
+        blob_in_conv = {str(p) for p in conv.to_dict()['in-conv']}
+        assert blob_in_conv == {'H0', 'H1'}  # no greedy floor in improved
+
+
+class TestThresholdMonotonicity:
+
+    @pytest.mark.parametrize('mode', ['improved', 'clojure-legacy'])
+    def test_qualifier_stays_in_across_ticks(self, monkeypatch, mode):
+        _mode(monkeypatch, mode)
+        conv = Conversation('mono').update_votes(_votes(_TICK1_SPECS))
+        assert 'H0' in _clustered_pids(conv)
+        # A later tick (new participants) never evicts an existing qualifier.
+        conv = conv.update_votes(_votes([(f'Q{i}', TOTAL_CMNTS) for i in range(3)]))
+        assert 'H0' in _clustered_pids(conv)
+
+
+class TestCarryPruneOnParticipantBan:
+    """T1 (#2623): a participant carried in self.in_conv and then BANNED
+    (mod_out_ptpts, a Python-only feature Clojure lacks) must be pruned from the
+    carry so the greedy floor tops the pool back up. Before the fix, the raw
+    carried set (`set(self.in_conv) | threshold_set`) still counted the banned
+    pids, so the size check saw a stale 15 while the actually-clustered pool
+    (proj ∩ in_conv, which excludes banned) had dropped below the floor — and
+    nothing topped it up. Legacy-mode-only; improved mode has no carry."""
+
+    def test_ban_after_carry_tops_floor_back_up(self, monkeypatch):
+        _mode(monkeypatch, 'clojure-legacy')
+        # Tick 1: greedy floor fills to 15 (H0,H1 + L0..L12) and persists them.
+        conv = Conversation('ban').update_votes(_votes(_TICK1_SPECS))
+        assert len(conv.in_conv) == 15
+        banned = {'L0', 'L1', 'L2', 'L3', 'L4'}
+        assert banned.issubset(conv.in_conv)  # all 5 are carried greedy admits
+
+        # Ban 5 of the 15 carried participants. vote_counts (from rating_mat.index)
+        # now excludes them, so the effective pool drops to 10 and the floor must
+        # re-admit 5 more of the remaining lows (L13..L19) to reach 15.
+        conv2 = conv.update_moderation({'mod_out_ptpts': list(banned)})
+
+        clustered = _clustered_pids(conv2)
+        # FIX: pruned carry -> floor re-fires -> back up to 15.
+        # BEFORE FIX: stale carry keeps count at 15 (floor never fires), but the
+        # banned rows aren't clustered -> only 10 clustered.
+        assert len(clustered) == 15
+        assert not (banned & clustered)          # banned never clustered
+        assert not (banned & conv2.in_conv)      # banned pruned from carry (no stale growth)
+        assert {'H0', 'H1'}.issubset(clustered)  # qualifiers retained
+        assert len(conv2.in_conv) == 15