"""CLI analytics helpers.""" from __future__ import annotations import os import traceback from collections.abc import Generator, Mapping from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass from typing import TYPE_CHECKING, Final from uuid import uuid4 from config.constants.investigation import MAX_INVESTIGATION_LOOPS from platform.analytics.events import Event from platform.analytics.investigation_loop import ( begin_investigation_loop_metrics_scope, bound_loop_metrics, loop_metrics_from_state, merge_loop_properties, reset_investigation_loop_metrics, ) from platform.analytics.provider import Properties, get_analytics from platform.analytics.repl_context import get_cli_session_id from platform.analytics.source import ( EntrypointSource, TriggerMode, build_source_properties, ) from platform.observability.errors.sentry import capture_exception if TYPE_CHECKING: from core.agent_harness.session import SessionCore EVAL_AND_TERMINAL_KPI_QUERIES: Final[dict[str, str]] = { "eval_pass_rate": """ SELECT round( 100.0 * countIf( event = 'eval_process_completed' AND (properties.overall_pass = true OR properties.overall_pass = 'true') ) / nullIf(countIf(event = 'eval_process_completed'), 0), 2 ) AS eval_pass_rate FROM events WHERE event IN ('eval_process_completed') """.strip(), "eval_latency_p50_p95_ms": """ SELECT quantile(0.50)(toFloat64OrNull(properties.duration_ms)) AS eval_latency_p50_ms, quantile(0.95)(toFloat64OrNull(properties.duration_ms)) AS eval_latency_p95_ms FROM events WHERE event = 'eval_process_completed' """.strip(), "eval_parse_error_rate": """ SELECT round( 100.0 * countIf(event = 'eval_process_parse_failed') / nullIf(countIf(event IN ('eval_process_completed', 'eval_process_parse_failed')), 0), 2 ) AS eval_parse_error_rate FROM events WHERE event IN ('eval_process_completed', 'eval_process_parse_failed') """.strip(), "terminal_action_execution_success_rate": """ SELECT round( 100.0 * sum(toFloat64OrNull(properties.executed_success_count)) / nullIf(sum(toFloat64OrNull(properties.executed_count)), 0), 2 ) AS terminal_action_execution_success_rate FROM events WHERE event = 'terminal_actions_executed' """.strip(), "terminal_fallback_rate": """ SELECT round( 100.0 * countIf( event = 'terminal_turn_summarized' AND (properties.fallback_to_llm = true OR properties.fallback_to_llm = 'true') ) / nullIf(countIf(event = 'terminal_turn_summarized'), 0), 2 ) AS terminal_fallback_rate FROM events WHERE event = 'terminal_turn_summarized' """.strip(), } EVAL_AND_TERMINAL_EVENT_CONTRACT: Final[dict[Event, frozenset[str]]] = { Event.EVAL_PROCESS_STARTED: frozenset( { "rubric_present", "rubric_length_bucket", "mode", } ), Event.EVAL_PROCESS_COMPLETED: frozenset( { "duration_bucket", "duration_ms", "overall_pass", "score_bucket", "rubric_item_count", "mode", } ), Event.EVAL_PROCESS_FAILED: frozenset( { "duration_bucket", "duration_ms", "failure_stage", "failure_type", "mode", } ), Event.EVAL_PROCESS_SKIPPED: frozenset({"skip_reason", "mode"}), Event.EVAL_PROCESS_PARSE_FAILED: frozenset({"failure_type", "mode"}), Event.TERMINAL_ACTIONS_PLANNED: frozenset({"planned_count", "has_unhandled_clause"}), Event.TERMINAL_ACTIONS_EXECUTED: frozenset( {"planned_count", "executed_count", "executed_success_count", "success_rate_bucket"} ), Event.TERMINAL_TURN_SUMMARIZED: frozenset( { "planned_count", "executed_count", "executed_success_count", "fallback_to_llm", "session_turn_index", "session_fallback_count", "session_action_success_bucket", "session_fallback_rate_bucket", } ), } _INVESTIGATION_TRACKING_DEPTH: ContextVar[int] = ContextVar( "investigation_tracking_depth", default=0, ) @dataclass class InvestigationTracker: """Holds shared context for investigation lifecycle captures.""" shared_properties: Properties enabled: bool completed: bool = False failed: bool = False investigation_loop_count: int | None = None investigation_iteration_cap: int = MAX_INVESTIGATION_LOOPS def record_loop_metrics_from_state(self, state: Mapping[str, object] | None) -> None: """Capture canonical loop metrics from the investigation final state.""" loop_count, iteration_cap = loop_metrics_from_state(state) self.investigation_loop_count = loop_count self.investigation_iteration_cap = iteration_cap def _string_value(value: object) -> str | None: return value if isinstance(value, str) and value else None def _mapping_value(mapping: Mapping[str, object], key: str) -> str | None: return _string_value(mapping.get(key)) def _resolve_investigation_loop_metrics( *, loop_count: int | None = None, iteration_cap: int | None = None, state: Mapping[str, object] | None = None, tracker: InvestigationTracker | None = None, ) -> tuple[int, int]: if loop_count is not None: resolved_cap = ( iteration_cap if iteration_cap is not None else ( tracker.investigation_iteration_cap if tracker is not None else MAX_INVESTIGATION_LOOPS ) ) return max(0, int(loop_count)), max(1, int(resolved_cap)) bound = bound_loop_metrics() if bound is not None: return bound if state is not None: return loop_metrics_from_state(state) if tracker is not None and tracker.investigation_loop_count is not None: return tracker.investigation_loop_count, tracker.investigation_iteration_cap return 0, MAX_INVESTIGATION_LOOPS def _with_investigation_loop_metrics( properties: Properties, *, loop_count: int | None = None, iteration_cap: int | None = None, state: Mapping[str, object] | None = None, tracker: InvestigationTracker | None = None, ) -> Properties: count, cap = _resolve_investigation_loop_metrics( loop_count=loop_count, iteration_cap=iteration_cap, state=state, tracker=tracker, ) return merge_loop_properties(properties, loop_count=count, iteration_cap=cap) def _onboard_completed_properties(config: Mapping[str, object]) -> Properties: properties: Properties = {} wizard_obj = config.get("wizard") if isinstance(wizard_obj, Mapping): wizard_mode = _mapping_value(wizard_obj, "mode") configured_target = _mapping_value(wizard_obj, "configured_target") if wizard_mode is not None: properties["wizard_mode"] = wizard_mode if configured_target is not None: properties["configured_target"] = configured_target targets_obj = config.get("targets") if isinstance(targets_obj, Mapping): local_obj = targets_obj.get("local") if isinstance(local_obj, Mapping): provider = _mapping_value(local_obj, "provider") model = _mapping_value(local_obj, "model") if provider is not None: properties["provider"] = provider if model is not None: properties["model"] = model return properties def _investigation_started_properties( *, input_path: str | None, input_json: str | None, interactive: bool, evaluate_requested: bool, shared_properties: Properties, ) -> Properties: properties: Properties = { **shared_properties, "has_input_file": input_path is not None, "has_inline_json": input_json is not None, "interactive": interactive, "evaluate_requested": evaluate_requested, } llm_provider = _string_value(os.getenv("LLM_PROVIDER")) llm_model = _string_value(os.getenv("ANTHROPIC_MODEL")) or _string_value( os.getenv("OPENAI_MODEL") ) if llm_provider is not None: properties["llm_provider"] = llm_provider if llm_model is not None: properties["llm_model"] = llm_model return _with_investigation_loop_metrics( properties, loop_count=0, iteration_cap=MAX_INVESTIGATION_LOOPS, ) def _investigation_completed_properties( *, shared_properties: Properties, tracker: InvestigationTracker | None = None, state: Mapping[str, object] | None = None, ) -> Properties: return _with_investigation_loop_metrics( {**shared_properties}, state=state, tracker=tracker, ) def _investigation_failed_properties( *, shared_properties: Properties, failure_type: str | None = None, failure_message: str | None = None, failure_detail: str | None = None, failure_category: str | None = None, integration_involved: str | None = None, integration_failure_message: str | None = None, investigation_target: str | None = None, state: Mapping[str, object] | None = None, tracker: InvestigationTracker | None = None, ) -> Properties: properties: Properties = {**shared_properties} if failure_type: properties["failure_type"] = failure_type if failure_message: properties["failure_message"] = failure_message if failure_detail: properties["failure_detail"] = failure_detail if failure_category: properties["failure_category"] = failure_category if integration_involved: properties["integration_involved"] = integration_involved if integration_failure_message: properties["integration_failure_message"] = integration_failure_message if investigation_target: properties["investigation_target"] = investigation_target return _with_investigation_loop_metrics(properties, state=state, tracker=tracker) def _investigation_outcome_properties( *, investigation_id: str, status: str, investigation_target: str, root_cause_excerpt: str = "", error_excerpt: str = "", failure_category: str | None = None, integration_involved: str | None = None, integration_failure_message: str | None = None, failure_detail: str | None = None, state: Mapping[str, object] | None = None, ) -> Properties: properties: Properties = { "investigation_id": investigation_id, "status": status, "investigation_target": investigation_target, } if root_cause_excerpt: properties["root_cause_excerpt"] = root_cause_excerpt if error_excerpt: properties["error_excerpt"] = error_excerpt if failure_category: properties["failure_category"] = failure_category if integration_involved: properties["integration_involved"] = integration_involved if integration_failure_message: properties["integration_failure_message"] = integration_failure_message if failure_detail: properties["failure_detail"] = failure_detail session_id = get_cli_session_id() if session_id: properties["cli_session_id"] = session_id return _with_investigation_loop_metrics(properties, state=state) def capture_investigation_lifecycle_event( event: Event, properties: Properties, *, state: Mapping[str, object] | None = None, tracker: InvestigationTracker | None = None, loop_count: int | None = None, iteration_cap: int | None = None, ) -> None: """Capture an investigation lifecycle event with canonical loop metrics.""" _capture( event, _with_investigation_loop_metrics( properties, loop_count=loop_count, iteration_cap=iteration_cap, state=state, tracker=tracker, ), ) def _capture(event: Event, properties: Properties | None = None) -> None: try: get_analytics().capture(event, properties) except Exception as exc: capture_exception(exc) def _integration_lifecycle_properties(service: str) -> Properties: properties: Properties = {"service": service} session_id = get_cli_session_id() if session_id: properties["cli_session_id"] = session_id return properties def _bucket_duration_ms(duration_ms: float) -> str: if duration_ms < 500: return "<500ms" if duration_ms < 1000: return "500ms-1s" if duration_ms < 3000: return "1s-3s" if duration_ms < 5000: return "3s-5s" return ">=5s" def _bucket_score(score_0_100: int) -> str: if score_0_100 < 50: return "0-49" if score_0_100 < 70: return "50-69" if score_0_100 < 85: return "70-84" return "85-100" def _bucket_percentage(percent: float) -> str: if percent < 25: return "0-24" if percent < 50: return "25-49" if percent < 75: return "50-74" if percent < 95: return "75-94" return "95-100" def _bucket_rubric_length(text: str) -> str: size = len(text.strip()) if size == 0: return "0" if size < 256: return "1-255" if size < 1024: return "256-1023" if size < 4096: return "1024-4095" return ">=4096" def build_cli_invoked_properties( *, entrypoint: str, command_parts: list[str], json_output: bool = False, verbose: bool = False, debug: bool = False, yes: bool = False, interactive: bool = True, ) -> Properties: """Build a structured ``cli_invoked`` payload for any CLI surface. Used by ``opensre`` (Click-driven) and the ``python -m app.*`` entrypoints so all three end up with the same property names. Records command names only — never raw argv values, option values, paths, URLs, or secrets. """ properties: Properties = { "entrypoint": entrypoint, "command_path": " ".join((entrypoint, *command_parts)), "command_family": command_parts[0] if command_parts else "root", "json_output": json_output, "verbose": verbose, "debug": debug, "yes": yes, "interactive": interactive, } if len(command_parts) > 1: properties["subcommand"] = command_parts[1] if command_parts: properties["command_leaf"] = command_parts[-1] return properties def capture_cli_invoked(properties: Properties | None = None) -> None: _capture(Event.CLI_INVOKED, properties) def capture_repl_execution_policy_decision(properties: Properties | None = None) -> None: _capture(Event.REPL_EXECUTION_POLICY_DECISION, properties) def capture_onboard_started() -> None: _capture(Event.ONBOARD_STARTED) def capture_onboard_completed(config: Mapping[str, object]) -> None: _capture(Event.ONBOARD_COMPLETED, _onboard_completed_properties(config)) def capture_onboard_failed() -> None: _capture(Event.ONBOARD_FAILED) def capture_investigation_started( *, input_path: str | None, input_json: str | None, interactive: bool, entrypoint: EntrypointSource = EntrypointSource.CLI_COMMAND, trigger_mode: TriggerMode = TriggerMode.FILE, investigation_id: str | None = None, evaluate_requested: bool = False, ) -> None: shared_properties = build_source_properties( entrypoint=entrypoint, trigger_mode=trigger_mode, investigation_id=investigation_id or str(uuid4()), ) _capture( Event.INVESTIGATION_STARTED, _investigation_started_properties( input_path=input_path, input_json=input_json, interactive=interactive, evaluate_requested=evaluate_requested, shared_properties=shared_properties, ), ) def capture_diagnosis_category_mismatch( *, root_cause_category: str, mismatch_reason: str | None = None, ) -> None: properties: Properties = { "category_text_mismatch": True, "root_cause_category": root_cause_category, } if mismatch_reason: properties["mismatch_reason"] = mismatch_reason _capture(Event.DIAGNOSIS_CATEGORY_MISMATCH, properties) def capture_investigation_completed(*, tracker: InvestigationTracker | None = None) -> None: if tracker is None: _capture(Event.INVESTIGATION_COMPLETED) return if tracker.completed: return if tracker.failed or not tracker.enabled: return _capture( Event.INVESTIGATION_COMPLETED, _investigation_completed_properties( shared_properties=tracker.shared_properties, tracker=tracker, ), ) tracker.completed = True def capture_investigation_failed( *, tracker: InvestigationTracker | None = None, failure_type: str | None = None, failure_message: str | None = None, failure_detail: str | None = None, failure_category: str | None = None, integration_involved: str | None = None, integration_failure_message: str | None = None, investigation_target: str | None = None, shared_properties: Properties | None = None, state: Mapping[str, object] | None = None, ) -> None: props = _investigation_failed_properties( shared_properties=shared_properties or (tracker.shared_properties if tracker else {}), failure_type=failure_type, failure_message=failure_message, failure_detail=failure_detail, failure_category=failure_category, integration_involved=integration_involved, integration_failure_message=integration_failure_message, investigation_target=investigation_target, state=state, tracker=tracker, ) if tracker is None: _capture(Event.INVESTIGATION_FAILED, props) return if tracker.failed or not tracker.enabled: tracker.failed = True return _capture(Event.INVESTIGATION_FAILED, props) tracker.failed = True def capture_investigation_cancelled( *, investigation_id: str, investigation_target: str = "", tracker: InvestigationTracker | None = None, state: Mapping[str, object] | None = None, ) -> None: shared = tracker.shared_properties if tracker is not None and tracker.enabled else {} if investigation_id and not shared.get("investigation_id"): shared = {**shared, "investigation_id": investigation_id} properties: Properties = { **shared, "failure_category": "user_cancelled", } if investigation_target: properties["investigation_target"] = investigation_target capture_investigation_lifecycle_event( Event.INVESTIGATION_CANCELLED, properties, state=state, tracker=tracker, ) def capture_investigation_outcome( *, investigation_id: str, status: str, investigation_target: str, root_cause_excerpt: str = "", error_excerpt: str = "", failure_category: str | None = None, integration_involved: str | None = None, integration_failure_message: str | None = None, failure_detail: str | None = None, state: Mapping[str, object] | None = None, ) -> None: if not investigation_id: return _capture( Event.INVESTIGATION_OUTCOME, _investigation_outcome_properties( investigation_id=investigation_id, status=status, investigation_target=investigation_target, root_cause_excerpt=root_cause_excerpt, error_excerpt=error_excerpt, failure_category=failure_category, integration_involved=integration_involved, integration_failure_message=integration_failure_message, failure_detail=failure_detail, state=state, ), ) @contextmanager def track_investigation( *, entrypoint: EntrypointSource, trigger_mode: TriggerMode, input_path: str | None = None, input_json: str | None = None, interactive: bool = False, evaluate_requested: bool = False, investigation_id: str | None = None, investigation_target: str | None = None, session: SessionCore | None = None, ) -> Generator[InvestigationTracker]: """Capture investigation lifecycle once, with nested-call dedupe.""" depth = _INVESTIGATION_TRACKING_DEPTH.get() token = _INVESTIGATION_TRACKING_DEPTH.set(depth + 1) loop_metrics_token = begin_investigation_loop_metrics_scope() if depth == 0 else None tracker: InvestigationTracker if depth > 0: tracker = InvestigationTracker(shared_properties={}, enabled=False) else: resolved_id = investigation_id or str(uuid4()) shared_properties = build_source_properties( entrypoint=entrypoint, trigger_mode=trigger_mode, investigation_id=resolved_id, ) if investigation_target: shared_properties["investigation_target"] = investigation_target if session is not None: session.last_investigation_id = resolved_id _capture( Event.INVESTIGATION_STARTED, _investigation_started_properties( input_path=input_path, input_json=input_json, interactive=interactive, evaluate_requested=evaluate_requested, shared_properties=shared_properties, ), ) tracker = InvestigationTracker(shared_properties=shared_properties, enabled=True) try: yielded = tracker yield yielded except Exception as exc: failure_message = str(exc).strip()[:500] failure_detail = "".join(traceback.format_exception_only(exc)).strip()[:500] capture_investigation_failed( tracker=yielded, failure_type=type(exc).__name__, failure_message=failure_message or type(exc).__name__, failure_detail=failure_detail or None, investigation_target=investigation_target, ) raise else: if not yielded.failed and not yielded.completed: capture_investigation_completed(tracker=yielded) finally: _INVESTIGATION_TRACKING_DEPTH.reset(token) if depth == 0 and loop_metrics_token is not None: reset_investigation_loop_metrics(loop_metrics_token) def capture_integration_setup_started(service: str) -> None: _capture(Event.INTEGRATION_SETUP_STARTED, _integration_lifecycle_properties(service)) def capture_integration_setup_completed(service: str) -> None: _capture(Event.INTEGRATION_SETUP_COMPLETED, _integration_lifecycle_properties(service)) def capture_integrations_listed() -> None: _capture(Event.INTEGRATIONS_LISTED) def capture_integration_removed(service: str) -> None: _capture(Event.INTEGRATION_REMOVED, _integration_lifecycle_properties(service)) def capture_integration_verified(service: str) -> None: _capture(Event.INTEGRATION_VERIFIED, _integration_lifecycle_properties(service)) def identify_saved_github_username() -> None: """Re-attach a previously saved GitHub handle to PostHog for this process. The integration store persists ``credentials.username`` across REPL sessions (used by the welcome banner), but analytics persistent properties are in-memory per CLI process. Call at REPL boot so events like ``$ai_generation`` include ``github_username`` without requiring a fresh device-flow login each session. """ from integrations.github.identity import saved_github_username identify_github_username(saved_github_username()) def identify_github_username(username: str) -> None: """Attach the authenticated GitHub username to PostHog. Calls :meth:`~platform.analytics.provider.Analytics.identify` to persist ``github_username`` on the person profile AND :meth:`~platform.analytics.provider.Analytics.set_persistent_property` so the property is stamped directly on every subsequent event. Both are needed: the ``$identify`` call keeps the person profile up-to-date for cohort queries, while the persistent property makes ``github_username`` queryable as a plain ``properties.github_username`` filter on any event without requiring a person-profile join. No-op for an empty username. Best-effort: telemetry kill-switches make the underlying calls no-ops, and any unexpected error is swallowed to Sentry. """ if not username: return try: analytics = get_analytics() analytics.identify({"github_username": username}) analytics.set_persistent_property("github_username", username) except Exception as exc: capture_exception(exc) def capture_github_login_completed(username: str) -> None: _capture(Event.GITHUB_LOGIN_COMPLETED, {"github_username": username}) def capture_tests_picker_opened() -> None: _capture(Event.TESTS_PICKER_OPENED) def capture_test_synthetic_started(scenario: str, *, mock_grafana: bool) -> None: _capture( Event.TEST_SYNTHETIC_STARTED, {"scenario": scenario, "mock_grafana": mock_grafana}, ) def capture_test_synthetic_completed(scenario: str, *, exit_code: int) -> None: _capture(Event.TEST_SYNTHETIC_COMPLETED, {"scenario": scenario, "exit_code": exit_code}) def capture_test_synthetic_failed(scenario: str, *, reason: str) -> None: _capture(Event.TEST_SYNTHETIC_FAILED, {"scenario": scenario, "reason": reason}) def capture_tests_listed(category: str, *, search: bool) -> None: _capture(Event.TESTS_LISTED, {"category": category, "search": search}) def capture_test_run_started(test_id: str, *, dry_run: bool) -> None: _capture(Event.TEST_RUN_STARTED, {"test_id": test_id, "dry_run": dry_run}) def capture_test_run_completed(test_id: str, *, dry_run: bool, exit_code: int) -> None: _capture( Event.TEST_RUN_COMPLETED, { "test_id": test_id, "dry_run": dry_run, "exit_code": exit_code, }, ) def capture_test_run_failed(test_id: str, *, dry_run: bool, reason: str) -> None: _capture( Event.TEST_RUN_FAILED, { "test_id": test_id, "dry_run": dry_run, "reason": reason, }, ) def capture_eval_process_started(*, rubric: str, mode: str) -> None: _capture( Event.EVAL_PROCESS_STARTED, { "rubric_present": bool(rubric.strip()), "rubric_length_bucket": _bucket_rubric_length(rubric), "mode": mode, }, ) def capture_eval_process_skipped(*, reason: str, mode: str) -> None: _capture( Event.EVAL_PROCESS_SKIPPED, { "skip_reason": reason, "mode": mode, }, ) def capture_eval_process_completed( *, duration_ms: float, overall_pass: bool, score_0_100: int, rubric_item_count: int, mode: str, ) -> None: _capture( Event.EVAL_PROCESS_COMPLETED, { "duration_ms": round(duration_ms, 2), "duration_bucket": _bucket_duration_ms(duration_ms), "overall_pass": overall_pass, "score_bucket": _bucket_score(score_0_100), "rubric_item_count": rubric_item_count, "mode": mode, }, ) def capture_eval_process_parse_failed(*, failure_type: str, mode: str) -> None: _capture( Event.EVAL_PROCESS_PARSE_FAILED, { "failure_type": failure_type, "mode": mode, }, ) def capture_eval_process_failed( *, duration_ms: float, failure_stage: str, failure_type: str, mode: str, ) -> None: _capture( Event.EVAL_PROCESS_FAILED, { "duration_ms": round(duration_ms, 2), "duration_bucket": _bucket_duration_ms(duration_ms), "failure_stage": failure_stage, "failure_type": failure_type, "mode": mode, }, ) def capture_terminal_actions_planned(*, planned_count: int, has_unhandled_clause: bool) -> None: _capture( Event.TERMINAL_ACTIONS_PLANNED, { "planned_count": planned_count, "has_unhandled_clause": has_unhandled_clause, }, ) def capture_terminal_actions_executed( *, planned_count: int, executed_count: int, executed_success_count: int, ) -> None: success_percent = 100.0 * executed_success_count / executed_count if executed_count > 0 else 0.0 _capture( Event.TERMINAL_ACTIONS_EXECUTED, { "planned_count": planned_count, "executed_count": executed_count, "executed_success_count": executed_success_count, "success_rate_bucket": _bucket_percentage(success_percent), }, ) def capture_react_turn_completed( *, phase: str, llm_iterations_used: int, llm_iteration_cap: int, hit_iteration_cap: bool, stop_reason: str, tool_calls_executed: int, duration_ms: int, cli_session_id: str, cli_turn_kind: str, llm_provider: str, llm_model: str, investigation_id: str | None = None, investigation_loop_count: int | None = None, prompt_turn_id: str | None = None, ) -> None: properties: Properties = { "phase": phase, "llm_iterations_used": llm_iterations_used, "llm_iteration_cap": llm_iteration_cap, "hit_iteration_cap": hit_iteration_cap, "stop_reason": stop_reason, "tool_calls_executed": tool_calls_executed, "duration_ms": duration_ms, "cli_session_id": cli_session_id, "cli_turn_kind": cli_turn_kind, "llm_provider": llm_provider, "llm_model": llm_model, } if investigation_id: properties["investigation_id"] = investigation_id if investigation_loop_count is not None: properties["investigation_loop_count"] = investigation_loop_count if prompt_turn_id: properties["prompt_turn_id"] = prompt_turn_id _capture(Event.REACT_TURN_COMPLETED, properties) def capture_terminal_turn_summarized( *, planned_count: int, executed_count: int, executed_success_count: int, fallback_to_llm: bool, session_turn_index: int, session_fallback_count: int, session_action_success_percent: float, session_fallback_rate_percent: float, ) -> None: _capture( Event.TERMINAL_TURN_SUMMARIZED, { "planned_count": planned_count, "executed_count": executed_count, "executed_success_count": executed_success_count, "fallback_to_llm": fallback_to_llm, "session_turn_index": session_turn_index, "session_fallback_count": session_fallback_count, "session_action_success_bucket": _bucket_percentage(session_action_success_percent), "session_fallback_rate_bucket": _bucket_percentage(session_fallback_rate_percent), }, ) def capture_deploy_started(*, target: str, dry_run: bool) -> None: _capture(Event.DEPLOY_STARTED, {"target": target, "dry_run": dry_run}) def capture_deploy_completed(*, target: str, dry_run: bool) -> None: _capture(Event.DEPLOY_COMPLETED, {"target": target, "dry_run": dry_run}) def capture_deploy_failed(*, target: str, dry_run: bool) -> None: _capture(Event.DEPLOY_FAILED, {"target": target, "dry_run": dry_run}) def capture_update_started(*, check_only: bool) -> None: _capture(Event.UPDATE_STARTED, {"check_only": check_only}) def capture_update_completed(*, check_only: bool, updated: bool) -> None: _capture(Event.UPDATE_COMPLETED, {"check_only": check_only, "updated": updated}) def capture_update_failed(*, check_only: bool, reason: str) -> None: _capture(Event.UPDATE_FAILED, {"check_only": check_only, "reason": reason}) def capture_agent_secret_detected( *, rule_names: tuple[str, ...], count: int, blocked: bool, ) -> None: _capture( Event.AGENT_SECRET_DETECTED, {"rule_names": ",".join(rule_names), "count": count, "blocked": blocked}, )