"""Module-level constants and free helper functions shared by the
AISegmentationPlugin mixins (extracted from ai_segmentation_plugin.py so
mixin modules can import them without a circular import)."""
from __future__ import annotations

import os
import sys
import time
import weakref

from qgis.core import QgsFeatureSink

from ...core.i18n import tr
from ...core.interaction_dials import (
    auto_pump_budget_s,
    live_frame_cost_ratio,
    live_repaint_max_ms,
    live_repaint_ms,
)
from ...core.qt_compat import field_type_double, field_type_int, field_type_string

# QgsField type args (QGIS 4 rejects raw int, #25/#36): resolved once in
# qt_compat (QVariant on QGIS 3, QMetaType on QGIS 4). Kept as module constants
# because several mixins import them by name.
_FIELD_TYPE_STRING = field_type_string()
_FIELD_TYPE_DOUBLE = field_type_double()
_FIELD_TYPE_INT = field_type_int()

# QSettings keys for tutorial flags
SETTINGS_KEY_TUTORIAL_SHOWN = "AISegmentation/tutorial_simple_shown"

# QSettings key for the last Manual-session timestamp (drives the predictive
# model warm-up on a returning Manual user, see env_setup._manual_used_recently).
SETTINGS_KEY_LAST_MANUAL_SESSION_TS = "AISegmentation/last_manual_session_ts"

# Free-trial zone cap: a free-tier user's Automatic zone may not exceed this
# geodesic area (km2). All features and detail levels stay available under
# the cap; subscribers are never capped. Enforced at zone commit time
# (interactive draw AND the MCP/headless paths). The surface IS the price:
# Precision changes the grid, never the bill. This is the fallback only: the
# figure that applies is the served free_zone_max_km2 dial, set against the
# monthly allowance so one drag can never take all of it and a second real
# test is always left. The allowance is a commercial number and is not
# written down here.
FREE_TRIAL_MAX_ZONE_KM2 = 1.5


def free_zone_cap_km2() -> float:
    """The free-trial zone cap (km2), resolved from the server detection policy
    when present, else the built-in FREE_TRIAL_MAX_ZONE_KM2. Lets the cap be
    tuned server-side (a growth dial) without a plugin release. Reads the cached
    config only (no network), so it is safe on the GUI thread at zone-commit
    time; fails to the constant so Manual/offline and older servers are
    unaffected."""
    try:
        from ...core.detection_policy import free_zone_max_km2
        return free_zone_max_km2(FREE_TRIAL_MAX_ZONE_KM2)
    except Exception:  # noqa: BLE001 -- policy is best-effort; fall to the constant
        return FREE_TRIAL_MAX_ZONE_KM2


# Longest served advisory the prompt line renders. A bound has to exist (the
# sentence arrives from outside and the dock has no room for a runaway string),
# but the old one cut the authored road advice mid-word, which reads as a bug in
# the plugin rather than as a long sentence.
SERVED_HINT_MAX_CHARS = 240


def clip_served_hint(text: str) -> str:
    """A served advisory bounded to SERVED_HINT_MAX_CHARS, never cut mid-word.

    Past the bound the cut falls back to the last space and carries an ellipsis,
    so a sentence that runs long is visibly unfinished instead of ending on half
    a word. Shorter text is returned stripped and unchanged.
    """
    text = (text or "").strip()
    if len(text) <= SERVED_HINT_MAX_CHARS:
        return text
    head = text[:SERVED_HINT_MAX_CHARS]
    cut = head.rfind(" ")
    return (head[:cut] if cut > 0 else head).rstrip(" ,;:.") + "..."


