"""Batched telemetry for the AI Segmentation plugin.

Design principles (do not deviate):
- Global opt-out: the shared TerraLab/telemetry_enabled QSettings key. When
  disabled, nothing is even queued. Fail-closed on read errors.
- Events batch in memory and flush once per generation cycle (or immediately
  for FLUSH_NOW milestones/failures). Batching collapses the power-user
  "hundreds of clicks per session" volume into one POST per cycle.
- Lifecycle events (NO_CONSENT_EVENTS) ship as soon as the plugin is activated;
  everything else additionally requires the user to have accepted the ToS.
  Pre-auth lifecycle events park in _pending_pre_auth until the first
  authenticated flush drains them.
- Relay model: plugin -> POST {base}/api/plugin/track -> our analytics relay. No analytics
  key in the plugin. Body shape {"events": [...]} matches the shared route.
- MAIN THREAD ONLY: flush() ends in QgsApplication.taskManager().addTask(),
  which is main-thread-only. Worker threads must only track(); the next
  main-thread flush ships the batch.
- Errors in telemetry never affect plugin functionality (fail silently).
- Payloads carry only OS/version/timing/counts/enums, never PII: no paths,
  coordinates, layer names, urls, or emails.

HTTP stack: QgsBlockingNetworkRequest (inside the flush QgsTask), so the
relay POST inherits QGIS proxy/TLS settings. No raw requests/urllib.
"""

from __future__ import annotations

import functools
import json
import platform
import sys
import threading
import uuid

from qgis.core import (
    Qgis,
    QgsApplication,
    QgsBlockingNetworkRequest,
    QgsTask,
)
from qgis.PyQt.QtCore import QByteArray, QSettings, QThread, QUrl
from qgis.PyQt.QtNetwork import QNetworkRequest

from . import telemetry_events as ev
from .qt_compat import HttpStatusCodeAttribute, silent_task_flags
from .telemetry_events import FLUSH_NOW, NO_CONSENT_EVENTS, REGISTRY_VERSION

_TIMEOUT_MS = 5_000
_BATCH_MAX = 10
_PENDING_PRE_AUTH_MAX = 50
_TELEMETRY_ENABLED_KEY = "TerraLab/telemetry_enabled"

# How much telemetry travels is a server dial: batch size, transport timeout,
# and a per-event sample rate for throttling one noisy event during an
# incident. All three default to exactly what ships.
#
# WHAT the plugin is allowed to send is NOT a dial and must never become one.
# The opt-out default, NO_CONSENT_EVENTS and the scrubbing patterns are
# privacy commitments published in the FAQ; a server that could widen them
# could collect, from an installed plugin, something the user never agreed to.
# Volume can be tuned remotely. Scope cannot.


def _batch_max() -> int:
    """Events queued before a flush goes out."""
    try:
        from .server_dials import dial_in_range

        return int(dial_in_range("telemetry.batch_max", _BATCH_MAX, 1, 200))
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return _BATCH_MAX


def _timeout_ms() -> int:
    """Transport timeout on one relay call."""
    try:
        from .server_dials import dial_in_range

        return int(dial_in_range("telemetry.timeout_ms", _TIMEOUT_MS, 500, 60_000))
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return _TIMEOUT_MS


def _event_sampled_in(event: str) -> bool:
    """Whether this occurrence of ``event`` is kept.

    ``telemetry.sample_rates`` is a ``{event_name: rate}`` map, every event
    defaulting to 1.0 (keep everything, today's behaviour). A rate below one
    drops that share of occurrences at random, which is how one noisy event
    gets throttled during an incident without shipping a release. Only a
    number in [0, 1] counts; anything else keeps the event.

    A dropped event is never queued, so this cannot be used to collect more.
    """
    try:
        from .server_dials import read_value

        rates = read_value("telemetry.sample_rates")
        if not isinstance(rates, dict):
            return True
        rate = rates.get(event)
        if not isinstance(rate, (int, float)) or isinstance(rate, bool):
            return True
        if not 0.0 <= rate < 1.0:
            return True
        import random  # noqa: PLC0415 -- only needed when a rate is served

        return random.random() < rate  # nosec B311 -- sampling, not security
    except Exception:  # noqa: BLE001 -- telemetry must never break a caller
        return True


# Guards _batch / _pending_pre_auth / _inflight: track() can run on a worker
# thread while the main thread flushes, so the list mutations must not race.
_lock = threading.Lock()
_batch: list[dict] = []
_pending_pre_auth: list[dict] = []
_inflight: list = []
_session_id = uuid.uuid4().hex

# Most-recent Automatic run correlation id (the id the server archives each
# billed run under). Kept so the error report can quote it for support; None
# until the first run this process. Not telemetry state, just a breadcrumb.
_last_run_id: str | None = None

# The plugin version and relay base URL come from static bundled files that
# never change during a session; memoize them so track()/flush() do not re-read
# metadata.txt (and .env.local) from disk on every event.
_plugin_version_cache: str | None = None
_base_url_cache: str | None = None


# --- Opt-out --------------------------------------------------------------


def is_telemetry_enabled() -> bool:
    """Whether anonymous usage telemetry is enabled. Opt-out: defaults to True.

    Reads the shared TerraLab/telemetry_enabled QSettings key (shared with
    AI Edit so the user opts out once). Fail-closed: if the preference cannot be
    read, do NOT send (privacy over a data point)."""
    try:
        return bool(QSettings().value(_TELEMETRY_ENABLED_KEY, True, type=bool))
    except Exception:  # nosec B110
        return False


def set_telemetry_enabled(enabled: bool) -> None:
    """Persist the global telemetry opt-out flag (shared across TerraLab plugins)."""
    try:
        QSettings().setValue(_TELEMETRY_ENABLED_KEY, bool(enabled))
    except Exception:  # nosec B110
        pass


