diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md
index 10d0c0fe6..e22371e71 100644
--- a/docs/release-control/v6/internal/subsystems/security-privacy.md
+++ b/docs/release-control/v6/internal/subsystems/security-privacy.md
@@ -1988,6 +1988,37 @@ actor, and every audit row read stay on the install.
`TestWithAuditReadActivity_RecordIsContentFree` and
`TestRecordAuditReadActivity_RejectsUnknownActivity` pin both properties.
+### Telemetry ingestion matches the released sender while storage stays compatible
+
+The active outbound contract remains schema v7. The draft schema-v8
+`business_estate` field was reverted before release and must not remain in the
+license server's accepted ping struct merely because a private receiver build
+and database migration briefly carried it. Existing deployed databases need no
+destructive column migration, while new databases do not recreate the retired
+column. Incoming draft-field values are ignored and can never enter adoption
+reporting. `TestTelemetryPing_IgnoresRetiredBusinessEstateField` pins that
+boundary, while `scripts/check_telemetry_schema_parity.py` requires the public
+sender, frontend preview, and active private receiver struct to stay exact apart
+from the two named legacy compatibility inputs.
+
+### Adoption reporting aggregates high-cardinality history in one pass
+
+`scripts/telemetry_adoption_report.py` must select only its explicit reporting
+projection instead of copying every receiver column over SSH. That projection
+includes the licensed-feature, availability-probe, and updater signals added to
+the released schema, while excluding the retired `business_estate` draft.
+SQLite reduces remote history to one latest-state row plus compact sufficient
+facts per install: first free, first paid, observed signal fields, and signal
+fields observed while free before the first paid posture. This preserves
+latest-state reporting, first-free/first-paid conversion, and all outcome
+cohort membership without sending raw heartbeat history over SSH. The local
+fallback must analyze each row once and keep only bounded per-install evidence
+sets and earliest observation times. It must not regroup rows into per-install
+history lists or sort those lists before producing the outcome cohorts and
+operations funnel. The production-scale guard exercises 120,000 rows across
+12,000 installs, verifies exactly one timestamp parse per row, and keeps the
+cohort plus funnel aggregation inside the bounded runtime budget.
+
### Licensed-feature adoption fields must discriminate
`telemetry.LicensedFeatureAdoptionFields` registers every telemetry field whose
diff --git a/scripts/telemetry_adoption_report.py b/scripts/telemetry_adoption_report.py
index 657420050..4081d0a32 100644
--- a/scripts/telemetry_adoption_report.py
+++ b/scripts/telemetry_adoption_report.py
@@ -10,7 +10,7 @@ from __future__ import annotations
import argparse
from collections import Counter
-from dataclasses import dataclass
+from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import gzip
import json
@@ -53,7 +53,19 @@ ADOPTION_COUNT_FIELDS = (
("vmware_vms", "VMware VMs"),
("vmware_datastores", "VMware datastores"),
("availability_targets", "Availability targets"),
+ ("availability_probe_targets", "Availability probe targets"),
+ ("availability_probe_agents", "Availability probe agents"),
("active_alerts", "Active alerts"),
+ ("rbac_custom_roles", "Custom RBAC roles"),
+ ("rbac_user_assignments", "RBAC user assignments"),
+ ("audit_reads_30d", "Audit reads (30d)"),
+ ("report_schedules", "Report schedules"),
+ ("report_schedules_enabled", "Enabled report schedules"),
+ ("report_schedules_run_30d", "Report schedule runs (30d)"),
+ ("agent_profiles", "Agent profiles"),
+ ("update_attempts_30d", "Update attempts (30d)"),
+ ("update_successes_30d", "Update successes (30d)"),
+ ("update_failures_30d", "Update failures (30d)"),
)
FEATURE_BOOL_FIELDS = (
("ai_enabled", "AI enabled"),
@@ -61,6 +73,7 @@ FEATURE_BOOL_FIELDS = (
("discovery_enabled", "Discovery enabled"),
("notifications_enabled", "Notifications enabled"),
("ai_actions_enabled", "AI actions enabled"),
+ ("alert_ai_enabled", "Alert AI enabled"),
("relay_enabled", "Relay enabled"),
("sso_enabled", "SSO enabled"),
("multi_tenant", "Multi-tenant"),
@@ -79,6 +92,7 @@ USER_BASE_CATEGORY_FIELDS = (
("activation_stage", "Highest observed activation stage"),
("time_to_first_monitored_resource_bucket", "Time to first monitored resource"),
("estate_size_bucket", "Estate size"),
+ ("update_last_failure_category", "Last update failure category"),
)
USER_BASE_BOOL_FIELDS = (
("auth_configured", "Authentication configured"),
@@ -925,10 +939,23 @@ DEEP_SIGNAL_FIELDS = (
("vmware_vms", "VMware VMs", "count"),
("vmware_datastores", "VMware datastores", "count"),
("availability_targets", "Availability targets", "count"),
+ ("availability_probe_targets", "Availability probe targets", "count"),
+ ("availability_probe_agents", "Availability probe agents", "count"),
+ ("rbac_custom_roles", "Custom RBAC roles", "count"),
+ ("rbac_user_assignments", "RBAC user assignments", "count"),
+ ("audit_reads_30d", "Audit reads (30d)", "count"),
+ ("report_schedules", "Report schedules", "count"),
+ ("report_schedules_enabled", "Enabled report schedules", "count"),
+ ("report_schedules_run_30d", "Report schedule runs (30d)", "count"),
+ ("agent_profiles", "Agent profiles", "count"),
+ ("update_attempts_30d", "Update attempts (30d)", "count"),
+ ("update_successes_30d", "Update successes (30d)", "count"),
+ ("update_failures_30d", "Update failures (30d)", "count"),
("patrol_enabled", "Patrol enabled", "bool"),
("discovery_enabled", "Discovery enabled", "bool"),
("notifications_enabled", "Notifications enabled", "bool"),
("ai_actions_enabled", "AI actions enabled", "bool"),
+ ("alert_ai_enabled", "Alert AI enabled", "bool"),
*(
(field, label, "bool")
for field, label in PULSE_INTELLIGENCE_BOOL_FIELDS
@@ -938,6 +965,39 @@ DEEP_SIGNAL_FIELDS = (
for field, label in PULSE_INTELLIGENCE_COUNT_FIELDS
),
)
+REPORT_ROW_COLUMNS = tuple(
+ dict.fromkeys(
+ (
+ "received_at",
+ "install_id",
+ "version",
+ "version_raw",
+ "schema_version",
+ "version_channel",
+ "version_build",
+ "version_is_development",
+ "version_is_published_release",
+ "platform",
+ "notification_failures_7d",
+ *(key for key, _ in ADOPTION_COUNT_FIELDS),
+ *(key for key, _ in FEATURE_BOOL_FIELDS),
+ *(key for key, _ in USER_BASE_CATEGORY_FIELDS),
+ *(key for key, _ in USER_BASE_BOOL_FIELDS),
+ *(key for key, _ in USER_BASE_COUNT_FIELDS),
+ *(key for key, _ in PULSE_INTELLIGENCE_BOOL_FIELDS),
+ *(key for key, _ in PULSE_INTELLIGENCE_COUNT_FIELDS),
+ )
+ )
+)
+REPORT_ROW_PROJECTION = ", ".join(REPORT_ROW_COLUMNS)
+REPORT_HISTORY_SIGNAL_COLUMNS = tuple(
+ dict.fromkeys(
+ (
+ *(key for key, _ in PULSE_INTELLIGENCE_BOOL_FIELDS),
+ *(key for key, _ in PULSE_INTELLIGENCE_COUNT_FIELDS),
+ )
+ )
+)
GIT_DESCRIBE_RE = re.compile(
r"^(?P\d+\.\d+\.\d+(?:-[0-9A-Za-z\.-]+)?)-(?P\d+)-g(?P[0-9a-fA-F]+)(?P-dirty)?$"
)
@@ -969,6 +1029,77 @@ class PulseIntelligenceInstallAnalysis:
free_signal_groups: frozenset[str]
+@dataclass
+class _PulseIntelligenceInstallAccumulator:
+ latest_received_at: datetime | None = None
+ first_free_at: datetime | None = None
+ first_paid_at: datetime | None = None
+ cohort_keys: set[str] = field(default_factory=set)
+ signal_groups: set[str] = field(default_factory=set)
+ earliest_free_cohort_at: dict[str, datetime] = field(default_factory=dict)
+ earliest_free_signal_group_at: dict[str, datetime] = field(default_factory=dict)
+
+ def observe(self, row: dict[str, Any]) -> None:
+ received_at = parse_received_at(str(row["received_at"]))
+ posture = parse_optional_bool(row.get("paid_license"))
+ if self.latest_received_at is None or received_at > self.latest_received_at:
+ self.latest_received_at = received_at
+ if posture is False and (
+ self.first_free_at is None or received_at < self.first_free_at
+ ):
+ self.first_free_at = received_at
+ if posture is True and (
+ self.first_paid_at is None or received_at < self.first_paid_at
+ ):
+ self.first_paid_at = received_at
+
+ row_cohort_keys, row_signal_groups = pulse_intelligence_row_analysis_keys(row)
+ self.cohort_keys.update(row_cohort_keys)
+ self.signal_groups.update(row_signal_groups)
+ if posture is not False:
+ return
+ for key in row_cohort_keys:
+ observed_at = self.earliest_free_cohort_at.get(key)
+ if observed_at is None or received_at < observed_at:
+ self.earliest_free_cohort_at[key] = received_at
+ for key in row_signal_groups:
+ observed_at = self.earliest_free_signal_group_at.get(key)
+ if observed_at is None or received_at < observed_at:
+ self.earliest_free_signal_group_at[key] = received_at
+
+ def finalize(self) -> PulseIntelligenceInstallAnalysis:
+ if self.latest_received_at is None:
+ raise ValueError("Pulse Intelligence install analysis requires at least one row")
+
+ first_paid_at = self.first_paid_at
+ free_cohort_keys = {
+ key
+ for key, observed_at in self.earliest_free_cohort_at.items()
+ if first_paid_at is None or observed_at < first_paid_at
+ }
+ free_signal_groups = {
+ key
+ for key, observed_at in self.earliest_free_signal_group_at.items()
+ if first_paid_at is None or observed_at < first_paid_at
+ }
+ signal_groups = set(self.signal_groups)
+ pulse_intelligence_derive_signal_groups(signal_groups)
+ pulse_intelligence_derive_signal_groups(free_signal_groups)
+ observed_free_start = self.first_free_at is not None and (
+ first_paid_at is None or self.first_free_at < first_paid_at
+ )
+ return PulseIntelligenceInstallAnalysis(
+ latest_received_at=self.latest_received_at,
+ first_paid_at=first_paid_at,
+ observed_free_start=observed_free_start,
+ observed_free_to_paid=observed_free_start and first_paid_at is not None,
+ cohort_keys=frozenset(self.cohort_keys),
+ free_cohort_keys=frozenset(free_cohort_keys),
+ signal_groups=frozenset(signal_groups),
+ free_signal_groups=frozenset(free_signal_groups),
+ )
+
+
PULSE_INTELLIGENCE_ANALYSIS_BOOL_FIELDS = tuple(
sorted(
{
@@ -1296,15 +1427,16 @@ def fetch_rows_local(db_path: str, since_days: int) -> dict[str, Any]:
"""
).fetchone()
)
+ rows_sql = (
+ f"SELECT {REPORT_ROW_PROJECTION} "
+ "FROM telemetry_pings "
+ "WHERE received_at >= datetime('now', ?) "
+ "ORDER BY received_at DESC"
+ )
rows = [
dict(row)
for row in conn.execute(
- """
- SELECT *
- FROM telemetry_pings
- WHERE received_at >= datetime('now', ?)
- ORDER BY received_at DESC
- """,
+ rows_sql,
(f"-{since_days} days",),
).fetchall()
]
@@ -1314,8 +1446,9 @@ def fetch_rows_local(db_path: str, since_days: int) -> dict[str, Any]:
def fetch_rows_remote(ssh_host: str, db_path: str, since_days: int) -> dict[str, Any]:
- # Streams JSON-lines (db_stats header, then one row per line) so the remote
- # process never holds the full result set — the license droplet has 1GB RAM.
+ # Let SQLite aggregate the history into sufficient per-install evidence.
+ # Only the latest row and compact Pulse Intelligence facts cross the
+ # network, not the full heartbeat history.
remote_script = """
import gzip
import json
@@ -1324,6 +1457,18 @@ import sys
db_path = sys.argv[1]
since_days = int(sys.argv[2])
+column_names = sys.argv[3].split(",")
+intelligence_columns = sys.argv[4].split(",")
+if not column_names or any(
+ not name or not name[0].isalpha() or not name.replace("_", "").isalnum()
+ for name in column_names
+):
+ raise ValueError("invalid telemetry report column projection")
+if not intelligence_columns or any(
+ name not in column_names
+ for name in intelligence_columns
+):
+ raise ValueError("invalid telemetry history signal projection")
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
db_stats_sql = (
@@ -1333,27 +1478,79 @@ db_stats_sql = (
"FROM telemetry_pings"
)
rows_sql = (
- "SELECT * "
+ "WITH ranked AS ("
+ "SELECT " + ", ".join(column_names) + ", "
+ "ROW_NUMBER() OVER ("
+ "PARTITION BY install_id ORDER BY received_at DESC, rowid DESC"
+ ") AS latest_rank "
+ "FROM telemetry_pings "
+ "WHERE received_at >= datetime('now', ?)"
+ ") SELECT " + ", ".join(column_names) + " "
+ "FROM ranked WHERE latest_rank = 1"
+)
+analysis_selects = [
+ "install_id",
+ "MAX(received_at)",
+ "MIN(CASE WHEN paid_license = 0 THEN received_at END)",
+ "MIN(CASE WHEN paid_license = 1 THEN received_at END)",
+]
+for name in intelligence_columns:
+ analysis_selects.append(
+ "MAX(CASE WHEN " + name + " <> 0 THEN 1 ELSE 0 END)"
+ )
+ analysis_selects.append(
+ "MIN(CASE WHEN paid_license = 0 AND " + name +
+ " <> 0 THEN received_at END)"
+ )
+analysis_sql = (
+ "SELECT " + ", ".join(analysis_selects) + " "
"FROM telemetry_pings "
"WHERE received_at >= datetime('now', ?) "
- "ORDER BY received_at DESC"
+ "GROUP BY install_id"
)
-output = gzip.GzipFile(fileobj=sys.stdout.buffer, mode="wb", compresslevel=6)
+output = gzip.GzipFile(fileobj=sys.stdout.buffer, mode="wb", compresslevel=1)
def emit(value):
output.write(json.dumps(value, separators=(",", ":")).encode("utf-8") + b"\\n")
try:
db_stats = dict(conn.execute(db_stats_sql).fetchone())
- emit({"db_stats": db_stats})
- for row in conn.execute(rows_sql, (f"-{since_days} days",)):
- emit(dict(row))
+ emit({"db_stats": db_stats, "row_columns": column_names})
+ cutoff = f"-{since_days} days"
+ for row in conn.execute(analysis_sql, (cutoff,)):
+ first_paid_at = row[3]
+ signal_fields = []
+ free_signal_fields = []
+ for signal_index, name in enumerate(intelligence_columns):
+ value_index = 4 + (signal_index * 2)
+ if row[value_index]:
+ signal_fields.append(name)
+ free_observed_at = row[value_index + 1]
+ if free_observed_at is not None and (
+ first_paid_at is None or free_observed_at < first_paid_at
+ ):
+ free_signal_fields.append(name)
+ emit({"a": [
+ row[0], row[1], row[2], first_paid_at,
+ signal_fields, free_signal_fields,
+ ]})
+ for row in conn.execute(rows_sql, (cutoff,)):
+ emit({"r": list(row)})
finally:
conn.close()
output.close()
"""
result = subprocess.run(
- ["ssh", ssh_host, "python3", "-", db_path, str(since_days)],
+ [
+ "ssh",
+ ssh_host,
+ "python3",
+ "-",
+ db_path,
+ str(since_days),
+ ",".join(REPORT_ROW_COLUMNS),
+ ",".join(REPORT_HISTORY_SIGNAL_COLUMNS),
+ ],
input=remote_script.encode("utf-8"),
capture_output=True,
check=True,
@@ -1367,7 +1564,39 @@ finally:
header = json.loads(next(lines))
except StopIteration:
raise RuntimeError(f"empty response from remote telemetry fetch on {ssh_host}") from None
- return {"db_stats": header["db_stats"], "rows": [json.loads(line) for line in lines]}
+ row_columns = header.get("row_columns")
+ if row_columns != list(REPORT_ROW_COLUMNS):
+ raise RuntimeError(f"invalid row schema from remote telemetry fetch on {ssh_host}")
+ rows: list[dict[str, Any]] = []
+ analysis_facts: list[dict[str, Any]] = []
+ for line in lines:
+ record = json.loads(line)
+ if "r" in record:
+ values = record["r"]
+ if len(values) != len(row_columns):
+ raise RuntimeError(f"invalid row from remote telemetry fetch on {ssh_host}")
+ rows.append(dict(zip(row_columns, values)))
+ elif "a" in record:
+ values = record["a"]
+ if len(values) != 6:
+ raise RuntimeError(f"invalid analysis from remote telemetry fetch on {ssh_host}")
+ analysis_facts.append(
+ {
+ "install_id": values[0],
+ "latest_received_at": values[1],
+ "first_free_at": values[2],
+ "first_paid_at": values[3],
+ "signal_fields": values[4],
+ "free_signal_fields": values[5],
+ }
+ )
+ else:
+ raise RuntimeError(f"invalid record from remote telemetry fetch on {ssh_host}")
+ return {
+ "db_stats": header["db_stats"],
+ "rows": rows,
+ "pulse_intelligence_analysis_facts": analysis_facts,
+ }
def counter_entries(counter: Counter[str], key_name: str) -> list[dict[str, Any]]:
@@ -1644,50 +1873,62 @@ def pulse_intelligence_row_analysis_keys(
return cohort_keys, signal_groups
+def pulse_intelligence_field_analysis_keys(
+ fields: Iterable[str],
+) -> tuple[set[str], set[str]]:
+ cohort_keys: set[str] = set()
+ signal_groups: set[str] = set()
+ for field in fields:
+ cohort_keys.update(PULSE_INTELLIGENCE_COHORT_BOOL_KEYS_BY_FIELD.get(field, ()))
+ cohort_keys.update(PULSE_INTELLIGENCE_COHORT_COUNT_KEYS_BY_FIELD.get(field, ()))
+ signal_groups.update(PULSE_INTELLIGENCE_SIGNAL_GROUP_BOOL_KEYS_BY_FIELD.get(field, ()))
+ signal_groups.update(PULSE_INTELLIGENCE_SIGNAL_GROUP_COUNT_KEYS_BY_FIELD.get(field, ()))
+ pulse_intelligence_derive_signal_groups(signal_groups)
+ return cohort_keys, signal_groups
+
+
+def analyze_pulse_intelligence_facts(
+ facts: Iterable[dict[str, Any]],
+) -> dict[str, PulseIntelligenceInstallAnalysis]:
+ analyses: dict[str, PulseIntelligenceInstallAnalysis] = {}
+ for fact in facts:
+ install_id = str(fact.get("install_id") or "").strip()
+ latest_raw = str(fact.get("latest_received_at") or "").strip()
+ if not install_id or not latest_raw:
+ continue
+ first_free_raw = str(fact.get("first_free_at") or "").strip()
+ first_paid_raw = str(fact.get("first_paid_at") or "").strip()
+ first_free_at = parse_received_at(first_free_raw) if first_free_raw else None
+ first_paid_at = parse_received_at(first_paid_raw) if first_paid_raw else None
+ cohort_keys, signal_groups = pulse_intelligence_field_analysis_keys(
+ fact.get("signal_fields") or ()
+ )
+ free_cohort_keys, free_signal_groups = pulse_intelligence_field_analysis_keys(
+ fact.get("free_signal_fields") or ()
+ )
+ observed_free_start = first_free_at is not None and (
+ first_paid_at is None or first_free_at < first_paid_at
+ )
+ analyses[install_id] = PulseIntelligenceInstallAnalysis(
+ latest_received_at=parse_received_at(latest_raw),
+ first_paid_at=first_paid_at,
+ observed_free_start=observed_free_start,
+ observed_free_to_paid=observed_free_start and first_paid_at is not None,
+ cohort_keys=frozenset(cohort_keys),
+ free_cohort_keys=frozenset(free_cohort_keys),
+ signal_groups=frozenset(signal_groups),
+ free_signal_groups=frozenset(free_signal_groups),
+ )
+ return analyses
+
+
def analyze_pulse_intelligence_install(
install_rows: Iterable[dict[str, Any]],
) -> PulseIntelligenceInstallAnalysis:
- timed_rows = pulse_intelligence_timed_rows(install_rows)
- if not timed_rows:
- raise ValueError("Pulse Intelligence install analysis requires at least one row")
-
- first_free_at = next(
- (received_at for received_at, _, posture in timed_rows if posture is False),
- None,
- )
- first_paid_at = next(
- (received_at for received_at, _, posture in timed_rows if posture is True),
- None,
- )
- observed_free_start = first_free_at is not None and (
- first_paid_at is None or first_free_at < first_paid_at
- )
-
- cohort_keys: set[str] = set()
- free_cohort_keys: set[str] = set()
- signal_groups: set[str] = set()
- free_signal_groups: set[str] = set()
-
- for received_at, row, posture in timed_rows:
- row_cohort_keys, row_signal_groups = pulse_intelligence_row_analysis_keys(row)
- cohort_keys.update(row_cohort_keys)
- signal_groups.update(row_signal_groups)
- if posture is False and (first_paid_at is None or received_at < first_paid_at):
- free_cohort_keys.update(row_cohort_keys)
- free_signal_groups.update(row_signal_groups)
-
- pulse_intelligence_derive_signal_groups(signal_groups)
- pulse_intelligence_derive_signal_groups(free_signal_groups)
- return PulseIntelligenceInstallAnalysis(
- latest_received_at=timed_rows[-1][0],
- first_paid_at=first_paid_at,
- observed_free_start=observed_free_start,
- observed_free_to_paid=observed_free_start and first_paid_at is not None,
- cohort_keys=frozenset(cohort_keys),
- free_cohort_keys=frozenset(free_cohort_keys),
- signal_groups=frozenset(signal_groups),
- free_signal_groups=frozenset(free_signal_groups),
- )
+ accumulator = _PulseIntelligenceInstallAccumulator()
+ for row in install_rows:
+ accumulator.observe(row)
+ return accumulator.finalize()
def pulse_intelligence_first_paid_at(
@@ -1903,23 +2144,22 @@ def summarize_pulse_intelligence_install_outcomes(
}
-def group_pulse_intelligence_rows_by_install(
- rows: Iterable[dict[str, Any]],
-) -> dict[str, list[dict[str, Any]]]:
- rows_by_install: dict[str, list[dict[str, Any]]] = {}
- for row in rows:
- install_id = str(row.get("install_id") or "").strip()
- if install_id:
- rows_by_install.setdefault(install_id, []).append(row)
- return rows_by_install
-
-
def analyze_pulse_intelligence_rows(
rows: Iterable[dict[str, Any]],
) -> dict[str, PulseIntelligenceInstallAnalysis]:
+ accumulators: dict[str, _PulseIntelligenceInstallAccumulator] = {}
+ for row in rows:
+ install_id = str(row.get("install_id") or "").strip()
+ if not install_id:
+ continue
+ accumulator = accumulators.get(install_id)
+ if accumulator is None:
+ accumulator = _PulseIntelligenceInstallAccumulator()
+ accumulators[install_id] = accumulator
+ accumulator.observe(row)
return {
- install_id: analyze_pulse_intelligence_install(install_rows)
- for install_id, install_rows in group_pulse_intelligence_rows_by_install(rows).items()
+ install_id: accumulator.finalize()
+ for install_id, accumulator in accumulators.items()
}
@@ -2087,7 +2327,12 @@ def summarize_user_base_signals(
schema_version = parse_optional_nonnegative_int(row.get("schema_version"))
schema_versions[str(schema_version or "legacy")] += 1
for field, _ in USER_BASE_CATEGORY_FIELDS:
- value = str(row.get(field) or "legacy_unknown").strip() or "legacy_unknown"
+ fallback = (
+ "not_reported"
+ if field == "update_last_failure_category"
+ else "legacy_unknown"
+ )
+ value = str(row.get(field) or fallback).strip() or fallback
categories[field][value] += 1
for field, _ in USER_BASE_BOOL_FIELDS:
if parse_optional_bool(row.get(field)):
@@ -2180,6 +2425,7 @@ def summarize_rows(
include_mock_fleet: bool = False,
*,
now: datetime | None = None,
+ pulse_intelligence_analysis_facts: Iterable[dict[str, Any]] | None = None,
) -> dict[str, Any]:
row_list: list[dict[str, Any]] = []
mock_fleet_rows = 0
@@ -2199,7 +2445,16 @@ def summarize_rows(
latest_by_install[install_id] = row
current_time = now or datetime.now(timezone.utc)
- pulse_intelligence_analysis = analyze_pulse_intelligence_rows(row_list)
+ pulse_intelligence_analysis = (
+ analyze_pulse_intelligence_facts(pulse_intelligence_analysis_facts)
+ if pulse_intelligence_analysis_facts is not None
+ else analyze_pulse_intelligence_rows(row_list)
+ )
+ pulse_intelligence_analysis = {
+ install_id: analysis
+ for install_id, analysis in pulse_intelligence_analysis.items()
+ if install_id in latest_by_install
+ }
latest_install_windows = summarize_latest_install_windows(
latest_by_install,
published_versions,
@@ -2380,7 +2635,7 @@ def format_text(summary: dict[str, Any], repo: str, since_days: int) -> str:
f" - {entry['version']}: {entry['installs']}"
for entry in user_base.get("schema_versions", [])
)
- lines.append("- lifecycle and audience buckets:")
+ lines.append("- lifecycle, audience, and update buckets:")
for signal in user_base.get("category_signals", []):
buckets = ", ".join(
f"{entry['bucket']} {entry['installs']}"
@@ -2597,6 +2852,9 @@ def main(argv: list[str] | None = None) -> int:
published_versions,
target_version=target_version,
include_mock_fleet=args.include_mock_fleet,
+ pulse_intelligence_analysis_facts=source.get(
+ "pulse_intelligence_analysis_facts"
+ ),
)
if args.format == "json":
diff --git a/scripts/tests/test_telemetry_adoption_report.py b/scripts/tests/test_telemetry_adoption_report.py
index 3d9ab430c..a3c4ab61e 100644
--- a/scripts/tests/test_telemetry_adoption_report.py
+++ b/scripts/tests/test_telemetry_adoption_report.py
@@ -7,8 +7,10 @@ from datetime import datetime, timedelta, timezone
from pathlib import Path
import gzip
import json
+import sqlite3
import subprocess
import sys
+import tempfile
import time
import unittest
from unittest import mock
@@ -37,7 +39,50 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
{"install_id": "a", "received_at": "2026-07-17 00:00:00"},
{"install_id": "b", "received_at": "2026-07-16 00:00:00"},
]
- stdout = "\n".join([json.dumps({"db_stats": db_stats}), *(json.dumps(row) for row in rows), ""])
+ analysis_facts = [
+ {
+ "install_id": "a",
+ "latest_received_at": "2026-07-17 00:00:00",
+ "first_free_at": "2026-07-17 00:00:00",
+ "first_paid_at": None,
+ "signal_fields": [],
+ "free_signal_fields": [],
+ }
+ ]
+ stdout = "\n".join(
+ [
+ json.dumps(
+ {
+ "db_stats": db_stats,
+ "row_columns": list(report.REPORT_ROW_COLUMNS),
+ }
+ ),
+ *(
+ json.dumps(
+ {
+ "a": [
+ fact["install_id"],
+ fact["latest_received_at"],
+ fact["first_free_at"],
+ fact["first_paid_at"],
+ fact["signal_fields"],
+ fact["free_signal_fields"],
+ ]
+ }
+ )
+ for fact in analysis_facts
+ ),
+ *(
+ json.dumps(
+ {
+ "r": [row.get(column) for column in report.REPORT_ROW_COLUMNS]
+ }
+ )
+ for row in rows
+ ),
+ "",
+ ]
+ )
completed = subprocess.CompletedProcess(
args=[],
returncode=0,
@@ -46,12 +91,153 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
)
with mock.patch.object(report.subprocess, "run", return_value=completed) as run_mock:
result = report.fetch_rows_remote("pulse-license", "/opt/licenses.sqlite", 30)
- self.assertEqual(result, {"db_stats": db_stats, "rows": rows})
+ expanded_rows = [
+ {column: row.get(column) for column in report.REPORT_ROW_COLUMNS}
+ for row in rows
+ ]
+ self.assertEqual(
+ result,
+ {
+ "db_stats": db_stats,
+ "rows": expanded_rows,
+ "pulse_intelligence_analysis_facts": analysis_facts,
+ },
+ )
remote_script = run_mock.call_args.kwargs["input"].decode("utf-8")
self.assertNotIn("fetchall", remote_script)
+ self.assertNotIn("SELECT *", remote_script)
self.assertIn("received_at >= datetime('now', ?)", remote_script)
+ self.assertEqual(
+ run_mock.call_args.args[0][-2],
+ ",".join(report.REPORT_ROW_COLUMNS),
+ )
+ self.assertEqual(
+ run_mock.call_args.args[0][-1],
+ ",".join(report.REPORT_HISTORY_SIGNAL_COLUMNS),
+ )
+ self.assertIn("ROW_NUMBER() OVER (", remote_script)
+ self.assertIn("GROUP BY install_id", remote_script)
+ self.assertIn("MIN(CASE WHEN paid_license = 0", remote_script)
compile(remote_script, "", "exec")
+ def test_report_projection_and_signal_specs_cover_recent_schema_fields(self) -> None:
+ recent_count_fields = {
+ "availability_probe_targets",
+ "availability_probe_agents",
+ "rbac_custom_roles",
+ "rbac_user_assignments",
+ "audit_reads_30d",
+ "report_schedules",
+ "report_schedules_enabled",
+ "report_schedules_run_30d",
+ "agent_profiles",
+ "update_attempts_30d",
+ "update_successes_30d",
+ "update_failures_30d",
+ }
+ projected = set(report.REPORT_ROW_COLUMNS)
+ self.assertTrue(recent_count_fields <= projected)
+ self.assertIn("alert_ai_enabled", projected)
+ self.assertIn("update_last_failure_category", projected)
+ self.assertNotIn("business_estate", projected)
+
+ specs = {entry["field"]: entry for entry in report.telemetry_signal_specs()}
+ for field in recent_count_fields:
+ self.assertEqual(specs[field]["type"], "count")
+ self.assertEqual(specs[field]["group"], "deep")
+ self.assertEqual(specs["alert_ai_enabled"]["type"], "bool")
+ self.assertEqual(specs["alert_ai_enabled"]["group"], "deep")
+
+ def test_compact_intelligence_facts_match_full_history_analysis(self) -> None:
+ numeric_columns = {
+ "schema_version",
+ "version_is_development",
+ "version_is_published_release",
+ *(key for key, _ in report.ADOPTION_COUNT_FIELDS),
+ *(key for key, _ in report.FEATURE_BOOL_FIELDS),
+ *(key for key, _ in report.USER_BASE_BOOL_FIELDS),
+ *(key for key, _ in report.USER_BASE_COUNT_FIELDS),
+ *(key for key, _ in report.PULSE_INTELLIGENCE_BOOL_FIELDS),
+ *(key for key, _ in report.PULSE_INTELLIGENCE_COUNT_FIELDS),
+ }
+ column_definitions = ", ".join(
+ f"{column} {'INTEGER' if column in numeric_columns else 'TEXT'}"
+ for column in report.REPORT_ROW_COLUMNS
+ )
+ now = datetime.now(timezone.utc).replace(microsecond=0)
+ rows = [
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=6)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 0,
+ },
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=5)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 0,
+ "pulse_intelligence_loop_configured": 1,
+ },
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=4)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 0,
+ },
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=3)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 1,
+ },
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 1,
+ "pulse_intelligence_approved_action_successes_30d": 1,
+ },
+ {
+ "install_id": "install-a",
+ "received_at": (now - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S"),
+ "paid_license": 1,
+ },
+ ]
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ db_path = str(Path(temp_dir) / "telemetry.sqlite")
+ conn = sqlite3.connect(db_path)
+ try:
+ conn.execute(f"CREATE TABLE telemetry_pings ({column_definitions})")
+ for row in rows:
+ columns = ", ".join(row)
+ placeholders = ", ".join("?" for _ in row)
+ conn.execute(
+ f"INSERT INTO telemetry_pings ({columns}) VALUES ({placeholders})",
+ tuple(row.values()),
+ )
+ conn.commit()
+ finally:
+ conn.close()
+
+ fetched_rows = report.fetch_rows_local(db_path, 1)["rows"]
+
+ self.assertEqual(len(fetched_rows), len(rows))
+ facts = [
+ {
+ "install_id": "install-a",
+ "latest_received_at": rows[-1]["received_at"],
+ "first_free_at": rows[0]["received_at"],
+ "first_paid_at": rows[3]["received_at"],
+ "signal_fields": [
+ "pulse_intelligence_loop_configured",
+ "pulse_intelligence_approved_action_successes_30d",
+ ],
+ "free_signal_fields": ["pulse_intelligence_loop_configured"],
+ }
+ ]
+ self.assertEqual(
+ report.analyze_pulse_intelligence_facts(facts),
+ report.analyze_pulse_intelligence_rows(rows),
+ )
+ self.assertEqual(fetched_rows[0]["received_at"], rows[-1]["received_at"])
+
def test_fetch_rows_remote_rejects_empty_response(self) -> None:
completed = subprocess.CompletedProcess(
args=[],
@@ -1445,8 +1631,10 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
now = datetime(2026, 7, 23, 12, tzinfo=timezone.utc)
rows: list[dict[str, object]] = []
latest_by_install: dict[str, dict[str, object]] = {}
- install_count = 500
- rows_per_install = 100
+ # Match the high-cardinality shape of the production 14-day window,
+ # where many installs contribute a small number of heartbeat rows.
+ install_count = 12_000
+ rows_per_install = 10
for install_index in range(install_count):
install_id = f"install-{install_index:04d}"
for row_index in range(rows_per_install):
@@ -1493,8 +1681,8 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
self.assertEqual(parse_received_at.call_count, len(rows))
self.assertLess(
elapsed,
- 5.0,
- f"50,000-row Pulse Intelligence aggregation took {elapsed:.3f}s",
+ 8.0,
+ f"120,000-row high-cardinality Pulse Intelligence aggregation took {elapsed:.3f}s",
)
def test_is_mock_fleet_row_matches_scaled_fixture_signature(self) -> None: