fix(signals): let the scout suggestion activity check answer for quiet projects

products/signals/backend/scout_harness/suggestions.py

@@ -76,13 +76,19 @@
 # Resolves a suggestion run to the `signals_scout_suggestions` gateway product.
 SUGGESTIONS_AI_STAGE = "scout_suggestions"
 
-# Row cap on the per-candidate activity read, so the check stays cheap on a large project. The
-# read refuses at the cap rather than truncating: the cap counts rows read while the query counts
-# the rows that pass the window filter, so a truncated read would come back as a small count that
-# reads like a quiet project. A project whose window does not fit the cap is active whatever the
-# refusal hid, and only a read that finishes has numbers — the small projects the check is about.
+# High bound for `min_events_in_window`. An event line above it would ask the check to tell
+# apart projects that the read cannot answer for anyway.
 ACTIVITY_READ_MAX_ROWS = 100_000
 
+# ClickHouse `max_rows_to_read` guard on the per-candidate activity read, so the check stays cheap
+# on a large project. The guard counts rows read, not rows that pass the filter. ClickHouse reads
+# whole granules (up to 8,192 rows), and a quiet project's 14-day window touches about one granule
+# in each of 60 to 100 parts, which other teams' rows mostly fill. So a quiet project reads several
+# hundred thousand rows to count a few dozen events, and the guard must sit well above that. The
+# read refuses at the guard rather than truncating, because a truncated count reads like a quiet
+# project. A project that trips the guard is active, and only a read that finishes has numbers.
+ACTIVITY_READ_MAX_ROWS_READ = 5_000_000
+
 # Wall-clock cap on the same read. `sync_execute` adds none of its own and the pooled client's
 # socket timeout is effectively infinite in production, so a stalled read would hold its worker
 # thread past the planning activity's own deadline, which Temporal cannot reclaim. A capped read
@@ -536,7 +542,7 @@ def stamp_requested(team_ids: list[int], now: datetime | None = None) -> None:
 LOW_ACTIVITY_SKIP_REASON = "low_activity"
 
 # TOO_MANY_ROWS / TOO_MANY_ROWS_OR_BYTES: what `read_overflow_mode: throw` raises at
-# `ACTIVITY_READ_MAX_ROWS`. Any other failure belongs to the caller's dispatch-anyway path.
+# `ACTIVITY_READ_MAX_ROWS_READ`. Any other failure belongs to the caller's dispatch-anyway path.
 _READ_CAP_ERROR_CODES = (158, 396)
 
 
@@ -546,7 +552,7 @@ class TeamActivity:
 
     event_count: int
     active_days: int
-    # The read refused at `ACTIVITY_READ_MAX_ROWS` rather than answer with a truncated count, so
+    # The read refused at `ACTIVITY_READ_MAX_ROWS_READ` rather than answer with a truncated count, so
     # the two fields above hold nothing the check may read.
     capped: bool
 
@@ -555,6 +561,11 @@ class TeamActivity:
 class ActivitySelection:
     dispatch: tuple[PlannedSuggestionRun, ...]
     skipped_team_ids: tuple[int, ...]
+    # How each activity read ended. A capped read and an answered read look the same downstream,
+    # so these counts are the only place a guard that trips on quiet projects shows.
+    reads_answered: int = 0
+    reads_capped: int = 0
+    reads_failed: int = 0
 
 
 def activity_check_enabled(settings: SuggestionSettings) -> bool:
@@ -590,7 +601,7 @@ def read_team_activity(team_id: int, *, window_days: int) -> TeamActivity:
             # candidate after another, and the per-tick cap is flag-tunable.
             workload=Workload.OFFLINE,
             settings={
-                "max_rows_to_read": ACTIVITY_READ_MAX_ROWS,
+                "max_rows_to_read": ACTIVITY_READ_MAX_ROWS_READ,
                 "read_overflow_mode": "throw",
                 "max_execution_time": ACTIVITY_READ_MAX_EXECUTION_S,
                 # Both overflow modes throw for the same reason: a partial aggregate reads as a
@@ -635,6 +646,7 @@ def select_teams_to_scan(
     exempt = canonical_team_ids(settings.team_allowlist)
     dispatch: list[PlannedSuggestionRun] = []
     skipped: list[int] = []
+    answered = capped = failed = 0
     for run in candidates:
         if len(dispatch) >= limit:
             break
@@ -646,14 +658,25 @@ def select_teams_to_scan(
         except Exception:
             # A read that cannot answer must not cost the project its refresh window.
             logger.warning("scout_suggestions: activity read failed", team_id=run.team_id, exc_info=True)
+            failed += 1
             dispatch.append(run)
             continue
+        if activity.capped:
+            capped += 1
+        else:
+            answered += 1
         if team_is_active_enough(activity, settings):
             dispatch.append(run)
             continue
         skipped.append(run.team_id)
         _record_low_activity(run.team_id, activity=activity, settings=settings, tier=run.tier)
-    return ActivitySelection(dispatch=tuple(dispatch), skipped_team_ids=tuple(skipped))
+    return ActivitySelection(
+        dispatch=tuple(dispatch),
+        skipped_team_ids=tuple(skipped),
+        reads_answered=answered,
+        reads_capped=capped,
+        reads_failed=failed,
+    )
 
 
 def _skip_event_uuid(team_id: int, *, requested_at: datetime | None) -> str:
@@ -1068,6 +1091,7 @@ def mark_stale_if_fleet_changed(team_id: int) -> None:
 
 __all__ = [
     "ACTIVITY_READ_MAX_ROWS",
+    "ACTIVITY_READ_MAX_ROWS_READ",
     "LOW_ACTIVITY_SKIP_REASON",
     "MAX_SUGGESTIONS_PER_BATCH",
     "SIGNALS_SCOUT_SUGGESTIONS_FLAG",

products/signals/backend/temporal/agentic/scout_suggestions.py

@@ -75,6 +75,9 @@ class PlanSuggestionsOutput:
     # they are stamped with the run so the planner leaves them alone for a refresh window.
     # Defaulted so a plan serialized before the check existed still deserializes across a deploy.
     skipped_team_ids: list[int] = field(default_factory=list)
+    activity_reads_answered: int = 0
+    activity_reads_capped: int = 0
+    activity_reads_failed: int = 0
 
 
 @dataclass(frozen=True)
@@ -93,6 +96,9 @@ class SuggestionsCoordinatorOutput:
     started_count: int
     skipped_count: int
     low_activity_count: int = 0
+    activity_reads_answered: int = 0
+    activity_reads_capped: int = 0
+    activity_reads_failed: int = 0
 
 
 @dataclass(frozen=True)
@@ -149,16 +155,26 @@ async def plan_scout_suggestion_runs_activity(_input: PlanSuggestionsInput) -> P
         selection = await database_sync_to_async(select_teams_to_scan, thread_sensitive=False)(
             planned, settings, limit=cap
         )
-    logger.info(
+    reads = selection.reads_answered + selection.reads_capped + selection.reads_failed
+    # A capped read counts as active, so a guard that trips on most reads turns the check off
+    # without an error.
+    log = logger.warning if reads and selection.reads_capped * 2 > reads else logger.info
+    log(
         "scout_suggestions coordinator: planned",
         count=len(selection.dispatch),
         low_activity=len(selection.skipped_team_ids),
+        activity_reads_answered=selection.reads_answered,
+        activity_reads_capped=selection.reads_capped,
+        activity_reads_failed=selection.reads_failed,
         enabled=settings.enabled,
     )
     return PlanSuggestionsOutput(
         planned=[PlannedSuggestion(team_id=run.team_id, tier=run.tier) for run in selection.dispatch],
         skipped_team_ids=list(selection.skipped_team_ids),
         settings_json=_settings_to_json(settings),
+        activity_reads_answered=selection.reads_answered,
+        activity_reads_capped=selection.reads_capped,
+        activity_reads_failed=selection.reads_failed,
     )
 
 
@@ -312,6 +328,9 @@ async def run(self, _input: SuggestionsCoordinatorInput) -> SuggestionsCoordinat
             started_count=started,
             skipped_count=skipped,
             low_activity_count=len(plan.skipped_team_ids),
+            activity_reads_answered=plan.activity_reads_answered,
+            activity_reads_capped=plan.activity_reads_capped,
+            activity_reads_failed=plan.activity_reads_failed,
         )
 
 

products/signals/backend/test/test_scout_suggestions.py

@@ -28,6 +28,8 @@
 from products.access_control.backend.models.access_control import AccessControl
 from products.signals.backend.models import SignalScoutConfig, SignalScoutSuggestionSet, SignalSourceConfig
 from products.signals.backend.scout_harness.suggestions import (
+    ACTIVITY_READ_MAX_ROWS,
+    ACTIVITY_READ_MAX_ROWS_READ,
     MAX_DESCRIPTION_CHARS,
     PlannedSuggestionRun,
     ScoutSuggestionBatch,
@@ -97,6 +99,7 @@ def test_malformed_payload_degrades_to_safe_defaults(self, _name, payload, enabl
             ("overridden", {"activity_window_days": 7, "min_events_in_window": 50}, 7, 50, 3),
             ("checks_off", {"min_events_in_window": 0, "min_active_days_in_window": 0}, 14, 0, 0),
             ("clamped", {"activity_window_days": 900, "min_active_days_in_window": -1}, 90, 100, 0),
+            ("events_clamped", {"min_events_in_window": 10**9}, 14, ACTIVITY_READ_MAX_ROWS, 3),
         ]
     )
     def test_activity_knobs(self, _name, extra, window_days, min_events, min_days):
@@ -665,6 +668,28 @@ def test_overselected_candidates_past_the_limit_are_left_untouched(self):
         self.assertEqual(selection.skipped_team_ids, ())
         self.assertIsNone(self._status(self.quiet))
 
+    def test_each_read_outcome_is_counted(self):
+        broken = self._team("broken-read")
+        reads = {
+            self.quiet.id: TeamActivity(event_count=12, active_days=1, capped=False),
+            self.busy.id: TeamActivity(event_count=ACTIVITY_READ_MAX_ROWS, active_days=0, capped=True),
+        }
+
+        def _read(team_id, **_):
+            if team_id == broken.id:
+                raise Exception("clickhouse is down")
+            return reads[team_id]
+
+        with patch("products.signals.backend.scout_harness.suggestions.read_team_activity", side_effect=_read):
+            selection = select_teams_to_scan(
+                self._planned(self.quiet, self.busy, broken), self.suggestion_settings, limit=10
+            )
+
+        self.assertEqual(
+            (selection.reads_answered, selection.reads_capped, selection.reads_failed),
+            (1, 1, 1),
+        )
+
 
 class TestReadTeamActivity(ClickhouseTestMixin, BaseTest):
     def test_counts_the_project_and_its_environments_inside_the_window(self):
@@ -698,6 +723,8 @@ def test_the_read_is_bounded_and_kept_off_the_interactive_cluster(self):
         self.assertEqual(kwargs["workload"], Workload.OFFLINE)
         self.assertGreater(kwargs["settings"]["max_execution_time"], 0)
         self.assertEqual(kwargs["settings"]["timeout_overflow_mode"], "throw")
+        self.assertEqual(kwargs["settings"]["max_rows_to_read"], ACTIVITY_READ_MAX_ROWS_READ)
+        self.assertGreater(ACTIVITY_READ_MAX_ROWS_READ, ACTIVITY_READ_MAX_ROWS)
 
     def test_the_read_is_attributed_to_the_product(self):
         seen: list[QueryTags] = []