diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index 1c66a3ff4..f6e10a365 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/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") diff --git a/delphi/tests/test_in_conv_greedy_carry.py b/delphi/tests/test_in_conv_greedy_carry.py new file mode 100644 index 000000000..7146ab3c1 --- /dev/null +++ b/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