live_inference_worker.py

"""Latest-frame live segmentation worker for RF-DETR Seg and YOLO Seg models."""

from __future__ import annotations

import threading
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

import cv2
import numpy as np
from PySide6.QtCore import QThread, Signal

# When the optional runtime is installed, activate its RF-DETR superset before
# any standard or unified model is loaded. This makes switching models in one
# process deterministic; importing mamir itself is intentionally torch-free.
try:
    import mamir as _mamir_runtime  # noqa: F401
except ImportError:
    _mamir_runtime = None

from live_detection_types import LiveDetectionResult, PreviewFramePacket
from live_tracking import LiveIdentityTracker, MamirLiveIdentityTracker, compute_body_center
from mask_storage import (
    CropMask,
    from_crop_like,
    is_crop_packed,
    mask_has_pixels,
    mask_region,
    mask_shape,
    occupied_crop,
)
from mask_skeleton import MaskSkeletonExtractor, repair_hip_keypoints_with_mask_geometry
from runtime_performance import configure_torch_runtime
from torch_runtime import import_torch

_PATH_EDGE_QUOTES = "\"'" + "".join(chr(code) for code in (0x201C, 0x201D, 0x2018, 0x2019))


def _normalize_checkpoint_path(value: object) -> str:
    """Accept plain or quoted model paths pasted into the UI."""
    return str(value or "").strip().strip(_PATH_EDGE_QUOTES).strip()


RFDETR_SEG_CLASS_NAME_MAP = {
    "rfdetr-seg-nano": "RFDETRSegNano",
    "rfdetr-seg-small": "RFDETRSegSmall",
    "rfdetr-seg-medium": "RFDETRSegMedium",
    "rfdetr-seg-large": "RFDETRSegLarge",
    "rfdetr-seg-xlarge": "RFDETRSegXLarge",
    "rfdetr-seg-2xlarge": "RFDETRSeg2XLarge",
}
RFDETR_SEG_POSITION_LENGTH_TO_MODEL_KEY = {
    677: "rfdetr-seg-nano",
    1025: "rfdetr-seg-small",
    1297: "rfdetr-seg-medium",
    1765: "rfdetr-seg-large",
    2705: "rfdetr-seg-xlarge",
    4097: "rfdetr-seg-2xlarge",
}
MAMIR_UNIFIED_MODEL_KEY = "mamir-unified"

# Upper bound for RF-DETR PostProcess num_select. PostProcess interpolates one mask
# per selected candidate to the full output resolution every frame; for a few-animal
# tracker, capping the candidate count trims a large amount of wasted upsampling
# without affecting real (above-threshold) detections.
_RFDETR_MAX_NUM_SELECT = 20


def _state_dict_from_checkpoint_payload(payload: object) -> object:
    if isinstance(payload, dict):
        for key in ("model", "state_dict"):
            nested = payload.get(key)
            if isinstance(nested, dict):
                return nested
    return payload


def _checkpoint_tensor_shape(state_dict: object, key_suffix: str) -> tuple[int, ...] | None:
    if not hasattr(state_dict, "items"):
        return None
    for key, value in state_dict.items():
        normalized_key = str(key).removeprefix("module.").removeprefix("model.")
        if normalized_key == key_suffix or normalized_key.endswith(f".{key_suffix}"):
            shape = getattr(value, "shape", None)
            if shape is not None:
                return tuple(int(dim) for dim in shape)
    return None


def _infer_rfdetr_seg_checkpoint_metadata_from_state_dict(
    state_dict: object,
) -> tuple[str | None, int | None]:
    """Infer RF-DETR Seg variant and train resolution from checkpoint tensors."""
    position_shape = _checkpoint_tensor_shape(
        state_dict,
        "backbone.0.encoder.encoder.embeddings.position_embeddings",
    )
    resolution = None
    if position_shape and len(position_shape) >= 2:
        position_count = int(position_shape[1])
        model_key = RFDETR_SEG_POSITION_LENGTH_TO_MODEL_KEY.get(position_count)
        patch_grid = int(round(float(max(0, position_count - 1)) ** 0.5))
        if patch_grid > 0 and (patch_grid * patch_grid + 1) == position_count:
            resolution = patch_grid * 12
        if model_key:
            return model_key, resolution

    refpoint_shape = _checkpoint_tensor_shape(state_dict, "refpoint_embed.weight")
    if refpoint_shape and len(refpoint_shape) >= 1:
        query_count = int(refpoint_shape[0])
        if query_count == 1300:
            return "rfdetr-seg-small", resolution
        if query_count == 2600:
            return "rfdetr-seg-large", resolution
        if query_count == 3900:
            return "rfdetr-seg-xlarge", resolution
    return None, resolution


def _infer_rfdetr_seg_checkpoint_metadata(
    checkpoint_path: str,
    torch_module,
) -> tuple[str | None, int | None]:
    checkpoint = _normalize_checkpoint_path(checkpoint_path)
    if not checkpoint:
        return None, None
    path = Path(checkpoint)
    if not path.exists():
        return None, None
    payload = torch_module.load(str(path), map_location="cpu")
    state_dict = _state_dict_from_checkpoint_payload(payload)
    return _infer_rfdetr_seg_checkpoint_metadata_from_state_dict(state_dict)