def new_session() -> None:
    """Rotate the per-session id. Call on dock open so events group by session."""
    global _session_id
    _session_id = uuid.uuid4().hex


def set_last_run_id(run_id: str | None) -> None:
    """Remember the most recent Automatic run's correlation id so a later error
    report can quote it. Best-effort breadcrumb, never raises."""
    global _last_run_id
    _last_run_id = run_id or None


def get_last_run_id() -> str | None:
    """The most recent Automatic run id this process, or None if no run yet."""
    return _last_run_id


# --- Payload helpers ------------------------------------------------------


def _base_properties() -> dict:
    """Properties shared by every event (computed once per call, cheap)."""
    try:
        qgis_version = Qgis.QGIS_VERSION
    except Exception:
        qgis_version = "unknown"
    props = {
        "product_id": "ai-segmentation",
        "plugin_version": _read_plugin_version(),
        "os": platform.system(),
        "os_version": platform.release(),
        "arch": platform.machine(),
        "python_version": f"{sys.version_info.major}.{sys.version_info.minor}",
        "qgis_version": qgis_version,
        "session_id": _session_id,
        "registry_version": REGISTRY_VERSION,
    }
    try:
        from .device_id import get_device_hash
        props["device_hash"] = get_device_hash()
    except Exception:  # nosec B110
        pass
    return props


def _read_plugin_version() -> str:
    global _plugin_version_cache
    if _plugin_version_cache is not None:
        return _plugin_version_cache
    import os
    version = "unknown"
    try:
        plugin_dir = os.path.dirname(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        )
        metadata_path = os.path.join(plugin_dir, "metadata.txt")
        with open(metadata_path, encoding="utf-8") as f:
            for line in f:
                if line.startswith("version="):
                    version = line.strip().split("=", 1)[1]
                    break
    except Exception:
        pass  # nosec B110
    # Cache only a real read: a transient failure ("unknown") is retried next
    # call rather than pinned for the whole session.
    if version != "unknown":
        _plugin_version_cache = version
    return version


