perf(delphi): resolve group member rows once per group in _compute_group_votes

delphi/polismath/conversation/conversation.py

@@ -1373,74 +1373,60 @@ def _compute_group_votes(self) -> Dict[str, Any]:
 
         group_votes = {}
 
-        # Helper to count votes of a specific type for a group
-        def count_votes_for_group(group_id, comment_id, vote_type):
-            group = next((g for g in unfolded if g.get('id') == group_id), None)
-            if not group:
-                return 0
-
-            # Get members of this group
+        # The A/D/S counts for a group are three column-wise reductions over
+        # that group's rows of the vote matrix, so resolve each group's member
+        # rows once and reduce over every comment in one pass. Doing the row
+        # lookup per (comment, vote-type) instead made this an O(groups ×
+        # comments × members) scan on every tick (#2587).
+        rating_values = self.rating_mat.values
+        comment_ids = list(self.rating_mat.columns)
+
+        # For each group, compute vote stats
+        for group in unfolded:
+            group_id = group.get('id')
+
+            # Skip groups without ID
+            if group_id is None:
+                continue
+
+            # Count members in this group
             members = group.get('members', [])
-
-            # If members list is empty, return 0
-            if not members:
-                return 0
-
+            n_members = len(members)
+
             # Get the row indices for these members
             row_indices = []
             for member in members:
                 try:
-                    member_idx = self.rating_mat.index.get_loc(member)
-                    row_indices.append(member_idx)
+                    row_indices.append(self.rating_mat.index.get_loc(member))
                 except ValueError:
                     # Skip members not found in matrix
                     continue
-
-            # Get the column index for this comment
-            try:
-                col_idx = self.rating_mat.columns.get_loc(comment_id)
-            except ValueError:
-                # If comment not found, return 0
-                return 0
-
-            # Count votes of specified type
-            votes = self.rating_mat.values[row_indices, col_idx]
-
-            if vote_type == 'A':  # Agree
-                return int(np.sum(np.abs(votes - 1.0) < 0.001))
-            elif vote_type == 'D':  # Disagree
-                return int(np.sum(np.abs(votes + 1.0) < 0.001))
-            elif vote_type == 'S':  # Total votes
-                return int(np.sum(~np.isnan(votes)))
-            else:
-                return 0
-
-        # For each group, compute vote stats
-        for group in unfolded:
-            group_id = group.get('id')
-
-            # Skip groups without ID
-            if group_id is None:
-                continue
-
-            # Count members in this group
-            n_members = len(group.get('members', []))
-
-            # Get vote counts for each comment
-            votes = {}
-            for comment_id in self.rating_mat.columns:
-                votes[comment_id] = {
-                    'A': count_votes_for_group(group_id, comment_id, 'A'),
-                    'D': count_votes_for_group(group_id, comment_id, 'D'),
-                    'S': count_votes_for_group(group_id, comment_id, 'S')
-                }
-
+
+            # One (members × comments) slice per group, reduced column-wise:
+            # empty membership yields an empty slice and therefore all-zero
+            # counts, matching the per-comment version.
+            group_ratings = rating_values[row_indices, :]
+            agree_counts = np.sum(
+                np.abs(group_ratings - 1.0) < 0.001, axis=0)
+            disagree_counts = np.sum(
+                np.abs(group_ratings + 1.0) < 0.001, axis=0)
+            # S counts every cast vote, PASS included (matches Clojure).
+            seen_counts = np.sum(~np.isnan(group_ratings), axis=0)
+
+            votes = {
+                comment_id: {'A': int(n_agree),
+                             'D': int(n_disagree),
+                             'S': int(n_seen)}
+                for comment_id, n_agree, n_disagree, n_seen in zip(
+                    comment_ids, agree_counts, disagree_counts, seen_counts)
+            }
+
             # Store results
             group_votes[str(group_id)] = {
                 'n-members': n_members,
                 'votes': votes
             }
-            
+
         return group_votes
 
     def _compute_user_vote_counts(self) -> Dict[str, int]:

delphi/tests/test_group_votes.py