def _running_from_pyinstaller_bundle() -> bool:
    return bool(getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"))


def _triton_nvidia_driver_source_available() -> bool:
    """Return whether torch.compile's Triton NVIDIA backend source is present."""
    try:
        import triton.backends.nvidia.driver as driver
    except Exception:
        return False
    try:
        driver_path = Path(getattr(driver, "__file__", "") or "")
    except Exception:
        return False
    return driver_path.is_file()


def _is_torchscript_source_lookup_error(exc: BaseException) -> bool:
    message = str(exc).lower()
    return "source" in message and (
        "can't get source" in message
        or "requires source access" in message
        or "could not get source code" in message
    )


def _patch_torchscript_source_lookup_for_pyinstaller(torch_module) -> None:
    """Let RF-DETR import in one-file bundles when TorchScript source is absent."""
    if not _running_from_pyinstaller_bundle():
        return
    jit_module = getattr(torch_module, "jit", None)
    script_fn = getattr(jit_module, "script", None)
    if script_fn is None or getattr(script_fn, "_pykaboo_source_fallback", False):
        return

    def _script_with_source_fallback(obj, *args, **kwargs):
        try:
            return script_fn(obj, *args, **kwargs)
        except Exception as exc:
            if _is_torchscript_source_lookup_error(exc):
                return obj
            raise

    _script_with_source_fallback._pykaboo_source_fallback = True
    jit_module.script = _script_with_source_fallback
    try:
        import torch.jit._script as jit_script_module

        jit_script_module.script = _script_with_source_fallback
    except Exception:
        pass

    script_if_tracing_fn = getattr(jit_module, "script_if_tracing", None)
    if script_if_tracing_fn is not None and not getattr(script_if_tracing_fn, "_pykaboo_source_fallback", False):

        def _script_if_tracing_with_source_fallback(obj, *args, **kwargs):
            try:
                return script_if_tracing_fn(obj, *args, **kwargs)
            except Exception as exc:
                if _is_torchscript_source_lookup_error(exc):
                    return obj
                raise

        _script_if_tracing_with_source_fallback._pykaboo_source_fallback = True
        jit_module.script_if_tracing = _script_if_tracing_with_source_fallback


def _patch_transformers_doc_source_lookup_for_pyinstaller() -> None:
    """Avoid RF-DETR import failures when transformers doc helpers cannot inspect frozen source."""
    if not _running_from_pyinstaller_bundle():
        return
    try:
        import transformers.utils.doc as transformers_doc
    except Exception:
        return

    original = getattr(transformers_doc, "get_docstring_indentation_level", None)
    if original is None or getattr(original, "_pykaboo_source_fallback", False):
        return

    def _get_docstring_indentation_level_with_fallback(func):
        try:
            return original(func)
        except Exception as exc:
            if _is_torchscript_source_lookup_error(exc):
                # The value is only used to re-indent generated docstrings.
                # Falling back to method indentation keeps model import safe.
                return 4
            raise

    _get_docstring_indentation_level_with_fallback._pykaboo_source_fallback = True
    transformers_doc.get_docstring_indentation_level = _get_docstring_indentation_level_with_fallback


@dataclass
class LiveInferenceConfig:
    model_key: str = "rfdetr-seg-medium"
    checkpoint_path: str = ""
    threshold: float = 0.35
    selected_class_ids: list[int] | None = None
    identity_mode: str = "tracker"
    expected_mouse_count: int = 1
    inference_max_width: int = 960
    # Optional YOLO pose model run on each segmentation bbox crop to attach
    # keypoints to detections without breaking identity tracking.
    keypoint_source: str = "yolo_pose"
    pose_checkpoint_path: str = ""
    pose_threshold: float = 0.25
    min_pose_keypoints: int = 0
    acceleration_mode: str = "auto"
    # Tracking mode runs the pose model on the full inference frame in a
    # parallel thread with segmentation, then matches poses to masks by IoU.
    # Combined latency becomes max(mask, pose) instead of mask + pose.
    tracking_mode: bool = False
    # Overlay-quality controls. ``clean_masks`` keeps one solid blob per
    # detection; ``clamp_pose_to_mask`` drops keypoints that fall outside the
    # body; ``smooth_keypoints`` removes per-frame jitter. ``pose_imgsz_cap``
    # bounds the full-frame pose inference size (smaller = faster).
    clean_masks: bool = True
    clamp_pose_to_mask: bool = True
    smooth_keypoints: bool = True
    pose_imgsz_cap: int = 640
    output_masks: bool = True
    # Realtime mask geometry can refresh different animals on alternating
    # inference frames. Cached keypoints are translated by the current tracked
    # center between full anatomy refreshes, so motion remains frame-rate smooth.
    geometry_refresh_interval: int = 1

    def normalized(self) -> "LiveInferenceConfig":
        acceleration_mode = str(self.acceleration_mode or "balanced").strip().lower().replace("-", "_").replace(" ", "_")
        if acceleration_mode not in {"auto", "balanced", "max_gpu", "max_gpu_trt", "compatibility"}:
            acceleration_mode = "auto"
        keypoint_source = str(self.keypoint_source or "yolo_pose").strip().lower().replace("-", "_").replace(" ", "_")
        keypoint_aliases = {
            "yolo": "yolo_pose",
            "pose": "yolo_pose",
            "yolo_pose": "yolo_pose",
            "mask": "mask_geometry",
            "geometry": "mask_geometry",
            "mask_pose": "mask_geometry",
            "mask_geometry": "mask_geometry",
            "none": "none",
            "off": "none",
        }
        keypoint_source = keypoint_aliases.get(keypoint_source, "yolo_pose")
        return LiveInferenceConfig(
            model_key=str(self.model_key or "rfdetr-seg-medium").strip(),
            checkpoint_path=_normalize_checkpoint_path(self.checkpoint_path),
            threshold=float(self.threshold),
            selected_class_ids=list(self.selected_class_ids or []),
            identity_mode=str(self.identity_mode or "tracker").strip().lower(),
            expected_mouse_count=max(1, int(self.expected_mouse_count or 1)),
            inference_max_width=max(0, int(self.inference_max_width or 0)),
            keypoint_source=keypoint_source,
            pose_checkpoint_path=_normalize_checkpoint_path(self.pose_checkpoint_path),
            pose_threshold=float(self.pose_threshold or 0.25),
            min_pose_keypoints=max(0, int(self.min_pose_keypoints or 0)),
            acceleration_mode=acceleration_mode,
            tracking_mode=bool(self.tracking_mode),
            clean_masks=bool(self.clean_masks),
            clamp_pose_to_mask=bool(self.clamp_pose_to_mask),
            smooth_keypoints=bool(self.smooth_keypoints),
            pose_imgsz_cap=max(256, int(self.pose_imgsz_cap or 640)),
            output_masks=bool(self.output_masks),
            geometry_refresh_interval=max(1, min(4, int(self.geometry_refresh_interval or 1))),
        )

    def signature(self) -> tuple:
        normalized = self.normalized()
        return (
            normalized.model_key,
            normalized.checkpoint_path,
            round(normalized.threshold, 4),
            tuple(normalized.selected_class_ids),
            normalized.identity_mode,
            normalized.expected_mouse_count,
            normalized.inference_max_width,
            normalized.keypoint_source,
            normalized.pose_checkpoint_path,
            round(normalized.pose_threshold, 4),
            normalized.min_pose_keypoints,
            normalized.acceleration_mode,
            normalized.output_masks,
        )


class LiveInferenceWorker(QThread):
    """Run live segmentation on the newest preview frame only."""

    result_ready = Signal(object)
    status_changed = Signal(str)
    error_occurred = Signal(str)

    def __init__(self, parent=None) -> None:
        super().__init__(parent)
        self._condition = threading.Condition()
        self._running = True
        self._active = False
        self._latest_packet: Optional[PreviewFramePacket] = None
        self._config = LiveInferenceConfig()
        self._model = None
        self._model_signature: Optional[tuple] = None
        # Signature of a model config that failed to load. While set, the run loop does
        # NOT re-attempt that exact load every frame (which floods the log and re-triggers
        # heavy weight downloads); it waits for the config to change (the user fixing the
        # checkpoint path). Cleared on each (re)arm and whenever the config changes.
        self._failed_model_signature: Optional[tuple] = None
        self._pose_model = None
        self._pose_model_signature: Optional[str] = None
        self._tracker = LiveIdentityTracker(expected_mice=1)
        self._tracker_model_key = ""
        self._rfdetr_direct_predict_enabled = True
        self._rfdetr_norm_cache: dict[tuple, tuple[object, object]] = {}
        self._rfdetr_scale_cache: dict[tuple, object] = {}
        self._pose_batch_predict_enabled = True
        self._pose_executor_instance = None
        self._unified_finalize_executor_instance = None
        self._pending_unified_finalize = None
        self._unified_finalize_lock = threading.Lock()
        self._stream_generation = 0
        self._mask_skeleton_extractor: MaskSkeletonExtractor | None = None
        self._mask_skeleton_signature: Optional[tuple] = None
        self._mask_geometry_cache: dict[int, tuple[np.ndarray, np.ndarray, tuple[float, float]]] = {}
        # Transient flips to apply once (a LIST, so two fast clicks really flip twice).
        self._pending_orientation_flips: list[int] = []
        # Net manual orientation per mouse (count mod 2): survives extractor rebuilds so a
        # user's head/tail correction is re-applied after a model reload / reconfigure.
        self._manual_orientation_flips: dict[int, int] = {}
        self._flip_applied_extractor: MaskSkeletonExtractor | None = None
        # Manual identity swaps requested from the GUI, drained on the inference
        # thread just before the next association so each pair of mice relabels live.
        self._pending_identity_swaps: list[tuple[int, int]] = []
        # Latest detailed timing sample for the realtime diagnostics harness.
        # A whole-dict assignment keeps readers from observing a partial update.
        self.postprocess_phase_ms: dict[str, float] = {}

    def request_flip_orientation(self, mouse_id: int) -> None:
        """Thread-safe request to manually swap a tracked mouse's head<->tail.

        Applied to the mask-skeleton extractor on the inference thread the next time
        that mouse is processed, so the user can correct a backwards orientation
        (e.g. a motionless subject seeded the wrong way) live from the GUI. The net
        correction is remembered and re-applied if the extractor is later rebuilt.
        """
        try:
            mid = int(mouse_id)
        except (TypeError, ValueError):
            return
        with self._condition:
            self._pending_orientation_flips.append(mid)
            self._manual_orientation_flips[mid] = (self._manual_orientation_flips.get(mid, 0) + 1) % 2
            self._condition.notify_all()

    def request_swap_identities(self, id_a: int, id_b: int) -> None:
        """Thread-safe request to swap two mice's identities live.

        Queued and applied on the inference thread just before the next frame's
        association, so the tracker (and the mask-skeleton orientation state)
        relabel each animal as the other from that frame on. Use it to undo an
        identity switch where the tracker crossed the two mice up.
        """
        try:
            a = int(id_a)
            b = int(id_b)
        except (TypeError, ValueError):
            return
        if a == b:
            return
        with self._condition:
            self._pending_identity_swaps.append((a, b))
            self._condition.notify_all()

    def _apply_pending_identity_swaps(self) -> None:
        """Drain and apply queued identity swaps to the tracker + skeleton state.

        Runs on the inference thread before association: swapping the tracker's
        remembered per-track state makes the next detections adopt the swapped
        labels, and swapping the extractor state carries each animal's settled
        head/tail lock across with it.
        """
        with self._condition:
            if not self._pending_identity_swaps:
                return
            swaps = self._pending_identity_swaps
            self._pending_identity_swaps = []
        for a, b in swaps:
            try:
                self._tracker.swap_identities(a, b)
            except Exception:
                pass
            extractor = self._mask_skeleton_extractor
            if extractor is not None:
                try:
                    extractor.swap_tracks(a, b)
                except Exception:
                    pass
            cached_a = self._mask_geometry_cache.pop(a, None)
            cached_b = self._mask_geometry_cache.pop(b, None)
            if cached_b is not None:
                self._mask_geometry_cache[a] = cached_b
            if cached_a is not None:
                self._mask_geometry_cache[b] = cached_a

    def start_inference(self, config: LiveInferenceConfig) -> None:
        self._wait_for_unified_finalize()
        normalized = config.normalized()
        with self._condition:
            self._stream_generation += 1
            self._config = normalized
            self._active = True
            self._select_tracker(normalized)
            self._tracker.reset(expected_mice=normalized.expected_mouse_count)
            self._reset_mask_skeleton_extractor()
            # Keep _manual_orientation_flips: a user's correction should survive a
            # reconfigure; the persistent set is re-applied to the rebuilt extractor.
            self._pending_orientation_flips = []
            self._failed_model_signature = None  # allow a fresh load attempt on re-arm
            self._latest_packet = None
            self._condition.notify_all()
        if not self.isRunning():
            self.start()
        self.status_changed.emit("Live inference armed")

    def stop_inference(self) -> None:
        with self._condition:
            self._stream_generation += 1
            self._active = False
            self._latest_packet = None
            self._condition.notify_all()
        self._wait_for_unified_finalize()
        self.status_changed.emit("Live inference stopped")

    def submit_preview(self, packet: object) -> None:
        if not isinstance(packet, PreviewFramePacket):
            return
        with self._condition:
            if not self._active or not self._running:
                return
            self._latest_packet = packet
            self._condition.notify_all()

    def shutdown(self) -> None:
        with self._condition:
            self._stream_generation += 1
            self._running = False
            self._active = False
            self._latest_packet = None
            self._condition.notify_all()
        self.wait(5000)
        self._wait_for_unified_finalize()
        if self._unified_finalize_executor_instance is not None:
            try:
                self._unified_finalize_executor_instance.shutdown(wait=True)
            except Exception:
                pass
            self._unified_finalize_executor_instance = None
        if self._pose_executor_instance is not None:
            try:
                self._pose_executor_instance.shutdown(wait=False)
            except Exception:
                pass
            self._pose_executor_instance = None
        self._release_accelerator_memory(self._model)
        self._model = None
        self._release_accelerator_memory(self._pose_model)
        self._pose_model = None
        self._pose_model_signature = None
        self._reset_mask_skeleton_extractor()

    def run(self) -> None:
        # Keep inference at normal priority. Demoting this thread starves Python
        # mask cleanup and geometry pose while camera conversion, overlay encoding,
        # and audio are active. The preview already has an independent, throttled
        # feed, so low priority makes overlays visibly stale without protecting
        # capture or recording throughput.
        try:
            self.setPriority(QThread.NormalPriority)
        except Exception:
            pass
        while True:
            with self._condition:
                while self._running and (not self._active or self._latest_packet is None):
                    self._condition.wait(timeout=0.2)
                if not self._running:
                    break
                packet = self._latest_packet
                self._latest_packet = None
                config = self._config.normalized()
                stream_generation = self._stream_generation

            try:
                signature = config.signature()
                if signature != self._model_signature or self._model is None:
                    self._wait_for_unified_finalize()
                    if signature == self._failed_model_signature:
                        # This exact model config already failed to load; do not retry
                        # every frame (it re-runs heavy weight downloads and floods the
                        # log). Wait for the config to change (the user fixing the path).
                        continue
                    try:
                        self.status_changed.emit(f"Loading {config.model_key} model")
                        self._release_accelerator_memory(self._model)
                        self._model = self._load_model(
                            config.model_key,
                            config.checkpoint_path,
                            acceleration_mode=config.acceleration_mode,
                        )
                    except Exception as load_exc:
                        self._failed_model_signature = signature
                        self._model = None
                        self._model_signature = None
                        self.error_occurred.emit(
                            f"Could not load model '{config.model_key}': {load_exc}. "
                            f"Check the checkpoint path; inference is paused until it is fixed."
                        )
                        continue
                    self._model_signature = signature
                    self._failed_model_signature = None
                    self._rfdetr_direct_predict_enabled = True
                    self._select_tracker(config)
                    self._tracker.reset(expected_mice=config.expected_mouse_count)
                    self.status_changed.emit(
                        f"Model ready: {self._describe_loaded_model(self._model, config.model_key, config.acceleration_mode)}"
                    )

                desired_pose_signature = config.pose_checkpoint_path if config.keypoint_source == "yolo_pose" else ""
                if desired_pose_signature != (self._pose_model_signature or ""):
                    self._release_accelerator_memory(self._pose_model)
                    self._pose_model = None
                    self._pose_model_signature = None
                    self._pose_batch_predict_enabled = True
                    if desired_pose_signature:
                        self.status_changed.emit("Loading pose checkpoint")
                        try:
                            self._pose_model = self._load_pose_model(
                                desired_pose_signature,
                                acceleration_mode=config.acceleration_mode,
                            )
                            self._pose_model_signature = desired_pose_signature
                            self.status_changed.emit("Pose model ready")
                        except Exception as exc:
                            self._pose_model = None
                            self._pose_model_signature = None
                            self.error_occurred.emit(f"Pose model load failed: {exc}")

                start_perf = time.perf_counter()
                start_wall = time.time()
                frame_rgb = self._ensure_rgb(packet.frame)
                inference_frame, scale_x, scale_y = self._prepare_inference_frame(
                    frame_rgb,
                    config.inference_max_width,
                )
                output_width, output_height, output_scale_x, output_scale_y = self._output_geometry(
                    packet,
                    frame_rgb.shape,
                    scale_x,
                    scale_y,
                )
                preprocess_ms = (time.perf_counter() - start_perf) * 1000.0

                # Tracking mode: launch full-frame pose inference in parallel
                # with segmentation so combined latency is max(), not sum().
                pose_future = None
                if (
                    config.tracking_mode
                    and config.keypoint_source == "yolo_pose"
                    and self._pose_model is not None
                ):
                    pose_future = self._pose_executor().submit(
                        self._predict_pose_fullframe,
                        inference_frame,
                        config.pose_threshold,
                        config.pose_imgsz_cap,
                    )

                predict_start_perf = time.perf_counter()
                if config.model_key == MAMIR_UNIFIED_MODEL_KEY:
                    self._model.topk = min(10, max(2, int(config.expected_mouse_count) + 2))
                detections = self._predict(
                    self._model,
                    config.model_key,
                    inference_frame,
                    config.threshold,
                )
                predict_ms = (time.perf_counter() - predict_start_perf) * 1000.0

                if (
                    config.model_key == MAMIR_UNIFIED_MODEL_KEY
                    and config.keypoint_source == "mask_geometry"
                ):
                    # Preserve latest-frame semantics across stop/re-arm while a
                    # GPU forward is in flight. A stale frame must not update the
                    # newly reset causal tracker.
                    with self._condition:
                        stream_is_current = (
                            self._running
                            and self._active
                            and stream_generation == self._stream_generation
                        )
                    if not stream_is_current:
                        continue

                    # Prediction for this frame has already overlapped the prior
                    # frame's CPU finalization. Retire that one ordered slot, then
                    # schedule this result. No scientific frame backlog is built.
                    self._wait_for_unified_finalize()
                    inference_height, inference_width = inference_frame.shape[:2]
                    with self._condition:
                        followup_ready = self._latest_packet is not None
                    finalize_kwargs = {
                        "detections": detections,
                        "config": config,
                        "packet": packet,
                        "output_width": int(output_width),
                        "output_height": int(output_height),
                        "output_scale_x": float(output_scale_x),
                        "output_scale_y": float(output_scale_y),
                        "inference_width": int(inference_width),
                        "inference_height": int(inference_height),
                        "start_perf": float(start_perf),
                        "start_wall": float(start_wall),
                        "preprocess_ms": float(preprocess_ms),
                        "predict_ms": float(predict_ms),
                    }
                    if not followup_ready:
                        # A paced source has no next GPU job to overlap. Finish
                        # inline to retain the minimum single-frame latency and
                        # avoid paying an executor hand-off for no throughput gain.
                        self._finalize_unified_geometry_result(**finalize_kwargs)
                        continue
                    future = self._unified_finalize_executor().submit(
                        self._finalize_unified_geometry_result,
                        **finalize_kwargs,
                    )
                    with self._unified_finalize_lock:
                        self._pending_unified_finalize = future
                    future.add_done_callback(
                        self._report_unified_finalize_failure
                    )
                    continue

                postprocess_start_perf = time.perf_counter()
                normalized = self._normalize_detections(detections)
                normalize_end_perf = time.perf_counter()
                # Apply any queued manual identity swaps before association so the
                # relabel takes effect from this frame on (both tracker branches).
                self._apply_pending_identity_swaps()
                if config.keypoint_source == "mask_geometry":
                    records = self._build_detection_records(normalized, config)
                    records_end_perf = time.perf_counter()
                    self._clean_record_masks(records, config)
                    clean_end_perf = time.perf_counter()
                    self._tracker.smooth_keypoints_enabled = bool(config.smooth_keypoints)
                    if config.identity_mode == "model_class":
                        tracked = self._tracker.assign_by_model_class(records, config.selected_class_ids or [])
                    else:
                        tracked = self._tracker.update(records)
                    track_end_perf = time.perf_counter()
                    self._attach_mask_skeleton_keypoints(
                        tracked,
                        config,
                        frame_index=int(packet.frame_index),
                    )
                    geometry_end_perf = time.perf_counter()
                    tracked = self._scale_tracked_states(
                        tracked,
                        scale_x=output_scale_x,
                        scale_y=output_scale_y,
                        keep_masks=bool(config.output_masks),
                        output_shape=(output_height, output_width),
                    )
                    scale_end_perf = time.perf_counter()
                    self.postprocess_phase_ms = {
                        "normalize": (normalize_end_perf - postprocess_start_perf) * 1000.0,
                        "records": (records_end_perf - normalize_end_perf) * 1000.0,
                        "clean_masks": (clean_end_perf - records_end_perf) * 1000.0,
                        "track": (track_end_perf - clean_end_perf) * 1000.0,
                        "geometry": (geometry_end_perf - track_end_perf) * 1000.0,
                        "scale": (scale_end_perf - geometry_end_perf) * 1000.0,
                    }
                else:
                    if output_scale_x != 1.0 or output_scale_y != 1.0:
                        normalized = self._rescale_detections(
                            normalized,
                            frame_shape=(output_height, output_width),
                            scale_x=output_scale_x,
                            scale_y=output_scale_y,
                        )
                    records = self._build_detection_records(normalized, config)
                    self._clean_record_masks(records, config)

                    if pose_future is not None:
                        try:
                            pose_result = pose_future.result(timeout=5.0)
                        except Exception as exc:
                            pose_result = None
                            self.status_changed.emit(f"Parallel pose inference failed: {exc}")
                        if records and pose_result is not None:
                            self._attach_pose_keypoints_fullframe(
                                records,
                                pose_result,
                                scale_x=output_scale_x,
                                scale_y=output_scale_y,
                                pose_threshold=config.pose_threshold,
                                min_confident_kp=config.min_pose_keypoints,
                            )
                    elif config.keypoint_source == "yolo_pose" and self._pose_model is not None and records:
                        # frame_rgb is the camera's downscaled inference frame, but
                        # records are in output (source-frame) space. Tell the pose
                        # attach how to map record coords onto frame_rgb for cropping
                        # so keypoints come back in output space (otherwise they fall
                        # outside the mask and clamping deletes them).
                        frame_h, frame_w = frame_rgb.shape[:2]
                        record_to_frame_scale = (
                            float(frame_w) / float(max(1, output_width)),
                            float(frame_h) / float(max(1, output_height)),
                        )
                        self._attach_pose_keypoints_in_bboxes(
                            frame_rgb,
                            records,
                            pose_threshold=config.pose_threshold,
                            min_confident_kp=config.min_pose_keypoints,
                            record_to_frame_scale=record_to_frame_scale,
                        )

                    self._clamp_record_keypoints(records, config)
                    self._tracker.smooth_keypoints_enabled = bool(config.smooth_keypoints)

                    if config.identity_mode == "model_class":
                        tracked = self._tracker.assign_by_model_class(records, config.selected_class_ids or [])
                    else:
                        tracked = self._tracker.update(records)
                    if config.keypoint_source == "mask_geometry":
                        self._attach_mask_skeleton_keypoints(tracked, config)
                    self.postprocess_phase_ms = {
                        "normalize": (normalize_end_perf - postprocess_start_perf) * 1000.0,
                        "other": (time.perf_counter() - normalize_end_perf) * 1000.0,
                    }
                postprocess_ms = (time.perf_counter() - postprocess_start_perf) * 1000.0

                completed_timestamp_s = time.time()
                inference_ms = (time.perf_counter() - start_perf) * 1000.0
                queue_wait_ms = max(0.0, (start_wall - float(packet.timestamp_s)) * 1000.0)
                end_to_end_ms = max(0.0, (completed_timestamp_s - float(packet.timestamp_s)) * 1000.0)
                inference_height, inference_width = inference_frame.shape[:2]
                self.result_ready.emit(
                    LiveDetectionResult(
                        frame_index=packet.frame_index,
                        timestamp_s=packet.timestamp_s,
                        width=int(output_width),
                        height=int(output_height),
                        inference_ms=float(inference_ms),
                        tracked_mice=tracked,
                        model_key=config.model_key,
                        status="ok",
                        predict_ms=float(predict_ms),
                        preprocess_ms=float(preprocess_ms),
                        postprocess_ms=float(postprocess_ms),
                        queue_wait_ms=float(queue_wait_ms),
                        end_to_end_ms=float(end_to_end_ms),
                        completed_timestamp_s=float(completed_timestamp_s),
                        inference_width=int(inference_width),
                        inference_height=int(inference_height),
                        runtime_provenance=self._model_runtime_provenance(),
                        identity_switch_alerts=list(self._tracker.last_identity_switch_alerts),
                    )
                )
            except Exception as exc:
                self.error_occurred.emit(f"Live inference error: {str(exc)}")

    def _finalize_unified_geometry_result(
        self,
        *,
        detections,
        config: LiveInferenceConfig,
        packet: PreviewFramePacket,
        output_width: int,
        output_height: int,
        output_scale_x: float,
        output_scale_y: float,
        inference_width: int,
        inference_height: int,
        start_perf: float,
        start_wall: float,
        preprocess_ms: float,
        predict_ms: float,
    ) -> LiveDetectionResult:
        """Finish one Unified frame in strict causal order and emit its result.

        Only this single-worker stage mutates identity and geometry state. The
        inference thread may run the next GPU forward concurrently, but a second
        finalization is never queued until this one has completed.
        """
        postprocess_start_perf = time.perf_counter()
        normalized = self._normalize_detections(detections)
        normalize_end_perf = time.perf_counter()
        self._apply_pending_identity_swaps()
        records = self._build_detection_records(normalized, config)
        records_end_perf = time.perf_counter()
        self._clean_record_masks(records, config)
        clean_end_perf = time.perf_counter()
        self._tracker.smooth_keypoints_enabled = bool(config.smooth_keypoints)
        if config.identity_mode == "model_class":
            tracked = self._tracker.assign_by_model_class(
                records,
                config.selected_class_ids or [],
            )
        else:
            tracked = self._tracker.update(records)
        track_end_perf = time.perf_counter()
        self._attach_mask_skeleton_keypoints(
            tracked,
            config,
            frame_index=int(packet.frame_index),
        )
        geometry_end_perf = time.perf_counter()
        tracked = self._scale_tracked_states(
            tracked,
            scale_x=output_scale_x,
            scale_y=output_scale_y,
            keep_masks=bool(config.output_masks),
            output_shape=(output_height, output_width),
        )
        scale_end_perf = time.perf_counter()
        self.postprocess_phase_ms = {
            "normalize": (normalize_end_perf - postprocess_start_perf) * 1000.0,
            "records": (records_end_perf - normalize_end_perf) * 1000.0,
            "clean_masks": (clean_end_perf - records_end_perf) * 1000.0,
            "track": (track_end_perf - clean_end_perf) * 1000.0,
            "geometry": (geometry_end_perf - track_end_perf) * 1000.0,
            "scale": (scale_end_perf - geometry_end_perf) * 1000.0,
        }
        postprocess_ms = (scale_end_perf - postprocess_start_perf) * 1000.0
        completed_timestamp_s = time.time()
        result = LiveDetectionResult(
            frame_index=packet.frame_index,
            timestamp_s=packet.timestamp_s,
            width=int(output_width),
            height=int(output_height),
            inference_ms=float((time.perf_counter() - start_perf) * 1000.0),
            tracked_mice=tracked,
            model_key=config.model_key,
            status="ok",
            predict_ms=float(predict_ms),
            preprocess_ms=float(preprocess_ms),
            postprocess_ms=float(postprocess_ms),
            queue_wait_ms=float(
                max(0.0, (start_wall - float(packet.timestamp_s)) * 1000.0)
            ),
            end_to_end_ms=float(
                max(0.0, (completed_timestamp_s - float(packet.timestamp_s)) * 1000.0)
            ),
            completed_timestamp_s=float(completed_timestamp_s),
            inference_width=int(inference_width),
            inference_height=int(inference_height),
            runtime_provenance=self._model_runtime_provenance(),
            identity_switch_alerts=list(self._tracker.last_identity_switch_alerts),
        )
        self.result_ready.emit(result)
        return result

    def _select_tracker(self, config: LiveInferenceConfig) -> None:
        """Use MAMIR's causal amodal association only for the unified model."""
        model_key = str(config.model_key or "")
        if model_key == self._tracker_model_key:
            return
        if model_key == MAMIR_UNIFIED_MODEL_KEY:
            self._tracker = MamirLiveIdentityTracker(expected_mice=config.expected_mouse_count)
        else:
            self._tracker = LiveIdentityTracker(expected_mice=config.expected_mouse_count)
        self._tracker_model_key = model_key

    def _model_runtime_provenance(self) -> dict[str, object]:
        """Return immutable scientific/runtime identity for the loaded model."""
        payload = getattr(self._model, "provenance", None)
        return dict(payload) if isinstance(payload, dict) else {}

    def _ensure_rgb(self, frame: np.ndarray) -> np.ndarray:
        if frame.ndim == 2:
            return cv2.cvtColor(frame, cv2.COLOR_GRAY2RGB)
        return np.asarray(frame)

    def _prepare_inference_frame(
        self,
        frame_rgb: np.ndarray,
        max_width: int,
    ) -> tuple[np.ndarray, float, float]:
        target_width = max(0, int(max_width or 0))
        height, width = frame_rgb.shape[:2]
        if target_width <= 0 or width <= target_width:
            return frame_rgb, 1.0, 1.0

        scale = target_width / float(width)
        target_height = max(1, int(round(height * scale)))
        resized = cv2.resize(frame_rgb, (target_width, target_height), interpolation=cv2.INTER_LINEAR)
        return resized, width / float(target_width), height / float(target_height)

    def _output_geometry(
        self,
        packet: PreviewFramePacket,
        frame_shape: tuple[int, ...],
        scale_x: float,
        scale_y: float,
    ) -> tuple[int, int, float, float]:
        """Return output size and inference-to-output coordinate scale."""
        frame_height = int(frame_shape[0])
        frame_width = int(frame_shape[1])
        metadata = getattr(packet, "metadata", {}) or {}
        output_width = self._metadata_int(
            metadata,
            ("source_frame_width", "record_frame_width", "original_width"),
            fallback=int(getattr(packet, "width", frame_width) or frame_width),
        )
        output_height = self._metadata_int(
            metadata,
            ("source_frame_height", "record_frame_height", "original_height"),
            fallback=int(getattr(packet, "height", frame_height) or frame_height),
        )
        output_scale_x = float(scale_x) * (float(output_width) / max(1.0, float(frame_width)))
        output_scale_y = float(scale_y) * (float(output_height) / max(1.0, float(frame_height)))
        return int(output_width), int(output_height), float(output_scale_x), float(output_scale_y)

    @staticmethod
    def _metadata_int(metadata: dict, keys: tuple[str, ...], *, fallback: int) -> int:
        for key in keys:
            try:
                value = int(metadata.get(key, 0) or 0)
            except Exception:
                value = 0
            if value > 0:
                return value
        return int(fallback)

    def _scale_tracked_states(
        self,
        tracked_mice: list,
        *,
        scale_x: float,
        scale_y: float,
        keep_masks: bool,
        output_shape: Optional[tuple[int, int]] = None,
    ) -> list:
        if not tracked_mice:
            return tracked_mice
        sx = float(scale_x)
        sy = float(scale_y)
        if sx == 1.0 and sy == 1.0 and keep_masks:
            return tracked_mice
        for mouse in tracked_mice:
            source_bbox = tuple(float(value) for value in getattr(mouse, "bbox", (0.0, 0.0, 0.0, 0.0)))
            if keep_masks and output_shape is not None and getattr(mouse, "mask", None) is not None:
                mouse.mask = self._resize_instance_mask_roi(
                    mouse.mask,
                    frame_shape=(int(output_shape[0]), int(output_shape[1])),
                    scale_x=sx,
                    scale_y=sy,
                    source_bbox=np.asarray(source_bbox, dtype=float),
                )
            try:
                cx, cy = mouse.center
                mouse.center = (float(cx) * sx, float(cy) * sy)
            except Exception:
                pass
            try:
                x1, y1, x2, y2 = source_bbox
                mouse.bbox = (float(x1) * sx, float(y1) * sy, float(x2) * sx, float(y2) * sy)
            except Exception:
                pass
            keypoints = getattr(mouse, "keypoints", None)
            if keypoints is not None:
                kp = np.asarray(keypoints, dtype=float).reshape(-1, 2).copy()
                kp[:, 0] *= sx
                kp[:, 1] *= sy
                mouse.keypoints = kp
            if not keep_masks:
                mouse.mask = None
        return tracked_mice

    @staticmethod
    def _resolve_engine_only_seg(checkpoint: str) -> Optional[str]:
        """Return an engine path when only a .engine is available (no .pth weights).

        - checkpoint is a ``.engine`` that exists -> use it directly.
        - checkpoint is a ``.pth``/``.pt`` that is MISSING but a sibling ``.engine``
          exists -> use the sibling.
        Otherwise return None (the normal torch checkpoint path is used).
        """
        path = str(checkpoint or "").strip()
        if not path:
            return None
        candidate = Path(path)
        if candidate.suffix.lower() == ".engine":
            return str(candidate) if candidate.is_file() else None
        if not candidate.is_file():
            sibling = candidate.with_suffix(".engine")
            if sibling.is_file():
                return str(sibling)
        return None

    @staticmethod
    def _resolve_preferred_rfdetr_engine(checkpoint: str, model_key: str) -> Optional[str]:
        """Prefer a compatible TensorRT RF-DETR engine for live CUDA inference."""
        checkpoint_path = Path(str(checkpoint or "").strip())
        candidates: list[Path] = []
        if str(checkpoint or "").strip():
            if checkpoint_path.suffix.lower() == ".engine":
                candidates.append(checkpoint_path)
            else:
                candidates.append(checkpoint_path.with_suffix(".engine"))

        if str(model_key or "").startswith("rfdetr"):
            bundled = Path(__file__).resolve().parent / "models" / "checkpoint_best_total.engine"
            candidates.append(bundled)

        seen: set[str] = set()
        for candidate in candidates:
            try:
                resolved = str(candidate.resolve())
            except Exception:
                resolved = str(candidate)
            if resolved in seen or not candidate.is_file():
                continue
            seen.add(resolved)
            if LiveInferenceWorker._rfdetr_engine_matches_checkpoint(candidate, checkpoint_path, model_key):
                return str(candidate)
        return None

    @staticmethod
    def _rfdetr_engine_matches_checkpoint(engine_path: Path, checkpoint_path: Path, model_key: str) -> bool:
        """Return True when engine metadata is compatible with the selected checkpoint."""
        if engine_path.suffix.lower() == ".engine" and checkpoint_path.suffix.lower() == ".engine":
            return True

        meta_path = engine_path.with_suffix(".engine.json")
        metadata: dict = {}
        if meta_path.is_file():
            try:
                import json

                payload = json.loads(meta_path.read_text(encoding="utf-8"))
                if isinstance(payload, dict):
                    metadata = payload
            except Exception:
                metadata = {}

        meta_model = str(metadata.get("model_key", "") or "").strip()
        if meta_model and str(model_key or "").strip() and meta_model != str(model_key or "").strip():
            return False

        selected = str(checkpoint_path).replace("\\", "/").lower()
        built_from = str(metadata.get("checkpoint_path", "") or "").replace("\\", "/").lower()
        if selected and built_from and selected == built_from:
            return True

        if checkpoint_path.name and engine_path.stem == checkpoint_path.stem:
            return True

        # The bundled default engine was built from the production medium checkpoint.
        if engine_path.name == "checkpoint_best_total.engine" and checkpoint_path.name == "checkpoint_best_total.pth":
            return True

        return not selected and bool(metadata)

    def _load_model(self, model_key: str, checkpoint: str, *, acceleration_mode: str = "balanced"):
        torch = import_torch()
        configure_torch_runtime(torch, gpu_active=bool(torch.cuda.is_available()))
        self._configure_torch_acceleration(torch, acceleration_mode)
        _patch_torchscript_source_lookup_for_pyinstaller(torch)

        if str(model_key) == MAMIR_UNIFIED_MODEL_KEY:
            from mamir_unified_runtime import MamirUnifiedLiveModel

            if not checkpoint:
                raise ValueError(
                    "Choose the MAMIR unified model "
                    "(a compatible Unified .pth or .unified.engine)."
                )
            return MamirUnifiedLiveModel(
                checkpoint,
                acceleration_mode=acceleration_mode,
                status_callback=self.status_changed.emit,
            )

        if str(model_key).startswith("rfdetr"):
            _patch_transformers_doc_source_lookup_for_pyinstaller()

            # Engine-only path: when there are no .pth weights (only a sidecar
            # .engine, e.g. the shipped default), run the engine standalone with a
            # weightless PostProcess instead of building the torch model.
            engine_only = (
                self._resolve_preferred_rfdetr_engine(checkpoint, model_key)
                if torch.cuda.is_available()
                else self._resolve_engine_only_seg(checkpoint)
            )
            if engine_only is not None:
                if not torch.cuda.is_available():
                    raise RuntimeError(
                        "This RF-DETR model ships only as a TensorRT .engine, which needs a CUDA GPU. "
                        "Provide the .pth checkpoint to run on CPU/other GPUs."
                    )
                import rfdetr_trt

                model = rfdetr_trt.RFDETRSegEngineModel(engine_only, device="cuda:0")
                self.status_changed.emit(
                    f"RF-DETR TensorRT engine ready ({model._optimized_resolution}px, fp16 compute; engine-only)"
                )
                return model

            import rfdetr as rfdetr_module

            checkpoint_model_key = None
            checkpoint_resolution = None
            if checkpoint:
                try:
                    checkpoint_model_key, checkpoint_resolution = _infer_rfdetr_seg_checkpoint_metadata(
                        checkpoint,
                        torch,
                    )
                except Exception as exc:
                    self.status_changed.emit(f"RF-DETR checkpoint inspection skipped: {exc}")
                if checkpoint_model_key and checkpoint_model_key != model_key:
                    self.status_changed.emit(
                        f"RF-DETR checkpoint matches {checkpoint_model_key}; "
                        f"using it instead of selected {model_key}."
                    )
                    model_key = checkpoint_model_key

            default_class_name = RFDETR_SEG_CLASS_NAME_MAP["rfdetr-seg-medium"]
            class_name = RFDETR_SEG_CLASS_NAME_MAP.get(model_key, default_class_name)
            model_cls = getattr(rfdetr_module, class_name, None)
            if model_cls is None:
                raise RuntimeError(
                    f"The installed rfdetr package does not expose {class_name}. "
                    "Upgrade rfdetr to use this live inference model."
                )
            if not checkpoint:
                model = model_cls()
                if hasattr(model, "eval"):
                    model.eval()
                model = self._align_rfdetr_postprocess_num_select(model)
                return self._optimize_loaded_model(
                model, model_key, acceleration_mode=acceleration_mode, checkpoint=checkpoint
            )
            architecture_kwargs = {}
            if checkpoint_resolution:
                architecture_kwargs["resolution"] = int(checkpoint_resolution)
            for key in ("pretrain_weights", "checkpoint_path", "weights"):
                try:
                    model = model_cls(**architecture_kwargs, **{key: checkpoint})
                    if hasattr(model, "eval"):
                        model.eval()
                    model = self._align_rfdetr_postprocess_num_select(model)
                    return self._optimize_loaded_model(
                model, model_key, acceleration_mode=acceleration_mode, checkpoint=checkpoint
            )
                except TypeError:
                    continue
            model = model_cls(**architecture_kwargs)
            if hasattr(model, "eval"):
                model.eval()
            model = self._align_rfdetr_postprocess_num_select(model)
            return self._optimize_loaded_model(
                model, model_key, acceleration_mode=acceleration_mode, checkpoint=checkpoint
            )

        from ultralytics import YOLO

        weight_path = checkpoint or "yolo11n-seg.pt"
        model = YOLO(weight_path)
        if hasattr(model, "model") and hasattr(model.model, "eval"):
            model.model.eval()
        return model

    def _load_pose_model(self, checkpoint: str, *, acceleration_mode: str = "balanced"):
        """Load a YOLO pose checkpoint for paired keypoint inference."""
        torch = import_torch()
        configure_torch_runtime(torch, gpu_active=bool(torch.cuda.is_available()))
        self._configure_torch_acceleration(torch, acceleration_mode)
        from ultralytics import YOLO

        path = str(checkpoint or "").strip()
        if not path:
            raise ValueError("Pose checkpoint path is empty")

        # In TensorRT mode, prefer a prebuilt .engine sitting next to the .pt
        # (ultralytics loads engines transparently). Falls back to the .pt when
        # no engine is present. Build one with scripts/build_yolo_pose_engine.py.
        mode = str(acceleration_mode or "balanced").strip().lower().replace("-", "_").replace(" ", "_")
        if mode == "max_gpu_trt":
            from pathlib import Path as _Path

            engine_candidate = _Path(path).with_suffix(".engine")
            if engine_candidate.is_file():
                path = str(engine_candidate)
                self.status_changed.emit(f"Pose: using TensorRT engine {engine_candidate.name}")
            else:
                self.status_changed.emit(
                    f"Pose: no TensorRT engine at {engine_candidate.name}; using {_Path(path).name}. "
                    "Build one with scripts/build_yolo_pose_engine.py."
                )

        # A TensorRT .engine carries no reliable task tag, so ultralytics assumes
        # 'detect' and never produces keypoints. Force the pose task explicitly.
        if path.lower().endswith(".engine"):
            model = YOLO(path, task="pose")
        else:
            model = YOLO(path)
        if hasattr(model, "model") and hasattr(model.model, "eval"):
            model.model.eval()
        return model

    def _attach_pose_keypoints_in_bboxes(
        self,
        frame_rgb: np.ndarray,
        records: list[dict],
        *,
        pose_threshold: float,
        min_confident_kp: int,
        record_to_frame_scale: tuple[float, float] = (1.0, 1.0),
    ) -> None:
        """Run the YOLO pose model on each detection's bbox crop, in place.

        Each record gains ``keypoints`` (Kx2, image-space pixels) and
        ``keypoint_scores`` (K,) when the pose model produces a detection
        inside the crop. Records that fail the ``min_confident_kp`` gate keep
        their seg-only state untouched (the seg detection is still emitted —
        we never drop identity for missing keypoints).
        """
        if self._pose_model is None or not records:
            return

        torch = import_torch(required=False)

        height, width = frame_rgb.shape[:2]
        # Records are in output (source-frame) space; frame_rgb is the downscaled
        # inference frame. Map record coords into frame_rgb space for cropping, and
        # map keypoints back to output space when storing them on the record.
        frame_scale_x = float(record_to_frame_scale[0]) or 1.0
        frame_scale_y = float(record_to_frame_scale[1]) or 1.0
        crop_jobs: list[dict] = []
        for record in records:
            try:
                bbox = record.get("bbox")
                if bbox is None:
                    continue
                x1, y1, x2, y2 = (float(value) for value in bbox)
                x1 *= frame_scale_x
                x2 *= frame_scale_x
                y1 *= frame_scale_y
                y2 *= frame_scale_y
                bw = max(1.0, x2 - x1)
                bh = max(1.0, y2 - y1)
                pad_x = max(12.0, bw * 0.12)
                pad_y = max(12.0, bh * 0.12)
                mask = record.get("mask")
                if mask_has_pixels(mask) and mask_shape(mask) == frame_rgb.shape[:2]:
                    mask_crop, mask_x, mask_y = occupied_crop(mask)
                    if mask_crop is not None:
                        x1 = min(x1, float(mask_x))
                        y1 = min(y1, float(mask_y))
                        x2 = max(x2, float(mask_x + mask_crop.shape[1]))
                        y2 = max(y2, float(mask_y + mask_crop.shape[0]))
                        bw = max(1.0, x2 - x1)
                        bh = max(1.0, y2 - y1)
                        pad_x = max(pad_x, bw * 0.08)
                        pad_y = max(pad_y, bh * 0.08)
                cx1 = max(0, int(round(x1 - pad_x)))
                cy1 = max(0, int(round(y1 - pad_y)))
                cx2 = min(width, int(round(x2 + pad_x)))
                cy2 = min(height, int(round(y2 + pad_y)))
                if cx2 - cx1 < 4 or cy2 - cy1 < 4:
                    continue
                crop_jobs.append(
                    {
                        "record": record,
                        "crop": frame_rgb[cy1:cy2, cx1:cx2],
                        "crop_origin": (float(cx1), float(cy1)),
                        "target_bbox_crop": np.asarray(
                            [x1 - float(cx1), y1 - float(cy1), x2 - float(cx1), y2 - float(cy1)],
                            dtype=float,
                        ),
                        "target_center_crop": np.asarray(
                            [
                                float(record.get("center", (0.0, 0.0))[0]) * frame_scale_x - float(cx1),
                                float(record.get("center", (0.0, 0.0))[1]) * frame_scale_y - float(cy1),
                            ],
                            dtype=float,
                        ),
                    }
                )
            except Exception:
                continue

        if not crop_jobs:
            return

        pose_results = self._predict_pose_batch(
            [job["crop"] for job in crop_jobs],
            pose_threshold=pose_threshold,
            torch_module=torch,
        )
        if not pose_results:
            return

        for job, pose_result in zip(crop_jobs, pose_results):
            try:
                if pose_result is None:
                    continue
                kp_obj = getattr(pose_result, "keypoints", None)
                if kp_obj is None:
                    continue

                kp_xy = self._to_numpy(getattr(kp_obj, "xy", None), dtype=float, shape=None)
                kp_conf = self._to_numpy(getattr(kp_obj, "conf", None), dtype=float, shape=None)
                if kp_xy is None or kp_xy.size == 0 or kp_xy.ndim < 2:
                    continue

                # Pick the highest-confidence pose detection inside the crop.
                pose_boxes = getattr(pose_result, "boxes", None)
                pose_scores = self._to_numpy(
                    getattr(pose_boxes, "conf", None) if pose_boxes is not None else None,
                    dtype=float,
                    shape=(-1,),
                )
                pose_boxes_xyxy = self._to_numpy(
                    getattr(pose_boxes, "xyxy", None) if pose_boxes is not None else None,
                    dtype=float,
                    shape=(-1, 4),
                )
                if kp_xy.ndim == 2:
                    selected_xy = kp_xy
                    selected_conf = kp_conf if (kp_conf is not None and kp_conf.ndim == 1) else None
                else:
                    best_index = self._select_best_pose_candidate(
                        kp_xy,
                        pose_scores,
                        pose_boxes_xyxy,
                        job["target_bbox_crop"],
                        job["target_center_crop"],
                    )
                    if best_index < 0 or best_index >= len(kp_xy):
                        continue
                    selected_xy = kp_xy[best_index]
                    if kp_conf is not None and kp_conf.ndim == 2 and best_index < len(kp_conf):
                        selected_conf = kp_conf[best_index]
                    else:
                        selected_conf = None

                if selected_xy.shape[-1] < 2:
                    continue
                keypoints_image = selected_xy.astype(float, copy=True)
                origin_x, origin_y = job["crop_origin"]
                keypoints_image[:, 0] = keypoints_image[:, 0] + float(origin_x)
                keypoints_image[:, 1] = keypoints_image[:, 1] + float(origin_y)
                # Map keypoints from frame_rgb space back to output space.
                keypoints_image[:, 0] = keypoints_image[:, 0] / frame_scale_x
                keypoints_image[:, 1] = keypoints_image[:, 1] / frame_scale_y
                if selected_conf is not None:
                    selected_conf = selected_conf.astype(float, copy=False).reshape(-1)

                if min_confident_kp > 0:
                    if selected_conf is None:
                        confident_count = int(np.sum(np.all(np.isfinite(keypoints_image), axis=1)))
                    else:
                        confident_count = int(
                            np.sum(np.asarray(selected_conf) >= float(pose_threshold))
                        )
                    if confident_count < int(min_confident_kp):
                        continue

                job["record"]["keypoints"] = keypoints_image
                job["record"]["keypoint_scores"] = selected_conf
            except Exception:
                # Per-detection pose failures must not abort live inference.
                continue

    def _predict_pose_batch(
        self,
        crops: list[np.ndarray],
        *,
        pose_threshold: float,
        torch_module,
    ) -> list:
        """Run YOLO pose on a batch of crops, falling back to per-crop mode if needed."""
        if self._pose_model is None or not crops:
            return []

        max_dim = max(max(int(crop.shape[0]), int(crop.shape[1])) for crop in crops)
        pose_imgsz = int(min(640, max(256, int(np.ceil(max_dim / 32.0) * 32))))

        def _run_predict(input_value):
            return self._pose_model.predict(
                input_value,
                conf=float(pose_threshold),
                imgsz=pose_imgsz,
                verbose=False,
            )

        if self._pose_batch_predict_enabled:
            try:
                if torch_module is not None:
                    with torch_module.inference_mode():
                        results = _run_predict(crops)
                else:
                    results = _run_predict(crops)
                if isinstance(results, (list, tuple)) and len(results) == len(crops):
                    return list(results)
            except Exception as exc:
                self._pose_batch_predict_enabled = False
                self.status_changed.emit(f"Pose batch path disabled: {exc}")

        results: list = []
        for crop in crops:
            try:
                if torch_module is not None:
                    with torch_module.inference_mode():
                        per_crop = _run_predict(crop)
                else:
                    per_crop = _run_predict(crop)
                results.append(per_crop[0] if per_crop else None)
            except Exception:
                results.append(None)
        return results

    def _pose_executor(self):
        """Single-thread executor that runs pose inference beside segmentation."""
        if self._pose_executor_instance is None:
            from concurrent.futures import ThreadPoolExecutor

            self._pose_executor_instance = ThreadPoolExecutor(
                max_workers=1,
                thread_name_prefix="pose-inference",
            )
        return self._pose_executor_instance

    def _unified_finalize_executor(self):
        """One ordered CPU stage overlapped with the next Unified GPU forward."""
        if self._unified_finalize_executor_instance is None:
            from concurrent.futures import ThreadPoolExecutor

            self._unified_finalize_executor_instance = ThreadPoolExecutor(
                max_workers=1,
                thread_name_prefix="unified-causal-finalize",
            )
        return self._unified_finalize_executor_instance

    def _wait_for_unified_finalize(self) -> None:
        """Wait for the one bounded finalization slot without duplicating errors."""
        with self._unified_finalize_lock:
            pending = self._pending_unified_finalize
        if pending is None:
            return
        try:
            pending.result()
        except Exception:
            # The completion callback reports the error through the normal signal.
            pass
        finally:
            with self._unified_finalize_lock:
                if self._pending_unified_finalize is pending:
                    self._pending_unified_finalize = None

    def _report_unified_finalize_failure(self, future) -> None:
        try:
            future.result()
        except Exception as exc:
            self.error_occurred.emit(f"Live inference error: {str(exc)}")

    def _predict_pose_fullframe(self, frame_rgb: np.ndarray, pose_threshold: float, imgsz_cap: int = 640):
        """Run the YOLO pose model once on the whole inference frame.

        Returns the raw ultralytics result (or None). Running full-frame keeps
        every animal in one forward pass and avoids crop-boundary keypoint
        loss, which makes tracking-mode pose both faster and more reliable
        than per-detection crops. ``imgsz_cap`` bounds the pose network size:
        640 keeps pose latency well under the segmentation forward so the two
        overlap cleanly instead of pose becoming the bottleneck.
        """
        if self._pose_model is None:
            return None
        torch_module = import_torch(required=False)
        use_half = bool(torch_module is not None and torch_module.cuda.is_available())
        height, width = frame_rgb.shape[:2]
        cap = int(max(320, imgsz_cap or 640))
        pose_imgsz = int(min(cap, max(320, int(np.ceil(max(height, width) / 32.0) * 32))))

        def _run():
            try:
                results = self._pose_model.predict(
                    frame_rgb,
                    conf=float(pose_threshold),
                    imgsz=pose_imgsz,
                    half=use_half,
                    verbose=False,
                )
            except TypeError:
                results = self._pose_model.predict(
                    frame_rgb,
                    conf=float(pose_threshold),
                    imgsz=pose_imgsz,
                    verbose=False,
                )
            return results[0] if results else None

        if torch_module is not None:
            with torch_module.inference_mode():
                return _run()
        return _run()

    def _attach_pose_keypoints_fullframe(
        self,
        records: list[dict],
        pose_result,
        *,
        scale_x: float,
        scale_y: float,
        pose_threshold: float,
        min_confident_kp: int,
    ) -> None:
        """Match full-frame pose candidates to segmentation records by IoU.

        Pose coordinates arrive in inference-frame space and are rescaled to
        full-frame space before matching, because records were already
        rescaled. Greedy assignment: highest-confidence record first, best
        remaining IoU candidate; one candidate is used at most once.
        """
        try:
            kp_obj = getattr(pose_result, "keypoints", None)
            if kp_obj is None:
                return
            kp_xy = self._to_numpy(getattr(kp_obj, "xy", None), dtype=float, shape=None)
            kp_conf = self._to_numpy(getattr(kp_obj, "conf", None), dtype=float, shape=None)
            if kp_xy is None or kp_xy.size == 0:
                return
            if kp_xy.ndim == 2:
                kp_xy = kp_xy[None, ...]
            if kp_conf is not None and kp_conf.ndim == 1:
                kp_conf = kp_conf[None, ...]

            pose_boxes_obj = getattr(pose_result, "boxes", None)
            pose_boxes = self._to_numpy(
                getattr(pose_boxes_obj, "xyxy", None) if pose_boxes_obj is not None else None,
                dtype=float,
                shape=(-1, 4),
            )
            if pose_boxes is None or len(pose_boxes) == 0:
                # Derive candidate boxes from keypoint extents when the model
                # exposes no boxes.
                finite = np.isfinite(kp_xy).all(axis=2)
                pose_boxes = np.zeros((len(kp_xy), 4), dtype=float)
                for index in range(len(kp_xy)):
                    points = kp_xy[index][finite[index]]
                    if len(points) == 0:
                        continue
                    pose_boxes[index] = [
                        float(np.min(points[:, 0])),
                        float(np.min(points[:, 1])),
                        float(np.max(points[:, 0])),
                        float(np.max(points[:, 1])),
                    ]

            # Rescale pose outputs from inference space to full-frame space.
            kp_xy = kp_xy.astype(float, copy=True)
            kp_xy[..., 0] *= float(scale_x)
            kp_xy[..., 1] *= float(scale_y)
            pose_boxes = pose_boxes.astype(float, copy=True)
            pose_boxes[:, [0, 2]] *= float(scale_x)
            pose_boxes[:, [1, 3]] *= float(scale_y)

            candidate_used = [False] * len(kp_xy)
            ordered_records = sorted(
                records,
                key=lambda record: float(record.get("confidence", 0.0) or 0.0),
                reverse=True,
            )
            for record in ordered_records:
                bbox = record.get("bbox")
                if bbox is None:
                    continue
                record_box = np.asarray([float(value) for value in bbox], dtype=float)
                best_index = -1
                best_iou = 0.0
                for index in range(len(kp_xy)):
                    if candidate_used[index]:
                        continue
                    iou = self._bbox_iou(record_box, pose_boxes[index])
                    if iou > best_iou:
                        best_iou = iou
                        best_index = index
                if best_index < 0 or best_iou < 0.05:
                    continue

                selected_xy = kp_xy[best_index]
                selected_conf = (
                    kp_conf[best_index].astype(float, copy=False).reshape(-1)
                    if (kp_conf is not None and best_index < len(kp_conf))
                    else None
                )
                if min_confident_kp > 0:
                    if selected_conf is None:
                        confident_count = int(np.sum(np.all(np.isfinite(selected_xy), axis=1)))
                    else:
                        confident_count = int(np.sum(selected_conf >= float(pose_threshold)))
                    if confident_count < int(min_confident_kp):
                        continue
                candidate_used[best_index] = True
                record["keypoints"] = selected_xy
                record["keypoint_scores"] = selected_conf
        except Exception:
            # Pose merge failures must never abort segmentation results.
            return

    def _select_best_pose_candidate(
        self,
        kp_xy: np.ndarray,
        pose_scores: np.ndarray,
        pose_boxes_xyxy: np.ndarray,
        target_bbox_crop: np.ndarray,
        target_center_crop: np.ndarray,
    ) -> int:
        """Choose the pose candidate that best matches the segmented animal geometry."""
        if kp_xy.ndim < 3 or len(kp_xy) == 0:
            return 0

        target_box = np.asarray(target_bbox_crop, dtype=float).reshape(4)
        target_center = np.asarray(target_center_crop, dtype=float).reshape(2)
        target_diag = max(
            1.0,
            float(np.hypot(max(1.0, target_box[2] - target_box[0]), max(1.0, target_box[3] - target_box[1]))),
        )
        best_index = -1
        best_score = -float("inf")

        for index, candidate_xy in enumerate(kp_xy):
            candidate_xy = np.asarray(candidate_xy, dtype=float).reshape(-1, candidate_xy.shape[-1])
            finite_mask = np.all(np.isfinite(candidate_xy[:, :2]), axis=1)
            candidate_points = candidate_xy[finite_mask, :2]
            if candidate_points.size == 0:
                continue

            if index < len(pose_boxes_xyxy):
                candidate_box = np.asarray(pose_boxes_xyxy[index], dtype=float).reshape(4)
            else:
                candidate_box = np.asarray(
                    [
                        float(np.min(candidate_points[:, 0])),
                        float(np.min(candidate_points[:, 1])),
                        float(np.max(candidate_points[:, 0])),
                        float(np.max(candidate_points[:, 1])),
                    ],
                    dtype=float,
                )

            candidate_center = np.asarray(
                [
                    float((candidate_box[0] + candidate_box[2]) * 0.5),
                    float((candidate_box[1] + candidate_box[3]) * 0.5),
                ],
                dtype=float,
            )
            center_distance = float(np.linalg.norm(candidate_center - target_center)) / target_diag
            iou = self._bbox_iou(candidate_box, target_box)

            expanded_target = np.asarray(
                [
                    target_box[0] - 0.1 * (target_box[2] - target_box[0]),
                    target_box[1] - 0.1 * (target_box[3] - target_box[1]),
                    target_box[2] + 0.1 * (target_box[2] - target_box[0]),
                    target_box[3] + 0.1 * (target_box[3] - target_box[1]),
                ],
                dtype=float,
            )
            inside_fraction = float(
                np.mean(
                    (candidate_points[:, 0] >= expanded_target[0])
                    & (candidate_points[:, 0] <= expanded_target[2])
                    & (candidate_points[:, 1] >= expanded_target[1])
                    & (candidate_points[:, 1] <= expanded_target[3])
                )
            )
            pose_conf = float(pose_scores[index]) if index < len(pose_scores) else 0.0
            score = (2.5 * iou) + (1.5 * inside_fraction) + (0.35 * pose_conf) - (1.25 * center_distance)
            if score > best_score:
                best_score = score
                best_index = index

        return best_index if best_index >= 0 else 0

    def _bbox_iou(self, box_a: np.ndarray, box_b: np.ndarray) -> float:
        """Compute IoU for two xyxy boxes."""
        a = np.asarray(box_a, dtype=float).reshape(4)
        b = np.asarray(box_b, dtype=float).reshape(4)
        inter_x1 = max(float(a[0]), float(b[0]))
        inter_y1 = max(float(a[1]), float(b[1]))
        inter_x2 = min(float(a[2]), float(b[2]))
        inter_y2 = min(float(a[3]), float(b[3]))
        inter_w = max(0.0, inter_x2 - inter_x1)
        inter_h = max(0.0, inter_y2 - inter_y1)
        inter_area = inter_w * inter_h
        area_a = max(0.0, float(a[2] - a[0])) * max(0.0, float(a[3] - a[1]))
        area_b = max(0.0, float(b[2] - b[0])) * max(0.0, float(b[3] - b[1]))
        union_area = area_a + area_b - inter_area
        if union_area <= 0.0:
            return 0.0
        return inter_area / union_area

    def _predict(self, model, model_key: str, frame_rgb: np.ndarray, threshold: float):
        torch = import_torch(required=False)

        def _run_prediction():
            if str(model_key) == MAMIR_UNIFIED_MODEL_KEY:
                return model.predict(frame_rgb, threshold=float(threshold))
            if str(model_key).startswith("rfdetr"):
                if torch is not None and self._rfdetr_direct_predict_enabled:
                    try:
                        return self._predict_rfdetr_direct(model, frame_rgb, threshold, torch)
                    except Exception as exc:
                        self._rfdetr_direct_predict_enabled = False
                        self.status_changed.emit(f"RF-DETR direct path disabled: {exc}")
                return model.predict(frame_rgb, threshold=float(threshold))

            try:
                results = model.predict(
                    frame_rgb,
                    conf=float(threshold),
                    verbose=False,
                    retina_masks=True,
                )
            except TypeError:
                results = model.predict(frame_rgb, conf=float(threshold), verbose=False)
            return results[0] if results else None

        if torch is not None:
            with torch.inference_mode():
                return _run_prediction()
        return _run_prediction()

    def _predict_rfdetr_direct(self, model, frame_rgb: np.ndarray, threshold: float, torch):
        model_context = getattr(model, "model", None)
        if model_context is None:
            raise RuntimeError("RF-DETR model context is unavailable")

        device = getattr(model_context, "device", None)
        if device is None:
            raise RuntimeError("RF-DETR device is unavailable")

        inference_module = getattr(model_context, "inference_model", None)
        use_optimized = bool(getattr(model, "_is_optimized_for_inference", False) and inference_module is not None)
        model_module = getattr(model_context, "model", None)
        if not use_optimized and model_module is None:
            raise RuntimeError("RF-DETR base model is unavailable")
        input_dtype = getattr(model, "_optimized_dtype", None) if use_optimized else None
        if not isinstance(input_dtype, torch.dtype):
            input_dtype = torch.float32

        target_resolution = int(
            getattr(model, "_optimized_resolution", 0) if use_optimized else getattr(model_context, "resolution", 0)
        )
        if target_resolution <= 0:
            target_resolution = int(getattr(model_context, "resolution", 0) or 0)
        if target_resolution <= 0:
            raise RuntimeError("RF-DETR inference resolution is unavailable")

        source_frame = np.ascontiguousarray(frame_rgb)
        if source_frame.ndim != 3 or source_frame.shape[2] != 3:
            raise ValueError(f"Expected RGB frame with shape HxWx3, got {source_frame.shape!r}")
        source_height, source_width = source_frame.shape[:2]

        batch_tensor = torch.from_numpy(source_frame).permute(2, 0, 1).unsqueeze(0).contiguous()
        batch_tensor = batch_tensor.to(device=device)
        batch_tensor = batch_tensor.to(dtype=input_dtype).div_(255.0)
        if batch_tensor.shape[-2] != target_resolution or batch_tensor.shape[-1] != target_resolution:
            batch_tensor = torch.nn.functional.interpolate(
                batch_tensor,
                size=(target_resolution, target_resolution),
                mode="bilinear",
                align_corners=False,
            )

        means = tuple(float(value) for value in getattr(model, "means", [0.485, 0.456, 0.406]))
        stds = tuple(float(value) for value in getattr(model, "stds", [0.229, 0.224, 0.225]))
        norm_key = (str(device), str(input_dtype), means, stds)
        norm_tensors = self._rfdetr_norm_cache.get(norm_key)
        if norm_tensors is None:
            norm_tensors = (
                torch.tensor(means, device=device, dtype=input_dtype).view(1, 3, 1, 1),
                torch.tensor(stds, device=device, dtype=input_dtype).view(1, 3, 1, 1),
            )
            self._rfdetr_norm_cache[norm_key] = norm_tensors
        mean_tensor, std_tensor = norm_tensors
        batch_tensor.sub_(mean_tensor).div_(std_tensor)

        if use_optimized:
            predictions = inference_module(batch_tensor.to(dtype=input_dtype))
        else:
            predictions = model_module(batch_tensor)
        if isinstance(predictions, tuple):
            normalized_predictions = {
                "pred_logits": predictions[1],
                "pred_boxes": predictions[0],
            }
            if len(predictions) >= 3:
                normalized_predictions["pred_masks"] = predictions[2]
            predictions = normalized_predictions

        postprocess = getattr(model_context, "postprocess", None)
        if postprocess is None:
            raise RuntimeError("RF-DETR postprocess module is unavailable")
        return self._postprocess_rfdetr_filter_first(
            predictions,
            height=source_height,
            width=source_width,
            threshold=float(threshold),
            num_select=int(getattr(postprocess, "num_select", _RFDETR_MAX_NUM_SELECT)),
            torch_module=torch,
        )

    def _postprocess_rfdetr_filter_first(
        self,
        predictions: dict,
        *,
        height: int,
        width: int,
        threshold: float,
        num_select: int,
        torch_module,
    ) -> dict[str, object]:
        """Match RF-DETR PostProcess while resizing only accepted masks."""
        torch = torch_module
        logits = predictions.get("pred_logits")
        boxes_cxcywh = predictions.get("pred_boxes")
        if logits is None or boxes_cxcywh is None or int(logits.shape[0]) != 1:
            return self._empty_detections()

        class_count = int(logits.shape[2])
        probabilities = logits.sigmoid().view(1, -1)
        select_count = min(max(1, int(num_select)), int(probabilities.shape[1]))
        scores, indices = torch.topk(probabilities, select_count, dim=1)
        scores = scores[0]
        indices = indices[0]
        query_indices = indices // class_count
        labels = indices % class_count
        keep = scores > float(threshold)
        if int(torch.count_nonzero(keep).item()) <= 0:
            return self._empty_detections()

        scores = scores[keep]
        query_indices = query_indices[keep]
        labels = labels[keep]
        selected_boxes = boxes_cxcywh[0, query_indices]
        center_x, center_y, box_width, box_height = selected_boxes.unbind(-1)
        boxes = torch.stack(
            (
                center_x - 0.5 * box_width,
                center_y - 0.5 * box_height,
                center_x + 0.5 * box_width,
                center_y + 0.5 * box_height,
            ),
            dim=-1,
        )
        scale_key = (str(boxes.device), str(boxes.dtype), int(height), int(width))
        scale = self._rfdetr_scale_cache.get(scale_key)
        if scale is None:
            scale = torch.tensor(
                [int(width), int(height), int(width), int(height)],
                device=boxes.device,
                dtype=boxes.dtype,
            )
            self._rfdetr_scale_cache[scale_key] = scale
        boxes = boxes * scale
        output: dict[str, object] = {
            "xyxy": boxes.detach().float().cpu().numpy(),
            "confidence": scores.detach().float().cpu().numpy(),
            "class_id": labels.detach().to(dtype=torch.int64).cpu().numpy(),
        }

        mask_logits = predictions.get("pred_masks")
        if mask_logits is not None:
            selected_masks = mask_logits[0, query_indices]
            resized_masks = torch.nn.functional.interpolate(
                selected_masks.unsqueeze(1),
                size=(int(height), int(width)),
                mode="bilinear",
                align_corners=False,
            )[:, 0] > 0.0
            # Standard live results are consumed immediately. A single batched
            # transfer is faster here than synchronizing once per crop; cleanup
            # converts the surviving instances to CropMask without allocating a
            # second full-frame output.
            output["mask"] = resized_masks.detach().cpu().numpy()
        return output

    def _build_detection_records(self, detections: dict[str, object], config: LiveInferenceConfig) -> list[dict]:
        boxes = np.asarray(detections.get("xyxy", np.empty((0, 4), dtype=float)), dtype=float).reshape(-1, 4)
        confidence = np.asarray(detections.get("confidence", np.empty((0,), dtype=float)), dtype=float).reshape(-1)
        class_id = np.asarray(detections.get("class_id", np.empty((0,), dtype=int)), dtype=int).reshape(-1)
        masks = self._to_masks(detections.get("mask"))
        amodal_masks = self._to_masks(detections.get("amodal_mask"))
        selected_classes = set(int(value) for value in (config.selected_class_ids or []))
        records: list[dict] = []
        for index, bbox in enumerate(boxes):
            cls = int(class_id[index]) if index < len(class_id) else 0
            if selected_classes and cls not in selected_classes:
                continue
            mask = None
            if masks is not None and index < len(masks):
                mask = masks[index]
            tracking_mask = mask
            if amodal_masks is not None and index < len(amodal_masks):
                tracking_mask = amodal_masks[index]
            bbox_tuple = tuple(float(value) for value in bbox)
            center = self._compute_body_center(mask, bbox_tuple)
            records.append(
                {
                    "bbox": bbox_tuple,
                    "confidence": float(confidence[index]) if index < len(confidence) else 0.0,
                    "class_id": cls,
                    "mask": mask,
                    "tracking_mask": tracking_mask,
                    "center": center,
                }
            )
        return records

    def _compute_body_center(
        self,
        mask: object,
        bbox: tuple[float, float, float, float],
        *,
        pad: int = 4,
    ) -> tuple[float, float]:
        """Return a mask centroid without scanning an entire full-frame mask."""
        if mask is None:
            return compute_body_center(None, bbox)
        h, w = mask_shape(mask)
        if h <= 0 or w <= 0:
            return compute_body_center(None, bbox)
        x1, y1, x2, y2 = (float(value) for value in bbox)
        ix1 = max(0, int(np.floor(min(x1, x2))) - int(pad))
        iy1 = max(0, int(np.floor(min(y1, y2))) - int(pad))
        ix2 = min(w, int(np.ceil(max(x1, x2))) + int(pad) + 1)
        iy2 = min(h, int(np.ceil(max(y1, y2))) + int(pad) + 1)
        if ix2 <= ix1 or iy2 <= iy1:
            return compute_body_center(None, bbox)
        crop = mask_region(mask, ix1, iy1, ix2, iy2)
        ys, xs = np.nonzero(crop)
        if len(xs) > 0 and len(ys) > 0:
            return float(np.mean(xs) + ix1), float(np.mean(ys) + iy1)
        return compute_body_center(None, bbox)

    def _refine_records_for_overlay(self, records: list[dict], config: LiveInferenceConfig) -> None:
        """Clean masks to one solid blob and clamp keypoints to the body in place.

        This is what makes the live overlay look "perfect": masks stop
        flickering into speckle, and pose keypoints never paint onto the
        background outside the animal's silhouette.
        """
        self._clean_record_masks(records, config)
        self._clamp_record_keypoints(records, config)

    def _clean_record_masks(self, records: list[dict], config: LiveInferenceConfig) -> None:
        """Clean masks and keep detection centers tied to the cleaned body."""
        if not records or not config.clean_masks:
            return
        from live_overlay_quality import clean_instance_mask

        for record in records:
            mask = record.get("mask")
            if mask is not None:
                cleaned = self._clean_mask_roi(mask, record.get("bbox", (0.0, 0.0, 0.0, 0.0)), clean_instance_mask)
                if cleaned is not None:
                    record["mask"] = cleaned
                    # Recompute the body center from the cleaned silhouette so
                    # the tracker and ROI tests use the solid blob.
                    record["center"] = self._compute_body_center(cleaned, record.get("bbox", (0.0, 0.0, 0.0, 0.0)))
        return

    def _clean_mask_roi(self, mask: object, bbox, clean_instance_mask_fn) -> Optional[object]:
        shape = mask_shape(mask)
        if shape[0] <= 0 or shape[1] <= 0:
            return None
        crop, origin = self._crop_mask_around_bbox(mask, bbox, pad=12)
        if crop is None or not bool(crop.any()):
            return None
        # The crop is already limited to the detection, so a lower area gate
        # keeps valid small animals from being rejected by frame-sized ratios.
        cleaned_crop = clean_instance_mask_fn(crop, min_area_ratio=0.001)
        if cleaned_crop is None:
            return None
        x0, y0 = origin
        return CropMask.from_crop(cleaned_crop, shape, y0, x0)

    def _clamp_record_keypoints(self, records: list[dict], config: LiveInferenceConfig) -> None:
        """Drop record keypoints that fall outside each cleaned mask."""
        if not records or not config.clamp_pose_to_mask:
            return
        from live_overlay_quality import clamp_keypoints_to_mask

        for record in records:
            mask = record.get("mask")
            if config.clamp_pose_to_mask and record.get("keypoints") is not None and mask is not None:
                kp, sc = clamp_keypoints_to_mask(
                    record.get("keypoints"),
                    record.get("keypoint_scores"),
                    mask,
                )
                kp, sc = repair_hip_keypoints_with_mask_geometry(kp, sc, mask)
                record["keypoints"] = kp
                record["keypoint_scores"] = sc

    def _reset_mask_skeleton_extractor(self) -> None:
        self._mask_skeleton_extractor = None
        self._mask_skeleton_signature = None
        self._mask_geometry_cache.clear()

    def _mask_skeleton_estimator(self, config: LiveInferenceConfig) -> MaskSkeletonExtractor:
        signature = ("pca", bool(config.smooth_keypoints))
        if self._mask_skeleton_extractor is None or self._mask_skeleton_signature != signature:
            self._mask_skeleton_extractor = MaskSkeletonExtractor(
                method="pca",
                # Live stabilization is applied after extraction so the core
                # geometry remains byte-compatible with MAMIR's scientific pass.
                smooth=False,
            )
            self._mask_skeleton_signature = signature
        return self._mask_skeleton_extractor

    @staticmethod
    def _stable_cached_geometry_translation(
        current_center,
        cached_center,
        bbox,
    ) -> tuple[float, float]:
        """Reject sub-pixel/scale mask-centroid noise between geometry refreshes."""
        try:
            dx = float(current_center[0]) - float(cached_center[0])
            dy = float(current_center[1]) - float(cached_center[1])
            x1, y1, x2, y2 = (float(value) for value in bbox)
        except Exception:
            return 0.0, 0.0
        body_diagonal = float(np.hypot(x2 - x1, y2 - y1))
        jitter_radius = float(np.clip(0.012 * body_diagonal, 0.75, 2.0))
        if float(np.hypot(dx, dy)) <= jitter_radius:
            return 0.0, 0.0
        return dx, dy

    def _attach_mask_skeleton_keypoints(
        self,
        tracked_mice: list,
        config: LiveInferenceConfig,
        *,
        frame_index: int = 0,
    ) -> None:
        """Attach eight geometry-derived keypoints to tracked mouse states."""
        if not tracked_mice:
            return
        extractor = self._mask_skeleton_estimator(config)
        force_refresh = False

        if extractor is not self._flip_applied_extractor:
            # Fresh extractor (first use / model reload / reconfigure): the persistent
            # NET correction is the single source of truth, so apply it and drop any
            # transient clicks queued for the previous extractor (already folded into
            # the net) to avoid a double flip.
            self._flip_applied_extractor = extractor
            with self._condition:
                self._pending_orientation_flips = []
                net_flips = [mid for mid, net in self._manual_orientation_flips.items() if net]
            for mid in net_flips:
                try:
                    extractor.flip_orientation(int(mid))
                except Exception:
                    pass
        elif self._pending_orientation_flips:
            force_refresh = True
            with self._condition:
                flips = self._pending_orientation_flips
                self._pending_orientation_flips = []
            for mid in flips:
                try:
                    extractor.flip_orientation(mid)
                except Exception:
                    pass

        for mouse in tracked_mice:
            mouse_id = int(getattr(mouse, "mouse_id", 0))
            refresh_interval = max(1, int(config.geometry_refresh_interval or 1))
            cached_geometry = self._mask_geometry_cache.get(mouse_id)
            refresh_geometry = (
                force_refresh
                or cached_geometry is None
                or refresh_interval <= 1
                or (int(frame_index) + mouse_id) % refresh_interval == 0
            )
            if not refresh_geometry and cached_geometry is not None:
                cached_keypoints, cached_scores, cached_center = cached_geometry
                current_center = getattr(mouse, "center", cached_center)
                dx, dy = self._stable_cached_geometry_translation(
                    current_center,
                    cached_center,
                    getattr(mouse, "bbox", (0.0, 0.0, 0.0, 0.0)),
                )
                keypoints = np.asarray(cached_keypoints, dtype=float).reshape(-1, 2).copy()
                if dx != 0.0 or dy != 0.0:
                    keypoints[:, 0] += dx
                    keypoints[:, 1] += dy
                mouse.keypoints = keypoints
                mouse.keypoint_scores = np.asarray(cached_scores, dtype=float).reshape(-1).copy()
                continue
            mask = getattr(mouse, "mask", None)
            if not mask_has_pixels(mask):
                continue
            if (
                config.model_key == MAMIR_UNIFIED_MODEL_KEY
                and is_crop_packed(mask)
            ):
                # Unified masks are already exact, tight, lossless crops. The
                # geometry extractor understands this representation directly.
                # Keep the same bbox-local coordinate system as the reference
                # dense crop so floating-point pose output stays bit-exact, but
                # retain only the positive pixels instead of allocating padded
                # zeros.
                geometry_mask, origin = self._crop_packed_mask_around_bbox(
                    mask,
                    getattr(mouse, "bbox", (0.0, 0.0, 0.0, 0.0)),
                    pad=12,
                )
                if geometry_mask is None:
                    continue
            else:
                geometry_mask, origin = self._crop_mask_around_bbox(
                    mask,
                    getattr(mouse, "bbox", (0.0, 0.0, 0.0, 0.0)),
                    pad=12,
                )
                if (
                    geometry_mask is None
                    or not bool(np.asarray(geometry_mask, dtype=bool).any())
                ):
                    continue
            try:
                skeleton = extractor.estimate(
                    geometry_mask,
                    track_id=mouse_id,
                    offset=(float(origin[0]), float(origin[1])),
                )
            except Exception:
                continue
            if skeleton is None:
                continue
            keypoints = np.asarray(skeleton.keypoints, dtype=float).reshape(-1, 2)
            scores = np.asarray(skeleton.scores, dtype=float).reshape(-1)
            if bool(config.smooth_keypoints) and cached_geometry is not None and not force_refresh:
                from live_overlay_quality import smooth_keypoints_adaptive

                stabilized = smooth_keypoints_adaptive(cached_geometry[0], keypoints)
                if stabilized is not None:
                    keypoints = stabilized
            mouse.keypoints = keypoints
            mouse.keypoint_scores = scores
            center = getattr(mouse, "center", (0.0, 0.0))
            self._mask_geometry_cache[mouse_id] = (
                keypoints.copy(),
                scores.copy(),
                (float(center[0]), float(center[1])),
            )

    def _crop_mask_around_bbox(
        self,
        mask: np.ndarray,
        bbox,
        *,
        pad: int = 8,
    ) -> tuple[Optional[np.ndarray], tuple[int, int]]:
        h, w = mask_shape(mask)
        if h <= 0 or w <= 0:
            return None, (0, 0)
        try:
            x1, y1, x2, y2 = (float(value) for value in bbox)
        except Exception:
            crop, x0, y0 = occupied_crop(mask)
            return crop, (x0, y0)
        ix1 = max(0, int(np.floor(min(x1, x2))) - int(pad))
        iy1 = max(0, int(np.floor(min(y1, y2))) - int(pad))
        ix2 = min(w, int(np.ceil(max(x1, x2))) + int(pad) + 1)
        iy2 = min(h, int(np.ceil(max(y1, y2))) + int(pad) + 1)
        if ix2 <= ix1 or iy2 <= iy1:
            return None, (0, 0)
        return mask_region(mask, ix1, iy1, ix2, iy2), (ix1, iy1)

    def _crop_packed_mask_around_bbox(
        self,
        mask: object,
        bbox,
        *,
        pad: int = 8,
    ) -> tuple[Optional[CropMask], tuple[int, int]]:
        """Return the reference bbox crop without allocating its padded zeros."""
        h, w = mask_shape(mask)
        bounds = getattr(mask, "crop_bbox", None)
        if h <= 0 or w <= 0 or bounds is None or not is_crop_packed(mask):
            return None, (0, 0)
        try:
            x1, y1, x2, y2 = (float(value) for value in bbox)
        except Exception:
            crop, x0, y0 = occupied_crop(mask)
            if crop is None:
                return None, (0, 0)
            return CropMask.from_crop(crop, crop.shape, 0, 0), (x0, y0)

        ix1 = max(0, int(np.floor(min(x1, x2))) - int(pad))
        iy1 = max(0, int(np.floor(min(y1, y2))) - int(pad))
        ix2 = min(w, int(np.ceil(max(x1, x2))) + int(pad) + 1)
        iy2 = min(h, int(np.ceil(max(y1, y2))) + int(pad) + 1)
        if ix2 <= ix1 or iy2 <= iy1:
            return None, (0, 0)

        my1, my2, mx1, mx2 = (int(value) for value in bounds)
        ox1, ox2 = max(ix1, mx1), min(ix2, mx2)
        oy1, oy2 = max(iy1, my1), min(iy2, my2)
        if ox2 <= ox1 or oy2 <= oy1:
            return None, (0, 0)
        source = np.asarray(mask.crop_array(), dtype=bool)[
            oy1 - my1:oy2 - my1,
            ox1 - mx1:ox2 - mx1,
        ]
        local = CropMask.from_crop(
            source,
            (iy2 - iy1, ix2 - ix1),
            oy1 - iy1,
            ox1 - ix1,
        )
        return local, (ix1, iy1)

    def _normalize_detections(self, detections) -> dict[str, object]:
        if detections is None:
            return self._empty_detections()

        if isinstance(detections, dict) and "xyxy" in detections:
            xyxy = self._to_numpy(detections.get("xyxy"), dtype=float, shape=(-1, 4))
            confidence = self._to_numpy(detections.get("confidence"), dtype=float, shape=(-1,))
            class_id = self._to_numpy(detections.get("class_id"), dtype=int, shape=(-1,))
            masks = self._to_masks(detections.get("mask", detections.get("masks")))
            amodal_masks = self._to_masks(detections.get("amodal_mask", detections.get("amodal_masks")))
            return self._coerce_lengths(xyxy, confidence, class_id, masks, amodal_masks)

        if hasattr(detections, "boxes"):
            boxes = detections.boxes
            xyxy = self._to_numpy(getattr(boxes, "xyxy", None), dtype=float, shape=(-1, 4))
            confidence = self._to_numpy(getattr(boxes, "conf", None), dtype=float, shape=(-1,))
            class_id = self._to_numpy(getattr(boxes, "cls", None), dtype=int, shape=(-1,))
            masks_obj = getattr(detections, "masks", None)
            masks = None
            if masks_obj is not None:
                masks = self._masks_from_ultralytics_polygons(masks_obj, detections)
                if masks is None:
                    masks = self._to_masks(getattr(masks_obj, "data", masks_obj))
            return self._coerce_lengths(xyxy, confidence, class_id, masks)

        xyxy = self._to_numpy(getattr(detections, "xyxy", None), dtype=float, shape=(-1, 4))
        confidence = self._to_numpy(
            getattr(detections, "confidence", getattr(detections, "conf", None)),
            dtype=float,
            shape=(-1,),
        )
        class_id = self._to_numpy(getattr(detections, "class_id", None), dtype=int, shape=(-1,))
        masks = self._to_masks(getattr(detections, "mask", getattr(detections, "masks", None)))
        return self._coerce_lengths(xyxy, confidence, class_id, masks)

    def _coerce_lengths(
        self,
        xyxy: np.ndarray,
        confidence: np.ndarray,
        class_id: np.ndarray,
        masks: object = None,
        amodal_masks: object = None,
    ) -> dict[str, object]:
        count = len(xyxy)
        if count == 0:
            return self._empty_detections()
        if len(confidence) != count:
            confidence = np.ones(count, dtype=float)
        if len(class_id) != count:
            class_id = np.zeros(count, dtype=int)
        result = {
            "xyxy": xyxy.astype(float, copy=False),
            "confidence": confidence.astype(float, copy=False),
            "class_id": class_id.astype(int, copy=False),
        }
        if masks is not None and self._is_packed_mask_sequence(masks, count):
            result["mask"] = list(masks)
        elif masks is not None:
            masks = np.asarray(masks)
            if masks.ndim == 2:
                masks = masks[None, ...]
            if masks.ndim == 4 and masks.shape[1] == 1:
                masks = masks[:, 0, :, :]
            if masks.ndim == 3 and len(masks) == count:
                result["mask"] = masks.astype(bool, copy=False)
        if amodal_masks is not None and self._is_packed_mask_sequence(amodal_masks, count):
            result["amodal_mask"] = list(amodal_masks)
        elif amodal_masks is not None:
            amodal_masks = np.asarray(amodal_masks)
            if amodal_masks.ndim == 2:
                amodal_masks = amodal_masks[None, ...]
            if amodal_masks.ndim == 4 and amodal_masks.shape[1] == 1:
                amodal_masks = amodal_masks[:, 0, :, :]
            if amodal_masks.ndim == 3 and len(amodal_masks) == count:
                result["amodal_mask"] = amodal_masks.astype(bool, copy=False)
        return result

    def _empty_detections(self) -> dict[str, np.ndarray]:
        return {
            "xyxy": np.empty((0, 4), dtype=float),
            "confidence": np.empty((0,), dtype=float),
            "class_id": np.empty((0,), dtype=int),
            "mask": np.empty((0, 0, 0), dtype=bool),
            "amodal_mask": np.empty((0, 0, 0), dtype=bool),
        }

    def _to_numpy(self, value, dtype=float, shape=None) -> np.ndarray:
        if value is None:
            arr = np.empty((0,), dtype=dtype)
        elif hasattr(value, "detach"):
            arr = value.detach().cpu().numpy()
        elif hasattr(value, "cpu") and hasattr(value, "numpy"):
            arr = value.cpu().numpy()
        else:
            arr = np.asarray(value)
        arr = np.asarray(arr, dtype=dtype)
        if shape is None:
            return arr
        if arr.size == 0:
            if shape == (-1, 4):
                return np.empty((0, 4), dtype=dtype)
            if shape == (-1,):
                return np.empty((0,), dtype=dtype)
        if shape == (-1, 4):
            return arr.reshape(-1, 4)
        if shape == (-1,):
            return arr.reshape(-1)
        return arr

    @staticmethod
    def _is_packed_mask_sequence(value: object, count: Optional[int] = None) -> bool:
        if not isinstance(value, (list, tuple)):
            return False
        if count is not None and len(value) != int(count):
            return False
        return bool(value) and all(is_crop_packed(mask) for mask in value)

    def _to_masks(self, value):
        if value is None:
            return None
        if self._is_packed_mask_sequence(value):
            return list(value)
        if hasattr(value, "detach"):
            value = value.detach().cpu().numpy()
        elif hasattr(value, "cpu") and hasattr(value, "numpy"):
            value = value.cpu().numpy()
        arr = np.asarray(value)
        if arr.size == 0:
            return None
        if arr.ndim == 2:
            arr = arr[None, ...]
        if arr.ndim == 4 and arr.shape[1] == 1:
            arr = arr[:, 0, :, :]
        if arr.ndim != 3:
            return None
        return arr.astype(bool, copy=False)

    def _masks_from_ultralytics_polygons(self, masks_obj, detections) -> Optional[list[CropMask]]:
        """Rasterize Ultralytics mask polygons in original-image coordinates."""
        polygons = getattr(masks_obj, "xy", None)
        if not polygons:
            return None
        orig_shape = getattr(detections, "orig_shape", None) or getattr(masks_obj, "orig_shape", None)
        if not orig_shape or len(orig_shape) < 2:
            return None
        height, width = int(orig_shape[0]), int(orig_shape[1])
        if height <= 0 or width <= 0:
            return None

        rendered: list[CropMask] = []
        for polygon in polygons:
            pts = np.asarray(polygon, dtype=np.float32).reshape(-1, 2)
            if len(pts) >= 3:
                pts[:, 0] = np.clip(pts[:, 0], 0, max(0, width - 1))
                pts[:, 1] = np.clip(pts[:, 1], 0, max(0, height - 1))
                points = np.rint(pts).astype(np.int32)
                x0, x1 = int(points[:, 0].min()), int(points[:, 0].max()) + 1
                y0, y1 = int(points[:, 1].min()), int(points[:, 1].max()) + 1
                crop = np.zeros((y1 - y0, x1 - x0), dtype=np.uint8)
                local_points = points - np.asarray([x0, y0], dtype=np.int32)
                cv2.fillPoly(crop, [local_points], 1)
                rendered.append(CropMask.from_crop(crop, (height, width), y0, x0))
            else:
                rendered.append(CropMask.from_crop(np.zeros((0, 0), dtype=bool), (height, width), 0, 0))
        return rendered or None

    def _align_rfdetr_postprocess_num_select(self, model):
        """Work around rfdetr building postprocess with an overly large num_select."""
        model_context = getattr(model, "model", None)
        postprocess = getattr(model_context, "postprocess", None)
        if postprocess is None:
            return model

        model_config = getattr(model, "model_config", None)
        desired_num_select = getattr(model_config, "num_select", None)
        query_count = getattr(getattr(model_context, "args", None), "num_queries", None)

        if not isinstance(desired_num_select, int) or desired_num_select <= 0:
            desired_num_select = query_count
        if not isinstance(desired_num_select, int) or desired_num_select <= 0:
            return model
        if isinstance(query_count, int) and query_count > 0:
            desired_num_select = min(desired_num_select, query_count)

        # PostProcess upsamples one mask per selected candidate to the full target
        # resolution every frame, but only detections above threshold are kept. For
        # this app (a handful of animals) hundreds of candidates is pure waste, so
        # cap num_select to bound that cost without affecting real detections.
        desired_num_select = min(desired_num_select, _RFDETR_MAX_NUM_SELECT)

        setattr(postprocess, "num_select", desired_num_select)
        args = getattr(model_context, "args", None)
        if args is not None:
            try:
                args.num_select = desired_num_select
            except Exception:
                pass
        return model

    def _optimize_loaded_model(
        self, model, model_key: str, *, acceleration_mode: str = "balanced", checkpoint: str = ""
    ):
        if not str(model_key).startswith("rfdetr"):
            return model

        torch = import_torch(required=False)
        if torch is None:
            return model

        mode = str(acceleration_mode or "balanced").strip().lower().replace("-", "_").replace(" ", "_")

        # TensorRT path: replace rfdetr's inference_model with a serialized engine.
        # Falls back to the torch max_gpu path if no compatible engine is present.
        if mode == "max_gpu_trt":
            if self._attach_trt_engine(model, model_key, checkpoint, torch):
                return model
            mode = "max_gpu"
            self.status_changed.emit("TensorRT engine unavailable; falling back to max GPU (torch).")

        if not hasattr(model, "optimize_for_inference"):
            return model

        self._configure_torch_acceleration(torch, mode)
        optimize_kwargs = {"batch_size": 1}
        try:
            model_device = str(getattr(getattr(model, "model", None), "device", "") or "").lower()
        except Exception:
            model_device = ""
        use_cuda = torch.cuda.is_available() and "cuda" in model_device
        compatibility_mode = mode == "compatibility"
        optimize_kwargs["dtype"] = torch.float16 if (use_cuda and not compatibility_mode) else torch.float32
        compile_requested = bool(use_cuda and not compatibility_mode and mode == "max_gpu")
        compile_available = bool(
            compile_requested
            and not _running_from_pyinstaller_bundle()
            and _triton_nvidia_driver_source_available()
        )
        optimize_kwargs["compile"] = compile_available

        try:
            self.status_changed.emit("Optimizing RF-DETR for live inference")
            model.optimize_for_inference(**optimize_kwargs)
            if compile_requested and not compile_available:
                self.status_changed.emit(
                    "RF-DETR torch.compile disabled; using CUDA eager mode to avoid Triton runtime errors."
                )
            self.status_changed.emit(
                f"RF-DETR optimized ({'fp16' if optimize_kwargs['dtype'] == torch.float16 else 'fp32'})"
            )
        except Exception as exc:
            self.status_changed.emit(f"RF-DETR optimization skipped: {exc}")
        return model

    def _attach_trt_engine(self, model, model_key: str, checkpoint: str, torch) -> bool:
        """Replace rfdetr's inference_model with a serialized TensorRT engine.

        Returns True when a compatible sidecar ``.engine`` is loaded and wired in,
        False otherwise (caller falls back to the torch optimization path).
        """
        if not hasattr(model, "optimize_for_inference"):
            return False
        if not (torch.cuda.is_available()):
            self.status_changed.emit("TensorRT needs a CUDA device; using torch path.")
            return False

        try:
            import rfdetr_trt
        except Exception as exc:  # pragma: no cover - import guard
            self.status_changed.emit(f"TensorRT runtime not installed ({exc}); using torch path.")
            return False

        if not checkpoint:
            self.status_changed.emit("TensorRT mode needs a checkpoint with a sidecar .engine; using torch path.")
            return False

        engine_path = rfdetr_trt.engine_path_for_checkpoint(checkpoint)
        if not engine_path.is_file():
            self.status_changed.emit(
                f"No TensorRT engine at {engine_path.name}. Build it with scripts/build_rfdetr_engine.py."
            )
            return False

        # Warn (but proceed) if the engine was built with a different TensorRT build.
        metadata = rfdetr_trt.read_engine_metadata(engine_path)
        current_trt = rfdetr_trt.tensorrt_version()
        if metadata and metadata.get("tensorrt_version") and current_trt:
            if str(metadata["tensorrt_version"]) != str(current_trt):
                self.status_changed.emit(
                    f"Engine built with TensorRT {metadata['tensorrt_version']} but {current_trt} is installed; "
                    "rebuild if loading fails."
                )

        model_context = getattr(model, "model", None)
        device = getattr(model_context, "device", None)
        if model_context is None or device is None or "cuda" not in str(device).lower():
            self.status_changed.emit("RF-DETR model is not on CUDA; cannot use TensorRT.")
            return False

        try:
            self.status_changed.emit("Loading RF-DETR TensorRT engine")
            engine_module = rfdetr_trt.RFDETRTensorRTModule(engine_path, device=str(device))
        except Exception as exc:
            self.status_changed.emit(f"TensorRT engine load failed: {exc}")
            return False

        # Wire the engine in where rfdetr's optimized inference_model would sit, so
        # _predict_rfdetr_direct drives it with the existing pre/post-processing.
        model.remove_optimized_model()
        model_context.inference_model = engine_module
        model._is_optimized_for_inference = True
        model._optimized_resolution = int(engine_module.resolution)
        model._optimized_dtype = engine_module.input_dtype
        model._optimized_has_been_compiled = False
        model._optimized_batch_size = int(engine_module.batch_size)
        model._using_tensorrt = True

        self._configure_torch_acceleration(torch, "max_gpu_trt")
        self.status_changed.emit(
            f"RF-DETR TensorRT engine ready ({engine_module.resolution}px, fp16 compute)"
        )
        return True

    def _configure_torch_acceleration(self, torch, acceleration_mode: str) -> None:
        mode = str(acceleration_mode or "balanced").strip().lower().replace("-", "_").replace(" ", "_")
        max_gpu = mode in ("auto", "max_gpu", "max_gpu_trt")
        compatibility = mode == "compatibility"

        benchmark_enabled = bool(max_gpu)
        tf32_enabled = bool(max_gpu and torch.cuda.is_available())
        matmul_precision = "highest" if compatibility else ("high" if max_gpu else "highest")

        try:
            torch.backends.cudnn.benchmark = benchmark_enabled
        except Exception:
            pass
        try:
            torch.backends.cuda.matmul.allow_tf32 = tf32_enabled
        except Exception:
            pass
        try:
            torch.backends.cudnn.allow_tf32 = tf32_enabled
        except Exception:
            pass
        try:
            torch.set_float32_matmul_precision(matmul_precision)
        except Exception:
            pass

    def _describe_loaded_model(self, model, model_key: str, acceleration_mode: str = "balanced") -> str:
        try:
            model_config = getattr(model, "model_config", None)
            resolution = int(getattr(model_config, "resolution", 0) or 0)
            num_queries = int(getattr(model_config, "num_queries", 0) or 0)
            details: list[str] = []
            if resolution > 0:
                details.append(f"{resolution}px")
            if num_queries > 0:
                details.append(f"{num_queries} queries")
            mode = str(acceleration_mode or "balanced").strip().lower().replace("-", "_").replace(" ", "_")
            if getattr(model, "_using_tensorrt", False):
                precision = str(getattr(model, "precision", ""))
                details.append(
                    f"TensorRT {precision}" if precision in {"fp16", "fp32"} else "TensorRT"
                )
            elif getattr(model, "_using_compiled", False):
                details.append(f"compiled {getattr(model, 'precision', 'PyTorch')}")
            elif mode in ("max_gpu", "max_gpu_trt"):
                details.append("max GPU")
            elif mode == "compatibility":
                details.append("compatibility")
            if details:
                return f"{model_key} ({', '.join(details)})"
        except Exception:
            pass
        return str(model_key or "")

    def _rescale_detections(
        self,
        detections: dict[str, object],
        *,
        frame_shape: tuple[int, int],
        scale_x: float,
        scale_y: float,
    ) -> dict[str, object]:
        if not detections:
            return self._empty_detections()

        height, width = int(frame_shape[0]), int(frame_shape[1])
        result = dict(detections)
        boxes = np.asarray(detections.get("xyxy", np.empty((0, 4), dtype=float)), dtype=float).reshape(-1, 4).copy()
        if len(boxes):
            boxes[:, [0, 2]] *= float(scale_x)
            boxes[:, [1, 3]] *= float(scale_y)
            boxes[:, [0, 2]] = np.clip(boxes[:, [0, 2]], 0.0, max(0.0, width - 1.0))
            boxes[:, [1, 3]] = np.clip(boxes[:, [1, 3]], 0.0, max(0.0, height - 1.0))
        result["xyxy"] = boxes

        for mask_key in ("mask", "amodal_mask"):
            masks = detections.get(mask_key)
            if masks is None:
                continue
            resized_masks: list[object] = []
            source_boxes = np.asarray(
                detections.get("xyxy", np.empty((0, 4), dtype=float)),
                dtype=float,
            ).reshape(-1, 4)
            source_is_packed = self._is_packed_mask_sequence(masks)
            mask_items = list(masks) if source_is_packed else list(np.asarray(masks, dtype=bool))
            for index, mask in enumerate(mask_items):
                resized_masks.append(
                    self._resize_instance_mask_roi(
                        mask,
                        frame_shape=(height, width),
                        scale_x=scale_x,
                        scale_y=scale_y,
                        source_bbox=source_boxes[index] if index < len(source_boxes) else None,
                    )
                )
            if source_is_packed:
                result[mask_key] = resized_masks
            else:
                result[mask_key] = (
                    np.asarray(resized_masks, dtype=bool)
                    if resized_masks
                    else np.empty((0, height, width), dtype=bool)
                )
        return result

    def _resize_instance_mask_roi(
        self,
        mask: object,
        *,
        frame_shape: tuple[int, int],
        scale_x: float,
        scale_y: float,
        source_bbox: Optional[np.ndarray],
    ) -> object:
        """Resize one instance mask by its occupied ROI instead of the full frame."""
        height, width = int(frame_shape[0]), int(frame_shape[1])
        source_is_packed = is_crop_packed(mask)
        src_h, src_w = mask_shape(mask)

        def empty_output():
            if source_is_packed:
                return from_crop_like(mask, np.zeros((0, 0), bool), (height, width), 0, 0)
            return np.zeros((height, width), dtype=bool)

        if src_h <= 0 or src_w <= 0:
            return empty_output()

        sx1, sy1, sx2, sy2 = 0, 0, src_w, src_h
        bbox_limited = False
        if source_bbox is not None:
            try:
                bx1, by1, bx2, by2 = (float(value) for value in source_bbox)
                sx1 = max(0, int(np.floor(min(bx1, bx2))) - 4)
                sy1 = max(0, int(np.floor(min(by1, by2))) - 4)
                sx2 = min(src_w, int(np.ceil(max(bx1, bx2))) + 5)
                sy2 = min(src_h, int(np.ceil(max(by1, by2))) + 5)
                bbox_limited = sx2 > sx1 and sy2 > sy1
            except Exception:
                sx1, sy1, sx2, sy2 = 0, 0, src_w, src_h
                bbox_limited = False

        if bbox_limited:
            roi = mask_region(mask, sx1, sy1, sx2, sy2)
            base_x, base_y = sx1, sy1
        else:
            roi, base_x, base_y = occupied_crop(mask)
            if roi is None:
                return empty_output()
        ys, xs = np.nonzero(roi)
        if len(xs) == 0 or len(ys) == 0:
            if bbox_limited:
                roi, base_x, base_y = occupied_crop(mask)
                if roi is None:
                    return empty_output()
                ys, xs = np.nonzero(roi)
                if len(xs) == 0 or len(ys) == 0:
                    return empty_output()
                sx1 = int(base_x + xs.min())
                sx2 = int(base_x + xs.max()) + 1
                sy1 = int(base_y + ys.min())
                sy2 = int(base_y + ys.max()) + 1
            else:
                return empty_output()
        else:
            sx1 = int(base_x + xs.min())
            sx2 = int(base_x + xs.max()) + 1
            sy1 = int(base_y + ys.min())
            sy2 = int(base_y + ys.max()) + 1

        sx1 = max(0, sx1 - 2)
        sy1 = max(0, sy1 - 2)
        sx2 = min(src_w, sx2 + 2)
        sy2 = min(src_h, sy2 + 2)
        if sx2 <= sx1 or sy2 <= sy1:
            return empty_output()

        dx1 = max(0, int(np.floor(sx1 * float(scale_x))))
        dy1 = max(0, int(np.floor(sy1 * float(scale_y))))
        dx2 = min(width, int(np.ceil(sx2 * float(scale_x))))
        dy2 = min(height, int(np.ceil(sy2 * float(scale_y))))
        if dx2 <= dx1 or dy2 <= dy1:
            return empty_output()

        crop = mask_region(mask, sx1, sy1, sx2, sy2).astype(np.uint8)
        resized = cv2.resize(
            crop,
            (dx2 - dx1, dy2 - dy1),
            interpolation=cv2.INTER_NEAREST,
        ).astype(bool)
        if source_is_packed:
            return from_crop_like(mask, resized, (height, width), dy1, dx1)
        output = np.zeros((height, width), dtype=bool)
        output[dy1:dy2, dx1:dx2] = resized
        return output

    def _release_accelerator_memory(self, model) -> None:
        del model
        torch = import_torch(required=False)
        if torch is not None:
            try:
                if torch.cuda.is_available():
                    torch.cuda.empty_cache()
            except Exception:
                pass