"""First-time setup, install/download workers, activation, pairing, settings.

Part of AISegmentationPlugin (see ai_segmentation_plugin.py); split out
so agents and humans can work on one concern per file. Methods here are
plain mixin members: state lives on the plugin instance (self).
"""
from __future__ import annotations

import sys

from qgis.core import (
    Qgis,
    QgsMessageLog,
)

from ...core.i18n import tr
from ..background_workers import DepsInstallWorker, DownloadWorker, VerifyWorker
from ..error_report_dialog import show_error_report
from .shared import (
    SETTINGS_KEY_LAST_MANUAL_SESSION_TS,
    _get_change_path_instructions,
)


def _notify_ui(callback, *args) -> None:
    """Call a dialog callback, tolerating a dialog that is already gone.

    The removal worker outlives the window that started it (QGIS can be closing
    while the delete finishes), so a dead Qt receiver must not turn into an
    unhandled exception inside a signal handler."""
    if callback is None:
        return
    try:
        callback(*args)
    except RuntimeError:
        pass  # nosec B110 - the receiving widget was destroyed


def _drop_untagged_account_history() -> None:
    """Forget every history blob that is not tied to one account.

    Called whenever the signed-in account changes (sign-in, sign-out, forced
    sign-out). The per-account stores are already invisible to the other
    accounts; what has to go is the run cache and the machine-wide Recent list
    older versions wrote with no account attached to them. Best-effort: a
    settings failure must never block a sign-in or a sign-out.
    """
    try:
        from ...core.presets.run_history_cache import (
            clear_run_history_cache,
            reset_account_fingerprint_cache,
        )
        from ...core.presets.segment_history import clear_unscoped_recent_objects

        # First, so every store below resolves against the account that is
        # signed in now rather than the one that just left.
        reset_account_fingerprint_cache()
        clear_run_history_cache()
        clear_unscoped_recent_objects()
    except Exception:
        pass  # nosec B110 - the cache is a warm start, never the source of truth