def _build_base_url() -> str:
    global _base_url_cache
    if _base_url_cache is not None:
        return _base_url_cache
    import os
    plugin_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
    env_path = os.path.join(plugin_dir, ".env.local")
    base = "https://terra-lab.ai"
    if os.path.isfile(env_path):
        try:
            with open(env_path, encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("TERRALAB_BASE_URL="):
                        base = line.split("=", 1)[1].strip().strip('"').strip("'")
                        break
        except Exception:
            pass  # nosec B110
    _base_url_cache = base
    return base


def _get_auth_header() -> dict | None:
    """Return {'Authorization': 'Bearer <key>'} if the plugin is activated."""
    try:
        from .activation_manager import get_auth_header, is_plugin_activated
        if not is_plugin_activated():
            return None
        hdr = get_auth_header()
        if hdr and hdr.get("Authorization"):
            return hdr
    except Exception:
        pass  # nosec B110
    return None


def _has_consent() -> bool:
    """Non-lifecycle events additionally require ToS acceptance (raw message
    fields can carry path fragments; the ToS is the user's data gate)."""
    try:
        from .activation_manager import has_tos_accepted, has_tos_locked
        return bool(has_tos_accepted() or has_tos_locked())
    except Exception:
        return False


def _on_main_thread() -> bool:
    try:
        app = QgsApplication.instance()
        return app is not None and QThread.currentThread() == app.thread()
    except Exception:
        return False


# --- Background flush task ------------------------------------------------


class _TelemetryFlushTask(QgsTask):
    """Sends one batch. Failures swallowed: telemetry must never break the plugin."""

    def __init__(self, events: list, auth: dict):
        super().__init__("AI Segmentation telemetry flush", silent_task_flags())
        self._events = events
        self._auth = auth

    def run(self) -> bool:
        if self.isCanceled():
            return False
        # One retry with a short backoff covers a transient network blip without
        # a disk queue; a hard-offline session still loses the batch (accepted).
        if not self._post() and not self.isCanceled():
            import time
            time.sleep(2)
            if self.isCanceled():
                return False
            self._post()
        return True

    def _post(self) -> bool:
        try:
            payload = json.dumps({"events": self._events}).encode("utf-8")
            url = f"{_build_base_url().rstrip('/')}/api/plugin/track"
            req = QNetworkRequest(QUrl(url))
            req.setRawHeader(b"Content-Type", b"application/json")
            if hasattr(req, "setTransferTimeout"):
                req.setTransferTimeout(_timeout_ms())
            for k, v in self._auth.items():
                req.setRawHeader(k.encode("utf-8"), v.encode("utf-8"))
            blocker = QgsBlockingNetworkRequest()
            err = blocker.post(req, QByteArray(payload))
            # ErrorCode 0 = NoError. Ignoring the result made run()'s single
            # retry dead code: a failed batch returned True and was dropped.
            if int(err) != 0:
                return False
            # A transport-level NoError still covers a 4xx/5xx response (the POST
            # reached the relay but it rejected the batch): treat those as a
            # failure too so run()'s single retry fires. An unreadable status
            # (None) stays a success, since the transport itself succeeded.
            status = self._http_status(blocker)
            return status is None or status < 400
        except Exception:
            return False  # nosec B110 - telemetry must never break the plugin

    @staticmethod
    def _http_status(blocker) -> int | None:
        """HTTP status of the reply, or None. Never raises (Qt can hand back a
        non-numeric attribute that int() would choke on)."""
        if HttpStatusCodeAttribute is None:
            return None
        try:
            reply = blocker.reply()
            if reply is None:
                return None
            attr = reply.attribute(HttpStatusCodeAttribute)
        except (RuntimeError, AttributeError):
            return None
        if attr is None:
            return None
        try:
            return int(attr)
        except (TypeError, ValueError):
            return None

    def finished(self, result: bool) -> None:
        return


def _drop_inflight(task: _TelemetryFlushTask) -> None:
    with _lock:
        try:
            _inflight.remove(task)
        except ValueError:
            pass


# --- Core API -------------------------------------------------------------


def track(event: str, properties: dict | None = None, flush_now: bool = False) -> None:
    """Queue an event. Global opt-out short-circuits before anything is queued.

    Ships immediately when flush_now is True, the event is a FLUSH_NOW milestone,
    or the batch is full; otherwise it waits for the next flush()."""
    if not is_telemetry_enabled():
        return
    # A milestone always travels: sampling is for volume, not for losing the
    # events the funnel is measured on.
    if not (flush_now or event in FLUSH_NOW) and not _event_sampled_in(event):
        return
    try:
        evt = {
            "event": event,
            "properties": {**_base_properties(), **(properties or {})},
        }
    except Exception:  # nosec B110
        return
    with _lock:
        _batch.append(evt)
        should_flush = flush_now or event in FLUSH_NOW or len(_batch) >= _batch_max()
    if should_flush:
        flush()


def flush() -> None:
    """Ship the queued batch. MAIN THREAD ONLY (no-ops off it). Lifecycle events
    ship pre-consent; everything else requires consent. Pre-auth lifecycle events
    park in _pending_pre_auth until the first authenticated flush."""
    if not _on_main_thread():
        return
    task = None
    with _lock:
        if not _batch and not _pending_pre_auth:
            return
        auth = _get_auth_header()
        if not auth:
            for evt in _batch:
                if evt["event"] in NO_CONSENT_EVENTS and len(_pending_pre_auth) < _PENDING_PRE_AUTH_MAX:
                    _pending_pre_auth.append(evt)
            _batch.clear()
            return
        consented = _has_consent()
        events_to_send = list(_pending_pre_auth) + [
            e for e in _batch
            if consented or e["event"] in NO_CONSENT_EVENTS
        ]
        _batch.clear()
        _pending_pre_auth.clear()
        if not events_to_send:
            return
        task = _TelemetryFlushTask(events_to_send, auth)
        _inflight.append(task)
    try:
        task.taskCompleted.connect(lambda t=task: _drop_inflight(t))
        task.taskTerminated.connect(lambda t=task: _drop_inflight(t))
    except Exception:  # nosec B110
        pass
    QgsApplication.taskManager().addTask(task)


# --- Payload scrubbing (kept as-is) ---------------------------------------


_COORD_PATTERN = None
_URL_PATTERN = None
_EMAIL_PATTERN = None


def _scrub_payload_value(value: str) -> str:
    """Strip path-like tokens, coordinate tuples, URLs and email addresses
    from telemetry strings.

    Applied defensively to any string leaving the machine. We already call
    log_scrub.anonymize_paths for filesystem paths, but this pass also catches
    leftover coordinate-like artefacts (crop bounds, click tuples, bbox
    extents) plus URLs/emails: the unhandled-error catch-all forwards raw
    third-party exception text, which can embed a host or an address, and the
    telemetry contract is no URLs and no emails, so redact rather than trust
    the source.
    """
    import re as _re
    global _COORD_PATTERN, _URL_PATTERN, _EMAIL_PATTERN
    if _COORD_PATTERN is None:
        _COORD_PATTERN = _re.compile(
            r"(?:[-+]?\d+(?:\.\d+)?)(?:\s*,\s*[-+]?\d+(?:\.\d+)?){1,}"
        )
        _URL_PATTERN = _re.compile(r"[a-zA-Z][a-zA-Z0-9+.-]*://[^\s'\"]+")
        _EMAIL_PATTERN = _re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")
    # core-to-core import, deliberately NOT wrapped in try/except: a broken
    # scrubber must fail loudly, never silently ship unscrubbed paths.
    from .log_scrub import anonymize_paths
    value = anonymize_paths(value)
    value = _URL_PATTERN.sub("<URL>", value or "")
    value = _EMAIL_PATTERN.sub("<EMAIL>", value)
    return _COORD_PATTERN.sub("<COORDS>", value)


# --- Public typed wrappers (delegate to track) ----------------------------
#
# Every event constant has a wrapper so call sites stay readable and prop names
# are centralized. flush_now is decided by FLUSH_NOW membership inside track().


# Lifecycle -----------------------------------------------------------------


_FIRST_OPEN_KEY = "AI_Segmentation/first_open_sent"


def track_plugin_first_open() -> None:
    """Fire exactly once, ever, the first time the dock is opened on this machine.

    Guarded by a persistent QSettings flag so an install -> first-open ->
    activation funnel has a clean entry marker. No consent needed (a lifecycle
    ping with no user content); parks pre-auth like plugin_opened."""
    try:
        settings = QSettings()
        if bool(settings.value(_FIRST_OPEN_KEY, False, type=bool)):
            return
        settings.setValue(_FIRST_OPEN_KEY, True)
    except Exception:  # nosec B110 - never break the open path on a settings error
        return
    track(ev.PLUGIN_FIRST_OPEN)


def track_plugin_opened() -> None:
    """Fire once per dock-open."""
    track(ev.PLUGIN_OPENED)


def track_plugin_activated(duration_ms: int | None = None) -> None:
    """Fire when the activation key is validated. duration_ms is the elapsed
    browser sign-in wait when the key came from the pairing flow."""
    track(ev.PLUGIN_ACTIVATED, {"duration_ms": duration_ms})


def track_pairing_started() -> None:
    """Fire when the browser sign-in poll starts (success = plugin_activated)."""
    track(ev.PAIRING_STARTED)


def track_pairing_failed(error_code: str, duration_ms: int | None = None) -> None:
    """error_code is the server rejection code, or "timeout" when the poll
    window expired (a stable machine string, never a localized message)."""
    track(ev.PAIRING_FAILED, {"error_code": error_code, "duration_ms": duration_ms})


def track_pairing_cancelled(duration_ms: int | None = None) -> None:
    track(ev.PAIRING_CANCELLED, {"duration_ms": duration_ms})


def track_mode_switched(to_mode: str, had_unsaved_manual: bool = False,
                        auto_step: int | None = None) -> None:
    """Fire when the user changes the Manual / Automatic toggle."""
    track(ev.MODE_SWITCHED, {
        "to_mode": to_mode,
        "had_unsaved_manual": bool(had_unsaved_manual),
        "auto_step": auto_step,
    })


def track_install_started() -> None:
    track(ev.INSTALL_STARTED)


def track_install_completed(duration_ms: int | None = None,
                            python_minor: int | None = None,
                            retry_count: int | None = None) -> None:
    track(ev.INSTALL_COMPLETED, {
        "duration_ms": duration_ms,
        "python_minor": python_minor,
        "retry_count": retry_count,
    })


def track_install_failed(error_class: str, duration_ms: int | None = None,
                         python_minor: int | None = None,
                         retry_count: int | None = None) -> None:
    track(ev.INSTALL_FAILED, {
        "error_class": error_class,
        "duration_ms": duration_ms,
        "python_minor": python_minor,
        "retry_count": retry_count,
    })


def track_install_cancelled(duration_ms: int | None = None) -> None:
    """Fire when the user cancels the dependency install mid-way."""
    track(ev.INSTALL_CANCELLED, {"duration_ms": duration_ms})


def track_model_download_completed(model: str, duration_ms: int | None = None) -> None:
    """model is "sam1" or "sam2" ONLY (never a checkpoint URL or file name)."""
    track(ev.MODEL_DOWNLOAD_COMPLETED, {"model": model, "duration_ms": duration_ms})


# Automatic funnel ----------------------------------------------------------


def track_auto_start_clicked(layer_kind: str, has_credits_known: bool = False) -> None:
    track(ev.AUTO_START_CLICKED, {
        "layer_kind": layer_kind,
        "has_credits_known": bool(has_credits_known),
    })


def track_zone_drawn(vertices: int, area_km2: float, zone_kind: str = "polygon") -> None:
    track(ev.ZONE_DRAWN, {
        "vertices": vertices,
        "area_km2": round(area_km2, 1),
        "zone_kind": zone_kind,
    })


def track_auto_zone_too_large(area_km2: float) -> None:
    """Free-trial zone cap hit: the zone exceeds the free-tier limit. Only
    the rounded area is sent, never coordinates."""
    track(ev.AUTO_ZONE_TOO_LARGE, {"area_km2": round(area_km2, 1)})


def track_auto_prompt_committed(prompt: str, from_library: bool = False) -> None:
    """prompt is what the user typed, verbatim.

    NOT the validated object class, whatever this docstring used to claim. The
    commit fires on a debounce and on focus loss, both of which run before the
    Detect-time guard, so free-form text reaches this call unchanged. That is
    deliberate: the demand mining behind the object catalogue and the pricing
    work reads these rows, and a cleaned value would answer a different
    question. It does mean the "no PII by construction" guarantee does not hold
    here, and the privacy FAQ has to keep covering it.
    """
    track(ev.AUTO_PROMPT_COMMITTED, {"prompt": prompt, "from_library": bool(from_library)})


def track_auto_prompt_steered(prompt: str, suggestion: str = "") -> None:
    """prompt is the weak 1-2 word object the user typed; suggestion is the term
    steered toward ("" = pointed at the Library). No PII by construction."""
    track(ev.AUTO_PROMPT_STEERED, {"prompt": prompt, "suggestion": suggestion or ""})


def track_auto_prompt_rewritten(kind: str, prompt: str = "") -> None:
    """A committed prompt was swapped for a cleaner run phrase. kind is one of
    "translated" / "plural" / "alias" (the commit-time guard) or
    "server_rewrite" (a server-side language-model rewrite from the run plan);
    prompt is the 1-2 word English token/phrase that will run (no PII by
    construction)."""
    track(ev.AUTO_PROMPT_REWRITTEN, {"kind": kind, "prompt": prompt or ""})


def track_auto_prompt_hint_shown(kind: str, prompt: str = "") -> None:
    """A non-blocking pre-Detect guidance note was shown. kind is
    "exemplar_boost" (a curated high-value-exemplar object with no example
    drawn), "unknown_object" (a word the model does not know well), "plan_hint"
    (a server run-plan hint under the prompt box), or "identical_rerun" (the
    next Detect would repeat the last run exactly); prompt is the 1-2 word
    object class (no PII by construction)."""
    track(ev.AUTO_PROMPT_HINT_SHOWN, {"kind": kind, "prompt": prompt or ""})


def track_tutorial_opened(source: str) -> None:
    """A tutorial/guide open. source is the touchpoint id (footer_tutorial,
    post_signin, zero_results); no PII by construction."""
    track(ev.TUTORIAL_OPENED, {"source": source})


def track_exemplar_added(count_after: int, label: str = "") -> None:
    """label is "include" or "exclude" (additive property; count_after stays
    the registry's only listed one)."""
    track(ev.EXEMPLAR_ADDED, {"count_after": count_after, "label": label})


def track_exemplar_removed(count_after: int) -> None:
    track(ev.EXEMPLAR_REMOVED, {"count_after": count_after})


def track_detail_changed(detail: int, tiles: int, source: str) -> None:
    """source: "auto_seeded" or "user"."""
    track(ev.DETAIL_CHANGED, {"detail": detail, "tiles": tiles, "source": source})


def track_auto_detect_started(run_id: str, tiles: int, zone_km2: float,
                              object_class: str, detail: int, exemplar_count: int,
                              est_credits: int, credits_before: int | None,
                              is_free_tier: bool,
                              merge_mode: str = "separate",
                              merge_mode_source: str = "prompt") -> None:
    """merge_mode is the count-vs-map policy the run picked ("separate"/"map");
    merge_mode_source says how it was decided: "prompt" (object token) or
    "signal" (exemplar-only, read from the run's own masks)."""
    track(ev.AUTO_DETECT_STARTED, {
        "run_id": run_id,
        "tiles": tiles,
        "zone_km2": round(zone_km2, 2),
        "object_class": object_class,
        "detail": detail,
        "exemplar_count": exemplar_count,
        "est_credits": est_credits,
        "credits_before": credits_before,
        "is_free_tier": bool(is_free_tier),
        "merge_mode": merge_mode,
        "merge_mode_source": merge_mode_source,
    })


def track_auto_detect_completed(run_id: str, duration_ms: int, tiles_done: int,
                                tiles_failed: int, instances_found: int,
                                instances_visible_at_default: int, zero_at_default: bool,
                                p50_tile_ms: int | None = None,
                                p95_tile_ms: int | None = None,
                                stop_reason: str = "completed",
                                warming_ms: int = 0,
                                merge_mode_final: str = "separate") -> None:
    """warming_ms is the wall time the run spent in the server waiting room
    (cold start / queue) as perceived by the user; 0 = the run never waited.
    Per-tile latency lives server-side keyed by run_id, so no client percentiles.
    merge_mode_final is the count-vs-map grouping the run finished on
    ("separate"/"map")."""
    track(ev.AUTO_DETECT_COMPLETED, {
        "run_id": run_id,
        "duration_ms": duration_ms,
        "tiles_done": tiles_done,
        "tiles_failed": tiles_failed,
        "instances_found": instances_found,
        "instances_visible_at_default": instances_visible_at_default,
        "zero_at_default": bool(zero_at_default),
        "p50_tile_ms": p50_tile_ms,
        "p95_tile_ms": p95_tile_ms,
        "stop_reason": stop_reason,
        "warming_ms": warming_ms,
        "merge_mode_final": merge_mode_final,
    })


def track_auto_gate_scan(run_id: str, tiles: int, group: int, scans: int,
                         blocks: int, tiles_skipped: int, tiles_prepaid: int,
                         tiles_unscanned: int, fallback: str,
                         scan_ms: int, tiles_prefiltered: int = 0) -> None:
    """Empty-tile scan gate outcome for one run. Emitted only when the server
    policy armed the gate for the run; fallback carries the stand-down reason
    (resolution / not_text_run / bad_config / no_blocks / offline) or "" when
    the scan phase actually ran. tiles_prefiltered counts the tiles the
    client degenerate prefilter settled during the scan phase (no request).
    Safe from the worker thread (track() only queues; the next main-thread
    flush ships it)."""
    track(ev.AUTO_GATE_SCAN, {
        "run_id": run_id,
        "tiles": tiles,
        "group": group,
        "scans": scans,
        "blocks": blocks,
        "tiles_skipped": tiles_skipped,
        "tiles_prepaid": tiles_prepaid,
        "tiles_unscanned": tiles_unscanned,
        "tiles_prefiltered": tiles_prefiltered,
        "fallback": fallback,
        "scan_ms": scan_ms,
    })


def track_auto_detect_failed(run_id: str, error_class: str, tiles_done: int,
                             duration_ms: int | None = None,
                             warming_ms: int = 0) -> None:
    """error_class: NETWORK/AUTH/CREDITS_EXHAUSTED/SERVER/CANCELLED/TIMEOUT/UNKNOWN."""
    track(ev.AUTO_DETECT_FAILED, {
        "run_id": run_id,
        "error_class": error_class,
        "tiles_done": tiles_done,
        "duration_ms": duration_ms,
        "warming_ms": warming_ms,
    })


def track_auto_detect_cancelled(run_id: str, tiles_done: int, tiles_total: int,
                                salvaged_to_review: bool,
                                duration_ms: int | None = None,
                                warming_ms: int = 0,
                                backend_stalled: bool = False,
                                submit_retries: int = 0) -> None:
    """duration_ms separates a reflex cancel from a gave-up-after-minutes one;
    warming_ms says how much of that wait was the server waiting room (this is
    the busy-time signal; no separate seconds field, so the *_ms convention
    holds). backend_stalled is True when the run billed ZERO tiles AND the
    service was unresponsive (waiting-room time and/or submit retries): the user
    cancelled a sick backend, not a healthy run. submit_retries is the run's
    total transient submit-retry count. Together they keep a backend outage from
    reading as a user-initiated cancel in analytics."""
    track(ev.AUTO_DETECT_CANCELLED, {
        "run_id": run_id,
        "tiles_done": tiles_done,
        "tiles_total": tiles_total,
        "salvaged_to_review": bool(salvaged_to_review),
        "duration_ms": duration_ms,
        "warming_ms": warming_ms,
        "backend_stalled": bool(backend_stalled),
        "submit_retries": int(submit_retries),
    })


def track_credits_exhausted(run_id: str, tiles_done: int, tiles_total: int,
                            is_free_tier: bool) -> None:
    track(ev.CREDITS_EXHAUSTED, {
        "run_id": run_id,
        "tiles_done": tiles_done,
        "tiles_total": tiles_total,
        "is_free_tier": bool(is_free_tier),
    })


def track_auto_tiles_degraded(run_id: str, skipped_tiles: int, timeout_tiles: int,
                              blank_tiles: int = 0,
                              render_failed_tiles: int = 0) -> None:
    track(ev.AUTO_TILES_DEGRADED, {
        "run_id": run_id,
        "skipped_tiles": skipped_tiles,
        "timeout_tiles": timeout_tiles,
        # Pre-submit, uncharged drops: blank/nodata skips (credits saved) and
        # render/provider holes (possible coverage gap). Additive; older keys
        # unchanged.
        "blank_tiles": blank_tiles,
        "render_failed_tiles": render_failed_tiles,
    })


def track_auto_zero_result(run_id: str, tiles: int, object_class: str,
                           had_exemplar: bool) -> None:
    track(ev.AUTO_ZERO_RESULT, {
        "run_id": run_id,
        "tiles": tiles,
        "object_class": object_class,
        "had_exemplar": bool(had_exemplar),
    })


def track_zero_assist_clicked(kind: str, from_prompt: str,
                              to_prompt: str = "") -> None:
    track(ev.ZERO_ASSIST_CLICKED, {
        "kind": kind,
        "from_prompt": from_prompt,
        "to_prompt": to_prompt,
    })


# Review / refine -----------------------------------------------------------


def track_review_opened(run_id: str, instances_found: int, visible_at_start: int,
                        start_confidence: int, auto_lowered: bool) -> None:
    track(ev.REVIEW_OPENED, {
        "run_id": run_id,
        "instances_found": instances_found,
        "visible_at_start": visible_at_start,
        "start_confidence": start_confidence,
        "auto_lowered": bool(auto_lowered),
    })


def track_review_confidence_final(run_id: str, final_pct: int, visible_count: int,
                                  moves: int) -> None:
    track(ev.REVIEW_CONFIDENCE_FINAL, {
        "run_id": run_id,
        "final_pct": final_pct,
        "visible_count": visible_count,
        "moves": moves,
    })


def track_review_display_mode(mode: str) -> None:
    track(ev.REVIEW_DISPLAY_MODE, {"mode": mode})


def track_review_shape_adjusted(control: str, value) -> None:
    track(ev.REVIEW_SHAPE_ADJUSTED, {"control": control, "value": value})


def track_refine_in_manual_entered(run_id: str, instances: int) -> None:
    track(ev.REFINE_IN_MANUAL_ENTERED, {"run_id": run_id, "instances": instances})


def track_refine_in_manual_back(run_id: str, validated_count: int,
                                duration_ms: int | None = None) -> None:
    track(ev.REFINE_IN_MANUAL_BACK, {
        "run_id": run_id,
        "validated_count": validated_count,
        "duration_ms": duration_ms,
    })


def track_auto_export_done(run_id: str, exported_count: int, visible_pct_of_found: int,
                           final_confidence: int, display_mode: str,
                           refined_in_manual: bool, autosave: bool = False) -> None:
    """autosave marks the passive leave-safety export (mode switch, new run,
    unload), as opposed to an explicit Finish / Save && exit."""
    track(ev.AUTO_EXPORT_DONE, {
        "run_id": run_id,
        "exported_count": exported_count,
        "visible_pct_of_found": visible_pct_of_found,
        "final_confidence": final_confidence,
        "display_mode": display_mode,
        "refined_in_manual": bool(refined_in_manual),
        "autosave": bool(autosave),
    })


def track_review_abandoned(run_id: str, instances_at_exit: int, refined: bool,
                           confidence_changed: bool, exit_path: str) -> None:
    """The user left the Automatic review without clicking Finish. exit_path is
    one of exit_button | new_run | mode_switch | zone_redraw | raster_removed |
    unload | other. No PII by construction."""
    track(ev.REVIEW_ABANDONED, {
        "run_id": run_id,
        "instances_at_exit": int(instances_at_exit),
        "refined": bool(refined),
        "confidence_changed": bool(confidence_changed),
        "exit_path": exit_path,
    })


def track_auto_retry_clicked(run_id: str, discarded_count: int, confirmed: bool) -> None:
    track(ev.AUTO_RETRY_CLICKED, {
        "run_id": run_id,
        "discarded_count": discarded_count,
        "confirmed": bool(confirmed),
    })


def track_auto_exit_clicked(from_step: int, autosaved_count: int) -> None:
    track(ev.AUTO_EXIT_CLICKED, {
        "from_step": from_step,
        "autosaved_count": autosaved_count,
    })


def track_review_correct_box(run_id: str, label: int, outcome: str,
                             objects: int, gesture: str = "box") -> None:
    """One completed correction gesture.

    gesture is box (the drawn add/remove rectangle) | merge | split, so the two
    free hand edits report through the same event instead of adding one.
    outcome is revealed | removed | queued | empty for a box, merged | split for
    a hand edit; objects counts the objects the gesture touched. label keeps the
    box convention (1 include, 0 exclude) and a hand edit reports 1, since it
    reshapes rather than drops.
    """
    track(ev.REVIEW_CORRECT_BOX, {
        "run_id": run_id,
        "label": int(label),
        "outcome": outcome,
        "objects": int(objects),
        "gesture": gesture,
    })


def track_review_correct_undo(run_id: str, kind: str) -> None:
    """kind is the undone journal entry's kind (merge | split), or
    clear_all for the whole-round clear."""
    track(ev.REVIEW_CORRECT_UNDO, {"run_id": run_id, "kind": kind})


def track_review_step(run_id: str, step: int) -> None:
    """Step navigation within the linear review; callers dedupe per run+step."""
    track(ev.REVIEW_STEP, {"run_id": run_id, "step": int(step)})


def track_qgis_edit_bridge(run_id: str, outcome: str,
                           duration_ms: int | None = None,
                           features: int | None = None) -> None:
    """QGIS digitizing bridge lifecycle from the review's Correct step.

    outcome is opened (armed native editing) | committed | rolled_back.
    duration_ms and features are absent on 'opened' and carried on the two
    terminal outcomes (how long the bridge stayed open, and how many features
    folded back).
    """
    props: dict = {"run_id": run_id, "outcome": outcome}
    if duration_ms is not None:
        props["duration_ms"] = int(duration_ms)
    if features is not None:
        props["features"] = int(features)
    track(ev.AUTO_EDIT_IN_QGIS, props)


# Manual --------------------------------------------------------------------


_segmentation_run_sent_this_session = False


def track_segmentation_run(success: bool, duration_ms: int | None = None) -> None:
    """Fire when a manual segmentation run completes (or fails).

    Success runs are sampled 1-in-10 (power users click hundreds of times per
    session); the first run per session is always sent (sample_rate 1). Failures
    are ALWAYS sent unsampled (sample_rate 1) so the failure rate is real."""
    global _segmentation_run_sent_this_session
    import random

    if not success:
        # Failures are never sampled: the failure signal must be complete.
        track(ev.SEGMENTATION_RUN, {
            "success": False, "duration_ms": duration_ms, "sample_rate": 1,
        })
        return

    if _segmentation_run_sent_this_session:
        if random.random() >= 0.1:  # nosec B311 - sampling, not crypto
            return
        sample_rate = 10
    else:
        _segmentation_run_sent_this_session = True
        sample_rate = 1
    track(ev.SEGMENTATION_RUN, {
        "success": True, "duration_ms": duration_ms, "sample_rate": sample_rate,
    })


def track_manual_export_done(
    polygon_count: int, refine_used: bool, destination: str = "new"
) -> None:
    # destination: "new" (a fresh layer) | "append" (added to an existing
    # layer). Additive optional property; older servers ignore it.
    track(ev.MANUAL_EXPORT_DONE, {
        "polygon_count": polygon_count,
        "refine_used": bool(refine_used),
        "destination": destination,
    })


def track_manual_session_summary(saves: int, undos: int,
                                 duration_ms: int | None = None) -> None:
    track(ev.MANUAL_SESSION_SUMMARY, {
        "saves": saves,
        "undos": undos,
        "duration_ms": duration_ms,
    })


def track_manual_abandoned(context: str, polygon_count: int) -> None:
    """Fire when the user CONFIRMS discarding unsaved manual work.
    context: "change_layer" | "stop"."""
    track(ev.MANUAL_ABANDONED, {
        "context": context,
        "polygon_count": polygon_count,
    })


_FIRST_SUCCESS_KEY = "AI_Segmentation/first_success_sent"


def track_first_generation_milestone(mode: str) -> None:
    """One-shot per machine: the user's first successful export ever (their
    first real value moment, in either mode). mode: "auto" | "manual"."""
    try:
        settings = QSettings()
        if bool(settings.value(_FIRST_SUCCESS_KEY, False, type=bool)):
            return
        settings.setValue(_FIRST_SUCCESS_KEY, True)
    except Exception:  # nosec B110 - never break the export path
        return
    track(ev.FIRST_GENERATION_MILESTONE, {"mode": mode})


# Monetization --------------------------------------------------------------


_upsell_viewed_triggers: set[str] = set()
_low_credit_banner_viewed_this_session = False


def track_pro_upsell_viewed(trigger: str = "free_exhausted") -> None:
    """Fire at most once per session PER TRIGGER when an upsell first renders.

    Deduplicated by trigger, not by process. A single flag for every trigger
    made the surfaces compete for one slot: whichever fired first in a QGIS
    session silenced the others for its whole lifetime, so the view count came
    out below the click count, which is impossible and made every view-to-click
    ratio unusable.
    """
    if trigger in _upsell_viewed_triggers:
        return
    _upsell_viewed_triggers.add(trigger)
    track(ev.PRO_UPSELL_VIEWED, {"trigger": trigger})


def track_pro_upsell_clicked(source: str = "upsell_card") -> None:
    """source: upsell_card / subscribe_pill / low_credit_banner / exhausted_status."""
    track(ev.PRO_UPSELL_CLICKED, {"source": source})


def track_free_taste_consumed(remaining: int, run_id: str = "") -> None:
    """remaining = free detections left after this one. run_id joins the taste
    to the run that consumed it (best effort: "" when unknown)."""
    track(ev.FREE_TASTE_CONSUMED, {"remaining": remaining, "run_id": run_id})


def track_low_credit_banner_viewed(remaining: int, total: int) -> None:
    """Fire at most once per session when the low-credit banner first shows."""
    global _low_credit_banner_viewed_this_session
    if _low_credit_banner_viewed_this_session:
        return
    _low_credit_banner_viewed_this_session = True
    track(ev.LOW_CREDIT_BANNER_VIEWED, {"remaining": remaining, "total": total})


def track_detect_blocked(reason: str) -> None:
    """reason: credits / zone_too_large / cost_over_balance / worker_busy /
    no_layer / raster_shape / not_activated / kill_switch / no_auth /
    prompt_<guard>."""
    track(ev.DETECT_BLOCKED, {"reason": reason})


# Errors --------------------------------------------------------------------


def track_plugin_error(
    stage: str,
    error_code: str,
    message: str,
    include_log_tail: bool = False,
    traceback_hash: str | None = None,
    module: str | None = None,
) -> None:
    """Fire when an error is shown to the user or an exception is caught.

    stage: install | download | activate | segment | export | other
    error_code: short machine-friendly id (e.g. "PIP_TIMEOUT", "RUNTIME_ERROR")
    message: first line of the error, truncated to 500 chars, path + coord scrubbed
    include_log_tail: OFF by default. When True, the last 20 anonymized log lines
        are capped to ~4KB and coordinate-scrubbed before being sent.
    traceback_hash: optional short sha of the normalized traceback (groups
        recurrences of the same crash). Additive; omitted when unknown.
    module: optional source module the exception was caught in. Additive.
    """
    props = {
        "stage": stage,
        "error_code": error_code,
        "message": _scrub_payload_value((message or "")[:500]),
    }
    if traceback_hash:
        props["traceback_hash"] = traceback_hash
    if module:
        props["module"] = module
    if include_log_tail:
        try:
            from .log_scrub import get_recent_logs
            tail_lines = get_recent_logs().splitlines()[-20:]
            scrubbed = _scrub_payload_value("\n".join(tail_lines))
            props["last_log_lines"] = scrubbed.encode("utf-8")[:4096].decode(
                "utf-8", errors="ignore"
            )
        except Exception:
            pass  # nosec B110
    track(ev.PLUGIN_ERROR, props)


# --- Error capture (top-level slots + worker bodies) ----------------------
#
# The only error -> telemetry path used to be an explicit show_error_report()
# call, so an uncaught exception in a Qt slot or worker run() body produced NO
# telemetry at all. These helpers give a standard capture pattern: track a
# plugin_error with a stable English error_code, a traceback_hash (so the
# analytics backend groups recurrences of the same crash), and the source
# module; log a line; and
# show a dialog ONLY when an explicit user_message is passed. No sys.excepthook
# (that conflicts with QGIS and other plugins) - wrap the entry points instead.


def _short_traceback_hash(exc: BaseException) -> str:
    """A short, path-free fingerprint of an exception's traceback.

    Each frame contributes basename:lineno:function; the exception class name
    is appended. Filenames are reduced to their basename so the hash is stable
    across machines. Returns "" if anything goes wrong (never raises)."""
    import hashlib
    import os as _os
    import traceback as _tb
    try:
        parts = [
            f"{_os.path.basename(fr.filename)}:{fr.lineno}:{fr.name}"
            for fr in _tb.extract_tb(exc.__traceback__)
        ]
        parts.append(exc.__class__.__name__)
        return hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()[:16]
    except Exception:
        return ""


def report_exception(
    exc: BaseException,
    stage: str,
    module: str = "",
    user_message: str | None = None,
    parent=None,
) -> None:
    """Capture an unhandled exception: track it, log it, optionally show it.

    Always tracks a plugin_error (error_code = exception class name, plus a
    traceback_hash and module) and writes one QgsMessageLog line. Shows the
    error-report dialog ONLY when user_message is given AND we are on the main
    thread. Never raises: this runs on failure paths where a second failure
    must stay invisible."""
    error_code = ""
    tb_hash = ""
    first_line = ""
    try:
        error_code = exc.__class__.__name__
        tb_hash = _short_traceback_hash(exc)
        text = str(exc)
        first_line = text.splitlines()[0] if text else ""
    except Exception:  # nosec B110
        pass
    try:
        track_plugin_error(
            stage=stage,
            error_code=error_code or "Exception",
            message=first_line,
            traceback_hash=tb_hash,
            module=module or None,
        )
    except Exception:  # nosec B110
        pass
    try:
        from qgis.core import Qgis, QgsMessageLog
        QgsMessageLog.logMessage(
            "Unhandled {code} in {mod} ({stage}) [{h}]".format(
                code=error_code or "Exception", mod=module or "?",
                stage=stage, h=tb_hash or "-"),
            "AI Segmentation", level=Qgis.MessageLevel.Warning,
        )
    except Exception:  # nosec B110
        pass
    if user_message and _on_main_thread():
        try:
            from ..ui.error_report_dialog import ErrorReportDialog
            dialog = ErrorReportDialog(user_message, user_message, parent)
            dialog.exec()
        except Exception:  # nosec B110
            pass


def slot_guard(stage: str, user_message: str | None = None):
    """Decorator for a top-level Qt slot: catch any unhandled exception, report
    it (telemetry + log, and a dialog only when user_message is given), and
    swallow it so a stray crash never leaves QGIS's console handler as the only
    trace. Do NOT stack on slots that already surface their own errors."""
    def deco(fn):
        module = (fn.__module__ or "").rsplit(".", 1)[-1]

        # Qt passes signal payloads (e.g. clicked's `checked` bool) into the
        # slot. Some PyQt builds introspect this wrapper's (*args) as "takes
        # anything" and forward them, so calling fn with the extra args raised
        # TypeError inside the guard and the slot silently did NOTHING (launch
        # bug: a dead Cancel button). Trim to fn's real positional arity.
        try:
            import inspect
            params = list(inspect.signature(fn).parameters.values())[1:]  # drop self
            _has_var = any(p.kind is inspect.Parameter.VAR_POSITIONAL for p in params)
            _max_pos = None if _has_var else sum(
                1 for p in params
                if p.kind in (inspect.Parameter.POSITIONAL_ONLY,
                              inspect.Parameter.POSITIONAL_OR_KEYWORD)
            )
        except Exception:  # noqa: BLE001 - introspection is best-effort
            _max_pos = None

        @functools.wraps(fn)
        def wrapper(self, *args, **kwargs):
            if _max_pos is not None and len(args) > _max_pos:
                args = args[:_max_pos]
            try:
                return fn(self, *args, **kwargs)
            except Exception as exc:  # noqa: BLE001 - top-level slot boundary
                parent = None
                if user_message is not None:
                    try:
                        parent = self.iface.mainWindow()
                    except Exception:  # nosec B110
                        parent = None
                report_exception(
                    exc, stage=stage, module=module,
                    user_message=user_message, parent=parent,
                )
                return None
        return wrapper
    return deco


# NOTE: worker run() bodies report inline (see AutoDetectionWorker.run): they
# pair the report with worker-specific cleanup (power inhibit, error signal),
# which a generic context manager cannot know about. slot_guard above is the
# one shared boundary; an unused error_boundary context manager was removed.
