tracer-cloud--opensre/platform/analytics/cli.py

"""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},
    )