@@ -0,0 +1,159 @@
+"""Tests for `Conversation._compute_group_votes`.
+
+`_compute_group_votes` runs twice per math tick — once from
+`_compute_comment_priorities` inside `recompute()`, once from
+`_compute_group_aware_consensus` on the DynamoDB write path — so it has to
+produce exactly the per-group A/D/S counts the Clojure `group-votes` did,
+and it has to do it without rescanning the vote matrix once per comment
+(#2587).
+"""
+
+import os
+import sys
+
+import numpy as np
+import pandas as pd
+
+# Add parent to path
+sys.path.append(os.path.abspath(os.path.dirname(os.path.dirname(__file__))))
+
+from polismath.conversation.conversation import Conversation
+
+
+class FakeConversation:
+    """Carries only the state `_compute_group_votes` reads.
+
+    Building the real thing through `recompute()` would pin group membership
+    to whatever k-means picks; these tests need known memberships and known
+    vote patterns.
+    """
+
+    def __init__(self, rating_mat, groups):
+        self.rating_mat = rating_mat
+        self.group_clusters = [{'id': g['id']} for g in groups]
+        self._groups = groups
+
+    def _unfolded_group_clusters(self):
+        return self._groups
+
+
+def reference_group_votes(conv):
+    """Straightforward per-group, per-comment A/D/S counts."""
+    result = {}
+    for group in conv._unfolded_group_clusters():
+        votes = {}
+        for comment_id in conv.rating_mat.columns:
+            column = conv.rating_mat.loc[group['members'], comment_id]
+            votes[comment_id] = {
+                'A': int((column == 1.0).sum()),
+                'D': int((column == -1.0).sum()),
+                'S': int(column.notna().sum()),
+            }
+        result[str(group['id'])] = {
+            'n-members': len(group['members']),
+            'votes': votes,
+        }
+    return result
+
+
+def make_conversation(n_participants, n_comments, n_groups, seed=0):
+    rng = np.random.default_rng(seed)
+    values = rng.choice([-1.0, 0.0, 1.0, np.nan],
+                        size=(n_participants, n_comments),
+                        p=[0.25, 0.3, 0.1, 0.35])
+    rating_mat = pd.DataFrame(values,
+                              index=list(range(n_participants)),
+                              columns=list(range(n_comments)))
+    assignment = rng.integers(0, n_groups, size=n_participants)
+    groups = [
+        {'id': gid,
+         'members': [int(pid) for pid in np.flatnonzero(assignment == gid)]}
+        for gid in range(n_groups)
+    ]
+    return FakeConversation(rating_mat, groups)
+
+
+def count_index_lookups(monkeypatch, conv):
+    """Run `_compute_group_votes` and return the `Index.get_loc` call count."""
+    calls = []
+    real_get_loc = pd.Index.get_loc
+
+    def counting_get_loc(self, key, *args, **kwargs):
+        calls.append(key)
+        return real_get_loc(self, key, *args, **kwargs)
+
+    monkeypatch.setattr(pd.Index, 'get_loc', counting_get_loc)
+    Conversation._compute_group_votes(conv)
+    monkeypatch.undo()
+    return len(calls)
+
+
+def test_group_votes_match_reference_counts():
+    """A/D/S per group and comment agree with a direct count."""
+    conv = make_conversation(60, 25, 3)
+
+    assert Conversation._compute_group_votes(conv) == reference_group_votes(conv)
+
+
+def test_group_votes_count_pass_in_s_but_not_in_a_or_d():
+    """S counts every cast vote, PASS included; A and D count only ±1."""
+    rating_mat = pd.DataFrame(
+        [[1.0, -1.0, 0.0, np.nan]],
+        index=[0],
+        columns=[10, 11, 12, 13],
+    )
+    conv = FakeConversation(rating_mat, [{'id': 0, 'members': [0]}])
+
+    votes = Conversation._compute_group_votes(conv)['0']['votes']
+
+    assert votes[10] == {'A': 1, 'D': 0, 'S': 1}
+    assert votes[11] == {'A': 0, 'D': 1, 'S': 1}
+    assert votes[12] == {'A': 0, 'D': 0, 'S': 1}
+    assert votes[13] == {'A': 0, 'D': 0, 'S': 0}
+
+
+def test_group_votes_with_no_members_are_all_zero():
+    """A group nobody landed in still reports every comment, at zero."""
+    rating_mat = pd.DataFrame([[1.0, -1.0]], index=[0], columns=[10, 11])
+    conv = FakeConversation(rating_mat, [{'id': 0, 'members': [0]},
+                                         {'id': 1, 'members': []}])
+
+    empty = Conversation._compute_group_votes(conv)['1']
+
+    assert empty['n-members'] == 0
+    assert empty['votes'] == {10: {'A': 0, 'D': 0, 'S': 0},
+                              11: {'A': 0, 'D': 0, 'S': 0}}
+
+
+def test_group_votes_counts_are_plain_ints():
+    """Counts are serialized downstream, so numpy scalars must not leak."""
+    conv = make_conversation(20, 5, 2)
+
+    for group in Conversation._compute_group_votes(conv).values():
+        for counts in group['votes'].values():
+            assert all(type(n) is int for n in counts.values())
+
+
+def test_group_votes_resolves_member_rows_once_per_group(monkeypatch):
+    """Member row lookup is hoisted out of the per-comment loop (#2587).
+
+    The pre-vectorization implementation resolved every member's row inside
+    a per-(comment, vote-type) helper, so `Index.get_loc` ran
+    `3 x groups x comments x members` times and the matrix was rescanned for
+    every comment. Counting the lookups pins the complexity without
+    depending on wall-clock time.
+    """
+    conv = make_conversation(40, 30, 2)
+    n_members = sum(len(g['members']) for g in conv._unfolded_group_clusters())
+
+    assert count_index_lookups(monkeypatch, conv) == n_members
+
+
+def test_group_votes_cost_is_independent_of_comment_count(monkeypatch):
+    """Quadrupling the comments must not multiply the index lookups."""
+    few = make_conversation(30, 10, 2)
+    many = make_conversation(30, 40, 2)
+
+    assert (count_index_lookups(monkeypatch, few)
+            == count_index_lookups(monkeypatch, many)
+            == 30)