class EnvSetupMixin:
    """First-time setup, install/download workers, activation, pairing, settings."""

    def toggle_dock_widget(self):
        just_created = not self._dock_created
        self._ensure_dock_widget()
        if self.dock_widget:
            if just_created or not self.dock_widget.isVisible():
                self.dock_widget.show()
                self.dock_widget.raise_()
                QgsMessageLog.logMessage(
                    f"Dock shown (first_create={just_created})",
                    "AI Segmentation", level=Qgis.MessageLevel.Info)
            else:
                self.dock_widget.hide()
                QgsMessageLog.logMessage(
                    "Dock hidden", "AI Segmentation", level=Qgis.MessageLevel.Info)

    def _do_first_time_setup(self):
        QgsMessageLog.logMessage(
            "Panel opened - checking dependencies...",
            "AI Segmentation",
            level=Qgis.MessageLevel.Info
        )

        # This runs only when Manual is on screen: the dock opened on it, or the
        # user switched to it (_ensure_interactive_setup). Both callers latch
        # their own guard BEFORE scheduling this, so it cannot fire twice in a
        # session, and the warm-up below starts no install and asks nothing.
        self._warm_local_ai_for_manual()

        # All environment checks (legacy cleanup, venv status, checkpoint
        # presence) run off the UI thread: get_venv_status walks the venv and
        # hashes dependency specs, which froze the dock for 1-2 s on open.
        if self._startup_check_worker is not None and self._startup_check_worker.isRunning():
            return
        from ..background_workers import StartupCheckWorker
        self._startup_check_worker = StartupCheckWorker()
        self._startup_check_worker.done.connect(self._on_startup_check_finished)
        self._startup_check_worker.start()

        # Check for plugin updates with retries - QGIS repo metadata may not
        # be ready after just a few seconds, especially on slower connections.
        from qgis.PyQt.QtCore import QTimer
        self._update_check_delays = [5000, 30000, 60000, 120000]
        self._update_check_index = 0
        QTimer.singleShot(
            self._update_check_delays[0], self._check_for_plugin_update)

    def _on_startup_check_finished(self, venv_ready: bool, message: str, checkpoint_ok: bool):
        # Cache the venv-installed state for the Refine-in-Manual env gate.
        self._env_ready = bool(venv_ready)
        if not self.dock_widget:
            return

        if message.startswith("startup_error:"):
            detail = message[len("startup_error:"):].strip()
            QgsMessageLog.logMessage(
                f"Dependency check error: {detail}",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            # A fixed sentence, not the first 50 characters of a Python
            # exception cut mid-word. The full detail is in the message log
            # two lines above, which is what a bug report carries.
            self.dock_widget.set_dependency_status(
                False, tr("Could not check the AI components. See the log for details."))
            # Do NOT leave the env-check latched done on an error: reset the guard
            # so a later Manual/Refine re-entry re-attempts the check instead of
            # stranding the error status until a manual Install click.
            self._interactive_setup_done = False
            return

        if venv_ready:
            self.dock_widget.set_dependency_status(True, "✓ " + tr("Dependencies ready"))
            QgsMessageLog.logMessage(
                "✓ Virtual environment verified successfully",
                "AI Segmentation",
                level=Qgis.MessageLevel.Success
            )
            self._start_device_info_worker()
            if checkpoint_ok:
                self.dock_widget.set_checkpoint_status(True, "SAM model ready")
                self._load_predictor()
                self._refresh_activation_async()
            else:
                # Model missing but deps ok.
                self.dock_widget.set_dependency_status(
                    True, tr("Dependencies ready, model not downloaded"))
                # A pending Refine handoff / background install is waiting on the
                # model: download it now so the deferred import can complete,
                # instead of stranding the user behind a manual Download button.
                if self._pending_refine_import or getattr(self, "_refine_install_pending", False):
                    self._auto_download_checkpoint()
                    return
                # Show the Download-Model button ONLY in Manual mode: without the
                # mode guard this install UI leaked onto the Automatic page when a
                # StartupCheckWorker finished after a switch back to Automatic.
                from ..ai_segmentation_dockwidget import Mode
                if self.dock_widget._mode == Mode.INTERACTIVE:
                    self.dock_widget.install_button.setVisible(True)
                    self.dock_widget.install_button.setEnabled(True)
                    self.dock_widget.install_button.setText(tr("Download AI model"))
                    self.dock_widget.setup_group.setVisible(True)
                    # Model download is pending: keep the setup section across a
                    # mode round trip (see dock _setup_section_wanted).
                    self.dock_widget._setup_section_wanted = True
        else:
            self.dock_widget.set_dependency_status(False, message)
            QgsMessageLog.logMessage(
                f"Virtual environment status: {message}",
                "AI Segmentation",
                level=Qgis.MessageLevel.Info
            )
            # Auto-trigger install for upgrades
            if "need updating" in message:
                self._on_install_requested()

    def _check_for_plugin_update(self):
        """Trigger the update check on the dock widget, retrying if needed."""
        if not self.dock_widget:
            return
        self.dock_widget.check_for_updates()

        # If notification is still hidden and we have retries left, schedule next
        if not self.dock_widget.update_notification_widget.isVisible() and hasattr(self, "_update_check_delays"):
            self._update_check_index += 1
            if self._update_check_index < len(self._update_check_delays):
                from qgis.PyQt.QtCore import QTimer
                delay = self._update_check_delays[self._update_check_index]
                QTimer.singleShot(delay, self._check_for_plugin_update)

    def _load_predictor(self):
        """Kick off predictor initialization in the background (#34)."""
        # Re-entry while a load is in flight must not drop the running thread's
        # last reference (GC of a live QThread hard-aborts QGIS).
        worker = getattr(self, "_predictor_worker", None)
        if worker is not None and worker.isRunning():
            return
        from ..background_workers import PredictorLoadWorker
        if self.dock_widget:
            self.dock_widget.set_checkpoint_status(True, tr("Loading AI model..."))
        self._predictor_worker = PredictorLoadWorker()
        self._predictor_worker.done.connect(self._on_predictor_loaded)
        self._predictor_worker.start()

    def _on_predictor_loaded(self, predictor, err_msg: str):
        if predictor is None:
            QgsMessageLog.logMessage(
                f"Failed to initialize predictor: {err_msg}",
                "AI Segmentation",
                level=Qgis.MessageLevel.Warning
            )
            if self.dock_widget:
                self.dock_widget.set_checkpoint_status(
                    False, tr("Model load failed"))
            # A background install started from the Automatic review can no longer
            # complete: drop the pending refine/add and re-enable the entry.
            self._abort_refine_install()
            self._abort_ai_add_install()
            return
        self.predictor = predictor
        QgsMessageLog.logMessage(
            "SAM predictor initialized (subprocess mode)",
            "AI Segmentation",
            level=Qgis.MessageLevel.Info
        )
        if self.dock_widget:
            self.dock_widget.set_checkpoint_status(True, tr("SAM model ready"))
            self.dock_widget.set_install_progress(100, tr("Ready"))
        # Warm the model NOW when Manual use is predictable, so the first
        # session's first click never waits out the model load: right after a
        # fresh install/model download (the user is onboarding, about to try),
        # when Manual mode is on screen at all (_warm_local_ai_for_manual: it
        # is the only mode that runs on this model), or when this machine ran a
        # Manual session recently (regular Manual users reach their
        # resident-model steady state at startup instead of at their first
        # click; the model stays loaded until QGIS closes anyway once a session
        # runs). Automatic-only users trip none of the three, so they never pay
        # the model's RAM. warm_up() is idempotent
        # and cheap on this thread (Popen + one init line; the load happens
        # inside the subprocess). Best-effort: the Start-time warm_up retries.
        try:
            if not self._headless:
                should_warm = getattr(self, "_warm_predictor_on_ready", False)
                should_warm = should_warm or self._manual_used_recently()
                if should_warm:
                    self._warm_predictor_on_ready = False
                    predictor.warm_up()
                    QgsMessageLog.logMessage(
                        "Pre-warming the SAM worker (Manual use predicted)",
                        "AI Segmentation", level=Qgis.MessageLevel.Info
                    )
        except Exception:  # noqa: BLE001 - prediction of intent must never break load
            pass  # nosec B110
        # D1: a background install kicked off from the Automatic review (the user
        # clicked Refine with no local AI) has finished. Clear the inline banner
        # and, since the user already asked to refine, open the handoff now on the
        # still-intact review. If they left the review meanwhile, just release the
        # flag (the AI is now ready for next time).
        if getattr(self, "_refine_install_pending", False):
            self._refine_install_pending = False
            if self.dock_widget:
                try:
                    self.dock_widget.set_auto_review_installing(False)
                except (RuntimeError, AttributeError):
                    pass
            if self._auto_review:
                self._on_reshape_ai_requested()
            return
        # D1 twin for the Add lane: a background install kicked off from the Add
        # button has finished. Clear the banner and arm the lane now on the
        # still-intact review (or just release the flag if the user left it).
        if getattr(self, "_ai_add_install_pending", False):
            self._ai_add_install_pending = False
            if self.dock_widget:
                try:
                    self.dock_widget.set_auto_review_installing(False)
                except (RuntimeError, AttributeError):
                    pass
            if self._auto_review:
                self._on_ai_add_requested()
            return
        # Complete a Refine-in-Manual handoff that arrived before the model was
        # loaded: now that the predictor is up, start the Manual session and
        # import the held detections as editable polygons.
        if self._pending_refine_import and self._refine_handoff_active:
            self._pending_refine_import = False
            layer = getattr(self, "_handoff_source_layer", None)
            review = self._auto_review
            if layer is not None and review:
                if self.dock_widget:
                    try:
                        self.dock_widget.set_refine_handoff_preparing(False)
                    except (RuntimeError, AttributeError):
                        pass
                    try:
                        combo = self.dock_widget.layer_combo
                        combo.blockSignals(True)
                        combo.setLayer(layer)
                        combo.blockSignals(False)
                    except (RuntimeError, AttributeError):
                        pass
                self._on_start_segmentation(layer)
                # Inherit the review's CURRENT refine settings, exactly like the
                # direct (predictor-already-loaded) reshape path does, so a run
                # whose model was still loading is not stranded on the generic
                # Manual defaults.
                self._seed_refine_from_review()
                self._import_review_geoms_as_saved(review)
                # Now the detections are loaded, open the one the user selected
                # for reshape (deferred until the predictor was ready).
                self._open_reshape_target()

    def _manual_used_recently(self, days: int = 14) -> bool:
        """True when this machine ran a Manual session within ``days``.

        The timestamp is written by _on_start_segmentation. Drives the
        predictive model warm-up in _on_predictor_loaded: a machine that
        segmented manually this fortnight will almost surely do it again,
        while an Automatic-only machine never gets the key at all."""
        try:
            import time

            from qgis.PyQt.QtCore import QSettings
            ts = QSettings().value(
                SETTINGS_KEY_LAST_MANUAL_SESSION_TS, 0, type=int)
            return ts > 0 and (time.time() - ts) < days * 86400
        except Exception:  # noqa: BLE001 - heuristic only
            return False

    def _start_device_info_worker(self):
        """Detect and log the compute device off the UI thread (diagnostics)."""
        if self._device_info_worker is not None and self._device_info_worker.isRunning():
            return
        from ..background_workers import DeviceInfoWorker
        self._device_info_worker = DeviceInfoWorker()
        self._device_info_worker.done.connect(self._on_device_info)
        self._device_info_worker.start()

    def _on_device_info(self, ok: bool, info: str):
        if ok:
            QgsMessageLog.logMessage(
                f"Device info: {info}",
                "AI Segmentation",
                level=Qgis.MessageLevel.Info
            )
            return

        # Check if this is a PyTorch DLL error (Windows)
        if "DLL" in info or "shm.dll" in info:
            QgsMessageLog.logMessage(
                f"PyTorch DLL error: {info}",
                "AI Segmentation",
                level=Qgis.MessageLevel.Critical
            )
            if self._headless:
                return
            # Show user-friendly error dialog (same converged helper every
            # other error path here uses: copy-logs + email, not a bare OK box)
            show_error_report(
                self.iface.mainWindow(),
                tr("PyTorch cannot load on Windows"),
                tr("The plugin requires Visual C++ Redistributables to run PyTorch.\n\n"
                   "Please download and install:\n"
                   "https://aka.ms/vs/17/release/vc_redist.x64.exe\n\n"
                   "After installation, restart QGIS and try again."),
                error_code="pytorch_dll_error",
            )
        else:
            QgsMessageLog.logMessage(
                f"Could not determine device info: {info}",
                "AI Segmentation",
                level=Qgis.MessageLevel.Warning
            )

    def _on_install_requested(self):
        # Manual (local) install needs to download/execute a standalone
        # Python and build a multi-GB venv, both unreliable or blocked under
        # Flatpak/Snap confinement. Fail fast with a clear message instead of
        # letting the install grind through a cryptic low-level failure;
        # Automatic mode needs no local install and works in the sandbox.
        from ...core.python_manager import is_sandboxed_linux
        if is_sandboxed_linux():
            QgsMessageLog.logMessage(
                "Manual install blocked: running inside a Flatpak/Snap sandbox",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            self.dock_widget.set_dependency_status(
                False, tr("Manual mode is not supported in this QGIS installation"))
            show_error_report(
                self.iface.mainWindow(),
                tr("Manual Mode Not Supported"),
                tr(
                    "Manual mode needs to install local dependencies, which is "
                    "not supported inside this sandboxed QGIS installation "
                    "(Flatpak or Snap). Please use Automatic mode instead, "
                    "which runs fully in the cloud and needs no local install."
                ),
                error_code="sandboxed_linux_manual_blocked",
            )
            return

        # Intel Macs on QGIS 4 (Python 3.13+) have no installable local runtime
        # (the Intel-Mac fallback has no build for this Python). Fail fast with a
        # clear message instead of a cryptic dependency-resolver error; Automatic
        # mode runs in the cloud and needs no local install.
        from ...core.model_config import MACOS_X86_NO_LOCAL_INFERENCE
        if MACOS_X86_NO_LOCAL_INFERENCE:
            QgsMessageLog.logMessage(
                "Manual install blocked: no local inference build for this Mac + Python",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            self.dock_widget.set_dependency_status(
                False, tr("Manual mode is not supported in this QGIS installation"))
            show_error_report(
                self.iface.mainWindow(),
                tr("Manual Mode Not Supported"),
                tr(
                    "Manual mode installs local components that are not available "
                    "for this Mac with this version of QGIS. Please use Automatic "
                    "mode instead, which runs fully in the cloud and needs no "
                    "local install."
                ),
                error_code="macos_x86_no_local_inference",
            )
            return

        # Guard: prevent concurrent installs
        if self.deps_install_worker is not None and self.deps_install_worker.isRunning():
            QgsMessageLog.logMessage(
                "Install already in progress, ignoring duplicate request",
                "AI Segmentation",
                level=Qgis.MessageLevel.Warning
            )
            return

        # Guard: never install into a tree that is being deleted. Same-process
        # concurrency has to be caught HERE, because the cross-process lock the
        # removal holds records this very process and would be broken as a dead
        # leftover by an install started from this window.
        remover = getattr(self, "_remove_data_worker", None)
        if remover is not None:
            try:
                running = remover.isRunning()
            except RuntimeError:
                running = False
            if running:
                QgsMessageLog.logMessage(
                    "Install refused: the downloaded AI data is being removed",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
                self.dock_widget.set_dependency_status(
                    False, tr("Removing the downloaded AI data..."))
                return

        from ...core.venv_manager import get_venv_status

        # This runs on the UI thread, on a button click: no subprocess probe,
        # which could freeze the window for a minute. An environment whose
        # packages are all on disk counts as ready even when the import probe
        # has not run yet, so a healthy install is not re-downloaded from
        # scratch; the background startup check does the real verification.
        is_ready, message = get_venv_status(allow_subprocess_probe=False)
        if is_ready:
            # Deps already installed, just need model download
            self.dock_widget.set_dependency_status(True, "✓ " + tr("Dependencies ready"))
            self._auto_download_checkpoint()
            return

        QgsMessageLog.logMessage(
            "Starting virtual environment creation and dependency installation...",
            "AI Segmentation",
            level=Qgis.MessageLevel.Info
        )
        QgsMessageLog.logMessage(
            f"Platform: {sys.platform}, Python: {sys.version}",
            "AI Segmentation",
            level=Qgis.MessageLevel.Info
        )

        self.dock_widget.set_install_progress(0, "Preparing installation...")

        import time as _time
        self._install_t0 = _time.monotonic()
        try:
            from ...core import telemetry
            telemetry.track_install_started()
        except Exception:
            pass  # nosec B110

        self.deps_install_worker = DepsInstallWorker()
        self.deps_install_worker.progress.connect(self._on_deps_install_progress)
        self.deps_install_worker.done.connect(self._on_deps_install_finished)
        self.deps_install_worker.start()

        # Surface the sign-in section 2 seconds after install starts
        from qgis.PyQt.QtCore import QTimer
        QTimer.singleShot(2000, self._refresh_activation_async)

    def _refresh_activation_async(self, force: bool = False):
        """Refresh the signed-in state without ever blocking the UI thread.

        Re-validates the stored key in a hidden QgsTask; the UI keeps its
        current state meanwhile (benefit of the doubt). Two safeguards for flaky
        networks: a connectivity failure never signs the user out (see
        _on_key_revalidate_failed), and the check is throttled to once per 15 min
        so a restored-open dock does not re-poll on every visibility event.
        """
        import time

        from ...core.activation_manager import is_plugin_activated
        if not self.dock_widget:
            return
        if not is_plugin_activated():
            if not self.dock_widget.is_activated():
                self.dock_widget._update_full_ui()
            return
        # Activated on startup (stored key): the dock restores straight to its
        # last tab, so if that tab is Automatic, _on_mode_changed never fires and
        # the upsell would stay visible for a Pro account until the user toggles
        # modes. Fetch usage now so the Pro/Free state is correct immediately.
        self._refresh_auto_credits()
        # Throttle: one revalidation per 15 min. Sign-in / sign-out reset the
        # stamp (to 0.0) so a fresh state revalidates immediately.
        if not force and (time.time() - self._last_key_validation_unix) < 900:
            return
        if self._key_revalidate_task is not None and self._key_revalidate_task.is_active():
            return
        from qgis.core import QgsApplication

        # Auth header is read here (main thread); the task only talks HTTP.
        from ...api.terralab_client import TerraLabClient
        from ...core.activation_manager import get_auth_header
        from ...workers.generic_request_task import GenericRequestTask
        auth = get_auth_header()
        client = TerraLabClient()
        self._key_revalidate_task = GenericRequestTask(
            tr("Checking your AI Segmentation subscription"),
            lambda: client.get_usage(auth=auth),
            hidden=True,
        )
        self._key_revalidate_task.succeeded.connect(self._on_key_revalidate_ok)
        self._key_revalidate_task.failed.connect(self._on_key_revalidate_failed)
        QgsApplication.taskManager().addTask(self._key_revalidate_task)

    def _on_key_revalidate_ok(self, _usage: object) -> None:
        import time
        self._key_revalidate_task = None
        self._last_key_validation_unix = time.time()

    def _on_key_revalidate_failed(self, message: str, code: str) -> None:
        self._key_revalidate_task = None
        from ...core.activation_manager import is_rejection_code
        # Only a genuine auth rejection signs the user out. A network blip must
        # NOT clear the stored key (that was the old bug that logged users out
        # offline); we keep the session and surface a non-blocking notice.
        if is_rejection_code(code):
            from ...core.activation_manager import clear_auth
            clear_auth()
            _drop_untagged_account_history()
            self._last_key_validation_unix = 0.0
            if self.dock_widget:
                self.dock_widget.set_activated_state(False)
            # A forced sign-out is a churn-grade moment (device limit hit,
            # subscription lapsed, key revoked): make it visible post-hoc.
            try:
                from ...core import telemetry
                telemetry.track_plugin_error(
                    stage="activate",
                    error_code=(code or "key_rejected").lower(),
                    message="stored key rejected on revalidation",
                )
            except Exception:
                pass  # nosec B110
            return
        self._notify_connection_issue(code, message)

    # --- One-click connect (browser pairing handoff) ------------------------

    def _on_pairing_requested(self, code: str):
        """Open the browser to /connect and start polling for the key.

        Re-entrant: if a poll is already running (the user clicked "open the
        page again"), we only re-open the browser instead of starting a second
        worker.
        """
        from qgis.PyQt.QtCore import QUrl
        from qgis.PyQt.QtGui import QDesktopServices

        from ...api.terralab_client import TerraLabClient

        client = TerraLabClient()
        # Build the connect URL from the client base so .env.local
        # TERRALAB_BASE_URL is honored in dev. Never log the code or full URL.
        url = (
            f"{client.base_url}/connect?code={code}&product=ai-segmentation"
            "&utm_source=qgis&utm_medium=plugin&utm_campaign=ai-segmentation"
            "&utm_content=connect"
        )
        opened = QDesktopServices.openUrl(QUrl(url))
        if not opened:
            # Browser pairing is the ONLY way in, so this branch used to end
            # the product: clicking again calls the same failing openUrl. On
            # Linux it goes through xdg-open, which returns False when that is
            # absent, when no browser is registered, or when a sandbox has no
            # portal, and none of those is a connection problem. So hand the
            # user the address instead of a diagnosis. It carries a short-lived
            # pairing code and lands on their own clipboard, which is no wider
            # than the browser we were trying to open it in.
            self.dock_widget.show_pairing_idle()
            copied = False
            try:
                from qgis.PyQt.QtWidgets import QApplication

                clipboard = QApplication.clipboard()
                if clipboard is not None:
                    clipboard.setText(url)
                    copied = True
            except (RuntimeError, AttributeError, ImportError):
                copied = False
            try:
                # Selectable, so the address can still be read and copied by
                # hand when the clipboard itself is unavailable.
                from qgis.PyQt.QtCore import Qt

                label = self.dock_widget.activation_message_label
                label.setWordWrap(True)
                label.setTextInteractionFlags(
                    Qt.TextInteractionFlag.TextSelectableByMouse)
            except (RuntimeError, AttributeError, ImportError):
                pass
            if copied:
                message = tr(
                    "QGIS could not open a browser. The sign-in address is "
                    "copied to your clipboard: paste it into a browser to "
                    "finish, then come back here.")
            else:
                message = tr(
                    "QGIS could not open a browser. Open this address to "
                    "finish signing in, then come back here:\n{}").format(url)
            self.dock_widget.set_activation_message(message, is_error=True)
            return

        if self._pairing_worker is not None and self._pairing_worker.is_active():
            # Already polling for this code; the browser was just re-opened.
            return

        from qgis.core import QgsApplication

        from ...workers.pairing_poll_task import PairingPollTask
        self._pairing_worker = PairingPollTask(client, code)
        self._pairing_worker.pairing_succeeded.connect(self._on_pairing_succeeded)
        self._pairing_worker.pairing_failed.connect(self._on_pairing_failed)
        self._pairing_worker.pairing_timeout.connect(self._on_pairing_timeout)
        # The task has hinted at a stalled sign-in since it was written, and
        # nothing listened: the user watched the waiting animation for the full
        # ten minutes and was then told it timed out.
        self._pairing_worker.pairing_stalled.connect(self._on_pairing_stalled)
        QgsApplication.taskManager().addTask(self._pairing_worker)
        import time as _time
        self._pairing_t0 = _time.monotonic()
        try:
            from ...core import telemetry
            telemetry.track_pairing_started()
        except Exception:
            pass  # nosec B110
        QgsMessageLog.logMessage(
            "Pairing started", "AI Segmentation", level=Qgis.MessageLevel.Info)

    def _pairing_elapsed_ms(self) -> int | None:
        """Elapsed ms since the current browser sign-in poll started."""
        try:
            import time as _time
            t0 = getattr(self, "_pairing_t0", None)
            return int((_time.monotonic() - t0) * 1000) if t0 else None
        except Exception:
            return None

    def _on_pairing_succeeded(self, key: str):
        from ...core.activation_manager import save_auth_token
        save_auth_token(key)
        # Whoever was signed in before must leave nothing behind in the library.
        _drop_untagged_account_history()
        # Fresh sign-in: clear the revalidation throttle so the next dock event
        # re-checks the new key immediately instead of waiting out the window.
        self._last_key_validation_unix = 0.0
        if self.dock_widget:
            self.dock_widget.set_activated_state(True)
            # Reflect the real plan (Pro vs free) in the Automatic panel right
            # away, so the upsell card is hidden immediately for a Pro account
            # instead of waiting for the next mode toggle.
            self._refresh_auto_credits()
        # Bring QGIS back to front so the user sees the activated dock.
        try:
            mw = self.iface.mainWindow()
            mw.activateWindow()
            mw.raise_()
            if self.dock_widget:
                self.dock_widget.raise_()
        except Exception:  # nosec B110
            pass
        try:
            from ...core.telemetry import track_plugin_activated
            track_plugin_activated(duration_ms=self._pairing_elapsed_ms())
        except Exception:  # nosec B110
            pass
        QgsMessageLog.logMessage(
            "Pairing successful", "AI Segmentation", level=Qgis.MessageLevel.Info)

    def _on_pairing_failed(self, message: str, code: str):
        if self.dock_widget:
            self.dock_widget.show_pairing_idle()
            self.dock_widget.set_activation_message(message, is_error=True)
        try:
            from ...core import telemetry
            telemetry.track_pairing_failed(
                error_code=code or "unknown",
                duration_ms=self._pairing_elapsed_ms(),
            )
        except Exception:
            pass  # nosec B110
        QgsMessageLog.logMessage(
            f"Pairing failed ({code})", "AI Segmentation",
            level=Qgis.MessageLevel.Warning)

    def _on_pairing_stalled(self):
        """The server has not seen the browser after a long wait.

        A hint, not a failure: the poll keeps running and can still succeed, so
        this only replaces the silent waiting animation with the two things the
        user can act on. Never switches the panel back to idle, which would
        take away the Cancel the poll is still honouring."""
        if not self.dock_widget:
            return
        self.dock_widget.set_activation_message(
            tr("Still waiting for the sign-in page. If no browser opened, or "
               "the page shows an error, click Cancel and try again."),
            is_error=False,
        )

    def _on_pairing_timeout(self):
        if self.dock_widget:
            self.dock_widget.show_pairing_idle()
            self.dock_widget.set_activation_message(
                tr("Sign-in timed out. Click Connect to try again."),
                is_error=True,
            )
        try:
            from ...core import telemetry
            telemetry.track_pairing_failed(
                error_code="timeout",
                duration_ms=self._pairing_elapsed_ms(),
            )
        except Exception:
            pass  # nosec B110
        QgsMessageLog.logMessage(
            "Pairing timed out", "AI Segmentation", level=Qgis.MessageLevel.Info)

    def _cancel_pairing_worker(self):
        """Cancel any in-flight pairing poll (never terminate())."""
        if self._pairing_worker is not None and self._pairing_worker.is_active():
            try:
                self._pairing_worker.cancel()
            except RuntimeError:
                pass

    def _cancel_task(self, attr: str):
        """Cooperatively cancel a GenericRequestTask held on ``self.<attr>``.

        Disconnect its signals so a late result can't fire into torn-down UI,
        then cancel it. We NEVER terminate a network-bound thread: the task
        manager drains run() on its own, and terminate() on a wedged socket
        crashes QGIS.
        """
        task = getattr(self, attr, None)
        if task is None:
            return
        try:
            task.succeeded.disconnect()
            task.failed.disconnect()
        except (RuntimeError, TypeError):  # nosec B110
            pass
        try:
            if task.is_active():
                task.cancel()
        except Exception:  # nosec B110
            pass
        setattr(self, attr, None)

    # Network error codes (from terralab_client._classify_network_error) that
    # mean "no/unstable connection" rather than an auth or app error.
    _CONNECTIVITY_CODES = frozenset({
        "DNS_ERROR", "CONNECTION_REFUSED", "TIMEOUT",
        "SSL_ERROR", "PROXY_ERROR", "NO_INTERNET",
    })
    _CONN_NOTICE_MIN_GAP_S = 60.0

    def _notify_connection_issue(self, code: str, message: str):
        """Show ONE transient, dismissible 'no connection' notice in the message
        bar. Non-blocking (never freezes), throttled so a flaky network does not
        spam the user, and silent for non-connectivity codes (auth/app errors
        surface through their own paths)."""
        import time

        if (code or "").strip().upper() not in self._CONNECTIVITY_CODES:
            return
        now = time.monotonic()
        if now - self._last_conn_notice_monotonic < self._CONN_NOTICE_MIN_GAP_S:
            return
        self._last_conn_notice_monotonic = now
        try:
            self.iface.messageBar().pushMessage(
                "AI Segmentation",
                message or tr("Network error. Check your internet connection."),
                level=Qgis.MessageLevel.Warning,
                duration=8,
            )
        except (RuntimeError, AttributeError):  # nosec B110
            pass

    def _on_cancel_pairing(self, code: str = ""):
        self._cancel_pairing_worker()
        try:
            from ...core import telemetry
            telemetry.track_pairing_cancelled(duration_ms=self._pairing_elapsed_ms())
        except Exception:
            pass  # nosec B110
        if code:
            # Retire the code server-side so a later Confirm in the browser
            # shows "expired" instead of binding a key nobody is polling for.
            from qgis.core import QgsApplication, QgsTask

            from ...api.terralab_client import TerraLabClient
            client = TerraLabClient()
            self._pairing_cancel_task = QgsTask.fromFunction(
                tr("Cancelling sign-in"),
                lambda task, c=code: client.cancel_pairing(c),
            )
            QgsApplication.taskManager().addTask(self._pairing_cancel_task)
        QgsMessageLog.logMessage(
            "Pairing cancelled", "AI Segmentation", level=Qgis.MessageLevel.Info)

    # --- Settings / sign out -------------------------------------------------

    def _on_settings_clicked(self):
        from ...core.activation_manager import get_auth_header, get_auth_token, is_plugin_activated
        if not is_plugin_activated():
            return
        from ...api.terralab_client import TerraLabClient
        from ..account_settings_dialog import AccountSettingsDialog
        client = TerraLabClient()
        dlg = AccountSettingsDialog(
            client=client,
            auth=get_auth_header(),
            activation_key=get_auth_token(),
            parent=self.iface.mainWindow(),
            on_remove_ai_data=self.remove_local_ai_data,
            is_busy_check=self.is_local_ai_busy,
        )
        dlg.sign_out_requested.connect(self._on_sign_out_requested)
        dlg.exec()

    def is_local_ai_busy(self) -> bool:
        """True while any local install/download/verify/model-load worker or a
        detection run is in flight, so removing the downloaded AI data would
        corrupt an operation. A loaded-but-idle predictor is NOT busy: it is
        shut down cleanly by remove_local_ai_data before any deletion. Fail
        safe: if the state cannot be read, report busy so nothing is deleted
        out from under a live operation."""
        try:
            for attr in (
                "deps_install_worker", "download_worker", "_verify_worker",
                "_predictor_worker", "_startup_check_worker", "_auto_worker",
                # The device probe imports torch from the venv; deleting the
                # venv under it locks files on Windows and faults the probe.
                "_device_info_worker",
                # The async crop encode talks to the SAM subprocess; deleting
                # the venv under it kills the process mid-round-trip.
                "_manual_encode_worker",
                # A removal already under way: the tree is disappearing, so
                # nothing may start against it (and no second removal).
                "_remove_data_worker",
            ):
                worker = getattr(self, attr, None)
                if worker is None:
                    continue
                try:
                    if worker.isRunning():
                        return True
                except (RuntimeError, AttributeError):
                    continue
            return False
        except Exception:  # nosec B110
            return True

    def remove_local_ai_data(self, on_progress=None, on_finished=None) -> tuple[bool, str]:
        """Start deleting every trace of the local AI install this plugin
        created: the isolated venv + downloaded model weights (all under
        PLUGIN_CACHE_DIR), the stored activation key (QgsAuthManager + the QSettings
        fallback) and this plugin's own QSettings groups. Signs the user out as
        a side effect.

        Returns (started, message), and the message always says where that
        leaves the user. started=False before the account state is cleared
        means nothing at all was touched: an install or a detection run is in
        flight, or another QGIS window holds the install lock. started=False
        after it (a worker that could not be created or started) means the key
        and settings are gone but the files are not, and the message says so.
        On True the settings and key are already cleared and the multi-GB
        delete runs on a worker thread, so
        the window stays responsive; `on_progress(text)` reports the current
        step and `on_finished(ok, message)` closes it out, both on the GUI
        thread. Never raises: any partial-failure detail is folded into the
        message passed to on_finished."""
        worker = getattr(self, "_remove_data_worker", None)
        if worker is not None:
            try:
                if worker.isRunning():
                    return False, tr("The removal is already running.")
            except RuntimeError:
                pass  # C++ side gone; the attribute is stale
        if self.is_local_ai_busy():
            return False, tr(
                "An install or detection is still running. Wait for it to "
                "finish, then try again.")

        # is_local_ai_busy only sees this process. The cache is shared across
        # QGIS windows and profiles, so the cross-process install lock is what
        # stops this delete from pulling the venv out from under another
        # window's running install. It is held until the worker reports back.
        from ...core.install_lock import InstallLock
        from ...core.venv_manager import INSTALL_LOCK_FILE

        lock = InstallLock(INSTALL_LOCK_FILE)
        if not lock.acquire():
            return False, tr(
                "Another QGIS window is installing the AI engine. Wait for it "
                "to finish, then try again.")

        errors = self._clear_local_ai_account_state()

        predictor = getattr(self, "predictor", None)
        self.predictor = None
        try:
            from ..background_workers import RemoveAiDataWorker
            from .shared import park_orphaned_worker

            worker = RemoveAiDataWorker(predictor)
            self._remove_data_worker = worker
            if on_progress is not None:
                worker.progress.connect(
                    lambda text, cb=on_progress: _notify_ui(cb, text))
            worker.done.connect(
                lambda nothing_left, freed, error, lk=lock, errs=errors, cb=on_finished:
                self._on_remove_ai_data_done(nothing_left, freed, error, lk, errs, cb))
            # Anchor the thread for its whole life: the dialog that started it
            # can be dismissed and the plugin unloaded before the delete is
            # over, and a QThread destroyed while running aborts all of QGIS.
            park_orphaned_worker(worker)
            worker.start()
            # start() does not raise when the OS refuses the thread: it warns
            # and returns, run() never executes and `done` never fires. Treat
            # that as a failed start rather than waiting forever for it.
            if not (worker.isRunning() or worker.isFinished()):
                raise RuntimeError("the removal thread did not start")
        except Exception as err:  # noqa: BLE001 - the lock must never leak
            # Nothing is running, so nothing will ever release the lock or
            # report back. Give everything back here: the lock is shared across
            # QGIS windows, and leaking it refuses every later install and
            # removal until QGIS is restarted.
            QgsMessageLog.logMessage(
                f"Could not start the AI data removal: {err}",
                "AI Segmentation", level=Qgis.MessageLevel.Critical)
            self._remove_data_worker = None
            self.predictor = predictor
            try:
                lock.release()
            except Exception:  # nosec B110 - the release is best-effort
                pass
            return False, tr(
                "The removal could not start. You are signed out, but the "
                "downloaded AI data is still on this computer. Try again.")
        return True, tr("Removing the downloaded AI data...")

    def _clear_local_ai_account_state(self) -> list:
        """Clear the stored key and this plugin's settings, and reflect the
        sign-out in the dock. GUI-thread work only (auth + QSettings + widgets),
        all of it instant, and it runs BEFORE the delete so the account is
        already clean if the file removal turns out to be partial. Returns the
        list of failure details, for the final message."""
        self._env_ready = False
        self._first_time_setup_done = False

        # Clear the activation key (removes the QgsAuthManager config via the
        # authcfg_id, which still lives in QSettings at this point) BEFORE
        # wiping the QSettings groups, else the stored config would leak.
        errors: list[str] = []
        try:
            from ...core.activation_manager import clear_auth
            clear_auth()
        except Exception as err:  # nosec B110
            errors.append(str(err)[:80])
        _drop_untagged_account_history()
        self._last_key_validation_unix = 0.0

        # Wipe this plugin's own QSettings groups. Leave the shared TerraLab/
        # keys (telemetry opt-out, device seed) untouched: they are shared with
        # the sibling plugin and are not "downloaded AI data".
        try:
            from qgis.PyQt.QtCore import QSettings
            settings = QSettings()
            for group in ("AISegmentation", "AI_Segmentation",
                          "TerraLab/AI_Segmentation"):
                settings.remove(group)
            settings.sync()
        except Exception as err:  # nosec B110
            errors.append(str(err)[:80])

        if self.dock_widget:
            try:
                self.dock_widget.set_activated_state(False)
            except (RuntimeError, AttributeError):
                pass
        return errors

    def _on_remove_ai_data_done(self, nothing_left, freed, error,
                                lock, errors, on_finished) -> None:
        """The delete worker finished: release the cross-process lock, drop the
        emptied cache directory and report. Runs on the GUI thread."""
        self._remove_data_worker = None
        errors = list(errors or [])
        if error:
            errors.append(str(error)[:80])
        elif not nothing_left:
            # Some files survived (locked / permissions): report it rather
            # than claim a clean removal.
            errors.append("leftover files")

        lock.release()
        # The lock file is the one thing kept alive inside the cache directory
        # while it is being emptied (see purge_cache_dir), so drop the now-empty
        # directory once release() has removed it.
        import os

        from ...core.cache_paths import PLUGIN_CACHE_DIR
        try:
            os.rmdir(PLUGIN_CACHE_DIR)
        except OSError:
            pass  # nosec B110 - not empty (reported) or already gone

        QgsMessageLog.logMessage(
            "Local AI data removed ({}), errors={}".format(
                freed or "0 MB", len(errors)),
            "AI Segmentation", level=Qgis.MessageLevel.Info)

        if errors:
            message = tr(
                "AI data removed, but some items could not be fully cleared. "
                "You can delete the folder manually.")
        else:
            message = tr("Downloaded AI data removed. You have been signed out.")
        _notify_ui(on_finished, True, message)

    def _on_sign_out_requested(self):
        from ...core.activation_manager import clear_auth
        clear_auth()
        _drop_untagged_account_history()
        self._last_key_validation_unix = 0.0
        if self.dock_widget:
            self.dock_widget.set_activated_state(False)

    def _on_deps_install_progress(self, percent: int, message: str):
        if not self.dock_widget:
            return
        # Scale deps progress to 0-80% (model download gets 80-100%)
        scaled = int(percent * 0.8)
        self.dock_widget.set_install_progress(scaled, message)

    def _on_deps_install_finished(self, success: bool, message: str):
        if not self.dock_widget:
            return

        if success:
            # install_completed fires in _on_verify_finished: a venv that fails
            # verification (broken torch DLL) is a FAILED install to the user,
            # so counting it completed here skewed the install funnel.
            # Run verification + device detection off main thread
            self.dock_widget.set_install_progress(80, tr("Verifying installation..."))
            # A verify pass can run for minutes, and a repair install can
            # finish while one is still going. Reassigning the attribute would
            # drop the last reference to a live QThread, which aborts QGIS.
            if self._verify_worker is not None and self._verify_worker.isRunning():
                return
            self._verify_worker = VerifyWorker()
            self._verify_worker.progress.connect(self._on_verify_progress)
            self._verify_worker.done.connect(self._on_verify_finished)
            self._verify_worker.start()
        else:
            # A user cancel reaches this handler through the SAME channel as a
            # real failure (the installer returns (False, "Installation
            # cancelled") from every one of its cancel checks). Left to fall
            # through, it showed an "Installation Failed" dialog with a bug
            # report form to someone who deliberately pressed Cancel, and
            # emitted install_failed on top of the install_cancelled that
            # _on_cancel_install already sent, so the two events double-counted
            # each other and inflated the measured install failure rate.
            # Cancels wind down quietly instead; the dock's own progress
            # handler already renders a "cancel" message as its neutral
            # cancelled state rather than the failed one.
            # The worker's own flag is the authoritative discriminator: WE set
            # it in _on_cancel_install, so it cannot be defeated by wrapping or
            # translation. The string check stays as a fallback for a cancel
            # that did not come through that button, and matches anywhere in
            # the message rather than only at the start: the uv path returns
            # the bare sentinel, but the pip fallback path (uv unavailable, so
            # exactly the antivirus and proxy-restricted population) wraps it
            # as "Failed to install <package>: Installation cancelled", which a
            # prefix test misses.
            _cancelled = bool(
                getattr(self.deps_install_worker, "_cancelled", False))
            _msg_lower_early = (message or "").lower()
            if (_cancelled or "installation cancelled" in _msg_lower_early or "download cancelled" in _msg_lower_early):
                self.dock_widget.set_install_progress(100, "Cancelled")
                self.dock_widget.set_dependency_status(
                    False, tr("Installation cancelled"))
                self._abort_refine_install()
                return

            self.dock_widget.set_install_progress(100, "Failed")
            error_msg = message[:300] if message else tr("Unknown error")
            self.dock_widget.set_dependency_status(False, tr("Installation failed"))
            # A background install from the Automatic review can no longer finish.
            self._abort_refine_install()

            error_title = tr("Installation Failed")
            error_code = "installation_failed"
            msg_lower = message.lower() if message else ""
            # The classifiers below live in pip_diagnostics; most were already
            # written and battle-tested but only a handful were wired in here.
            # Routing the failure through ALL of them turns the opaque
            # "installation_failed" bucket into a specific error_class (better
            # telemetry) AND gives the user the correct fix instead of a generic
            # one. Order matters: most specific first, and disk/app-control are
            # checked before the antivirus/permission branches (a full disk and
            # a managed policy both surface as access errors on Windows).
            from ...core.cache_paths import PLUGIN_CACHE_DIR
            from ...core.pip_diagnostics import (
                get_app_control_help,
                get_broken_python_runtime_help,
                get_corrupt_venv_help,
                get_crash_help,
                get_dependency_conflict_help,
                get_file_locked_help,
                get_glibc_too_old_help,
                get_invalid_path_help,
                get_macos_intel_help,
                get_pip_antivirus_help,
                get_vcpp_help,
                is_antivirus_error,
                is_app_control_error,
                is_broken_python_runtime,
                is_corrupt_venv,
                is_dependency_conflict,
                is_disk_full,
                is_dll_init_error,
                is_file_locked_error,
                is_glibc_too_old,
                is_invalid_path_error,
                is_macos_intel_no_wheel,
                is_proxy_auth_error,
                is_rename_or_record_error,
                is_unable_to_create_process,
            )
            from ...core.venv_manager import mark_venv_for_rebuild
            if "another qgis window is installing" in msg_lower:
                # Cross-process lock contention: nothing failed, another live
                # install owns the build. An info box is the right weight; the
                # bug-report dialog made people file reports for waiting. No
                # install_failed here either: counting a wait as a failure is
                # the same metric-inflation trap the user-cancel branch avoids.
                self.dock_widget.set_dependency_status(
                    False, tr("Installation running in another window"))
                from qgis.PyQt.QtWidgets import QMessageBox
                QMessageBox.information(
                    self.iface.mainWindow(),
                    tr("Installation Already Running"),
                    tr(
                        "Another QGIS window is installing the AI components. "
                        "Wait for it to finish, then try again."
                    ),
                )
                return
            if "not enough free disk space" in msg_lower or is_disk_full(msg_lower):
                # Preflight or mid-install disk exhaustion: the message already
                # carries the free-up / AI_SEGMENTATION_CACHE_DIR guidance.
                error_title = tr("Not Enough Disk Space")
                error_code = "disk_space"
            elif is_broken_python_runtime(msg_lower):
                # The local runtime cannot boot (gutted stdlib). Marking the
                # venv for rebuild + the standalone re-verification at install
                # start make the next attempt self-heal instead of looping on
                # the same wall.
                error_title = tr("AI Environment Damaged")
                error_msg = get_broken_python_runtime_help(PLUGIN_CACHE_DIR)
                error_code = "broken_python_runtime"
                mark_venv_for_rebuild()
            elif is_corrupt_venv(msg_lower):
                # The venv exists but its interpreter or pip is gone; a plain
                # retry reused it forever. The marker forces a from-scratch
                # rebuild on the next install run (worker thread, not UI).
                error_title = tr("AI Environment Damaged")
                error_msg = get_corrupt_venv_help()
                error_code = "corrupt_venv"
                mark_venv_for_rebuild()
            elif is_invalid_path_error(msg_lower):
                # Windows refused the path itself (cloud-synced placeholder,
                # odd characters, 260-char limit), not a permission problem.
                error_title = tr("Installation Path Problem")
                error_msg = get_invalid_path_help(PLUGIN_CACHE_DIR)
                error_code = "invalid_install_path"
            elif is_dependency_conflict(msg_lower):
                error_title = tr("Package Versions Conflict")
                error_msg = get_dependency_conflict_help()
                error_code = "dependency_conflict"
            elif any(p in msg_lower for p in [
                "ssl", "certificate verify", "sslerror",
                "unable to get local issuer",
                # uv (rustls) wording, no "ssl" substring anywhere
                "invalid peer certificate", "unknownissuer",
            ]):
                error_title = tr("SSL Certificate Error")
                error_code = "ssl_certificate_error"
            elif "file in use by qgis" in msg_lower or is_file_locked_error(msg_lower):
                # Native module (.pyd/.dll) locked by the running QGIS process
                # during a torch upgrade - no dialog-splash of AV advice here,
                # the actionable fix is just "restart QGIS".
                error_title = tr("Restart QGIS Required")
                error_msg = get_file_locked_help()
                error_code = "restart_qgis_required"
            elif is_app_control_error(msg_lower):
                # AppLocker/WDAC style managed policy: checked BEFORE the
                # generic "blocked" branch below, whose change-path advice is
                # a dead end (the policy blocks any user-writable location).
                error_title = tr("Blocked by IT Security Policy")
                error_msg = get_app_control_help(PLUGIN_CACHE_DIR)
                error_code = "app_control_policy"
            elif is_dll_init_error(msg_lower):
                # A DLL init failure (missing VC++ runtime). is_dll_init_error
                # self-excludes policy/AV blocks, so this is genuinely a missing
                # redistributable, not a quarantine.
                error_title = tr("Missing System Component")
                error_msg = get_vcpp_help()
                error_code = "dll_init_error"
            elif any(p in msg_lower for p in [
                "access is denied", "winerror 5", "winerror 225",
                "permission denied", "blocked",
                "cannot write to install",
                "cannot open the device or file",
            ]):
                error_title = tr("Installation Blocked")
                error_msg = f"{error_msg}\n\n{_get_change_path_instructions()}"
                error_code = "installation_blocked"
            elif is_antivirus_error(msg_lower):
                # Localized "access denied" and antivirus-quarantine wordings the
                # English "blocked" branch above misses (uv surfaces the raw OS
                # message in the system language). Antivirus/exclusion guidance.
                error_title = tr("Blocked by Antivirus or Security Software")
                error_msg = get_pip_antivirus_help(PLUGIN_CACHE_DIR)
                error_code = "antivirus_blocked"
            elif is_proxy_auth_error(msg_lower):
                # HTTP 407 behind a corporate proxy: distinct from a generic
                # network failure - the fix is credentials in QGIS proxy settings.
                error_title = tr("Proxy Authentication Required")
                error_msg = "{}\n\n{}".format(
                    error_msg,
                    tr(
                        "Your network proxy requires a username and password. "
                        "Enter them in QGIS > Settings > Options > Network, "
                        "then restart QGIS and try again."
                    ),
                )
                error_code = "proxy_auth_required"
            elif any(p in msg_lower for p in [
                "network error", "connection aborted", "connection reset",
                "timed out", "timeout", "network connection failed",
                "connection broken", "could not resolve",
            ]):
                error_title = tr("Network Connection Problem")
                error_msg = "{}\n\n{}".format(
                    error_msg,
                    tr(
                        "Your connection appears unstable or blocked. "
                        "Check: (1) your internet is working, "
                        "(2) QGIS > Settings > Options > Network has a proxy "
                        "configured if you are on a corporate network, "
                        "(3) your firewall allows connections to pypi.org "
                        "and files.pythonhosted.org."
                    ),
                )
                error_code = "network_connection_problem"
            elif is_unable_to_create_process(msg_lower):
                # Broken Python launcher shim (Windows): rebuilding the venv fixes it.
                error_title = tr("Installation Failed")
                error_msg = "{}\n\n{}".format(
                    error_msg,
                    tr(
                        "The installer could not start a helper process (a "
                        "damaged Python launcher). Click Reinstall Dependencies "
                        "to rebuild the environment from scratch."
                    ),
                )
                error_code = "broken_pip_shim"
            elif is_rename_or_record_error(msg_lower):
                # dist-info rename/RECORD error mid torch upgrade (Windows): a
                # stale, partially-written package. A clean rebuild resolves it.
                error_title = tr("Restart QGIS Required")
                error_msg = get_crash_help(PLUGIN_CACHE_DIR)
                error_code = "dist_info_rename"
            elif is_glibc_too_old(msg_lower):
                # Linux distro older than the current PyTorch wheels support.
                error_title = tr("Linux System Too Old")
                error_msg = get_glibc_too_old_help()
                error_code = "glibc_too_old"
            elif is_macos_intel_no_wheel(msg_lower):
                # Intel Mac + a Python newer than the last Intel torch wheel.
                error_title = tr("Unsupported Mac and Python Combination")
                error_msg = get_macos_intel_help()
                error_code = "macos_intel_no_wheel"

            show_error_report(
                self.iface.mainWindow(),
                error_title,
                error_msg,
                error_code=error_code,
            )
            try:
                import time as _time

                from ...core import telemetry
                t0 = getattr(self, "_install_t0", 0.0)
                telemetry.track_install_failed(
                    error_class=error_code,
                    duration_ms=int((_time.monotonic() - t0) * 1000) if t0 else None,
                    python_minor=sys.version_info.minor,
                )
            except Exception:
                pass  # nosec B110

    def _on_verify_progress(self, percent: int, message: str):
        if not self.dock_widget:
            return
        # Scale verify progress (0-100%) into the 80-95% range
        scaled = 80 + int(percent * 0.15)
        self.dock_widget.set_install_progress(scaled, message)

    def _on_verify_finished(self, is_valid: bool, message: str):
        if not self.dock_widget:
            return
        if is_valid:
            # The install only counts as completed once the venv VERIFIES:
            # firing at deps-install success counted broken-torch installs
            # (DLL blocked) as completed and understated install_failed.
            try:
                import time as _time

                from ...core import telemetry
                t0 = getattr(self, "_install_t0", 0.0)
                if t0:
                    telemetry.track_install_completed(
                        duration_ms=int((_time.monotonic() - t0) * 1000),
                        python_minor=sys.version_info.minor,
                    )
                    self._install_t0 = 0.0
            except Exception:
                pass  # nosec B110
            self.dock_widget.set_dependency_status(True, "✓ " + tr("Dependencies ready"))
            if message and not message.startswith("device_error"):
                QgsMessageLog.logMessage(
                    f"Device info: {message}",
                    "AI Segmentation", level=Qgis.MessageLevel.Info)
            elif message.startswith("device_error"):
                QgsMessageLog.logMessage(
                    "Could not determine device info: {}".format(
                        message.replace("device_error: ", "")),
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
            try:
                self._auto_download_checkpoint()
            except Exception as e:
                QgsMessageLog.logMessage(
                    f"Auto-download checkpoint failed: {e}",
                    "AI Segmentation", level=Qgis.MessageLevel.Warning)
                self.dock_widget.set_install_progress(100, "Failed")
                self.dock_widget.set_dependency_status(
                    True, tr("Dependencies ready, model download failed"))
                self._abort_refine_install()
        else:
            self.dock_widget.set_install_progress(100, "Failed")
            self.dock_widget.set_dependency_status(
                False, "{} {}".format(tr("Verification failed:"), message))
            self._abort_refine_install()
            # A broken native module (torch DLL blocked by a missing VC++
            # runtime) is the dominant verify failure: route it to the
            # actionable fix-it steps instead of a dead-end generic dialog.
            from ...core.pip_diagnostics import (
                get_app_control_help,
                get_vcpp_help,
                is_app_control_error,
                is_dll_init_error,
            )
            error_title = tr("Verification Failed")
            error_code = "verification_failed"
            body = "{}\n{}".format(
                tr("Virtual environment was created but verification failed:"),
                message)
            low = (message or "").lower()
            if is_app_control_error(low):
                # A managed policy (AppLocker/WDAC) blocking a DLL also prints
                # "DLL load failed": checked first so the user gets the IT
                # allow-rule guidance, not the VC++ runtime steps.
                from ...core.cache_paths import PLUGIN_CACHE_DIR
                error_title = tr("Blocked by IT Security Policy")
                error_code = "app_control_policy"
                body = f"{message}\n\n{get_app_control_help(PLUGIN_CACHE_DIR)}"
            elif is_dll_init_error(low) or "required dll failed to initialize" in low:
                error_title = tr("A Component Failed to Load")
                error_code = "dll_init_failed"
                body = f"{message}\n\n{get_vcpp_help()}"
            show_error_report(
                self.iface.mainWindow(),
                error_title,
                body,
                error_code=error_code)
            try:
                import time as _time

                from ...core import telemetry
                t0 = getattr(self, "_install_t0", 0.0)
                telemetry.track_install_failed(
                    error_class=error_code,
                    duration_ms=int((_time.monotonic() - t0) * 1000) if t0 else None,
                    python_minor=sys.version_info.minor,
                )
            except Exception:
                pass  # nosec B110

    def _on_cancel_install(self):
        # The model download is the second half of the same install, so Cancel
        # must stop it too: without this the worker kept retrying for up to an
        # hour after the user asked it to stop.
        if self.download_worker is not None and self.download_worker.isRunning():
            self._download_cancelled = True
            try:
                from ...core.checkpoint_manager import request_download_cancel
                request_download_cancel()
            except Exception:
                pass  # nosec B110
            QgsMessageLog.logMessage(
                "Model download cancelled by user",
                "AI Segmentation",
                level=Qgis.MessageLevel.Warning
            )
        if self.deps_install_worker and self.deps_install_worker.isRunning():
            self.deps_install_worker.cancel()
            try:
                import time as _time

                from ...core import telemetry
                t0 = getattr(self, "_install_t0", 0.0)
                telemetry.track_install_cancelled(
                    duration_ms=int((_time.monotonic() - t0) * 1000) if t0 else None)
            except Exception:
                pass  # nosec B110
            QgsMessageLog.logMessage(
                "Installation cancelled by user",
                "AI Segmentation",
                level=Qgis.MessageLevel.Warning
            )

    def _auto_download_checkpoint(self):
        """Auto-download model after deps install if not already present."""
        # Guard: reachable from four paths (install finish, verify finish,
        # dependency-ready shortcut, retry); a second start while one is
        # already running would corrupt the checkpoint download.
        if self.download_worker is not None and self.download_worker.isRunning():
            return

        # This path only runs at the tail of a fresh install/repair: the user
        # is onboarding and will click Start within moments, so ask
        # _on_predictor_loaded to warm the model as soon as it is ready.
        self._warm_predictor_on_ready = True
        from ...core.checkpoint_manager import checkpoint_exists
        try:
            if checkpoint_exists():
                self.dock_widget.set_install_progress(95, tr("Loading AI model..."))
                self._load_predictor()
                self._refresh_activation_async()
                return
        except Exception:
            pass  # nosec B110

        self.dock_widget.set_install_progress(80, tr("Downloading AI model..."))
        self._download_cancelled = False
        try:
            import time as _time
            self._model_download_t0 = _time.monotonic()
            self.download_worker = DownloadWorker()
            self.download_worker.progress.connect(self._on_download_progress)
            self.download_worker.done.connect(self._on_download_finished)
            self.download_worker.start()
        except Exception as e:
            QgsMessageLog.logMessage(
                f"Failed to start model download: {e}",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
            self.dock_widget.set_install_progress(100, "Failed")
            self.dock_widget.set_dependency_status(
                True, tr("Dependencies ready, model download failed"))
            self._abort_refine_install()

    def _on_download_progress(self, percent: int, message: str):
        if not self.dock_widget:
            return
        # Scale model download to 80-100% of the unified progress
        scaled = 80 + int(percent * 0.2)
        self.dock_widget.set_install_progress(scaled, message)

    def _on_download_finished(self, success: bool, message: str):
        if not self.dock_widget:
            return
        if success:
            try:
                import time as _time

                from ...core import telemetry
                from ...core.model_config import USE_SAM2
                t0 = getattr(self, "_model_download_t0", 0.0)
                telemetry.track_model_download_completed(
                    model="sam2" if USE_SAM2 else "sam1",
                    duration_ms=int((_time.monotonic() - t0) * 1000) if t0 else None)
            except Exception:
                pass  # nosec B110
            self.dock_widget.set_install_progress(95, tr("Loading AI model..."))
            self._load_predictor()
            self._refresh_activation_async()
        else:
            # A cancel comes back through the same channel as a real failure.
            # Wind it down quietly instead of accusing the user of a failed
            # download they deliberately stopped.
            if getattr(self, "_download_cancelled", False):
                self._download_cancelled = False
                self.dock_widget.set_install_progress(100, tr("Cancelled"))
                self.dock_widget.set_dependency_status(
                    True, tr("Dependencies ready, model not downloaded"))
                self._abort_refine_install()
                return

            self.dock_widget.set_install_progress(100, "Failed")
            self.dock_widget.set_dependency_status(
                True, tr("Dependencies ready, model download failed"))
            self._abort_refine_install()

            show_error_report(
                self.iface.mainWindow(),
                tr("Download Failed"),
                "{}\n{}".format(
                    tr("Failed to download model:"),
                    message),
                error_code="download_failed",
            )

    def _recover_corrupt_checkpoint(self, deleted: bool) -> bool:
        """Recover from a corrupt model checkpoint detected during encoding.

        ``deleted`` reports whether the bad file was removed. When it was, we
        kick off a fresh download (which re-verifies the hash and reloads the
        predictor on success). When it could not be removed, we ask the user
        to clear the cache folder manually. Always returns False so the caller
        aborts the current encode. (#65)
        """
        from ...core.checkpoint_manager import get_checkpoints_dir

        if not deleted:
            msg = tr(
                "The AI model file is corrupted but could not be removed "
                "automatically. Please delete this folder and restart QGIS:"
            ) + "\n" + get_checkpoints_dir()
            if self._headless:
                self._headless_error = msg
                return False
            show_error_report(
                self.iface.mainWindow(),
                tr("Model File Corrupted"),
                msg,
                error_code="checkpoint_corrupt",
            )
            return False

        msg = tr(
            "The AI model file was corrupted and is being re-downloaded. "
            "Please try your selection again once it finishes."
        )
        if self._headless:
            self._headless_error = msg
            return False

        from qgis.PyQt.QtWidgets import QMessageBox
        QMessageBox.information(
            self.iface.mainWindow(), tr("Re-downloading Model"), msg)
        try:
            self._auto_download_checkpoint()
        except Exception as e:
            QgsMessageLog.logMessage(
                f"Failed to re-download checkpoint after corruption: {e}",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False

    def _recover_broken_venv(self, error: str) -> bool:
        """Recover from a venv that can no longer run.

        Happens when the base Python the venv points at was deleted by a
        cleanup tool or its recorded path got corrupted: site-packages look
        complete, the panel says ready, but every worker spawn fails.
        Reinstalling recreates the runtime in place; the package cache makes
        it a short repair, not a full re-download. Always returns False so
        the caller aborts the current encode. (#64)
        """
        QgsMessageLog.logMessage(
            f"Broken Python runtime detected, starting repair: {error[:200]}",
            "AI Segmentation", level=Qgis.MessageLevel.Warning)

        msg = tr(
            "The Python runtime used by the AI engine is damaged "
            "(this can be caused by a disk cleanup tool or antivirus). "
            "It will now be repaired automatically. "
            "Please try your selection again once the repair finishes."
        )
        if self._headless:
            self._headless_error = msg
            return False

        from qgis.PyQt.QtWidgets import QMessageBox
        QMessageBox.information(
            self.iface.mainWindow(), tr("Repairing Installation"), msg)
        self.dock_widget.set_dependency_status(
            False, tr("Repairing installation..."))
        try:
            self._on_install_requested()
        except Exception as e:
            QgsMessageLog.logMessage(
                f"Failed to start repair install: {e}",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
        return False
