diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 368ac1fe9..08652de3f 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -848,6 +848,22 @@ adoption baselines. `scripts/telemetry_adoption_report.py` must emit windowed 24h, 72h, and 7d latest-install snapshots that split published versions from unpublished or development builds, so RC adoption reads stop depending on ad hoc SQL or one-off local helper scripts. +Target-release activity reads must not attribute a rolling counter already +present on an install's first heartbeat after upgrade to the target release. +The first target-version heartbeat in the selected source window is a +non-attributable baseline; activity +requires consecutive heartbeats from the same pseudonymous install on the same +version and is reported as the observed counter increase across those pairs. A +version departure breaks the comparison chain, so returning to the target +version starts with another non-attributable observation. Counter decreases +remain separate because a rolling window or local reset can reduce a value. If +the latest later heartbeat reports a different version, the report must +classify that departure as a semantic-version rollback, a forward transition, +or unclassified development/version drift and must expose aggregate +destination-version counts without install identifiers. Activity from an +install whose latest later heartbeat departed the target must remain separate +from activity on installs still reporting the target version, so a rolled-back +install cannot supply positive evidence for the currently running cohort. Pulse Intelligence derived governed-operation booleans must treat content-free MCP / external-agent capability-class counters as external-agent collaboration activity, not only the legacy `pulse_intelligence_external_agent_used_30d` @@ -2109,9 +2125,15 @@ 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 +fields observed while free before the first paid posture. For an explicitly +selected target release, those compact facts also include the first and latest +target-version heartbeat times, consecutive same-version pair counts, the +first-heartbeat values, and separately accumulated increases and decreases for +the closed activity-counter projection. This preserves latest-state reporting, first-free/first-paid conversion, and all outcome -cohort membership without sending raw heartbeat history over SSH. The local +cohort membership while allowing inherited first-heartbeat totals, +same-version net changes, and later version departures to remain distinct, +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 diff --git a/scripts/telemetry_adoption_report.py b/scripts/telemetry_adoption_report.py index 4081d0a32..9696507c5 100644 --- a/scripts/telemetry_adoption_report.py +++ b/scripts/telemetry_adoption_report.py @@ -998,6 +998,37 @@ REPORT_HISTORY_SIGNAL_COLUMNS = tuple( ) ) ) +TARGET_RELEASE_ACTIVITY_COUNT_FIELDS = tuple( + dict.fromkeys( + ( + "alerts_fired_30d", + "alerts_acknowledged_30d", + "alerts_resolved_30d", + "notification_attempts_7d", + "notification_deliveries_7d", + "notification_failures_7d", + "notification_failures_authentication_7d", + "notification_failures_rate_limited_7d", + "notification_failures_connectivity_7d", + "notification_failures_tls_7d", + "notification_failures_configuration_7d", + "notification_failures_rejected_7d", + "notification_failures_unknown_7d", + "audit_reads_30d", + "report_schedules_run_30d", + "update_attempts_30d", + "update_successes_30d", + "update_failures_30d", + *(key for key, _ in PULSE_INTELLIGENCE_COUNT_FIELDS), + ) + ) +) +TARGET_RELEASE_ACTIVITY_LABELS = { + **{key: label for key, label in USER_BASE_COUNT_FIELDS}, + **{key: label for key, label in ADOPTION_COUNT_FIELDS}, + **{key: label for key, label in PULSE_INTELLIGENCE_COUNT_FIELDS}, + "notification_failures_7d": "Notification failures (7d; schema-dependent semantics)", +} GIT_DESCRIBE_RE = re.compile( r"^(?P\d+\.\d+\.\d+(?:-[0-9A-Za-z\.-]+)?)-(?P\d+)-g(?P[0-9a-fA-F]+)(?P-dirty)?$" ) @@ -1029,6 +1060,83 @@ class PulseIntelligenceInstallAnalysis: free_signal_groups: frozenset[str] +@dataclass(frozen=True) +class TargetReleaseInstallAnalysis: + first_received_at: datetime + latest_received_at: datetime + heartbeat_count: int + same_version_pair_count: int + first_counts: tuple[int, ...] + increase_totals: tuple[int, ...] + decrease_totals: tuple[int, ...] + + +@dataclass +class _TargetReleaseInstallAccumulator: + first_received_at: datetime | None = None + latest_received_at: datetime | None = None + heartbeat_count: int = 0 + same_version_pair_count: int = 0 + first_counts: tuple[int, ...] = () + increase_totals: list[int] = field( + default_factory=lambda: [0] * len(TARGET_RELEASE_ACTIVITY_COUNT_FIELDS) + ) + decrease_totals: list[int] = field( + default_factory=lambda: [0] * len(TARGET_RELEASE_ACTIVITY_COUNT_FIELDS) + ) + adjacent_received_at: datetime | None = None + adjacent_is_target: bool = False + adjacent_counts: tuple[int, ...] = () + + def observe(self, row: dict[str, Any], received_at: datetime, is_target: bool) -> None: + counts = tuple( + parse_optional_nonnegative_int(row.get(field)) + for field in TARGET_RELEASE_ACTIVITY_COUNT_FIELDS + ) + if is_target: + self.heartbeat_count += 1 + if self.first_received_at is None or received_at < self.first_received_at: + self.first_received_at = received_at + self.first_counts = counts + if self.latest_received_at is None or received_at > self.latest_received_at: + self.latest_received_at = received_at + + if ( + is_target + and self.adjacent_is_target + and self.adjacent_received_at is not None + and received_at != self.adjacent_received_at + ): + self.same_version_pair_count += 1 + if received_at > self.adjacent_received_at: + earlier_counts, later_counts = self.adjacent_counts, counts + else: + earlier_counts, later_counts = counts, self.adjacent_counts + for index, (earlier, later) in enumerate(zip(earlier_counts, later_counts)): + change = later - earlier + if change > 0: + self.increase_totals[index] += change + elif change < 0: + self.decrease_totals[index] += -change + + self.adjacent_received_at = received_at + self.adjacent_is_target = is_target + self.adjacent_counts = counts + + def finalize(self) -> TargetReleaseInstallAnalysis: + if self.first_received_at is None or self.latest_received_at is None: + raise ValueError("target release analysis requires at least one heartbeat") + return TargetReleaseInstallAnalysis( + first_received_at=self.first_received_at, + latest_received_at=self.latest_received_at, + heartbeat_count=self.heartbeat_count, + same_version_pair_count=self.same_version_pair_count, + first_counts=self.first_counts, + increase_totals=tuple(self.increase_totals), + decrease_totals=tuple(self.decrease_totals), + ) + + @dataclass class _PulseIntelligenceInstallAccumulator: latest_received_at: datetime | None = None @@ -1349,6 +1457,45 @@ def normalize_release_tag(tag: str) -> str: return version +def compare_semver_precedence(left: str, right: str) -> int | None: + """Compare SemVer precedence while ignoring build metadata.""" + + left_match = SEMVER_RE.match(normalize_release_tag(left)) + right_match = SEMVER_RE.match(normalize_release_tag(right)) + if left_match is None or right_match is None: + return None + + left_core = tuple(int(left_match.group(name)) for name in ("major", "minor", "patch")) + right_core = tuple(int(right_match.group(name)) for name in ("major", "minor", "patch")) + if left_core != right_core: + return -1 if left_core < right_core else 1 + + left_prerelease = left_match.group("prerelease") + right_prerelease = right_match.group("prerelease") + if left_prerelease is None and right_prerelease is None: + return 0 + if left_prerelease is None: + return 1 + if right_prerelease is None: + return -1 + + left_identifiers = left_prerelease.split(".") + right_identifiers = right_prerelease.split(".") + for left_identifier, right_identifier in zip(left_identifiers, right_identifiers): + if left_identifier == right_identifier: + continue + left_numeric = left_identifier.isdigit() + right_numeric = right_identifier.isdigit() + if left_numeric and right_numeric: + return -1 if int(left_identifier) < int(right_identifier) else 1 + if left_numeric != right_numeric: + return -1 if left_numeric else 1 + return -1 if left_identifier < right_identifier else 1 + if len(left_identifiers) == len(right_identifiers): + return 0 + return -1 if len(left_identifiers) < len(right_identifiers) else 1 + + def fetch_published_releases(repo: str) -> list[dict[str, Any]]: releases: list[dict[str, Any]] = [] page = 1 @@ -1445,10 +1592,15 @@ def fetch_rows_local(db_path: str, since_days: int) -> dict[str, Any]: conn.close() -def fetch_rows_remote(ssh_host: str, db_path: str, since_days: int) -> dict[str, Any]: +def fetch_rows_remote( + ssh_host: str, + db_path: str, + since_days: int, + target_version: str | None = None, +) -> dict[str, Any]: # 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. + # Only the latest row and compact per-install facts cross the network, not + # the full heartbeat history. remote_script = """ import gzip import json @@ -1459,6 +1611,8 @@ db_path = sys.argv[1] since_days = int(sys.argv[2]) column_names = sys.argv[3].split(",") intelligence_columns = sys.argv[4].split(",") +target_version = sys.argv[5] +target_activity_columns = sys.argv[6].split(",") if sys.argv[6] else [] if not column_names or any( not name or not name[0].isalpha() or not name.replace("_", "").isalnum() for name in column_names @@ -1469,6 +1623,8 @@ if not intelligence_columns or any( for name in intelligence_columns ): raise ValueError("invalid telemetry history signal projection") +if any(name not in column_names for name in target_activity_columns): + raise ValueError("invalid target release activity projection") conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row db_stats_sql = ( @@ -1508,6 +1664,70 @@ analysis_sql = ( "WHERE received_at >= datetime('now', ?) " "GROUP BY install_id" ) +target_analysis_sql = None +if target_version: + target_match = "(ordered_version = ? OR ordered_version = ?)" + previous_target_match = "(previous_version = ? OR previous_version = ?)" + pair_match = previous_target_match + " AND received_at <> previous_received_at" + target_selects = [ + "install_id", + "COUNT(*)", + "SUM(CASE WHEN " + pair_match + " THEN 1 ELSE 0 END)", + "MIN(received_at)", + "MAX(received_at)", + ] + for name in target_activity_columns: + target_selects.append( + "MAX(CASE WHEN first_rank = 1 THEN COALESCE(" + name + ", 0) END)" + ) + target_selects.append( + "SUM(CASE WHEN " + pair_match + " AND " + name + " > previous_" + name + + " THEN " + name + " - previous_" + name + " ELSE 0 END)" + ) + target_selects.append( + "SUM(CASE WHEN " + pair_match + " AND " + name + " < previous_" + name + + " THEN previous_" + name + " - " + name + " ELSE 0 END)" + ) + ordered_columns = [ + "install_id", + "received_at", + "rowid AS source_rowid", + "TRIM(version) AS ordered_version", + *target_activity_columns, + "LAG(received_at) OVER install_order AS previous_received_at", + "LAG(TRIM(version)) OVER install_order AS previous_version", + *( + "LAG(COALESCE(" + name + ", 0)) OVER install_order AS previous_" + name + for name in target_activity_columns + ), + ] + ordered_output_columns = [ + "install_id", + "received_at", + "source_rowid", + "ordered_version", + *target_activity_columns, + "previous_received_at", + "previous_version", + *("previous_" + name for name in target_activity_columns), + ] + target_analysis_sql = ( + "WITH ordered AS (" + "SELECT " + ", ".join(ordered_columns) + " " + "FROM telemetry_pings " + "WHERE received_at >= datetime('now', ?) " + "WINDOW install_order AS (" + "PARTITION BY install_id ORDER BY received_at ASC, rowid ASC" + ")" + "), target_rows AS (" + "SELECT " + ", ".join(ordered_output_columns) + ", " + "ROW_NUMBER() OVER (" + "PARTITION BY install_id ORDER BY received_at ASC, source_rowid ASC" + ") AS first_rank " + "FROM ordered WHERE " + target_match + + ") SELECT " + ", ".join(target_selects) + " " + "FROM target_rows GROUP BY install_id" + ) output = gzip.GzipFile(fileobj=sys.stdout.buffer, mode="wb", compresslevel=1) def emit(value): @@ -1534,6 +1754,32 @@ try: row[0], row[1], row[2], first_paid_at, signal_fields, free_signal_fields, ]}) + if target_analysis_sql is not None: + for row in conn.execute( + target_analysis_sql, + ( + cutoff, + target_version, + "v" + target_version, + *( + value + for _ in range(1 + (len(target_activity_columns) * 2)) + for value in (target_version, "v" + target_version) + ), + ), + ): + first_counts = [] + increase_totals = [] + decrease_totals = [] + for signal_index in range(len(target_activity_columns)): + value_index = 5 + (signal_index * 3) + first_counts.append(row[value_index]) + increase_totals.append(row[value_index + 1]) + decrease_totals.append(row[value_index + 2]) + emit({"t": [ + row[0], row[1], row[2], row[3], row[4], first_counts, + increase_totals, decrease_totals, + ]}) for row in conn.execute(rows_sql, (cutoff,)): emit({"r": list(row)}) finally: @@ -1550,6 +1796,8 @@ finally: str(since_days), ",".join(REPORT_ROW_COLUMNS), ",".join(REPORT_HISTORY_SIGNAL_COLUMNS), + normalize_release_tag(target_version or ""), + ",".join(TARGET_RELEASE_ACTIVITY_COUNT_FIELDS), ], input=remote_script.encode("utf-8"), capture_output=True, @@ -1569,6 +1817,7 @@ finally: raise RuntimeError(f"invalid row schema from remote telemetry fetch on {ssh_host}") rows: list[dict[str, Any]] = [] analysis_facts: list[dict[str, Any]] = [] + target_release_facts: list[dict[str, Any]] = [] for line in lines: record = json.loads(line) if "r" in record: @@ -1590,12 +1839,29 @@ finally: "free_signal_fields": values[5], } ) + elif "t" in record: + values = record["t"] + if len(values) != 8: + raise RuntimeError(f"invalid target release analysis from remote telemetry fetch on {ssh_host}") + target_release_facts.append( + { + "install_id": values[0], + "heartbeat_count": values[1], + "same_version_pair_count": values[2], + "first_received_at": values[3], + "latest_received_at": values[4], + "first_counts": values[5], + "increase_totals": values[6], + "decrease_totals": values[7], + } + ) 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, + "target_release_analysis_facts": target_release_facts, } @@ -2369,6 +2635,216 @@ def summarize_user_base_signals( } +def analyze_target_release_rows( + rows: Iterable[dict[str, Any]], + published_versions: set[str], + target_version: str, +) -> dict[str, TargetReleaseInstallAnalysis]: + normalized_target = normalize_release_tag(target_version) + accumulators: dict[str, _TargetReleaseInstallAccumulator] = {} + for row in rows: + install_id = str(row["install_id"]) + accumulator = accumulators.setdefault( + install_id, + _TargetReleaseInstallAccumulator(), + ) + accumulator.observe( + row, + parse_received_at(str(row["received_at"])), + classify_row_version(row, published_versions).version == normalized_target, + ) + return { + install_id: accumulator.finalize() + for install_id, accumulator in accumulators.items() + if accumulator.heartbeat_count > 0 + } + + +def analyze_target_release_facts( + facts: Iterable[dict[str, Any]], +) -> dict[str, TargetReleaseInstallAnalysis]: + analyses: dict[str, TargetReleaseInstallAnalysis] = {} + expected_count_length = len(TARGET_RELEASE_ACTIVITY_COUNT_FIELDS) + for fact in facts: + first_counts = tuple( + parse_optional_nonnegative_int(value) + for value in fact.get("first_counts") or () + ) + increase_totals = tuple( + parse_optional_nonnegative_int(value) + for value in fact.get("increase_totals") or () + ) + decrease_totals = tuple( + parse_optional_nonnegative_int(value) + for value in fact.get("decrease_totals") or () + ) + if ( + len(first_counts) != expected_count_length + or len(increase_totals) != expected_count_length + or len(decrease_totals) != expected_count_length + ): + raise ValueError("invalid target release activity fact count projection") + analyses[str(fact["install_id"])] = TargetReleaseInstallAnalysis( + first_received_at=parse_received_at(str(fact["first_received_at"])), + latest_received_at=parse_received_at(str(fact["latest_received_at"])), + heartbeat_count=parse_optional_nonnegative_int(fact.get("heartbeat_count")), + same_version_pair_count=parse_optional_nonnegative_int( + fact.get("same_version_pair_count") + ), + first_counts=first_counts, + increase_totals=increase_totals, + decrease_totals=decrease_totals, + ) + return analyses + + +def summarize_target_release_followup( + latest_by_install: dict[str, dict[str, Any]], + analyses_by_install: dict[str, TargetReleaseInstallAnalysis], + published_versions: set[str], + target_version: str, +) -> dict[str, Any]: + normalized_target = normalize_release_tag(target_version) + target_identity = classify_reported_version(normalized_target, published_versions) + analyses = { + install_id: analysis + for install_id, analysis in analyses_by_install.items() + if install_id in latest_by_install + } + signals = { + field: { + "field": field, + "label": TARGET_RELEASE_ACTIVITY_LABELS[field], + "first_heartbeat_baseline_nonzero_installs": 0, + "first_heartbeat_baseline_total": 0, + "same_version_increased_installs": 0, + "same_version_total_increase": 0, + "same_version_decreased_installs": 0, + "same_version_total_decrease": 0, + "same_version_unchanged_installs": 0, + "current_target_increased_installs": 0, + "current_target_total_increase": 0, + "current_target_decreased_installs": 0, + "current_target_total_decrease": 0, + "departed_increased_installs": 0, + "departed_total_increase": 0, + "departed_decreased_installs": 0, + "departed_total_decrease": 0, + } + for field in TARGET_RELEASE_ACTIVITY_COUNT_FIELDS + } + current_target_installs = 0 + same_version_followup_installs = 0 + current_target_followup_installs = 0 + departed_followup_installs = 0 + without_later_same_version_heartbeat = 0 + target_heartbeat_rows = 0 + transition_counts: dict[str, Counter[str]] = { + "rollback": Counter(), + "forward": Counter(), + "unclassified": Counter(), + } + + for install_id, analysis in analyses.items(): + target_heartbeat_rows += analysis.heartbeat_count + has_same_version_followup = analysis.same_version_pair_count > 0 + latest_row = latest_by_install[install_id] + latest_identity = classify_row_version(latest_row, published_versions) + is_current_target = latest_identity.version == normalized_target + if is_current_target: + current_target_installs += 1 + if has_same_version_followup: + same_version_followup_installs += 1 + if is_current_target: + current_target_followup_installs += 1 + else: + departed_followup_installs += 1 + else: + without_later_same_version_heartbeat += 1 + + for index, field in enumerate(TARGET_RELEASE_ACTIVITY_COUNT_FIELDS): + signal = signals[field] + first_value = analysis.first_counts[index] + if first_value > 0: + signal["first_heartbeat_baseline_nonzero_installs"] += 1 + signal["first_heartbeat_baseline_total"] += first_value + if not has_same_version_followup: + continue + increase = analysis.increase_totals[index] + decrease = analysis.decrease_totals[index] + if increase > 0: + signal["same_version_increased_installs"] += 1 + signal["same_version_total_increase"] += increase + destination = "current_target" if is_current_target else "departed" + signal[f"{destination}_increased_installs"] += 1 + signal[f"{destination}_total_increase"] += increase + if decrease > 0: + signal["same_version_decreased_installs"] += 1 + signal["same_version_total_decrease"] += decrease + destination = "current_target" if is_current_target else "departed" + signal[f"{destination}_decreased_installs"] += 1 + signal[f"{destination}_total_decrease"] += decrease + if increase == 0 and decrease == 0: + signal["same_version_unchanged_installs"] += 1 + + if is_current_target: + continue + latest_received_at = parse_received_at(str(latest_row["received_at"])) + if latest_received_at <= analysis.latest_received_at: + continue + transition_kind = "unclassified" + if ( + target_identity.channel in {"stable", "rc", "prerelease"} + and latest_identity.channel in {"stable", "rc", "prerelease"} + and latest_identity.is_published_release + ): + comparison = compare_semver_precedence(latest_identity.version, normalized_target) + if comparison is not None and comparison < 0: + transition_kind = "rollback" + elif comparison is not None and comparison > 0: + transition_kind = "forward" + transition_counts[transition_kind][latest_identity.version] += 1 + + def transition_entries(kind: str) -> list[dict[str, Any]]: + return counter_entries(transition_counts[kind], "destination_version") + + return { + "version": normalized_target, + "installs_seen": len(analyses), + "target_heartbeat_rows": target_heartbeat_rows, + "current_target_installs": current_target_installs, + "same_version_followup_installs": same_version_followup_installs, + "current_target_followup_installs": current_target_followup_installs, + "departed_followup_installs": departed_followup_installs, + "without_later_same_version_heartbeat": without_later_same_version_heartbeat, + "departed_after_target_installs": sum( + sum(counter.values()) for counter in transition_counts.values() + ), + "rollback_installs": sum(transition_counts["rollback"].values()), + "forward_transition_installs": sum(transition_counts["forward"].values()), + "unclassified_transition_installs": sum(transition_counts["unclassified"].values()), + "rollback_transitions": transition_entries("rollback"), + "forward_transitions": transition_entries("forward"), + "unclassified_transitions": transition_entries("unclassified"), + "activity_signals": list(signals.values()), + "interpretation": { + "first_heartbeat": ( + "Rolling counters on the first target-version heartbeat in the source window " + "are baseline only and are not attributed to the target release." + ), + "same_version_change": ( + "Activity is the observed counter change across consecutive target-version " + "heartbeats from the same pseudonymous install; a version departure breaks " + "the comparison chain." + ), + "counter_decrease": ( + "Decreases are reported separately because rolling windows and local resets can " + "reduce a counter; they are never subtracted from observed increases." + ), + }, + } + + def summarize_target_version_coverage( latest_by_install: dict[str, dict[str, Any]], published_versions: set[str], @@ -2425,7 +2901,9 @@ def summarize_rows( include_mock_fleet: bool = False, *, now: datetime | None = None, + source_window_days: int | None = None, pulse_intelligence_analysis_facts: Iterable[dict[str, Any]] | None = None, + target_release_analysis_facts: Iterable[dict[str, Any]] | None = None, ) -> dict[str, Any]: row_list: list[dict[str, Any]] = [] mock_fleet_rows = 0 @@ -2455,6 +2933,17 @@ def summarize_rows( for install_id, analysis in pulse_intelligence_analysis.items() if install_id in latest_by_install } + target_release_analysis: dict[str, TargetReleaseInstallAnalysis] = {} + if target_version: + target_release_analysis = ( + analyze_target_release_facts(target_release_analysis_facts) + if target_release_analysis_facts is not None + else analyze_target_release_rows( + row_list, + published_versions, + target_version, + ) + ) latest_install_windows = summarize_latest_install_windows( latest_by_install, published_versions, @@ -2462,6 +2951,16 @@ def summarize_rows( ) summary_72h = latest_install_windows["72h"] summary_7d = latest_install_windows["7d"] + target_release_followup = None + if target_version: + target_release_followup = summarize_target_release_followup( + latest_by_install, + target_release_analysis, + published_versions, + target_version, + ) + if source_window_days is not None: + target_release_followup["source_window_days"] = source_window_days return { "db_stats": db_stats, @@ -2504,6 +3003,7 @@ def summarize_rows( ) if target_version else None, + "target_release_followup": target_release_followup, "active_latest": { "active_24h": latest_install_windows["24h"]["active_installs"], "active_72h": summary_72h["active_installs"], @@ -2744,12 +3244,120 @@ def format_text(summary: dict[str, Any], repo: str, since_days: int) -> str: else: lines.append(" - none") + target_followup = summary.get("target_release_followup") + if target_followup: + lines.extend( + [ + "", + f"Target release follow-up ({target_followup['version']}):", + "- source window: last " + f"{target_followup.get('source_window_days', since_days)} day(s)", + f"- installs seen on target: {target_followup['installs_seen']}", + f"- target heartbeat rows: {target_followup['target_heartbeat_rows']}", + f"- latest version still target: {target_followup['current_target_installs']}", + "- attribution-ready installs with a later heartbeat on the same version: " + f"{target_followup['same_version_followup_installs']}", + " - still running target: " + f"{target_followup['current_target_followup_installs']}", + " - later departed target: " + f"{target_followup['departed_followup_installs']}", + "- installs without a later same-version heartbeat: " + f"{target_followup['without_later_same_version_heartbeat']}", + "- first target-version heartbeat in the source window is baseline only; its " + "rolling counters are excluded from " + "target-release activity", + "- observed same-version counter increases on installs still running target:", + ] + ) + increased_signals = [ + signal + for signal in target_followup.get("activity_signals", []) + if signal["current_target_total_increase"] > 0 + ] + if increased_signals: + lines.extend( + " - " + f"{signal['label']}: +{signal['current_target_total_increase']} across " + f"{signal['current_target_increased_installs']} install(s)" + for signal in increased_signals + ) + else: + lines.append(" - none") + + departed_increased_signals = [ + signal + for signal in target_followup.get("activity_signals", []) + if signal["departed_total_increase"] > 0 + ] + lines.append("- same-version counter increases on installs that later departed target:") + if departed_increased_signals: + lines.extend( + " - " + f"{signal['label']}: +{signal['departed_total_increase']} across " + f"{signal['departed_increased_installs']} install(s)" + for signal in departed_increased_signals + ) + else: + lines.append(" - none") + + decreased_signals = [ + signal + for signal in target_followup.get("activity_signals", []) + if signal["current_target_total_decrease"] > 0 + ] + lines.append( + "- rolling-counter decreases on installs still running target " + "(reported separately, never netted against increases):" + ) + if decreased_signals: + lines.extend( + " - " + f"{signal['label']}: -{signal['current_target_total_decrease']} across " + f"{signal['current_target_decreased_installs']} install(s)" + for signal in decreased_signals + ) + else: + lines.append(" - none") + + departed_decreased_signals = [ + signal + for signal in target_followup.get("activity_signals", []) + if signal["departed_total_decrease"] > 0 + ] + lines.append("- rolling-counter decreases on installs that later departed target:") + if departed_decreased_signals: + lines.extend( + " - " + f"{signal['label']}: -{signal['departed_total_decrease']} across " + f"{signal['departed_decreased_installs']} install(s)" + for signal in departed_decreased_signals + ) + else: + lines.append(" - none") + + for key, heading in ( + ("rollback_transitions", "rollback transitions"), + ("forward_transitions", "forward transitions"), + ("unclassified_transitions", "unclassified version departures"), + ): + transitions = target_followup.get(key, []) + lines.append(f"- {heading}:") + if transitions: + lines.extend( + f" - {entry['destination_version']}: {entry['installs']} install(s)" + for entry in transitions + ) + else: + lines.append(" - none") + target_coverage = summary.get("target_release_coverage_7d") if target_coverage: lines.extend( [ "", - f"Target release signal coverage (7d, {target_coverage['version']}):", + f"Target release latest-state signal coverage (7d, {target_coverage['version']}):", + "- interpretation: latest rolling totals show signal availability, not activity " + "caused by this release", f"- active installs: {target_coverage['active_installs']}", "- platforms:", ] @@ -2812,7 +3420,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument( "--target-version", help=( - "release version to highlight for per-signal coverage; defaults to " + "release version to highlight for latest-state coverage and same-version follow-up; defaults to " "the latest published stable release, falling back to the latest RC" ), ) @@ -2842,7 +3450,12 @@ def main(argv: list[str] | None = None) -> int: published_versions = {release["version"] for release in published_releases} target_version = args.target_version or latest_target_release_version(published_releases) source = ( - fetch_rows_remote(args.ssh_host, args.db_path, args.since_days) + fetch_rows_remote( + args.ssh_host, + args.db_path, + args.since_days, + target_version=target_version, + ) if args.ssh_host else fetch_rows_local(args.db_path, args.since_days) ) @@ -2852,9 +3465,13 @@ def main(argv: list[str] | None = None) -> int: published_versions, target_version=target_version, include_mock_fleet=args.include_mock_fleet, + source_window_days=args.since_days, pulse_intelligence_analysis_facts=source.get( "pulse_intelligence_analysis_facts" ), + target_release_analysis_facts=source.get( + "target_release_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 a3c4ab61e..2e69c4c52 100644 --- a/scripts/tests/test_telemetry_adoption_report.py +++ b/scripts/tests/test_telemetry_adoption_report.py @@ -49,6 +49,22 @@ class TelemetryAdoptionReportTest(unittest.TestCase): "free_signal_fields": [], } ] + first_counts = [0] * len(report.TARGET_RELEASE_ACTIVITY_COUNT_FIELDS) + increase_totals = list(first_counts) + increase_totals[0] = 2 + decrease_totals = list(first_counts) + target_release_facts = [ + { + "install_id": "a", + "heartbeat_count": 2, + "same_version_pair_count": 1, + "first_received_at": "2026-07-16 23:00:00", + "latest_received_at": "2026-07-17 00:00:00", + "first_counts": first_counts, + "increase_totals": increase_totals, + "decrease_totals": decrease_totals, + } + ] stdout = "\n".join( [ json.dumps( @@ -72,6 +88,23 @@ class TelemetryAdoptionReportTest(unittest.TestCase): ) for fact in analysis_facts ), + *( + json.dumps( + { + "t": [ + fact["install_id"], + fact["heartbeat_count"], + fact["same_version_pair_count"], + fact["first_received_at"], + fact["latest_received_at"], + fact["first_counts"], + fact["increase_totals"], + fact["decrease_totals"], + ] + } + ) + for fact in target_release_facts + ), *( json.dumps( { @@ -90,7 +123,12 @@ class TelemetryAdoptionReportTest(unittest.TestCase): stderr=b"", ) with mock.patch.object(report.subprocess, "run", return_value=completed) as run_mock: - result = report.fetch_rows_remote("pulse-license", "/opt/licenses.sqlite", 30) + result = report.fetch_rows_remote( + "pulse-license", + "/opt/licenses.sqlite", + 30, + target_version="v6.3.0-rc.3", + ) expanded_rows = [ {column: row.get(column) for column in report.REPORT_ROW_COLUMNS} for row in rows @@ -101,6 +139,7 @@ class TelemetryAdoptionReportTest(unittest.TestCase): "db_stats": db_stats, "rows": expanded_rows, "pulse_intelligence_analysis_facts": analysis_facts, + "target_release_analysis_facts": target_release_facts, }, ) remote_script = run_mock.call_args.kwargs["input"].decode("utf-8") @@ -108,16 +147,22 @@ class TelemetryAdoptionReportTest(unittest.TestCase): self.assertNotIn("SELECT *", remote_script) self.assertIn("received_at >= datetime('now', ?)", remote_script) self.assertEqual( - run_mock.call_args.args[0][-2], + run_mock.call_args.args[0][-4], ",".join(report.REPORT_ROW_COLUMNS), ) self.assertEqual( - run_mock.call_args.args[0][-1], + run_mock.call_args.args[0][-3], ",".join(report.REPORT_HISTORY_SIGNAL_COLUMNS), ) + self.assertEqual(run_mock.call_args.args[0][-2], "6.3.0-rc.3") + self.assertEqual( + run_mock.call_args.args[0][-1], + ",".join(report.TARGET_RELEASE_ACTIVITY_COUNT_FIELDS), + ) 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) + self.assertIn("target_analysis_sql", remote_script) compile(remote_script, "", "exec") def test_report_projection_and_signal_specs_cover_recent_schema_fields(self) -> None: @@ -148,6 +193,114 @@ class TelemetryAdoptionReportTest(unittest.TestCase): self.assertEqual(specs["alert_ai_enabled"]["type"], "bool") self.assertEqual(specs["alert_ai_enabled"]["group"], "deep") + def test_remote_target_release_query_executes_and_emits_compact_pair_deltas(self) -> None: + empty_header = gzip.compress( + ( + json.dumps( + { + "db_stats": {}, + "row_columns": list(report.REPORT_ROW_COLUMNS), + } + ) + + "\n" + ).encode("utf-8") + ) + completed = subprocess.CompletedProcess( + args=[], + returncode=0, + stdout=empty_header, + stderr=b"", + ) + with mock.patch.object(report.subprocess, "run", return_value=completed) as run_mock: + report.fetch_rows_remote( + "pulse-license", + "/unused.sqlite", + 7, + target_version="6.3.0-rc.3", + ) + remote_script = run_mock.call_args.kwargs["input"] + + numeric_columns = { + "schema_version", + "version_is_development", + "version_is_published_release", + "notification_failures_7d", + *(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", + "version": "6.3.0-rc.3", + "received_at": (now - timedelta(hours=3)).strftime("%Y-%m-%d %H:%M:%S"), + "alerts_fired_30d": 4, + }, + { + "install_id": "install-a", + "version": "6.3.0-rc.3", + "received_at": (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S"), + "alerts_fired_30d": 6, + }, + { + "install_id": "install-a", + "version": "6.2.1", + "received_at": (now - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S"), + "alerts_fired_30d": 6, + }, + ] + 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() + + result = subprocess.run( + [ + sys.executable, + "-", + db_path, + "7", + ",".join(report.REPORT_ROW_COLUMNS), + ",".join(report.REPORT_HISTORY_SIGNAL_COLUMNS), + "6.3.0-rc.3", + ",".join(report.TARGET_RELEASE_ACTIVITY_COUNT_FIELDS), + ], + input=remote_script, + capture_output=True, + check=True, + ) + + records = [ + json.loads(line) + for line in gzip.decompress(result.stdout).decode("utf-8").splitlines() + ] + target_record = next(record["t"] for record in records if "t" in record) + alerts_index = report.TARGET_RELEASE_ACTIVITY_COUNT_FIELDS.index("alerts_fired_30d") + self.assertEqual(target_record[1], 2) + self.assertEqual(target_record[2], 1) + self.assertEqual(target_record[5][alerts_index], 4) + self.assertEqual(target_record[6][alerts_index], 2) + self.assertEqual(target_record[7][alerts_index], 0) + def test_compact_intelligence_facts_match_full_history_analysis(self) -> None: numeric_columns = { "schema_version", @@ -356,6 +509,157 @@ class TelemetryAdoptionReportTest(unittest.TestCase): "6.2.0-rc.1", ) + def test_compare_semver_precedence_orders_rc_stable_and_patch_rollbacks(self) -> None: + self.assertLess(report.compare_semver_precedence("6.3.0-rc.2", "6.3.0-rc.3"), 0) + self.assertGreater(report.compare_semver_precedence("6.3.0", "6.3.0-rc.3"), 0) + self.assertLess(report.compare_semver_precedence("6.2.1", "6.3.0-rc.3"), 0) + self.assertEqual(report.compare_semver_precedence("6.3.0+build.2", "6.3.0"), 0) + + def test_target_release_followup_excludes_first_heartbeat_baselines_and_flags_rollbacks(self) -> None: + now = datetime(2026, 8, 19, 12, tzinfo=timezone.utc) + + def row( + install_id: str, + version: str, + hour: int, + **signals: object, + ) -> dict[str, object]: + return { + "install_id": install_id, + "version": version, + "platform": "binary", + "received_at": now.replace(hour=hour).strftime("%Y-%m-%d %H:%M:%S"), + **signals, + } + + rows = [ + row("followup", "6.3.0-rc.2", 7, alerts_fired_30d=38), + row( + "followup", + "6.3.0-rc.3", + 8, + alerts_fired_30d=40, + pulse_intelligence_patrol_ai_calls_30d=100, + ), + row( + "followup", + "6.3.0-rc.3", + 10, + alerts_fired_30d=43, + pulse_intelligence_patrol_ai_calls_30d=100, + ), + row("baseline-only", "6.3.0-rc.3", 9, alerts_fired_30d=500), + row("rollback", "6.3.0-rc.3", 7, pulse_intelligence_patrol_runs_30d=10), + row("rollback", "6.3.0-rc.3", 8, pulse_intelligence_patrol_runs_30d=12), + row("rollback", "6.2.1", 11, pulse_intelligence_patrol_runs_30d=12), + row("forward", "6.3.0-rc.3", 8), + row("forward", "6.3.0", 11), + row("drift", "6.3.0-rc.3", 8), + row("drift", "feature/local-build", 11), + ] + full_summary = report.summarize_rows( + { + "latest_ping": rows[-1]["received_at"], + "total_rows": len(rows), + "total_distinct_installs": 5, + }, + rows, + published_versions={"6.2.1", "6.3.0-rc.2", "6.3.0-rc.3", "6.3.0"}, + target_version="v6.3.0-rc.3", + now=now, + ) + summary = full_summary["target_release_followup"] + + self.assertEqual(summary["installs_seen"], 5) + self.assertEqual(summary["current_target_installs"], 2) + self.assertEqual(summary["same_version_followup_installs"], 2) + self.assertEqual(summary["current_target_followup_installs"], 1) + self.assertEqual(summary["departed_followup_installs"], 1) + self.assertEqual(summary["without_later_same_version_heartbeat"], 3) + self.assertEqual(summary["rollback_installs"], 1) + self.assertEqual( + summary["rollback_transitions"], + [{"destination_version": "6.2.1", "installs": 1}], + ) + self.assertEqual( + summary["forward_transitions"], + [{"destination_version": "6.3.0", "installs": 1}], + ) + self.assertEqual( + summary["unclassified_transitions"], + [{"destination_version": "0.0.0-feature-local-build", "installs": 1}], + ) + + signals = {entry["field"]: entry for entry in summary["activity_signals"]} + alerts = signals["alerts_fired_30d"] + self.assertEqual(alerts["first_heartbeat_baseline_total"], 540) + self.assertEqual(alerts["same_version_total_increase"], 3) + self.assertEqual(alerts["same_version_increased_installs"], 1) + self.assertEqual(alerts["current_target_total_increase"], 3) + self.assertEqual(alerts["departed_total_increase"], 0) + patrol_calls = signals["pulse_intelligence_patrol_ai_calls_30d"] + self.assertEqual(patrol_calls["first_heartbeat_baseline_total"], 100) + self.assertEqual(patrol_calls["same_version_total_increase"], 0) + self.assertEqual(patrol_calls["same_version_unchanged_installs"], 2) + patrol_runs = signals["pulse_intelligence_patrol_runs_30d"] + self.assertEqual(patrol_runs["same_version_total_increase"], 2) + self.assertEqual(patrol_runs["current_target_total_increase"], 0) + self.assertEqual(patrol_runs["departed_total_increase"], 2) + + rendered = report.format_text(full_summary, "rcourtman/Pulse", 7) + self.assertIn("Target release follow-up (6.3.0-rc.3):", rendered) + self.assertIn("first target-version heartbeat in the source window is baseline only", rendered) + self.assertIn("Alerts fired (30d): +3 across 1 install(s)", rendered) + self.assertIn("rollback transitions:", rendered) + self.assertIn("6.2.1: 1 install(s)", rendered) + + def test_compact_target_release_facts_match_local_one_pass_analysis(self) -> None: + rows = [ + { + "install_id": "install-a", + "version": "6.3.0-rc.3", + "received_at": "2026-08-19 08:00:00", + "alerts_fired_30d": 7, + }, + { + "install_id": "install-a", + "version": "6.3.0-rc.3", + "received_at": "2026-08-19 09:00:00", + "alerts_fired_30d": 9, + }, + { + "install_id": "install-a", + "version": "6.2.1", + "received_at": "2026-08-19 10:00:00", + "alerts_fired_30d": 9, + }, + ] + original_parse_received_at = report.parse_received_at + with mock.patch.object( + report, + "parse_received_at", + wraps=original_parse_received_at, + ) as parse_received_at: + local = report.analyze_target_release_rows( + rows, + {"6.3.0-rc.3", "6.2.1"}, + "6.3.0-rc.3", + ) + self.assertEqual(parse_received_at.call_count, 3) + + analysis = local["install-a"] + facts = [{ + "install_id": "install-a", + "heartbeat_count": analysis.heartbeat_count, + "same_version_pair_count": analysis.same_version_pair_count, + "first_received_at": analysis.first_received_at.strftime("%Y-%m-%d %H:%M:%S"), + "latest_received_at": analysis.latest_received_at.strftime("%Y-%m-%d %H:%M:%S"), + "first_counts": list(analysis.first_counts), + "increase_totals": list(analysis.increase_totals), + "decrease_totals": list(analysis.decrease_totals), + }] + self.assertEqual(report.analyze_target_release_facts(facts), local) + def test_value_loop_reconciles_schema_v4_action_outcomes_and_refusals(self) -> None: now = datetime.now(timezone.utc).replace(microsecond=0) summary = report.summarize_pulse_intelligence_value_loop( @@ -2102,7 +2406,8 @@ class TelemetryAdoptionReportTest(unittest.TestCase): self.assertIn("Deep telemetry signal sources (7d):", rendered) self.assertIn("- Agent hosts: 6.0.0-rc.2: 4 installs, total 18", rendered) self.assertIn("- Patrol enabled: 6.0.0-rc.2: 2 installs", rendered) - self.assertIn("Target release signal coverage (7d, 6.0.0-rc.6):", rendered) + self.assertIn("Target release latest-state signal coverage (7d, 6.0.0-rc.6):", rendered) + self.assertIn("latest rolling totals show signal availability, not activity caused by this release", rendered) self.assertIn(" - PVE nodes: 55 installs, total 131", rendered) self.assertIn(" - AI enabled: 19 installs", rendered) self.assertIn(" - Agent hosts, Patrol enabled", rendered)