Avoid parsing unavailable telemetry signals per heartbeat

Intersect supplied signal keys with the existing field maps and skip explicit nulls before coercion. Preserve analyses, histories, cohort counts, timestamp parsing and the unchanged eight-second production-scale guard; add deterministic sparse/full-null cost and all-field coercion controls.

Contract-Neutral: In-memory reporting optimisation only; no telemetry schema, report semantics, collection, privacy, runtime or subsystem contract changes.
Change-source: pulse-maintainer

scripts/telemetry_adoption_report.py

@@ -2279,13 +2279,18 @@ def pulse_intelligence_row_analysis_keys(
 ) -> tuple[set[str], set[str]]:
     cohort_keys: set[str] = set()
     signal_groups: set[str] = set()
-    for field in PULSE_INTELLIGENCE_ANALYSIS_BOOL_FIELDS:
-        if not parse_optional_bool(row.get(field)):
+    # Older schemas and sparse exports omit most signals. Intersect in C rather
+    # than parsing every absent field for every heartbeat; explicit nulls have
+    # the same no-signal meaning. Keep the existing parsers for supplied values.
+    for field in row.keys() & PULSE_INTELLIGENCE_COHORT_BOOL_KEYS_BY_FIELD.keys():
+        value = row[field]
+        if value is None or not parse_optional_bool(value):
             continue
         cohort_keys.update(PULSE_INTELLIGENCE_COHORT_BOOL_KEYS_BY_FIELD[field])
         signal_groups.update(PULSE_INTELLIGENCE_SIGNAL_GROUP_BOOL_KEYS_BY_FIELD[field])
-    for field in PULSE_INTELLIGENCE_ANALYSIS_COUNT_FIELDS:
-        if parse_optional_nonnegative_int(row.get(field)) <= 0:
+    for field in row.keys() & PULSE_INTELLIGENCE_COHORT_COUNT_KEYS_BY_FIELD.keys():
+        value = row[field]
+        if value is None or parse_optional_nonnegative_int(value) <= 0:
             continue
         cohort_keys.update(PULSE_INTELLIGENCE_COHORT_COUNT_KEYS_BY_FIELD[field])
         signal_groups.update(PULSE_INTELLIGENCE_SIGNAL_GROUP_COUNT_KEYS_BY_FIELD[field])

scripts/tests/test_telemetry_adoption_report.py

@@ -2120,6 +2120,76 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
                     expected_signal_conversion,
                 )
 
+    def test_pulse_intelligence_sparse_rows_parse_only_supplied_signals(self) -> None:
+        row = {
+            "install_id": "synthetic",
+            "received_at": "2026-07-23 12:00:00",
+            "paid_license": False,
+            "pulse_intelligence_loop_configured": "yes",
+            "pulse_intelligence_patrol_new_findings_30d": "2",
+            "pulse_intelligence_approved_action_successes_30d": "invalid",
+            "unrelated": 99,
+        }
+        # Full projections include explicit nulls; sparse older exports omit
+        # them. Neither should spend parser calls on unavailable signals.
+        projected = dict.fromkeys(report.REPORT_ROW_COLUMNS)
+        projected.update(row)
+        for supplied in (row, projected):
+            with self.subTest(projected=len(supplied) > len(row)):
+                with mock.patch.object(
+                    report, "parse_optional_bool", wraps=report.parse_optional_bool
+                ) as parse_bool, mock.patch.object(
+                    report,
+                    "parse_optional_nonnegative_int",
+                    wraps=report.parse_optional_nonnegative_int,
+                ) as parse_count:
+                    actual = report.pulse_intelligence_row_analysis_keys(supplied)
+                self.assertEqual(parse_bool.call_args_list, [mock.call("yes")])
+                self.assertCountEqual(
+                    parse_count.call_args_list, [mock.call("2"), mock.call("invalid")]
+                )
+                expected_cohorts = {
+                    key for key, _, bool_fields, count_fields
+                    in report.PULSE_INTELLIGENCE_OUTCOME_COHORTS
+                    if report.pulse_intelligence_row_matches_cohort(
+                        row, bool_fields, count_fields
+                    )
+                }
+                expected_groups = report.pulse_intelligence_row_signal_groups(row)
+                self.assertEqual(actual[0], expected_cohorts)
+                self.assertEqual(
+                    report.pulse_intelligence_derive_signal_groups(actual[1]),
+                    expected_groups,
+                )
+
+    def test_pulse_intelligence_sparse_analysis_preserves_all_field_coercions(self) -> None:
+        values = (None, True, False, 0, 1, -1, 0.5, "2", "yes", "no", " ", "bad", [], {})
+        fields = set(report.PULSE_INTELLIGENCE_ANALYSIS_BOOL_FIELDS) | set(
+            report.PULSE_INTELLIGENCE_ANALYSIS_COUNT_FIELDS
+        )
+        # The independent helpers walk the declared cohorts/groups, not the
+        # optimized field maps. Cover every field, including group-only fields.
+        for field in sorted(fields):
+            for value in values:
+                with self.subTest(field=field, value=value):
+                    row = {field: value, "undeclared_signal": True}
+                    expected_cohorts = {
+                        key for key, _, bool_fields, count_fields
+                        in report.PULSE_INTELLIGENCE_OUTCOME_COHORTS
+                        if report.pulse_intelligence_row_matches_cohort(
+                            row, bool_fields, count_fields
+                        )
+                    }
+                    expected_groups = report.pulse_intelligence_row_signal_groups(row)
+                    actual_cohorts, actual_groups = (
+                        report.pulse_intelligence_row_analysis_keys(row)
+                    )
+                    self.assertEqual(actual_cohorts, expected_cohorts)
+                    self.assertEqual(
+                        report.pulse_intelligence_derive_signal_groups(actual_groups),
+                        expected_groups,
+                    )
+
     def test_pulse_intelligence_production_scale_performance_guard(self) -> None:
         now = datetime(2026, 7, 23, 12, tzinfo=timezone.utc)
         rows: list[dict[str, object]] = []