def max_tiles_per_run_cap(zone_km2: float | None = None) -> int:
    """Hard per-run tile ceiling, resolved from the server detection policy
    when present, else the built-in constants. Cached config only (no network),
    so it is safe on the GUI thread and in the live credit estimate; fails to
    the constants offline and on older servers.

    Two limits, and the smaller one wins.

    - The absolute ceiling bounds one run's wall clock and our own compute.
    - The per-km2 ceiling bounds what a run may spend for each km2 it bills.
      A run is priced on the surface drawn, so this is the one ratio that can
      be driven to an absurd value on purpose: without it a pinhole zone at
      the finest grid buys a large amount of compute for almost no surface.
      It is sized on the finest ground tile any class is measurably answered
      at, so it refuses no precision anybody can use.

    ``zone_km2`` is the surface the user drew. Without it only the absolute
    ceiling is known, which is what every caller had before the second limit
    existed.
    """
    from ...core.tile_manager import (
        MAX_TILES,
        MAX_TILES_FLOOR,
        MAX_TILES_PER_KM2,
    )
    try:
        from ...core.detection_policy import (
            max_tiles_floor,
            max_tiles_per_km2,
            max_tiles_per_run,
        )
        ceiling = max_tiles_per_run(MAX_TILES)
        if zone_km2 is None or zone_km2 <= 0:
            return ceiling
        per_km2 = max_tiles_per_km2(MAX_TILES_PER_KM2)
        floor = max_tiles_floor(MAX_TILES_FLOOR)
    except Exception:  # noqa: BLE001 -- policy is best-effort; fall to the constants
        ceiling = MAX_TILES
        if zone_km2 is None or zone_km2 <= 0:
            return ceiling
        per_km2 = MAX_TILES_PER_KM2
        floor = MAX_TILES_FLOOR
    try:
        allowed = int(float(zone_km2) * float(per_km2))
    except (TypeError, ValueError):
        return ceiling
    return max(int(floor), min(int(ceiling), allowed))


# -- the two zone refusals -------------------------------------------------
# Both sentences quote a number that is already a server dial, so they are
# built from the LIVE value here rather than from a copy of it. A refusal that
# names a different limit from the one that refused is worse than no number.
# The wording itself is served copy (ids ``zone.too_large`` and
# ``zone.free_cap``), so a confusing refusal is one deploy away from fixed.
# Placeholders are filled with str.replace, never format(), so a stray brace in
# a served sentence cannot raise on the draw path.


def zone_too_large_message(max_tiles: int, fallback: str | None = None) -> str:
    """Why a zone was refused for exceeding the per-run tile ceiling.

    ``fallback`` lets a caller keep its own shipped wording while still reading
    the one served id: the dock's cost row sits in a header that cannot wrap,
    so it ships a shorter sentence than the plain paths do. A served override
    replaces both.
    """
    from ...core.server_dials import dial_copy

    if fallback is None:
        fallback = tr("Zone too large. Draw a zone of {max} tiles or fewer.")
    text = dial_copy("zone.too_large", fallback)
    return text.replace("{max}", str(int(max_tiles)))


def zone_over_free_cap_message(area_km2: float) -> str:
    """Why a free-tier zone was refused for exceeding the free-trial cap.

    The one-sentence form, for the paths that answer with a plain string (the
    MCP API and the headless run). Those answers go to a program, not to a
    dock label, so the shipped wording is untranslated exactly as it was. A
    newline is the dock's line break (see ``zone_over_free_cap_lines``), so it
    flattens to a space here rather than reaching a program as two lines.
    """
    from ...core.server_dials import dial_copy

    text = dial_copy(
        "zone.free_cap",
        "Zone is {area} km2; free trial zones go up to {max} km2. "
        "Use a smaller zone, or subscribe to segment areas of any size.")
    return _fill_free_cap_text(text.replace("\n", " "), area_km2, "")


def zone_over_free_cap_lines(area_km2: float, upgrade_url: str) -> tuple[str, str]:
    """The same refusal as the two rich-text lines the dock draws.

    Reads the same ``zone.free_cap`` id as the one-sentence form, so one deploy
    fixes both surfaces at once. Served copy replaces the FIRST line only: the
    action line is always the shipped one, because served text has its angle
    brackets removed and so can never carry the subscribe link. Taking the
    second line from the served text is what would drop that link, which is
    the one thing on this label the user has to be able to click.

    Never networks and never raises, so it is safe on the zone-draw path.
    """
    from ...core.server_dials import dial_copy

    # Two lines: the gap, then the action. The served head owns the numbers AND
    # the reason to pay, so the tail carries the two ways out and nothing else.
    # It used to carry an offer line of its own, quoting the MONTHLY quota next
    # to a SIZE refusal, in a unit the product no longer uses, and repeating the
    # size promise the head already makes. No figure belongs here: the tail
    # cannot be reached by a deploy, so any number in it goes stale on the next
    # pricing change and stays stale until the user updates the plugin.
    shipped_head = tr(
        "This zone is {area} km². Free zones stop at {max} km².")
    shipped_tail = tr(
        '<a href="{url}">Upgrade to Pro</a> to run this zone as drawn, or make '
        'it smaller.')
    # escape=True: the two lines go into a RichText label, where a served "&"
    # has to be an entity. The shipped fallback comes back untouched, so its
    # own markup survives.
    text = dial_copy("zone.free_cap", shipped_head + "\n" + shipped_tail,
                     escape=True)
    head, _, _served_tail = text.partition("\n")
    return (_fill_free_cap_text(head, area_km2, upgrade_url),
            _fill_free_cap_text(shipped_tail, area_km2, upgrade_url))


def _fill_free_cap_text(text: str, area_km2: float, upgrade_url: str) -> str:
    """Fill the free-cap placeholders. str.replace, never format(): a stray
    brace in a served sentence would raise on the draw path."""
    return (text
            .replace("{area}", f"{area_km2:.1f}")
            .replace("{max}", f"{free_zone_cap_km2():g}")
            .replace("{url}", upgrade_url)
            .strip())


def _zone_shape_key(plugin) -> str | None:
    """The drawn zone as one comparable string, or None when there is no zone.

    The SHAPE, not its box: two zones that share a bounding box can cover very
    different ground, and a key built from the box alone called the second one
    a repeat of the first. The outline is written at a fixed precision, because
    the same shape redrawn by hand never reproduces its float tail, and it is
    hashed so a traced outline of a thousand points stays one short value. The
    canvas CRS is part of the key: the zone is stored in canvas numbers, and
    the same numbers in another CRS are another place.
    """
    import hashlib

    crs_id = ""
    try:
        crs_id = plugin.iface.mapCanvas().mapSettings().destinationCrs().authid()
    except (RuntimeError, AttributeError):
        crs_id = ""
    outline = ""
    polygon = getattr(plugin, "_auto_zone_polygon", None)
    if polygon is not None:
        try:
            outline = polygon.asWkt(6) or ""
        except (RuntimeError, AttributeError, TypeError):
            outline = ""
    if not outline:
        # No polygon: the MCP rectangle path and any run started from a box.
        zone = getattr(plugin, "_auto_zone", None)
        if zone is None:
            return None
        try:
            outline = (f"{zone.xMinimum():.6f} {zone.yMinimum():.6f} "
                       f"{zone.xMaximum():.6f} {zone.yMaximum():.6f}")
        except (AttributeError, TypeError, ValueError):
            return None
    digest = hashlib.sha256(outline.encode("utf-8")).hexdigest()[:16]
    return f"{crs_id}:{digest}"


def auto_rerun_scope(plugin) -> tuple:
    """Where a finished Automatic run stood: its zone, and the raster it read."""
    zone_key = _zone_shape_key(plugin)
    layer_id = ""
    try:
        layer = plugin._get_active_raster_layer()
        if layer is not None:
            layer_id = layer.id()
    except (RuntimeError, AttributeError):
        layer_id = ""
    return (zone_key, layer_id)


class AutoRerunSignature(tuple):
    """A finished run's inputs, as the identical-re-run note compares them.

    It IS the (prompt, detail, example count) tuple the note builds from the
    current panel, so the comparison stays a plain tuple compare wherever it
    is read. What it adds is the run's SCOPE: the same prompt over a different
    zone, or against a different raster, gives a different answer, so equality
    also asks where the zone and the layer stand now. Without that the note
    called the second zone of a prompt a repeat of the first.
    """

    def __new__(cls, plugin, fields, scope):
        obj = super().__new__(cls, fields)
        obj._plugin = weakref.ref(plugin)
        obj._scope = scope
        return obj

    def __eq__(self, other):
        if not isinstance(other, tuple):
            return NotImplemented
        if tuple(self) != tuple(other):
            return False
        plugin = self._plugin()
        return plugin is not None and self._scope == auto_rerun_scope(plugin)

    def __ne__(self, other):
        equal = self.__eq__(other)
        return equal if equal is NotImplemented else not equal

    __hash__ = tuple.__hash__


def backend_stalled_flag(tiles_done: int, warming_ms: int,
                         submit_retries: int, tiles_skipped_network: int,
                         tiles_timed_out: int = 0) -> bool:
    """Whether an ending run was a STALLED BACKEND rather than a healthy run the
    user simply cancelled.

    A run that billed ZERO tiles is stalled as soon as the service showed any
    distress: time spent in the waiting room (warming_ms), transient submit
    retries, tiles dropped after their submit-retry budget was exhausted, or
    replies the worker gave up waiting on.

    A run that DID produce tiles is a healthy cancel unless the service left at
    least as many accepted tiles unanswered as it answered. Those tiles are
    billed and delivered nothing, so past that line the user gave up on a sick
    service, not on a run going well. A stricter rule would let a run that
    served 5 tiles and swallowed 60 read as healthy; a looser one would flag a
    long run over a single slow tile.

    Carried on auto_detect_cancelled so a service outage the user gives up on
    does not read as a user-initiated cancel in analytics.

    ``tiles_timed_out`` defaults to 0 so an older caller keeps working. It is
    the one signal for a service that accepts a request and then answers a byte
    at a time: nothing retries, nothing is skipped, and without it a run where
    every reply stalled read as a healthy cancel.
    """
    if tiles_done > 0:
        return tiles_timed_out >= tiles_done
    return bool(warming_ms > 0 or submit_retries > 0
                or tiles_skipped_network > 0 or tiles_timed_out > 0)


# The low recall floor sent to the server so every plausible mask comes back and
# the review confidence slider can re-filter client-side with no re-detection
# (free, instant). The UI default cutoff (_AUTO_DEFAULT_CONFIDENCE) is imported
# from core/review_defaults.py, the single source shared with the dock.
_RECALL_FLOOR = 0.10
# Exemplar-ONLY runs (a drawn example, no text prompt) get a higher floor:
# without the open-vocab text prior the model floods low-confidence visual
# matches (big context blobs), which drowns the merger and the review in
# junk. 0.20 keeps the review slider meaningful while cutting the noise tier;
# a text prompt restores the full 0.10 recall floor.
_RECALL_FLOOR_EXEMPLAR_ONLY = 0.20

# The object currently OPEN for editing (refine/handoff) reuses pending-blue
# (PENDING_STROKE / PENDING_FILL in ui/canvas_palette.py) with a bolder outline,
# so the canvas keeps a two-color language: blue = editable, green = validated.

# EPSG:3857 ground resolution (map units/px) at XYZ zoom 0 with 256 px tiles:
# 2*pi*6378137 / 256. Divided by 2**zmax it gives an online basemap's finest
# native resolution, used to clamp the render so an online source is never
# upsampled past the detail it actually has (mirrors the local native clamp).
_WEBMERC_MUPP_Z0 = 156543.033928

# Automatic-review shape-refine defaults (Simplify / Clean / Round corners /
# Fill holes / Expand) now live in core/review_defaults.py, imported at the top
# as the private _AUTO_REVIEW_* aliases so both this file and the dock share one
# source. No local copies to keep in sync.

# The finalize and reslice chains are cooperatively time-sliced on the GUI
# thread so a big run never freezes QGIS. Each turn works for at most this many
# seconds, then yields to the event loop (cursor, repaints, window switches stay
# live) and reschedules itself while work remains.
_AUTO_PUMP_BUDGET_S = 0.02
# The live preview writes arriving objects to the provider on a coalesced tick
# rather than once per completed tile. The tick applies a DELTA, so a burst of
# tiles costs one write of what changed; this is how often the canvas is asked
# to redraw, not how much work the GUI does per object.
_AUTO_LIVE_REPAINT_MS = 300
# What the live preview may cost the run. A frame's draw time grows with the
# objects already found, and the canvas draws on the same machine as the tiles
# are folded on, so a preview that repaints as fast as the map can draw ends up
# owning the machine for the whole run. After each frame the preview sits out
# this many times its own measured cost before asking for the next one: at 40 ms
# a frame it is imperceptible, at 1.5 s it settles to a repaint every few
# seconds and gives the run its cores back.
_AUTO_LIVE_FRAME_COST_RATIO = 3.0
# However slow one frame gets, the preview still shows something this often.
_AUTO_LIVE_REPAINT_MAX_MS = 6000


def auto_pump_budget() -> float:
    """Seconds one finalize or reslice turn works before it yields. Served
    as ui.auto_pump_budget_s, read per turn."""
    return auto_pump_budget_s(_AUTO_PUMP_BUDGET_S)


def auto_live_repaint_settings() -> tuple[int, float, int]:
    """(repaint interval ms, frame cost ratio, repaint ceiling ms) for the
    live preview, each served under ui.live_*, read when a tick is scheduled."""
    return (
        live_repaint_ms(_AUTO_LIVE_REPAINT_MS),
        live_frame_cost_ratio(_AUTO_LIVE_FRAME_COST_RATIO),
        live_repaint_max_ms(_AUTO_LIVE_REPAINT_MAX_MS),
    )


# Bulk-insert flag: tells the provider not to populate feature IDs back onto the
# input features, which we never read. A recognised QGIS speed-up for bulk
# addFeatures (live rebuilds + export). Version-defensive (Qt5 flat / Qt6 scoped).
_FAST_INSERT = getattr(
    getattr(QgsFeatureSink, "Flag", QgsFeatureSink), "FastInsert", None
)


def _get_change_path_instructions():
    """Return platform-specific instructions for changing the install path."""
    if sys.platform == "win32":
        steps = tr(
            "1. Open Windows Settings > System > Advanced system settings\n"
            "2. Click 'Environment Variables'\n"
            "3. Under 'User variables', click 'New'\n"
            "4. Variable name: AI_SEGMENTATION_CACHE_DIR\n"
            "5. Variable value: the folder path you want to use\n"
            "6. Click OK and restart QGIS"
        )
    elif sys.platform == "darwin":
        steps = tr(
            "Run this command in Terminal, then restart QGIS:\n\n"
            "launchctl setenv AI_SEGMENTATION_CACHE_DIR /your/path"
        )
    else:
        steps = tr(
            "Add this line to your ~/.bashrc or ~/.profile, "
            "then restart QGIS:\n\n"
            "export AI_SEGMENTATION_CACHE_DIR=/your/path"
        )
    return "{}\n\n{}".format(
        tr("To install in a different folder, set the environment "
           "variable AI_SEGMENTATION_CACHE_DIR:"),
        steps)


def _apply_fast_render(layer) -> None:
    """Make a detection result layer cheap to pan and zoom. The UI-side name for
    ``core.output_store.apply_fast_canvas_render``, which holds the whole
    implementation and the tolerance behind it; read that one.

    The forwarder stays because half a dozen mixins already import this name,
    and because the same settings have to reach layers created in ``src/core``
    (the autosave recovery), which cannot import from ``src/ui``.
    """
    from ...core.output_store import apply_fast_canvas_render

    apply_fast_canvas_render(layer)


def _notify_provider_write(layer) -> None:
    """Tell QGIS that features changed under it after a PROVIDER-level write.

    We rewrite the review layer through the data provider (deleteFeatures /
    addFeatures / truncate / changeGeometryValues) because it is far cheaper than
    an edit session. The catch: a provider write emits ONLY ``repaintRequested``.
    None of the five signals ``QgsPointLocator`` invalidates on (featureAdded,
    featureDeleted, geometryChanged, attributeValueChanged, dataChanged) are
    emitted, because those come from the edit buffer. So QGIS's snapping index
    and the vertex tool's geometry cache keep describing the PREVIOUS features.

    Measured on QGIS 3.44: after a delete + re-add, ``snapToMap`` still returned
    a confident match on a feature id that no longer existed, so clicking a
    vertex did nothing. After ``changeGeometryValues`` it returned a LIVE id
    carrying the OLD shape, so an edit landed on the wrong vertex.
    ``startEditing()`` does not rebuild the index, which is why simply
    re-entering the hand-edit bridge never cleared it.

    ``dataChanged`` is the one call that clears both (the locator destroys its
    index, the vertex tool clears its geometry cache). It is layer-scoped and
    lazy (the index rebuilds on the next snap), unlike ``clearAllLocators()``,
    which would also throw away every other layer's index. Never raises."""
    try:
        layer.dataChanged.emit()
    except (RuntimeError, AttributeError):
        pass


def _clear_all_features(provider) -> None:
    """Empty a provider we rewrite in full, WITHOUT using ``truncate()``.

    ``QgsMemoryProvider::truncate()`` drops the feature map but leaves every
    entry in the spatial index it built. The provider then hands out the same
    small feature ids again, so after K truncate + re-add cycles a bbox request
    returns each feature K times and the renderer paints each polygon K times.
    ``featureCount()`` stays correct throughout, which is why this reads as
    "QGIS got slow" rather than as a bug, and why the stacked translucent fills
    start looking opaque. Calling ``createSpatialIndex()`` again does not repair
    it: the provider returns early when an index already exists.

    Deleting by id keeps the index consistent. It costs more than a truncate on
    a large set, but it runs on the full-rebuild path only, which already
    follows a whole reslice, and it buys back the per-frame multiplication.
    """
    try:
        ids = list(provider.allFeatureIds())
    except (AttributeError, RuntimeError):
        ids = []
    if ids:
        provider.deleteFeatures(ids)
    else:
        provider.truncate()


def _add_features_fast(provider, features) -> bool:
    """Bulk-add features with the FastInsert flag when the running QGIS exposes
    it. FastInsert skips writing provider feature IDs back onto the input
    features (which we never read), the standard speed-up for bulk loads.

    Returns whether the provider took the batch. A provider that refuses one
    (a schema it rejected, so every setAttributes mismatched) reports nothing
    else, and a caller that drops this answer goes on to tell the user it saved
    polygons over an empty file.
    """
    if _FAST_INSERT is not None:
        result = provider.addFeatures(features, _FAST_INSERT)
    else:
        result = provider.addFeatures(features)
    # Some bindings answer (ok, added_features), the rest a plain bool.
    if isinstance(result, tuple):
        return bool(result[0]) if result else False
    return bool(result)


def _add_features_with_ids(provider, features):
    """Bulk-add features and return (ok, added) where ``added`` holds the
    provider's ASSIGNED feature copies. PyQGIS never writes the new ids back
    onto the features you pass in (the C++ in-out list arrives as a sip copy,
    so reading .id() on an input after the call yields the invalid sentinel):
    any caller that needs the provider fids must read them off ``added``."""
    res = provider.addFeatures(features)
    if isinstance(res, tuple):
        ok, added = res
        return bool(ok), list(added or [])
    return bool(res), []


def _debounce_timer(owner, attr_name: str, parent, interval_ms: int, slot) -> None:
    """Lazily create (once) and (re)start a single-shot debounce timer stored
    as ``owner.<attr_name>``. The plugin controller is a plain class, not a
    QObject, so ``parent`` must be a QObject the caller already relies on to
    own the timer's lifetime (usually ``self.dock_widget``). The slot is
    connected only at creation, never re-connected on later calls; every call
    restarts the countdown (trailing-edge debounce), so the slot fires once
    ``interval_ms`` has passed with no further call."""
    from qgis.PyQt.QtCore import QTimer

    timer = getattr(owner, attr_name, None)
    if timer is None:
        timer = QTimer(parent)
        timer.setSingleShot(True)
        timer.timeout.connect(slot)
        setattr(owner, attr_name, timer)
    timer.start(interval_ms)


def _provider_name_for_log(layer) -> str:
    """Best-effort data-provider name of a raster layer for diagnostic logs
    (e.g. 'wms' for XYZ/WMS basemaps, 'gdal' for local files). Online providers
    make the upfront zone render slow (basemap tile fetches), so this tag is the
    first clue when 'generating tiles' is slow. Never raises; production-safe
    (no URLs/keys, just the provider type)."""
    try:
        prov = layer.dataProvider()
        return prov.name() if prov is not None else "unknown"
    except (RuntimeError, AttributeError):
        return "unknown"


# Workers whose OS thread outlived plugin unload (blocked in a long network
# call). Holding the last Python reference here prevents the C++ QThread
# from being garbage-collected while running, which would hard-abort QGIS
# ("QThread: Destroyed while thread is still running"). Entries remove
# themselves when the thread finishes; on interpreter exit the daemon-like
# leak is bounded by the client's longest timeout (110 s).
_ORPHANED_WORKERS: list = []


