From a099823eab2df1d9381fe35c469f9cfcf43a2c2c Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 26 Jul 2026 20:51:36 -0400 Subject: [PATCH 1/2] feat: add privacy-safe automation failure signals --- README.md | 28 +++ src/openadapt_telemetry/__init__.py | 31 +++ src/openadapt_telemetry/failure_signals.py | 277 +++++++++++++++++++++ src/openadapt_telemetry/posthog.py | 43 +++- tests/test_failure_signals.py | 107 ++++++++ 5 files changed, 474 insertions(+), 12 deletions(-) create mode 100644 src/openadapt_telemetry/failure_signals.py create mode 100644 tests/test_failure_signals.py diff --git a/README.md b/README.md index 54b7eb7..86624cb 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,9 @@ Unified telemetry and error tracking for OpenAdapt packages. - **Configurable Opt-Out**: Respects `DO_NOT_TRACK` and custom environment variables - **Internal Usage Tagging**: Explicit flags + CI detection with optional git heuristic - **GlitchTip/Sentry Compatible**: Uses the Sentry SDK for maximum compatibility +- **Failure recurrence without customer evidence**: Closed categorical failure + signatures omit application, tenant, workflow, step, text, input, screenshot, + origin, and exception details ## Installation @@ -70,6 +73,31 @@ capture_usage_event( ) ``` +### Capture an automation failure safely + +Use the closed-schema failure API rather than sending an exception message or +run report as analytics. Its grouping key is derived only from categorical +runtime facts; it is not an installation or tenant identifier. + +```python +from openadapt_telemetry import ( + ActionKind, AutomationFailureSignal, DeliveryState, ExecutionOutcome, + FailureKind, RiskClass, Substrate, capture_automation_failure, +) + +capture_automation_failure(AutomationFailureSignal( + failure_kind=FailureKind.DELIVERY_UNCERTAIN, + substrate=Substrate.CITRIX, + action_kind=ActionKind.CLICK, + risk_class=RiskClass.CONSEQUENTIAL, + delivery_state=DeliveryState.UNCERTAIN, + outcome=ExecutionOutcome.HALTED, +)) +``` + +This signal supports aggregate discovery only. Full evidence remains local, +and no signal can authorize or promote a repair. + ### Using Decorators ```python diff --git a/src/openadapt_telemetry/__init__.py b/src/openadapt_telemetry/__init__.py index 81f2e36..d20fc00 100644 --- a/src/openadapt_telemetry/__init__.py +++ b/src/openadapt_telemetry/__init__.py @@ -80,6 +80,22 @@ def add_demo(demo_id, task): track_shutdown, track_startup, ) +from openadapt_telemetry.failure_signals import ( + FAILURE_SIGNAL_EVENT, + FAILURE_SIGNAL_SCHEMA, + ActionKind, + AutomationFailureSignal, + DeliveryState, + EffectTier, + ExecutionOutcome, + ExecutionProfile, + FailureKind, + IdentityState, + ResolutionRung, + RiskClass, + Substrate, + capture_automation_failure, +) from openadapt_telemetry.posthog import ( capture_event as capture_posthog_event, ) @@ -138,6 +154,21 @@ def add_demo(demo_id, task): "track_command", "track_operation", "track_error", + # Privacy-safe automation failure discovery + "FAILURE_SIGNAL_EVENT", + "FAILURE_SIGNAL_SCHEMA", + "AutomationFailureSignal", + "FailureKind", + "Substrate", + "ActionKind", + "RiskClass", + "ResolutionRung", + "IdentityState", + "EffectTier", + "DeliveryState", + "ExecutionProfile", + "ExecutionOutcome", + "capture_automation_failure", # PostHog usage events "capture_posthog_event", "capture_usage_event", diff --git a/src/openadapt_telemetry/failure_signals.py b/src/openadapt_telemetry/failure_signals.py new file mode 100644 index 0000000..a14e69e --- /dev/null +++ b/src/openadapt_telemetry/failure_signals.py @@ -0,0 +1,277 @@ +"""Closed, privacy-safe automation failure signals. + +The full run report, screenshots, OCR, workflow parameters, application names, +target origins, exception messages, and exact step/workflow identifiers stay in +the declared execution boundary. This module emits only a coarse structural +signature assembled from enums. The signature can reveal recurrence without +becoming a customer evidence channel. + +This is discovery telemetry, not repair authorization. A recurring signal may +open a reviewed repair candidate, but it can never promote a patch or weaken an +identity, effect, risk, or policy contract. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from enum import Enum +from typing import Any + +from .posthog import _base_properties, _queue_capture_payload, _usage_enabled + +FAILURE_SIGNAL_SCHEMA = "openadapt.automation-failure-signal/v1" +FAILURE_SIGNAL_EVENT = "automation_failure_observed" +_RELEASE_RE = re.compile(r"^[0-9A-Za-z][0-9A-Za-z.+_-]{0,63}$") + + +class _ValueEnum(str, Enum): + def __str__(self) -> str: + return self.value + + +class FailureKind(_ValueEnum): + RUNTIME_BUG = "runtime_bug" + RESOLUTION_AMBIGUOUS = "resolution_ambiguous" + TARGET_NOT_FOUND = "target_not_found" + STALE_STATE = "stale_state" + IDENTITY_REFUTED = "identity_refuted" + IDENTITY_UNVERIFIABLE = "identity_unverifiable" + DELIVERY_UNCERTAIN = "delivery_uncertain" + EFFECT_REFUTED = "effect_refuted" + EFFECT_UNVERIFIABLE = "effect_unverifiable" + AUTHORIZATION_REFUSED = "authorization_refused" + INFRASTRUCTURE_FAILURE = "infrastructure_failure" + OPERATOR_OVER_HALT = "operator_over_halt" + WRONG_EFFECT_DETECTED = "wrong_effect_detected" + + +class Substrate(_ValueEnum): + WEB = "web" + WINDOWS = "windows" + MACOS = "macos" + LINUX = "linux" + RDP = "rdp" + CITRIX = "citrix" + MIXED = "mixed" + UNKNOWN = "unknown" + + +class ActionKind(_ValueEnum): + CLICK = "click" + DOUBLE_CLICK = "double_click" + RIGHT_CLICK = "right_click" + DRAG = "drag" + TYPE = "type" + KEY = "key" + HOTKEY = "hotkey" + WAIT = "wait" + SCROLL = "scroll" + NAVIGATE = "navigate" + API = "api" + MCP = "mcp" + TOOL = "tool" + FILE = "file" + CLIPBOARD = "clipboard" + UNKNOWN = "unknown" + + +class RiskClass(_ValueEnum): + READ_ONLY = "read_only" + STATE_CHANGING = "state_changing" + CONSEQUENTIAL = "consequential" + IRREVERSIBLE = "irreversible" + UNKNOWN = "unknown" + + +class ResolutionRung(_ValueEnum): + STRUCTURAL = "structural" + TEMPLATE = "template" + OCR = "ocr" + GEOMETRY = "geometry" + GROUNDER = "grounder" + NONE = "none" + UNKNOWN = "unknown" + + +class IdentityState(_ValueEnum): + VERIFIED = "verified" + REFUTED = "refuted" + UNVERIFIABLE = "unverifiable" + NOT_REQUIRED = "not_required" + UNKNOWN = "unknown" + + +class EffectTier(_ValueEnum): + INDEPENDENT_SYSTEM = "tier_1" + INDEPENDENT_SESSION = "tier_2" + PERSISTED_REACQUISITION = "tier_3" + IMMEDIATE_SCREEN = "tier_4" + NONE = "none" + UNKNOWN = "unknown" + + +class DeliveryState(_ValueEnum): + NOT_ATTEMPTED = "not_attempted" + CONFIRMED = "confirmed" + UNCERTAIN = "uncertain" + UNKNOWN = "unknown" + + +class ExecutionProfile(_ValueEnum): + DEMO = "demo" + STANDARD = "standard" + REGULATED = "regulated" + UNKNOWN = "unknown" + + +class ExecutionOutcome(_ValueEnum): + VERIFIED = "VERIFIED" + COMPLETED_UNVERIFIED = "COMPLETED_UNVERIFIED" + HALTED = "HALTED" + FAILED = "FAILED" + ROLLED_BACK = "ROLLED_BACK" + + +_SEVERITY = { + FailureKind.RUNTIME_BUG: "error", + FailureKind.RESOLUTION_AMBIGUOUS: "warning", + FailureKind.TARGET_NOT_FOUND: "warning", + FailureKind.STALE_STATE: "warning", + FailureKind.IDENTITY_REFUTED: "warning", + FailureKind.IDENTITY_UNVERIFIABLE: "warning", + FailureKind.DELIVERY_UNCERTAIN: "critical", + FailureKind.EFFECT_REFUTED: "critical", + FailureKind.EFFECT_UNVERIFIABLE: "error", + FailureKind.AUTHORIZATION_REFUSED: "warning", + FailureKind.INFRASTRUCTURE_FAILURE: "error", + FailureKind.OPERATOR_OVER_HALT: "info", + FailureKind.WRONG_EFFECT_DETECTED: "critical", +} + + +def _hour_bucket(value: datetime | None) -> str: + current = value or datetime.now(timezone.utc) + if current.tzinfo is None: + current = current.replace(tzinfo=timezone.utc) + current = current.astimezone(timezone.utc).replace(minute=0, second=0, microsecond=0) + return current.isoformat().replace("+00:00", "Z") + + +@dataclass(frozen=True) +class AutomationFailureSignal: + """A coarse failure observation safe for aggregate recurrence analysis.""" + + failure_kind: FailureKind + substrate: Substrate + action_kind: ActionKind = ActionKind.UNKNOWN + risk_class: RiskClass = RiskClass.UNKNOWN + resolution_rung: ResolutionRung = ResolutionRung.UNKNOWN + identity_state: IdentityState = IdentityState.UNKNOWN + effect_tier: EffectTier = EffectTier.UNKNOWN + delivery_state: DeliveryState = DeliveryState.UNKNOWN + execution_profile: ExecutionProfile = ExecutionProfile.UNKNOWN + outcome: ExecutionOutcome = ExecutionOutcome.HALTED + runtime_version: str = "unknown" + model_calls: int = 0 + external_network_calls: str = "unknown" + occurred_at: str = "" + + def __post_init__(self) -> None: + if not isinstance(self.failure_kind, FailureKind): + raise TypeError("failure_kind must be a FailureKind") + for name, enum_type in ( + ("substrate", Substrate), + ("action_kind", ActionKind), + ("risk_class", RiskClass), + ("resolution_rung", ResolutionRung), + ("identity_state", IdentityState), + ("effect_tier", EffectTier), + ("delivery_state", DeliveryState), + ("execution_profile", ExecutionProfile), + ("outcome", ExecutionOutcome), + ): + if not isinstance(getattr(self, name), enum_type): + raise TypeError(f"{name} must be a {enum_type.__name__}") + if self.runtime_version != "unknown" and not _RELEASE_RE.fullmatch(self.runtime_version): + raise ValueError("runtime_version must be a bounded release identifier") + if not isinstance(self.model_calls, int) or not 0 <= self.model_calls <= 1_000_000: + raise ValueError("model_calls must be a bounded non-negative integer") + if self.external_network_calls not in {"none", "observed", "unknown"}: + raise ValueError("external_network_calls must be none, observed, or unknown") + if self.occurred_at: + parsed = datetime.fromisoformat(self.occurred_at.replace("Z", "+00:00")) + object.__setattr__(self, "occurred_at", _hour_bucket(parsed)) + else: + object.__setattr__(self, "occurred_at", _hour_bucket(None)) + + @property + def failure_signature(self) -> str: + """Cross-install recurrence key derived from closed categories only.""" + basis = { + "schema": FAILURE_SIGNAL_SCHEMA, + "failure_kind": self.failure_kind.value, + "substrate": self.substrate.value, + "action_kind": self.action_kind.value, + "risk_class": self.risk_class.value, + "resolution_rung": self.resolution_rung.value, + "identity_state": self.identity_state.value, + "effect_tier": self.effect_tier.value, + "delivery_state": self.delivery_state.value, + "execution_profile": self.execution_profile.value, + "outcome": self.outcome.value, + } + encoded = json.dumps(basis, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(encoded.encode()).hexdigest() + + def to_envelope(self) -> dict[str, Any]: + """Return the exact schema-limited egress envelope.""" + values = asdict(self) + return { + "schema": FAILURE_SIGNAL_SCHEMA, + "failure_signature": self.failure_signature, + "failure_kind": self.failure_kind.value, + "severity": _SEVERITY[self.failure_kind], + "substrate": self.substrate.value, + "action_kind": self.action_kind.value, + "risk_class": self.risk_class.value, + "resolution_rung": self.resolution_rung.value, + "identity_state": self.identity_state.value, + "effect_tier": self.effect_tier.value, + "delivery_state": self.delivery_state.value, + "execution_profile": self.execution_profile.value, + "outcome": self.outcome.value, + "runtime_version": values["runtime_version"], + "model_calls": values["model_calls"], + "external_network_calls": values["external_network_calls"], + "occurred_at": values["occurred_at"], + } + + +def capture_automation_failure( + signal: AutomationFailureSignal, + *, + package_name: str = "openadapt-flow", +) -> bool: + """Queue a closed failure signal while honoring the standard opt-out. + + The PostHog ``distinct_id`` is the coarse failure signature itself, not an + installation, user, workflow, or tenant pseudonym. Consequently this event + supports recurrence counts but cannot be used to reconstruct who observed + it. No raw evidence is accepted by this API. + """ + if not isinstance(signal, AutomationFailureSignal): + raise TypeError("signal must be an AutomationFailureSignal") + if not _usage_enabled(): + return False + return _queue_capture_payload( + event=FAILURE_SIGNAL_EVENT, + distinct_id=f"failure:{signal.failure_signature}", + properties={ + **_base_properties(package_name), + **signal.to_envelope(), + }, + ) diff --git a/src/openadapt_telemetry/posthog.py b/src/openadapt_telemetry/posthog.py index 014a24e..8f7658b 100644 --- a/src/openadapt_telemetry/posthog.py +++ b/src/openadapt_telemetry/posthog.py @@ -150,6 +150,32 @@ def _base_properties(package_name: str) -> dict[str, Any]: } +def _queue_capture_payload( + *, + event: str, + distinct_id: str, + properties: dict[str, Any], +) -> bool: + """Queue an already-bounded event payload. + + This is deliberately private. Public callers use :func:`capture_event`, + which supplies the installation pseudonym. Closed-schema aggregate events + (for example automation failure signatures) may instead use a non-user + grouping key without exposing an installation or tenant identifier. + """ + payload = { + "api_key": _posthog_project_api_key(), + "event": event, + "distinct_id": distinct_id, + "properties": properties, + } + try: + _ensure_worker().put_nowait(payload) + return True + except queue.Full: + return False + + def _send_payload(payload: dict[str, Any]) -> None: timeout_seconds = float(os.getenv("OPENADAPT_TELEMETRY_TIMEOUT_SECONDS", "1.0")) req = urllib.request.Request( @@ -203,21 +229,14 @@ def capture_event( if not event_name or not _usage_enabled(): return False - payload = { - "api_key": _posthog_project_api_key(), - "event": event_name, - "distinct_id": _get_distinct_id(), - "properties": { + return _queue_capture_payload( + event=event_name, + distinct_id=_get_distinct_id(), + properties={ **_base_properties(package_name), **_sanitize_properties(properties), }, - } - - try: - _ensure_worker().put_nowait(payload) - return True - except queue.Full: - return False + ) def capture_usage_event( diff --git a/tests/test_failure_signals.py b/tests/test_failure_signals.py new file mode 100644 index 0000000..16e7091 --- /dev/null +++ b/tests/test_failure_signals.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import json +import os +from unittest.mock import patch + +import pytest + +from openadapt_telemetry.failure_signals import ( + FAILURE_SIGNAL_EVENT, + ActionKind, + AutomationFailureSignal, + DeliveryState, + EffectTier, + ExecutionOutcome, + ExecutionProfile, + FailureKind, + IdentityState, + ResolutionRung, + RiskClass, + Substrate, + capture_automation_failure, +) + + +class _CaptureQueue: + def __init__(self) -> None: + self.payload = None + + def put_nowait(self, payload): # noqa: ANN001 + self.payload = payload + + +def _signal(**overrides) -> AutomationFailureSignal: # noqa: ANN003 + values = { + "failure_kind": FailureKind.DELIVERY_UNCERTAIN, + "substrate": Substrate.CITRIX, + "action_kind": ActionKind.CLICK, + "risk_class": RiskClass.CONSEQUENTIAL, + "resolution_rung": ResolutionRung.OCR, + "identity_state": IdentityState.VERIFIED, + "effect_tier": EffectTier.PERSISTED_REACQUISITION, + "delivery_state": DeliveryState.UNCERTAIN, + "execution_profile": ExecutionProfile.REGULATED, + "outcome": ExecutionOutcome.HALTED, + "runtime_version": "1.23.0", + "model_calls": 0, + "external_network_calls": "none", + "occurred_at": "2026-07-26T19:37:22+00:00", + } + values.update(overrides) + return AutomationFailureSignal(**values) + + +def test_envelope_is_closed_coarse_and_deterministic() -> None: + signal = _signal() + envelope = signal.to_envelope() + assert envelope["occurred_at"] == "2026-07-26T19:00:00Z" + assert ( + envelope["failure_signature"] + == _signal(runtime_version="1.24.0", occurred_at="2026-08-01T10:00:00Z").failure_signature + ) + forbidden = { + "tenant", + "workflow", + "step", + "application", + "origin", + "message", + "exception", + "text", + "screenshot", + "parameter", + "evidence", + } + serialized = json.dumps(envelope, sort_keys=True).lower() + assert all(term not in serialized for term in forbidden) + + +def test_invalid_free_form_inputs_fail_closed() -> None: + with pytest.raises(TypeError, match="failure_kind"): + _signal(failure_kind="patient john smith") + with pytest.raises(ValueError, match="runtime_version"): + _signal(runtime_version="patient@example.com") + + +def test_capture_groups_by_failure_not_installation() -> None: + queue = _CaptureQueue() + with patch.dict( + os.environ, + { + "OPENADAPT_TELEMETRY_ENABLED": "true", + "OPENADAPT_TELEMETRY_DISTINCT_ID": "must-not-leave", + }, + clear=False, + ): + with patch("openadapt_telemetry.posthog._ensure_worker", return_value=queue): + signal = _signal() + assert capture_automation_failure(signal) is True + assert queue.payload["event"] == FAILURE_SIGNAL_EVENT + assert queue.payload["distinct_id"] == f"failure:{signal.failure_signature}" + assert "must-not-leave" not in json.dumps(queue.payload) + + +def test_capture_respects_do_not_track() -> None: + with patch.dict(os.environ, {"DO_NOT_TRACK": "1"}, clear=False): + assert capture_automation_failure(_signal()) is False From 6ab42a02c05f5cb4f6804a6fbeea40788f745087 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 26 Jul 2026 20:59:25 -0400 Subject: [PATCH 2/2] fix: close failure signal free-form seams --- src/openadapt_telemetry/failure_signals.py | 6 ++---- tests/test_failure_signals.py | 2 ++ 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/openadapt_telemetry/failure_signals.py b/src/openadapt_telemetry/failure_signals.py index a14e69e..9355d73 100644 --- a/src/openadapt_telemetry/failure_signals.py +++ b/src/openadapt_telemetry/failure_signals.py @@ -25,7 +25,7 @@ FAILURE_SIGNAL_SCHEMA = "openadapt.automation-failure-signal/v1" FAILURE_SIGNAL_EVENT = "automation_failure_observed" -_RELEASE_RE = re.compile(r"^[0-9A-Za-z][0-9A-Za-z.+_-]{0,63}$") +_RELEASE_RE = re.compile(r"^(?:unknown|[0-9]+(?:\.[0-9]+){1,3}(?:[-+][0-9A-Za-z.-]+)?)$") class _ValueEnum(str, Enum): @@ -253,8 +253,6 @@ def to_envelope(self) -> dict[str, Any]: def capture_automation_failure( signal: AutomationFailureSignal, - *, - package_name: str = "openadapt-flow", ) -> bool: """Queue a closed failure signal while honoring the standard opt-out. @@ -271,7 +269,7 @@ def capture_automation_failure( event=FAILURE_SIGNAL_EVENT, distinct_id=f"failure:{signal.failure_signature}", properties={ - **_base_properties(package_name), + **_base_properties("openadapt-flow"), **signal.to_envelope(), }, ) diff --git a/tests/test_failure_signals.py b/tests/test_failure_signals.py index 16e7091..a3b3d3d 100644 --- a/tests/test_failure_signals.py +++ b/tests/test_failure_signals.py @@ -82,6 +82,8 @@ def test_invalid_free_form_inputs_fail_closed() -> None: _signal(failure_kind="patient john smith") with pytest.raises(ValueError, match="runtime_version"): _signal(runtime_version="patient@example.com") + with pytest.raises(ValueError, match="runtime_version"): + _signal(runtime_version="JohnSmith") def test_capture_groups_by_failure_not_installation() -> None: