"""The review's geometry work: ground-metre to CRS-unit conversions, the
per-object shape refine and its cache, and shared borders over the whole set.

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

from qgis.core import (
    Qgis,
    QgsCoordinateReferenceSystem,
    QgsMessageLog,
    QgsPointXY,
)

from ...core.interaction_dials import (
    ground_scale_band_deg,
    ground_scale_band_m,
    live_refiner_memo_max,
    rescue_refine_budget_s,
    review_reslice_parked_geoms_max,
    review_reslice_parked_keys_max,
)
from ...core.live_refine import (
    LiveRefiner,
    plain_outline_geom,
    points_dial_fraction,
    refine_review_geom,
)

# How many run refiners the review keeps alive at once. One covers a whole
# reslice; the spare room absorbs the Correct step's per-shape overrides, which
# hand this mixin a different params dict for one object inside a pass, and the
# ground-scale bands below, which give a tall zone one refiner per band. Over
# the bound the least recently built goes, so a run degrades to re-measuring
# rather than to a memo that grows with the zone.
_LIVE_REFINER_MEMO_MAX = 8

# How far a run may move along y before its ground dials are re-measured, in
# the two CRS families where one unit is not one ground metre everywhere: half
# a degree of latitude in a geographic CRS, and the northing that spans about
# as much in a Mercator one. Inside a band the metres-per-unit factor moves by
# well under a percent, which is smaller than the dead band the aspect
# correction already ignores. A CRS that measures in ground metres gets no
# band at all: one refiner serves the whole run, as before.
_GROUND_SCALE_BAND_DEG = 0.5
_GROUND_SCALE_BAND_M = 50000.0

# How many SUPERSEDED shape keys keep their refined geometry, and how many
# geometries all of them together may hold. Moving a Shape control and moving
# it back is the commonest gesture in the step, so the way back should cost
# nothing; the geometry budget is what stops a huge run from paying for that in
# memory instead.
_RESLICE_PARKED_KEYS_MAX = 2
_RESLICE_PARKED_GEOMS_MAX = 40000

# How long the safety-net export may spend shaping objects the Confidence
# cutoff was hiding. Those were never refined, so a dense review can owe
# thousands of GEOS passes at the exact moment the user is closing QGIS, on a
# path with no event loop to yield to. Windows calls a window that stops
# answering for a few seconds "not responding", and a user who force-quits
# there loses the whole rescue, so past this budget the remaining objects are
# saved with their traced outline instead. Losing the polish beats losing the
# detection.
_RESCUE_REFINE_BUDGET_S = 3.0


def rescue_refine_budget() -> float:
    """The rescue export's shaping budget, served as
    ui.review.rescue_refine_budget_s. Read once per rescue."""
    return rescue_refine_budget_s(_RESCUE_REFINE_BUDGET_S)


def _geom_centre_xy(geom) -> tuple[float, float] | None:
    """Where a per-CRS ground measure is taken for ONE object: the centre of
    its bounding box, or None when there is no box to read.

    None rather than the CRS origin. An empty geometry answers a null box, and
    the centre of a null box is (0, 0), which on a geographic CRS is a point in
    the Atlantic: the factor measured there converts a whole run's ground dials
    at the wrong latitude, and nothing says so.
    """
    try:
        if geom is None or geom.isEmpty():
            return None
        bbox = geom.boundingBox()
        if bbox.isNull():
            return None
        centre = bbox.center()
        return float(centre.x()), float(centre.y())
    except (AttributeError, RuntimeError, TypeError):
        return None


class AutoReviewGeometryMixin:
    """Refine one object, cache the result, then snap the assembled set."""

    def _auto_crs_metres_per_unit(self, ref_x: float, ref_y: float) -> float:
        """Ground metres per unit of the RUN CRS, near (ref_x, ref_y).

        A run happens in the layer CRS, which is often Web Mercator, where one
        unit is well under a metre away from the equator (and a geographic CRS
        counts in degrees). Any setting expressed in ground metres has to cross
        that gap before it touches a geometry. Measured on the ellipsoid, so a
        projected metre CRS answers ~1.0 whatever its zone. Answers 1.0 on any
        failure, which reads the metre setting as CRS units: the behaviour
        before this conversion existed.

        Memoized per CRS and per coarse position: the factor moves with
        latitude, but not across one drawn zone.
        """
        authid = self._auto_crs_authid or "EPSG:4326"
        try:
            crs = QgsCoordinateReferenceSystem(authid)
            if not crs.isValid():
                return 1.0
            geographic = bool(crs.isGeographic())
        except Exception:  # noqa: BLE001 -- an unusable CRS means no conversion
            return 1.0
        # Degrees move fast, projected units do not: bucket each at the scale
        # where the factor is still flat (about 1 km / 10 km on the ground).
        bucket = round(ref_y, 2) if geographic else round(ref_y / 10000.0)
        key = (authid, bucket)
        cached = getattr(self, "_auto_crs_mpu", None)
        if cached is not None and cached[0] == key:
            return cached[1]
        factor = 1.0
        try:
            from ...core.layer_conventions import make_area_measurer
            step = 0.001 if geographic else 1.0
            metres = float(make_area_measurer(crs).measureLine(
                QgsPointXY(ref_x, ref_y), QgsPointXY(ref_x + step, ref_y)))
            if metres > 0:
                factor = metres / step
        except Exception:  # noqa: BLE001 -- never block a refine on a measure
            factor = 1.0
        self._auto_crs_mpu = (key, factor)
        return factor

    def _auto_crs_unit_aspect(self, ref_x: float, ref_y: float) -> float:
        """How much longer one y unit of the RUN CRS is than one x unit, near
        (ref_x, ref_y). See core.layer_conventions.ground_unit_aspect: 1.0 in a
        projected CRS, above 1 in a geographic one, where squaring a footprint
        on raw coordinates leaves every corner tilted. Memoized like the ground
        measure above, and on the same coarse position."""
        authid = self._auto_crs_authid or "EPSG:4326"
        try:
            crs = QgsCoordinateReferenceSystem(authid)
            if not crs.isValid():
                return 1.0
            geographic = bool(crs.isGeographic())
        except Exception:  # noqa: BLE001 -- an unusable CRS means no correction
            return 1.0
        bucket = round(ref_y, 2) if geographic else round(ref_y / 10000.0)
        key = (authid, bucket)
        cached = getattr(self, "_auto_crs_aspect", None)
        if cached is not None and cached[0] == key:
            return cached[1]
        try:
            from ...core.layer_conventions import ground_unit_aspect
            aspect = ground_unit_aspect(crs, ref_x, ref_y)
        except Exception:  # noqa: BLE001 -- never block a refine on a measure
            aspect = 1.0
        self._auto_crs_aspect = (key, aspect)
        return aspect

    def _auto_ground_scale_band(self) -> float:
        """How far along y the RUN CRS may move before its ground dials have to
        be measured again, in CRS units. 0.0 means never.

        0.0 wherever a coordinate difference is already a ground metre, which is
        every national grid and every UTM zone: the factor is 1.0 from one end
        of the run to the other. A geographic CRS and any Mercator answer a real
        band, because their unit follows latitude. Memoized per CRS: the answer
        cannot change inside a run."""
        authid = getattr(self, "_auto_crs_authid", None) or "EPSG:4326"
        cached = getattr(self, "_auto_crs_scale_band", None)
        if cached is not None and cached[0] == authid:
            return cached[1]
        band = ground_scale_band_m(_GROUND_SCALE_BAND_M)
        try:
            from ...core.layer_conventions import crs_measures_in_ground_metres
            crs = QgsCoordinateReferenceSystem(authid)
            if crs_measures_in_ground_metres(crs):
                band = 0.0
            elif crs.isGeographic():
                band = ground_scale_band_deg(_GROUND_SCALE_BAND_DEG)
        except Exception:  # noqa: BLE001 -- an unreadable CRS keeps the band
            band = ground_scale_band_m(_GROUND_SCALE_BAND_M)
        self._auto_crs_scale_band = (authid, band)
        return band

    def _auto_ground_scale_bucket(self, ref_y: float) -> int:
        """Which band of ground scale a northing falls in. Always 0 in a CRS
        that measures in ground metres, so those runs keep one refiner."""
        band = self._auto_ground_scale_band()
        if band <= 0.0:
            return 0
        try:
            return int(float(ref_y) // band)
        except (TypeError, ValueError, OverflowError):
            return 0

    def _refine_geom_for_review(self, base, params: dict, pixel_size: float):
        """Apply the review shape-refine controls to a base object geometry.
        simplify/expand are px in the UI, converted to ground units by the run's
        detection pixel size (_auto_refine_pixel_size). Returns a NEW geometry
        (base is never mutated).

        Every dial behind those controls is constant for a run, so they are all
        resolved once into a core.live_refine.LiveRefiner and reused: this
        method only picks the right refiner and hands it the object.
        """
        refiner = self._review_refiner_for(base, params, pixel_size)
        if refiner is None:
            return None
        return refiner.refine(base)

    def _review_refiner_for(self, base, params: dict, pixel_size: float,
                            shape_key: tuple | None = None):
        """The refiner that shapes THIS object under ``params``, or None when
        there is no position to measure the run's ground dials at.

        ``shape_key`` lets a caller that has already computed the shape key hand
        it in. The key walks a dozen params, and the refine path computed it
        once for the cache and again here for every object of the set.

        Split out from the refine itself because the pick reads the project
        (the ground measure) and the shape does not: the review's off-GUI
        refine thread takes the refiner from here and runs it on its own
        thread. A refiner carries no plugin state and is read-only once built,
        so several threads may share one.
        """
        # The object's own position decides which refiner it gets: a refiner
        # holds ground dials already converted, at the factor measured where it
        # was built, so a zone tall enough to cross a band must not be served
        # the band it started in. One band, one refiner; a CRS that measures in
        # ground metres has a single band and keeps a single refiner.
        ref = _geom_centre_xy(base)
        if ref is None:
            # Nothing to measure at, and the CRS origin is not a stand-in.
            # Fall back to the last object that did carry a position, which is
            # at worst one band away; with none, there is no shape to refine.
            ref = getattr(self, "_review_ground_ref_xy", None)
            if ref is None:
                return None
        else:
            self._review_ground_ref_xy = ref
        # _review_shape_key covers exactly the params the refiner reads
        # (confidence and the size filters change no shape), so it doubles as
        # the refiner's identity; the run CRS and the band join it because the
        # ground measure below depends on both. Read off the class, not off
        # self: the tests bind this method onto bare stubs.
        key = (getattr(self, "_auto_crs_authid", None),
               self._auto_ground_scale_bucket(ref[1]),
               shape_key if shape_key is not None
               else AutoReviewGeometryMixin._review_shape_key(params, pixel_size))
        refiners = getattr(self, "_review_live_refiners", None)
        if not isinstance(refiners, dict):
            refiners = {}
            self._review_live_refiners = refiners
        refiner = refiners.get(key)
        if refiner is None:
            # Dicts keep insertion order, so the first key is the oldest.
            while len(refiners) >= live_refiner_memo_max(_LIVE_REFINER_MEMO_MAX):
                refiners.pop(next(iter(refiners)))
            refiner = LiveRefiner(
                params, pixel_size,
                self._auto_crs_metres_per_unit(ref[0], ref[1]),
                self._auto_crs_unit_aspect(ref[0], ref[1]))
            refiners[key] = refiner
        return refiner

    @staticmethod
    def _review_shape_key(params: dict, pixel_size: float) -> tuple:
        """Hashable identity of the SHAPE portion of the review params (+ the
        px scale they are applied at). Confidence and Min/Max size are filters,
        not shape ops, so they stay OUT: two reslices that differ only by a
        filter share every refined geometry through _auto_reslice_cache."""
        fill_holes = bool(params.get("fill_holes", False))
        return (
            round(float(params.get("simplify_px", 0) or 0.0), 4),
            bool(params.get("smooth", False)),
            round(float(params.get("expand_px", 0) or 0.0), 4),
            fill_holes,
            # Without this the reslice cache serves geometry refined at the old
            # threshold and the Fill-under spinbox reads as a dead control.
            # Reported as 0 while Fill holes is off, because the refine does not
            # read it then: a key that moved anyway threw away every refined
            # object for a change that cannot alter one.
            round(float(params.get("fill_max_m2", 0) or 0.0), 4) if fill_holes
            else 0.0,
            round(float(params.get("open_px", 0) or 0.0), 4),
            # Notch closing is a ground distance and a shape op like the rest.
            # Without it the reslice cache serves outlines closed at the
            # previous distance, so a class whose preset moves it reads as a
            # control that does nothing.
            round(float(params.get("close_notches_m", 0) or 0.0), 4),
            bool(params.get("ortho", False)),
            round(float(pixel_size or 0.0), 6),
            # The point budget changes every cached outline, so a run whose
            # spacing or Points dial differs must not be served the previous
            # geometry.
            round(float(params.get("vertex_spacing_m", 0) or 0.0), 4),
            round(points_dial_fraction(params), 4),
            # Shared borders is applied to the WHOLE set, after this cache, so
            # it changes no cached per-object geometry. It still belongs in the
            # key: the key is also the provenance stamp the incremental push
            # diffs on, and without it a toggle would leave the previous,
            # unsnapped shapes on the layer as "already current".
            bool(params.get("snap_boundaries", False)),
        )

    def _boundary_snap_offered(self) -> bool:
        """Whether this run may show the shared-borders control. Body in
        review_boundary_snap."""
        from .review_boundary_snap import boundary_snap_is_offered

        return boundary_snap_is_offered(self)

    def _boundary_snap_tolerance_units(self, geoms: list) -> float:
        """The shared-borders tolerance for this set, in RUN CRS UNITS. Body in
        review_boundary_snap."""
        from .review_boundary_snap import boundary_snap_tolerance_in_units

        return boundary_snap_tolerance_in_units(self, geoms)

    def _apply_boundary_snap(self, geoms: list, params: dict) -> list:
        """Give the VISIBLE set exact shared borders, once, as a whole set.

        Every other review control acts on ONE object, so it lives in the
        per-object refine (_refine_geom_for_review). Sharing a border needs
        every neighbour at once, so it runs on the assembled visible set, right
        before the review adopts it, which is also the set the Export commits.
        Body in review_boundary_snap.
        """
        from .review_boundary_snap import apply_boundary_snap_to_set

        return apply_boundary_snap_to_set(self, geoms, params)

    def _begin_boundary_snap(self, geoms: list, params: dict) -> tuple:
        """``(answer, started)``: the sliced half of _apply_boundary_snap, for
        the finalize pump. Body in review_boundary_snap."""
        from .review_boundary_snap import begin_boundary_snap_pass

        return begin_boundary_snap_pass(self, geoms, params)

    def _finish_boundary_snap(self, geoms: list, started: tuple) -> list:
        """The snapped set from a pass stepped to the end. Body in
        review_boundary_snap."""
        from .review_boundary_snap import finish_boundary_snap_pass

        return finish_boundary_snap_pass(self, geoms, started)

    def _set_boundary_snap_skip_reason(self, reason: str) -> None:
        """Tell the panel why shared borders did nothing, once. Body in
        review_boundary_snap."""
        from .review_boundary_snap import set_boundary_snap_skip_reason

        set_boundary_snap_skip_reason(self, reason)

    def _reset_review_refine_cache(self) -> None:
        """Drop the reslice refine cache. Must run whenever _auto_objects is
        rebuilt or cleared (finalize build, exemplar re-group, export, discard,
        run start): the cache is keyed by object INDEX, so a stale entry after
        a rebuild would render the wrong object. The review provider map goes
        with it (det_ids from a rebuilt object set can collide with the old
        ones), which forces the next push to a full rebuild of the layer.

        The run refiners go with it: they hold the run's pixel size and its
        ground measure, and a rebuild can be the start of a different run. So
        do the parked shape keys, for the same index-collision reason as the
        current one.

        The off-GUI refine thread goes FIRST, before the dict it fills is
        replaced: it holds jobs keyed by the object indices being invalidated,
        so an answer landing after this point would shape the wrong object."""
        stop_refine_thread = getattr(self, "_stop_review_refine_thread", None)
        if stop_refine_thread is not None:
            stop_refine_thread()
        self._auto_reslice_cache = {
            "key": None, "geoms": {}, "parked": {}, "areas": {}}
        # Per-object edit sequences belong to the object set that just went.
        self._review_refine_seq = {}
        self._review_fid_map = {}
        self._review_live_refiners = {}
        # The whole-set shared-borders and gap-fill answers describe the
        # geometry that just went, and their keys hold those geometries alive.
        self._boundary_snap_memo = None
        self._gap_fill_memo = None
        # The position the ground measure falls back to when an object carries
        # none. It belongs to the run the refiners belonged to.
        self._review_ground_ref_xy = None

    def _invalidate_review_refine(self, indices) -> None:
        """Drop only ``indices`` from the reslice refine cache, keeping every
        other object's cached geometry, and force the next push to re-diff.

        A hand edit (merge / split) rewrites just a few rows and, by the index
        discipline in core.shape_edits, never shifts the others: a merge
        overwrites its target row, a split overwrites the target and APPENDS the
        extra pieces. So the untouched objects keep their cached shape instead
        of being re-refined, which matters because the right-angle squaring runs
        at a few ms per object and a full reset re-squared the whole set on
        every click.

        The provider diff map keeps every other entry: only the rewritten rows
        get their stamp voided, which is what makes the diff rewrite exactly
        their geometry. Clearing the whole map instead sent the next push down
        the full truncate + re-add path, rebuilding a review layer of tens of
        thousands of features after a single click (and silencing the
        progressive mid-reslice pushes, which need the diff). Voiding rather
        than dropping the entry matters: a dropped det_id reads as an ADD and
        would leave the pre-edit feature orphaned on the layer.

        Falls back to nothing when the cache is absent (the next reslice
        rebuilds it)."""
        cache = getattr(self, "_auto_reslice_cache", None)
        touched = [int(idx) for idx in indices]
        # An answer for the shape BEFORE this edit may still be on its way back
        # from the refine thread. The shape key has not moved (no control was
        # touched), so only the per-object sequence can tell the two apart.
        bump = getattr(self, "_bump_review_refine_seq", None)
        if bump is not None:
            bump(touched)
        # The whole-set shared-borders and gap-fill answers described the
        # shapes that just changed.
        self._boundary_snap_memo = None
        self._gap_fill_memo = None
        if isinstance(cache, dict):
            areas = cache.get("areas")
            if isinstance(areas, dict):
                for idx in touched:
                    areas.pop(idx, None)
            # The parked keys go with the current one. They exist so that
            # returning to a previous setting is free, and a hand edit rewrote
            # the object: without this, moving a control back would redraw the
            # shape the user just edited away.
            entries = [cache.get("geoms")]
            entries.extend((cache.get("parked") or {}).values())
            for geoms in entries:
                if isinstance(geoms, dict):
                    for idx in touched:
                        geoms.pop(idx, None)
        fid_map = getattr(self, "_review_fid_map", None)
        if not touched or not isinstance(fid_map, dict) or not fid_map:
            return
        for idx in touched:
            det_id = self._object_fid_for(idx)
            rec = fid_map.get(det_id)
            if rec is not None:
                # (provider_fid, stamp, is_full, score): a None stamp can never
                # equal the next push's, so the diff reports a geometry change.
                fid_map[det_id] = (rec[0], None, rec[2], rec[3])

    def _adopt_reslice_shape_key(self, cache: dict, key: tuple) -> None:
        """Make ``key`` the cache's current shape key, parking the outgoing one.

        A Shape control is used by moving it, looking, and moving it back or
        on. With one key held, every one of those steps re-ran the GEOS refine
        over the whole visible set, including the step that returns to a shape
        already computed a second earlier. Parking the superseded keys makes
        the way back a dict lookup.

        Bounded twice, because a review can hold tens of thousands of objects:
        by how many keys are parked, and by how many geometries they hold in
        total. Over the budget the oldest go first, so a big run degrades to
        today's behaviour instead of growing without limit.
        """
        parked = cache.get("parked")
        if not isinstance(parked, dict):
            parked = {}
            cache["parked"] = parked
        old_key, old_geoms = cache.get("key"), cache.get("geoms")
        if old_key is not None and old_geoms:
            # Re-inserting moves it to the end: dicts keep insertion order, so
            # "the oldest" below is the least recently current key.
            parked.pop(old_key, None)
            parked[old_key] = old_geoms
        revived = parked.pop(key, None)
        held = sum(len(g) for g in parked.values())
        keys_max = review_reslice_parked_keys_max(_RESLICE_PARKED_KEYS_MAX)
        geoms_max = review_reslice_parked_geoms_max(_RESLICE_PARKED_GEOMS_MAX)
        while parked and (len(parked) > keys_max or held > geoms_max):
            held -= len(parked.pop(next(iter(parked))))
        cache["key"] = key
        cache["geoms"] = revived if revived is not None else {}
        # The measured areas describe the geometry under the OLD key. They are
        # cheap to rebuild and wrong to keep, so they go with the swap.
        cache["areas"] = {}

    @staticmethod
    def _plain_outline_geom(base):
        """``base`` repaired and coerced to MultiPolygon, with no shape control
        applied: the object exactly as the run traced it.

        The same normalize the refine cache applies at fill time, so an object
        that takes this path is byte-comparable with a refined one everywhere
        downstream. Used where a shaped outline is wanted but cannot be had:
        the budgeted rescue export, and a refine that failed or emptied
        (_review_refined_geom). Returns None when the repair empties it.

        The body lives in core.live_refine so the off-GUI refine thread reaches
        the same one: an object must come out the same shape whichever thread
        shaped it."""
        return plain_outline_geom(base)

    def _review_refined_geom(self, det_idx: int, base, params: dict,
                             pixel_size: float):
        """Refined + normalized (repaired, MultiPolygon-coerced) geometry for
        one canonical object, memoized in _auto_reslice_cache. A cache hit is a
        dict lookup, so filter-only reslices (Confidence / Min / Max size) never
        re-run the GEOS refine on objects whose shape params did not change.

        A refine that fails or empties falls back to the object as the run
        traced it, and only a base that will not repair returns None. Shaping is
        cosmetic (right angles, simplify, fill holes) and the live preview drew
        the raw outline, so dropping the object instead made a paid detection
        disappear at the end of the run over a step that was never load-bearing.
        Right angles are on by default for built classes, which is exactly where
        the regularizer is most likely to empty a shape.

        Callers must not mutate the returned geometry (copy first)."""
        cache = self._auto_reslice_cache
        key = self._review_shape_key(params, pixel_size)
        if cache.get("key") != key:
            self._adopt_reslice_shape_key(cache, key)
        geoms = cache["geoms"]
        if det_idx in geoms:
            return geoms[det_idx]
        # No provenance check here, on purpose. Every edit the Correct step
        # makes (AI reshape, native vertex work, a hand-drawn add, a merge, a
        # split) becomes the object's canonical base, and the Shapes step reads
        # those bases. So one outline can never sit outside the settings the
        # user is moving, whatever produced it.
        # One object may carry its own Simplify / Right angles (the Correct
        # step's per-shape settings). Merged HERE so every consumer of the
        # visible set, the export included, sees the same shape; a change to an
        # override drops that object's cache entry, so the shared key stays a
        # whole-set identity.
        _per_shape = getattr(self, "_shape_params_for_object", None)
        _shared_params = params
        if _per_shape is not None:
            params = _per_shape(det_idx, params)
        # Guarded inside refine_review_geom: one pathological geometry
        # (self-intersecting, NaN coords) raising out of the GEOS refine/repair
        # must cost only ITSELF, never the finalize/reslice chain. The result is
        # cached either way, so a retried pass does not repeat the failure.
        # Normalized (repair + MultiPolygon coerce) ONCE at cache-fill time, so
        # every later push of this geometry skips both.
        # The key above already describes these params; hand it over rather than
        # walking every dial a second time. An object carrying its OWN Shape
        # settings is a different identity, so it recomputes.
        refiner = self._review_refiner_for(
            base, params, pixel_size,
            shape_key=key if params is _shared_params else None)
        if refiner is None:
            result = self._plain_outline_geom(base)
            geoms[det_idx] = result
            return result
        result, err = refine_review_geom(refiner, base)
        if err is not None:
            self._log_review_refine_failure(det_idx, err)
        geoms[det_idx] = result
        return result

    def _log_review_refine_failure(self, det_idx: int, exc: Exception) -> None:
        """Say that one object would not shape and kept its traced outline.

        Its own method because the off-GUI refine thread hands its failures
        back rather than logging them: QgsMessageLog belongs on the GUI thread,
        and the two paths must say the same thing."""
        try:
            QgsMessageLog.logMessage(
                f"Auto review: refine failed on object {det_idx}, "
                f"kept as traced ({exc})",
                "AI Segmentation", level=Qgis.MessageLevel.Warning)
        except Exception:  # nosec B110
            pass

    def _note_stitch_shapes_dirty(self, fids) -> None:
        """Record fids whose object changed after the live stitch thread shaped
        it, so the finalize seeds its refine cache from every OTHER object
        instead of giving the whole set up.

        Two things move an object after that thread has stopped: the partition
        restore, and the run-wide footprint alignment. Both touch a small part
        of the run, and both report exactly what they touched.
        """
        if not fids:
            return
        dirty = getattr(self, "_auto_stitch_dirty_fids", None)
        if dirty is None:
            dirty = set()
            self._auto_stitch_dirty_fids = dirty
        dirty.update(fids)

    def _stitch_shapes_are_reusable(self, pixel_size: float,
                                    objects: list) -> bool:
        """Whether the shapes the live stitch thread built still describe THIS
        set, at the scale the finalize is about to refine it at.

        Three things can make them a different shape from what the review would
        compute, and all three are silent: the run's detection pixel size moves
        (the thread latches its rescale once, the finalize reads a running
        maximum), the ground metres-per-CRS-unit factor is resolved at a
        different place, and the object under a fid changes after the thread has
        stopped. The last one is flagged by its two causes; the first two are
        compared here. Any doubt returns False and the refine runs as before.
        """
        if getattr(self, "_auto_stitch_shapes_stale", True):
            return False
        if not objects:
            # Nothing to compare the thread's factor against, so nothing can be
            # said about it. An empty set also has no ends to index.
            return False
        stitch_px = float(getattr(self, "_auto_stitch_shape_px", 0.0) or 0.0)
        if abs(stitch_px - float(pixel_size or 0.0)) > 1e-9:
            return False
        factor = float(getattr(self, "_auto_stitch_shape_mpu", 0.0) or 0.0)
        if factor <= 0.0:
            return False
        # The thread resolved ONE factor and shaped the whole set with it. The
        # review works per ground-scale band, giving each band its own refiner
        # (see _refine_geom_for_review), so the thread's single factor is only
        # the right one when the whole set sits in ONE band. EVERY object is
        # checked, not the two ends: the set is not ordered by position, so an
        # object in another band can sit between two that share one, and it
        # would come back shaped at a ground scale the review never used. A
        # band is arithmetic on the centre, so the walk costs one bounding box
        # per object; only the two ends are measured.
        band = None
        try:
            # A CRS whose coordinate difference IS a ground metre has one band
            # from end to end, so every object answers 0 and the walk can only
            # ever confirm it. Skip it: this runs once per finalize over the
            # whole set.
            if self._auto_ground_scale_band() > 0.0:
                for row in objects:
                    ref = _geom_centre_xy(row[0])
                    if ref is None:
                        return False
                    here = self._auto_ground_scale_bucket(ref[1])
                    if band is None:
                        band = here
                    elif here != band:
                        return False
            for row in (objects[0], objects[-1]):
                ref = _geom_centre_xy(row[0])
                if ref is None:
                    return False
                here = self._auto_crs_metres_per_unit(ref[0], ref[1])
                if abs(float(here) - factor) > factor * 1e-6:
                    return False
        except (RuntimeError, AttributeError, TypeError, ValueError):
            return False
        return True

    def _seed_review_refine_cache(self, params: dict, pixel_size: float,
                                  objects: list, object_fids: list) -> int:
        """Fill the refine cache with the shapes the live stitch thread already
        built, and return how many landed.

        That thread applies the run's shape preset to every object it settles,
        off the GUI thread, while the run is still going. The filter phase then
        ran the same refine over the same bases on the GUI thread with an empty
        cache, which is most of the arithmetic behind "Almost done" and all of
        it twice.

        Object by object, so one that cannot be matched costs only itself:
        anything not seeded falls through to the refine exactly as before, and
        the worst case is the behaviour this replaces.
        """
        shapes = getattr(self, "_auto_stitch_shapes", None)
        if not shapes or not objects or len(object_fids) != len(objects):
            return 0
        if not self._stitch_shapes_are_reusable(pixel_size, objects):
            return 0
        from ...core.layer_conventions import repair_polygon, to_multipolygon
        cache = self._auto_reslice_cache
        key = self._review_shape_key(params, pixel_size)
        if cache.get("key") != key:
            self._adopt_reslice_shape_key(cache, key)
        geoms = cache["geoms"]
        per_shape = getattr(self, "_shape_params_for_object", None)
        # Objects that moved after the thread shaped them. Skipping them here
        # is what keeps the seeded output identical to the refined one: they
        # fall through and get shaped from the geometry they now carry.
        dirty = getattr(self, "_auto_stitch_dirty_fids", None) or ()
        seeded = 0
        for det_idx, fid in enumerate(object_fids):
            if det_idx in geoms or fid in dirty:
                continue
            shape = shapes.get(fid)
            if shape is None:
                continue
            # An outline carrying its own Shape settings is not what the run-wide
            # preset produced. It cannot exist on a fresh finalize, which is the
            # only caller today, so this is a guard for a future one.
            if per_shape is not None and per_shape(det_idx, params) is not params:
                continue
            try:
                # The same normalize the refine path applies at cache-fill time.
                # The thread coerces to MultiPolygon but does not repair, so
                # without this a seeded object and a refined one would not be
                # byte-identical.
                g = to_multipolygon(repair_polygon(shape) or shape)
            except Exception:  # noqa: BLE001 -- fall through to the refine  # nosec B112
                continue
            if g is None or g.isEmpty():
                continue
            geoms[det_idx] = g
            seeded += 1
        return seeded

    def _compute_visible_objects(
        self, params: dict, pixel_size: float, with_scores: bool = False,
        refine_budget_s: float = 0.0, with_ids: bool = False,
    ) -> list | tuple[list, list] | tuple[list, list, list]:
        """Synchronous filter+refine of _auto_objects into the visible geom set:
        whole-object confidence + size filter, then the shape refine. Used by the
        headless path; the interactive path does the same cooperatively.
        ``with_scores`` also returns the parallel per-object score list, so the
        headless export can fill the layer's `score` field like the review path
        does (it silently exported NULL scores before).

        ``refine_budget_s`` bounds the time spent shaping objects that are NOT
        already in the reslice cache (see _RESCUE_REFINE_BUDGET_S). A cache hit
        is a dict lookup and never touches the budget, so the set the user was
        actually looking at always comes out fully shaped; only the cohort the
        Confidence cutoff had hidden can fall back to its traced outline. 0
        means no bound, which is what the headless path wants.

        ``with_ids`` also returns the parallel canonical det_id list, so a caller
        can tell two same-sized sets apart by identity instead of by count."""
        import time as _t

        removed = self._review_removed_fids()
        # Settle the cache's shape key BEFORE the loop so the peek below reads
        # the dict that _review_refined_geom will fill, not one it is about to
        # swap out.
        cache = self._auto_reslice_cache
        shape_key = self._review_shape_key(params, pixel_size)
        if cache.get("key") != shape_key:
            self._adopt_reslice_shape_key(cache, shape_key)
        cached_geoms = cache["geoms"]
        deadline = (_t.monotonic() + refine_budget_s) if refine_budget_s > 0 else None
        unshaped = 0
        unrepairable = 0
        out = []
        out_scores = []
        out_ids = []
        for det_idx, (base, score, area) in enumerate(self._auto_objects):
            if det_idx in removed or base is None or base.isEmpty():
                continue
            if not (self._object_is_manual(det_idx) or self._passes_review_filters(score, area, params)):
                continue
            if deadline is not None and det_idx not in cached_geoms and _t.monotonic() >= deadline:
                g = self._plain_outline_geom(base)
                unshaped += 1
            else:
                g = self._review_refined_geom(det_idx, base, params, pixel_size)
            if g is not None:
                out.append(g)
                out_scores.append(float(score))
                if with_ids:
                    out_ids.append(self._object_fid_for(det_idx))
            else:
                unrepairable += 1
        if unshaped:
            QgsMessageLog.logMessage(
                f"Auto review: rescue export ran out of its {refine_budget_s:.0f}s "
                f"shaping budget; {unshaped} hidden object(s) saved with their "
                f"traced outline", "AI Segmentation", level=Qgis.MessageLevel.Info)
        if unrepairable:
            # A base that will not repair is the one case that loses a billed
            # object, so it is counted and said rather than dropped in silence.
            QgsMessageLog.logMessage(
                f"Auto review: {unrepairable} object(s) left out; their geometry "
                f"could not be repaired", "AI Segmentation",
                level=Qgis.MessageLevel.Warning)
        # The set is complete here, which is the only point a whole-set
        # operation can run: shared borders needs every neighbour at once.
        out = self._apply_boundary_snap(out, params)
        if with_ids:
            return out, out_scores, out_ids
        if with_scores:
            return out, out_scores
        return out