def park_orphaned_worker(worker) -> None:
    """Keep `worker` alive until its finished signal fires, then release it.

    Two callers rely on this: a still-running orphan at teardown, and a
    NOT-YET-STARTED worker parked as a lifetime anchor right before its
    start() call (e.g. the manual encode worker). Both must be safe here.
    """
    # Capture the list object itself (not the module-global name) so a
    # reload of this module can never rebind `_ORPHANED_WORKERS` out from
    # under an already-parked worker and drop its last strong reference.
    bucket = _ORPHANED_WORKERS
    bucket.append(worker)
    released = {"done": False}

    def _release() -> None:
        if released["done"]:
            return  # idempotent: the re-check below and the signal can both fire
        released["done"] = True
        try:
            bucket.remove(worker)
        except ValueError:
            pass
        # Join the native thread before destruction: finished() is emitted a
        # hair before the thread has fully stopped, and destroying a QThread
        # inside that window hard-aborts all of QGIS. wait() returns
        # immediately for a thread that never started or already stopped.
        try:
            worker.wait(10000)
        except (RuntimeError, AttributeError):
            pass
        try:
            worker.deleteLater()
        except RuntimeError:
            pass  # C++ object already gone

    worker.finished.connect(_release)
    # A thread can finish between the caller's isRunning() check and the
    # connect above: its finished signal never reaches the slot and the
    # worker would leak in the bucket forever. Release now ONLY when the
    # thread actually ran to completion: a parked-before-start worker
    # (isRunning False, isFinished False) must stay anchored, or the pending
    # delete would destroy the thread the moment start() runs and abort QGIS.
    try:
        finished_in_gap = worker.isFinished() and not worker.isRunning()
    except (RuntimeError, AttributeError):
        finished_in_gap = True  # dead C++ object: just drop the anchor
    if finished_in_gap:
        _release()


def join_orphaned_workers(budget_seconds: float) -> int:
    """Wait, inside one budget, on every worker parked anywhere in the plugin.

    Parking keeps a worker alive. It does not wait for it, and a parked thread
    still running when Python shuts down is destroyed by ~QThread, which aborts
    all of QGIS. Teardown reads the bucket itself rather than a list one caller
    built, because parking happens in a dozen places across the mixins and only
    two of them were ever on that list: the online tile fetch, the live
    stitcher, the review refine thread and the diagnostics collector all parked
    and then nothing ever joined them.

    A worker parked before start() has never run, and wait() returns at once for
    it, so an anchor costs nothing here. The budget is shared, so a set of stuck
    threads cannot add up to a frozen quit.

    Asking first is what makes the wait mean anything. Teardown's cancel loop
    only reaches the workers the controller holds, and most of this bucket was
    parked at its own call site by code that never went through that loop. A
    thread nobody asked to stop is still working when the wait starts, so it
    runs the budget out and gets destroyed anyway. Both stop channels are used:
    the Qt flag a run loop polls, and the cancel() method the plugin's own
    workers expose.

    Returns how many stopped inside the budget, for the teardown log.
    """
    workers = list(_ORPHANED_WORKERS)
    if not workers:
        return 0
    for worker in workers:
        try:
            worker.requestInterruption()
        except (RuntimeError, AttributeError):
            pass  # dead C++ wrapper, or not a QThread
        try:
            cancel = getattr(worker, "cancel", None)
            if callable(cancel):
                cancel()
        except (RuntimeError, AttributeError, TypeError):
            pass  # a cancel that refuses must never abort teardown
    deadline = time.monotonic() + budget_seconds
    joined = 0
    for worker in workers:
        left_ms = int(max(0.0, deadline - time.monotonic()) * 1000)
        if left_ms <= 0:
            break
        try:
            if worker.wait(left_ms):
                joined += 1
        except (RuntimeError, AttributeError):
            pass  # dead C++ wrapper, or not a QThread: nothing to join
    return joined


def detach_widget_from_main_window(widget) -> bool:
    """Take a top-level plugin widget off the QGIS main window and schedule
    its destruction. Returns True when the widget is on its way out.

    Unparenting is the part that matters. `iface.removeDockWidget` and `hide()`
    both leave the widget a child of the main window, and `deleteLater()` only
    posts an event that nothing processes before unload returns. So the widget
    is still there, under its object name, when the next instance of the plugin
    builds its own: two widgets, one name.

    Ownership goes back to C++ after the unparent, so dropping the last Python
    reference cannot run the destructor inside unload. The delete happens on the
    next pass of the event loop instead.

    Never raises: this runs inside unload, where a raise skips the steps below.
    """
    if widget is None:
        return False
    from qgis.PyQt import sip
    try:
        if sip.isdeleted(widget):
            return False
    except (RuntimeError, TypeError):
        return False
    try:
        widget.hide()
    except (RuntimeError, AttributeError):
        pass
    try:
        widget.setParent(None)
    except (RuntimeError, AttributeError):
        pass
    try:
        sip.transferto(widget, None)
    except (RuntimeError, TypeError, ValueError):
        pass
    try:
        widget.deleteLater()
    except (RuntimeError, AttributeError):
        return False
    return True


def dir_size_label(path: str) -> str:
    """Human-readable total size of a directory tree (cache-size labels,
    removal log lines). Best-effort, never raises."""
    total = 0
    try:
        for root, _dirs, files in os.walk(path):
            for name in files:
                try:
                    total += os.path.getsize(os.path.join(root, name))
                except OSError:  # nosec B112
                    continue
    except OSError:
        return "-"
    mb = total / (1024 * 1024)
    if mb >= 1024:
        return f"{mb / 1024:.1f} GB"
    return f"{mb:.0f} MB"


# A flat grid of pixels with no place on the earth: what Manual writes when the
# raster has no CRS. Two spellings of the same thing, newest first, because an
# older PROJ only parses the second one.
_PIXEL_GRID_WKT = (
    'ENGCRS["Pixel grid",EDATUM["Image origin"],CS[Cartesian,2],'
    'AXIS["x",east,ORDER[1]],AXIS["y",north,ORDER[2]],'
    'LENGTHUNIT["metre",1]]',
    'LOCAL_CS["Pixel grid",UNIT["metre",1],AXIS["X",EAST],AXIS["Y",NORTH]]',
)


def pixel_grid_crs():
    """The CRS a non-georeferenced Manual export is written in.

    It has to be one the writer leaves alone: the saved layer is moved onto a
    ground-metre CRS unless its own already measures in ground metres, and
    Pseudo-Mercator never does, so declaring pixels as EPSG:3857 gets every
    coordinate reprojected and the mode's one promise is gone. A local grid
    measures in metres by declaration, so nothing moves, and QGIS draws it at
    its own coordinates, on top of the image it came from.

    Falls back to EPSG:3857 if neither spelling parses, which is what this
    mode used before.
    """
    from qgis.core import QgsCoordinateReferenceSystem

    for wkt in _PIXEL_GRID_WKT:
        crs = QgsCoordinateReferenceSystem(wkt)
        if crs.isValid():
            return crs
    return QgsCoordinateReferenceSystem("EPSG:3857")


# Plain image formats that carry no georeferencing of their own (a world file
# or an explicit CRS is what makes one usable on a map).
NON_GEOREF_IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".bmp", ".gif")


def looks_like_pixel_image(layer) -> bool:
    """True when the layer's source is a plain image format. Used to tell an
    expected pixel-mode run (a PNG screenshot) apart from a surprising one (a
    GeoTIFF whose CRS is missing), so only the latter warrants a heads-up."""
    try:
        source = (layer.source() or "").lower()
    except (RuntimeError, AttributeError):
        return False
    return any(source.endswith(ext) for ext in NON_GEOREF_IMAGE_EXTENSIONS)


def is_layer_georeferenced(layer) -> bool:
    """True when a raster layer is properly georeferenced.

    A missing CRS is decisive on its OWN, whatever the file extension: without
    one there is no pixel-to-ground mapping, so the layer belongs in Manual's
    pixel-coordinate mode. Deciding this by extension alone (the previous rule)
    classified a CRS-less GeoTIFF/IMG/JP2 as georeferenced, which then failed
    the caller's CRS guard and left Manual with no way to run at all.

    The extension list stays as a SECOND, narrower test: an image format can
    carry a valid CRS and still be laid out at bare pixel coordinates (extent
    anchored at the origin), which the CRS check alone cannot catch.

    Guarded against a layer deleted mid-check (RuntimeError from the sip
    wrapper)."""
    from qgis.core import QgsRasterLayer
    if layer is None or not isinstance(layer, QgsRasterLayer):
        return False
    try:
        if not layer.crs().isValid():
            return False
    except RuntimeError:
        return False
    try:
        source = layer.source().lower()
    except RuntimeError:
        return False
    if any(source.endswith(ext) for ext in NON_GEOREF_IMAGE_EXTENSIONS):
        try:
            extent = layer.extent()
            if extent.xMinimum() == 0 and extent.yMinimum() == 0:
                # Likely bare pixel dimensions, not ground coordinates.
                return False
        except RuntimeError:
            return False
    return True
