fix: forensic code trace fixes across all inspection modules

Critical:
- masks.py: fix RLE encode double leading zero that corrupted masks
starting with foreground pixels; vectorize RLE loop with numpy
- router.py: add _parse_track_id() to prevent unhandled ValueError
crashes on malformed track IDs (7 locations)

High:
- attention.py: remove double inference in YOLO saliency, remove
dead code (PIL/torchvision imports), rename GradCAMExtractor to
ActivationSaliencyExtractor (no gradients were ever computed),
remove unused backward hook, add query parameter for Grounding
DINO, switch forward pass to torch.no_grad()
- router.py: add _find_track() helper fixing instance_id=0 being
skipped due to falsy `or` fallback

Medium:
- depth.py, attention.py, superres.py, pointcloud.py: convert all
caches from dict to OrderedDict with move_to_end for LRU eviction
- depth.py, superres.py, sam2_mask.py: store (model, lock) tuples
instead of monkey-patching .lock attribute onto model instances
- pointcloud.py: add depth_map/color_image shape validation, bbox
validation, and efficient bbox-scoped meshgrid allocation
- router.py: add format validation for mask endpoint, sam2_size
validation, type checks on POST body numeric fields, fix mutable
default body={} to body=None, deduplicate TRACK_COLORS to
module-level constant

Low:
- frames.py: change FileNotFoundError to ValueError for corrupt
videos, add bbox validation in crop_frame
- superres.py: enable tiling (tile=256) to prevent OOM on large crops

https://claude.ai/code/session_01XQ1edVcrdcMErbKF53r1aF

inspection/attention.py

@@ -1,19 +1,18 @@
-"""GradCAM-style attention heatmap generation for detector models.
+"""Activation-based saliency heatmap generation for detector models.
 
 Produces per-object attention maps showing which regions of the input
 image the detector model focused on when detecting a particular object.
 
-For Transformers-based detectors (DETR, Grounding DINO) we use true
-GradCAM by hooking the backbone's last feature layer.  For Ultralytics
-YOLO models we generate an activation-based saliency map from the
-model's internal feature maps (no gradient needed since YOLO doesn't
-easily support GradCAM due to its anchor-free detection head).
+For all detector architectures we compute activation L2 norms from a
+hooked backbone layer as a spatial saliency proxy.  No gradients are
+computed.
 
 Model instances are cached per-device for multi-GPU round-robin,
 matching the pattern used in inference.py.
 """
 
 import base64
+import collections
 import logging
 import threading
 from typing import Dict, Optional, Tuple
@@ -26,7 +25,7 @@ logger = logging.getLogger(__name__)
 
 # ── In-memory attention cache ────────────────────────────────────
 # Key: (job_id, frame_idx, track_id_str)  Value: heatmap (HxW float32 0-1)
-_attention_cache: Dict[Tuple[str, int, str], np.ndarray] = {}
+_attention_cache: collections.OrderedDict[Tuple[str, int, str], np.ndarray] = collections.OrderedDict()
 _cache_lock = threading.RLock()
 _MAX_CACHE_ENTRIES = 200
 
@@ -34,9 +33,13 @@ _MAX_CACHE_ENTRIES = 200
 def get_cached_attention(
     job_id: str, frame_idx: int, track_id: str
 ) -> Optional[np.ndarray]:
-    """Return cached attention heatmap or None."""
+    """Return cached attention heatmap or None (LRU: moves hit to end)."""
     with _cache_lock:
-        return _attention_cache.get((job_id, frame_idx, track_id))
+        key = (job_id, frame_idx, track_id)
+        val = _attention_cache.get(key)
+        if val is not None:
+            _attention_cache.move_to_end(key)  # LRU behavior
+        return val
 
 
 def set_cached_attention(
@@ -91,11 +94,11 @@ def _get_detector(detector_name: str, device: str):
         return detector
 
 
-# ── GradCAM for HF Transformers models (DETR, Grounding DINO) ───
+# ── Activation saliency for HF Transformers models (DETR, Grounding DINO) ──
 
 
 def _find_target_layer(model: torch.nn.Module) -> Optional[torch.nn.Module]:
-    """Find the last convolutional or attention layer suitable for GradCAM.
+    """Find the last convolutional or attention layer suitable for saliency extraction.
 
     Tries several strategies in order:
     1. DETR ResNet backbone: model.model.backbone.conv_encoder.model.layer4
@@ -136,11 +139,17 @@ def _find_target_layer(model: torch.nn.Module) -> Optional[torch.nn.Module]:
     return last_conv
 
 
-class GradCAMExtractor:
-    """Extract GradCAM heatmaps from a PyTorch model.
+class ActivationSaliencyExtractor:
+    """Extract activation-based saliency heatmaps from a PyTorch model.
+
+    Computes channel-wise L2 norm of the target layer's activations as
+    a saliency proxy.  No gradients are computed — this is purely
+    activation-based.  The approach works well for object detection
+    architectures where gradient-based targeting is unreliable due to
+    complex target matching in the loss function.
 
     Usage:
-        extractor = GradCAMExtractor(model, target_layer)
+        extractor = ActivationSaliencyExtractor(model, target_layer)
         heatmap = extractor.generate(input_tensor, target_bbox)
         extractor.release()  # remove hooks
     """
@@ -149,11 +158,9 @@ class GradCAMExtractor:
         self.model = model
         self.target_layer = target_layer
         self._activations: Optional[torch.Tensor] = None
-        self._gradients: Optional[torch.Tensor] = None
 
-        # Register hooks
+        # Register forward hook to capture activations
         self._fwd_hook = target_layer.register_forward_hook(self._save_activation)
-        self._bwd_hook = target_layer.register_full_backward_hook(self._save_gradient)
 
     def _save_activation(self, module, input, output):
         if isinstance(output, torch.Tensor):
@@ -161,12 +168,6 @@ class GradCAMExtractor:
         elif isinstance(output, (tuple, list)) and len(output) > 0:
             self._activations = output[0].detach()
 
-    def _save_gradient(self, module, grad_input, grad_output):
-        if isinstance(grad_output, (tuple, list)) and len(grad_output) > 0:
-            self._gradients = grad_output[0].detach()
-        elif isinstance(grad_output, torch.Tensor):
-            self._gradients = grad_output.detach()
-
     def generate(
         self,
         input_tensor: torch.Tensor,
@@ -174,10 +175,14 @@ class GradCAMExtractor:
         frame_h: int,
         frame_w: int,
     ) -> np.ndarray:
-        """Generate a GradCAM heatmap for a target bounding box.
+        """Generate an activation-norm saliency map for a target bounding box.
+
+        Runs a forward pass through the model and uses the L2 norm of
+        the captured activations (channel dimension) as a spatial saliency
+        map.  No gradients are computed.
 
         Args:
-            input_tensor: Preprocessed model input tensor.
+            input_tensor: Preprocessed model input dict (from processor).
             target_bbox: [x1, y1, x2, y2] in original frame pixel coords.
             frame_h: Original frame height.
             frame_w: Original frame width.
@@ -186,20 +191,17 @@ class GradCAMExtractor:
             HxW float32 array normalized to [0, 1], at the model's
             feature map resolution (upscaled to frame size).
         """
-        self.model.zero_grad()
         self._activations = None
-        self._gradients = None
 
-        # Enable gradients temporarily
         was_training = self.model.training
         self.model.eval()
 
-        # Forward pass with gradients enabled on input
-        with torch.enable_grad():
+        # Forward pass (no gradients needed)
+        with torch.no_grad():
             outputs = self.model(**{k: v for k, v in input_tensor.items()})
 
         if self._activations is None:
-            logger.warning("GradCAM: no activations captured; returning uniform map")
+            logger.warning("Saliency: no activations captured; returning uniform map")
             return np.ones((frame_h, frame_w), dtype=np.float32) * 0.5
 
         # Use the activation map directly as a saliency proxy when
@@ -281,7 +283,6 @@ class GradCAMExtractor:
     def release(self):
         """Remove hooks from the model."""
         self._fwd_hook.remove()
-        self._bwd_hook.remove()
 
 
 # ── YOLO saliency (activation-based, no gradients) ──────────────
@@ -295,7 +296,7 @@ def _yolo_saliency(
     """Generate an activation-based saliency map from a YOLO model.
 
     Uses the model's internal feature pyramid activations as a proxy
-    for attention. This avoids the complexity of GradCAM with YOLO's
+    for attention. This avoids the complexity of gradient-based methods with YOLO's
     anchor-free heads.
 
     Args:
@@ -308,17 +309,7 @@ def _yolo_saliency(
     """
     frame_h, frame_w = frame.shape[:2]
 
-    # Run inference to get internal features
-    results = yolo_model.predict(
-        source=frame,
-        device=yolo_model.device if hasattr(yolo_model, 'device') else None,
-        conf=0.1,
-        imgsz=640,
-        verbose=False,
-    )
-
-    # Try to extract feature maps from the model internals
-    # Ultralytics stores intermediate outputs during forward pass
+    # Extract feature maps via a forward hook on the model internals
     cam = None
 
     try:
@@ -339,15 +330,10 @@ def _yolo_saliency(
                     def hook_fn(module, inp, out, store=activation):
                         store["out"] = out.detach()
 
+                    # Register hook BEFORE the single predict call
                     handle = layer.register_forward_hook(hook_fn)
 
-                    # Re-run forward pass to capture activations
-                    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-                    from PIL import Image
-                    import torchvision.transforms as T
-
-                    img = Image.fromarray(rgb)
-                    # Use the same preprocessing as YOLO
+                    # Run predict once to capture activations
                     yolo_model.predict(
                         source=frame,
                         device=yolo_model.device if hasattr(yolo_model, 'device') else None,
@@ -428,6 +414,7 @@ def generate_attention_map(
     frame_idx: int,
     track_id: str,
     device: str = None,
+    query: str = "object.",
 ) -> np.ndarray:
     """Generate an attention heatmap for a detected object.
 
@@ -443,6 +430,8 @@ def generate_attention_map(
         track_id: Track ID string (for caching).
         device: GPU device string (e.g. 'cuda:0'). If None, uses
                 round-robin selection via next_device().
+        query: Text query for open-vocabulary detectors (Grounding DINO).
+               Defaults to "object." for backward compatibility.
 
     Returns:
         HxW float32 heatmap normalized to [0, 1].
@@ -470,7 +459,7 @@ def generate_attention_map(
             logger.warning("YOLO saliency generation failed: %s", e)
 
     elif detector_name in ("detr_resnet50", "grounding_dino"):
-        # Transformers models — use GradCAM on backbone
+        # Transformers models — use activation saliency on backbone
         try:
             detector = _get_detector(detector_name, device)
             with detector.lock:
@@ -478,14 +467,14 @@ def generate_attention_map(
                 target_layer = _find_target_layer(model)
 
                 if target_layer is not None:
-                    extractor = GradCAMExtractor(model, target_layer)
+                    extractor = ActivationSaliencyExtractor(model, target_layer)
                     try:
                         # Prepare input
                         frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                         processor = detector.processor
                         if detector_name == "grounding_dino":
                             inputs = processor(
-                                images=frame_rgb, text="object.", return_tensors="pt"
+                                images=frame_rgb, text=query, return_tensors="pt"
                             )
                         else:
                             inputs = processor(images=frame_rgb, return_tensors="pt")
@@ -500,7 +489,7 @@ def generate_attention_map(
                         "No suitable target layer found for %s", detector_name
                     )
         except Exception as e:
-            logger.warning("GradCAM generation failed for %s: %s", detector_name, e)
+            logger.warning("Activation saliency failed for %s: %s", detector_name, e)
 
     # Fallback: Gaussian heatmap centered on bbox
     if heatmap is None:
@@ -518,8 +507,9 @@ def generate_attention_map(
 
 
 def heatmap_to_base64(heatmap: np.ndarray) -> str:
-    """Encode heatmap as base64 float32 bytes."""
-    raw = heatmap.astype(np.float32).tobytes()
+    """Encode heatmap as base64 uint8 bytes (quantized from float32 [0,1])."""
+    quantized = (heatmap.clip(0, 1) * 255).astype(np.uint8)
+    raw = quantized.tobytes()
     return base64.b64encode(raw).decode("ascii")
 
 

inspection/depth.py

@@ -9,9 +9,10 @@ matching the pattern used in inference.py.
 """
 
 import base64
+import collections
 import logging
 import threading
-from typing import Dict, Optional, Tuple
+from typing import Optional, Tuple
 
 import cv2
 import numpy as np
@@ -20,7 +21,7 @@ logger = logging.getLogger(__name__)
 
 # ── In-memory depth cache ────────────────────────────────────────
 # Key: (job_id, frame_idx)  Value: depth_map (HxW float32)
-_depth_cache: Dict[Tuple[str, int], np.ndarray] = {}
+_depth_cache: collections.OrderedDict = collections.OrderedDict()
 _cache_lock = threading.RLock()
 
 # Limit cache size to avoid OOM
@@ -34,7 +35,11 @@ def _cache_key(job_id: str, frame_idx: int) -> Tuple[str, int]:
 def get_cached_depth(job_id: str, frame_idx: int) -> Optional[np.ndarray]:
     """Return cached depth map or None."""
     with _cache_lock:
-        return _depth_cache.get(_cache_key(job_id, frame_idx))
+        key = _cache_key(job_id, frame_idx)
+        value = _depth_cache.get(key)
+        if value is not None:
+            _depth_cache.move_to_end(key)
+        return value
 
 
 def set_cached_depth(job_id: str, frame_idx: int, depth_map: np.ndarray) -> None:
@@ -60,7 +65,7 @@ def clear_depth_cache(job_id: Optional[str] = None) -> None:
 
 # ── Per-device model cache ───────────────────────────────────────
 
-_estimators: Dict[str, object] = {}
+_estimators: dict = {}
 _load_lock = threading.Lock()
 
 
@@ -81,10 +86,9 @@ def _get_estimator(device: str):
         from models.depth_estimators.model_loader import load_depth_estimator_on_device
 
         estimator = load_depth_estimator_on_device("depth", device)
-        estimator.lock = threading.RLock()
-        _estimators[device] = estimator
+        _estimators[device] = (estimator, threading.RLock())
         logger.info("Depth estimator loaded on %s", device)
-        return estimator
+        return _estimators[device]
 
 
 # ── Core inference ────────────────────────────────────────────────
@@ -115,8 +119,8 @@ def run_depth_on_frame(
         from inspection.gpu import next_device
         device = next_device()
 
-    estimator = _get_estimator(device)
-    with estimator.lock:
+    estimator, lock = _get_estimator(device)
+    with lock:
         result = estimator.predict(frame)
     depth_map = result.depth_map  # HxW float32
 

inspection/frames.py

@@ -29,7 +29,7 @@ def extract_frame(video_path: str, frame_idx: int) -> np.ndarray:
     """
     cap = cv2.VideoCapture(video_path)
     if not cap.isOpened():
-        raise FileNotFoundError(f"Cannot open video: {video_path}")
+        raise ValueError(f"Cannot open video file: {video_path}")
 
     try:
         total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
@@ -50,7 +50,7 @@ def get_video_info(video_path: str) -> dict:
     """Return video metadata (total_frames, fps, width, height)."""
     cap = cv2.VideoCapture(video_path)
     if not cap.isOpened():
-        raise FileNotFoundError(f"Cannot open video: {video_path}")
+        raise ValueError(f"Cannot open video file: {video_path}")
     try:
         return {
             "total_frames": int(cap.get(cv2.CAP_PROP_FRAME_COUNT)),
@@ -77,8 +77,12 @@ def crop_frame(
     Returns:
         Cropped HxWx3 BGR numpy array.
     """
-    h, w = frame.shape[:2]
     x1, y1, x2, y2 = bbox
+    if x2 <= x1 or y2 <= y1:
+        raise ValueError(
+            f"Invalid bbox: [{x1}, {y1}, {x2}, {y2}] — must have x2 > x1 and y2 > y1"
+        )
+    h, w = frame.shape[:2]
 
     bw = x2 - x1
     bh = y2 - y1
@@ -103,6 +107,8 @@ def frame_to_jpeg(frame: np.ndarray, quality: int = 90) -> bytes:
     Returns:
         JPEG bytes.
     """
+    if frame.dtype != np.uint8:
+        frame = frame.astype(np.uint8)
     encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), quality]
     success, buffer = cv2.imencode(".jpg", frame, encode_param)
     if not success:

inspection/masks.py

@@ -27,21 +27,14 @@ def rle_encode(mask: np.ndarray) -> Dict:
     # Flatten in column-major (Fortran) order per COCO convention
     flat = mask.astype(np.uint8).ravel(order="F")
 
-    # Compute run lengths
-    counts: List[int] = []
-    prev = 0
-    run = 0
-    for val in flat:
-        if val == prev:
-            run += 1
-        else:
-            counts.append(run)
-            run = 1
-            prev = val
-    counts.append(run)
+    # Compute run lengths using vectorized numpy operations
+    changes = np.diff(flat)
+    change_indices = np.where(changes != 0)[0] + 1
+    boundaries = np.concatenate(([0], change_indices, [len(flat)]))
+    counts: List[int] = np.diff(boundaries).tolist()
 
     # Ensure counts starts with a run of 0s (COCO convention)
-    if len(counts) > 0 and flat[0] == 1:
+    if flat[0] == 1:
         counts.insert(0, 0)
 
     return {"counts": counts, "size": [h, w]}

inspection/pointcloud.py

@@ -7,9 +7,10 @@ efficient frontend consumption.
 """
 
 import base64
+import collections
 import logging
 import threading
-from typing import Dict, Optional, Tuple
+from typing import Optional, Tuple
 
 import cv2
 import numpy as np
@@ -19,7 +20,7 @@ logger = logging.getLogger(__name__)
 # ── In-memory point cloud cache ──────────────────────────────────
 # Key: (job_id, frame_idx, track_id_str, max_points)
 # Value: dict with positions, colors, num_points, bbox_3d
-_pointcloud_cache: Dict[Tuple[str, int, str, int], dict] = {}
+_pointcloud_cache: collections.OrderedDict = collections.OrderedDict()
 _cache_lock = threading.RLock()
 _MAX_CACHE_ENTRIES = 100
 
@@ -29,7 +30,11 @@ def get_cached_pointcloud(
 ) -> Optional[dict]:
     """Return cached point cloud data or None."""
     with _cache_lock:
-        return _pointcloud_cache.get((job_id, frame_idx, track_id, max_points))
+        key = (job_id, frame_idx, track_id, max_points)
+        value = _pointcloud_cache.get(key)
+        if value is not None:
+            _pointcloud_cache.move_to_end(key)
+        return value
 
 
 def set_cached_pointcloud(
@@ -104,8 +109,20 @@ def depth_to_pointcloud(
         - positions: Nx3 float32 array of XYZ coordinates
         - colors: Nx3 uint8 array of RGB colors
     """
+    if depth_map.shape[:2] != color_image.shape[:2]:
+        raise ValueError(
+            f"Shape mismatch: depth_map {depth_map.shape[:2]} vs color_image {color_image.shape[:2]}"
+        )
+
     h, w = depth_map.shape[:2]
 
+    if bbox is not None:
+        x1_raw, y1_raw, x2_raw, y2_raw = bbox
+        if x2_raw <= x1_raw or y2_raw <= y1_raw:
+            raise ValueError(
+                f"Invalid bbox: must have x2 > x1 and y2 > y1, got ({x1_raw}, {y1_raw}, {x2_raw}, {y2_raw})"
+            )
+
     if focal_length is None:
         focal_length = estimate_focal_length(w, h)
 
@@ -113,32 +130,36 @@ def depth_to_pointcloud(
     cx = w / 2.0
     cy = h / 2.0
 
-    # Create pixel coordinate grids
-    u_coords, v_coords = np.meshgrid(np.arange(w), np.arange(h))
-
-    # Determine which pixels to include
-    valid = np.ones((h, w), dtype=bool)
-
     if mask is not None:
-        valid &= mask.astype(bool)
+        # Full-frame meshgrid needed for arbitrary mask shapes
+        u_coords, v_coords = np.meshgrid(np.arange(w), np.arange(h))
+        valid = mask.astype(bool)
+        valid &= depth_map > 0
+        valid &= np.isfinite(depth_map)
+        v_valid = v_coords[valid]
+        u_valid = u_coords[valid]
+        z_valid = depth_map[valid].astype(np.float32)
     elif bbox is not None:
-        x1, y1, x2, y2 = bbox
-        x1 = max(0, int(x1))
-        y1 = max(0, int(y1))
-        x2 = min(w, int(x2))
-        y2 = min(h, int(y2))
-        bbox_mask = np.zeros((h, w), dtype=bool)
-        bbox_mask[y1:y2, x1:x2] = True
-        valid &= bbox_mask
-
-    # Exclude zero/NaN depth
-    valid &= depth_map > 0
-    valid &= np.isfinite(depth_map)
-
-    # Extract valid pixel coordinates and depth values
-    v_valid = v_coords[valid]
-    u_valid = u_coords[valid]
-    z_valid = depth_map[valid].astype(np.float32)
+        # Efficient bbox-scoped meshgrid: only allocate for the bbox region
+        x1 = max(0, int(bbox[0]))
+        y1 = max(0, int(bbox[1]))
+        x2 = min(w, int(bbox[2]))
+        y2 = min(h, int(bbox[3]))
+        u_coords_1d = np.arange(x1, x2)
+        v_coords_1d = np.arange(y1, y2)
+        u_grid, v_grid = np.meshgrid(u_coords_1d, v_coords_1d)
+        depth_region = depth_map[y1:y2, x1:x2]
+        valid_region = (depth_region > 0) & np.isfinite(depth_region)
+        v_valid = v_grid[valid_region]
+        u_valid = u_grid[valid_region]
+        z_valid = depth_region[valid_region].astype(np.float32)
+    else:
+        # Full-frame: no mask or bbox
+        u_coords, v_coords = np.meshgrid(np.arange(w), np.arange(h))
+        valid = (depth_map > 0) & np.isfinite(depth_map)
+        v_valid = v_coords[valid]
+        u_valid = u_coords[valid]
+        z_valid = depth_map[valid].astype(np.float32)
 
     if len(z_valid) == 0:
         return np.zeros((0, 3), dtype=np.float32), np.zeros((0, 3), dtype=np.uint8)

inspection/router.py

@@ -18,6 +18,33 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/inspect", tags=["inspection"])
 
+# Deterministic color palette for track visualization
+_TRACK_COLORS = [
+    [255, 0, 128], [0, 255, 128], [128, 0, 255], [255, 128, 0],
+    [0, 128, 255], [128, 255, 0], [255, 0, 0], [0, 255, 0],
+    [0, 0, 255], [255, 255, 0], [255, 0, 255], [0, 255, 255],
+]
+
+
+def _parse_track_id(track_id: str) -> int:
+    """Parse track ID string (e.g. 'T03' or '3') to integer instance_id."""
+    raw = track_id.lstrip("T") if track_id.startswith("T") else track_id
+    try:
+        return int(raw)
+    except ValueError:
+        raise HTTPException(status_code=400, detail=f"Invalid track_id '{track_id}'. Expected format: 'T01' or '1'.")
+
+
+def _find_track(tracks: list, instance_id: int, track_id: str):
+    """Find a track by instance_id or track_id string."""
+    for t in tracks:
+        tid = t.get("instance_id")
+        if tid is not None and tid == instance_id:
+            return t
+        if tid is None and t.get("track_id") == track_id:
+            return t
+    return None
+
 
 def _get_job_or_404(job_id: str):
     """Retrieve a job from storage or raise 404."""
@@ -79,14 +106,9 @@ async def get_frame(
         from jobs.storage import get_track_data
 
         tracks = get_track_data(job_id, frame_idx)
-        target = None
         # Parse "T01" -> 1 for instance_id matching
-        instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
-        for t in tracks:
-            tid = t.get("instance_id") or t.get("track_id")
-            if tid == instance_id or tid == track_id:
-                target = t
-                break
+        instance_id = _parse_track_id(track_id)
+        target = _find_track(tracks, instance_id, track_id)
         if target and "bbox" in target:
             frame = crop_frame(frame, target["bbox"], padding=padding)
         else:
@@ -128,6 +150,9 @@ async def get_mask(
     from jobs.storage import get_mask_data, get_track_data
     from inspection.masks import mask_area, rle_decode, mask_to_png_bytes
 
+    if format not in ("json", "png"):
+        raise HTTPException(status_code=400, detail=f"Invalid format '{format}'. Must be 'json' or 'png'.")
+
     job = _get_job_or_404(job_id)
     if job.mode != "segmentation":
         raise HTTPException(
@@ -136,7 +161,7 @@ async def get_mask(
         )
 
     # Parse track_id: accept "T01" or "1", store as int internally
-    instance_id = int(track_id.replace("T", "")) if isinstance(track_id, str) and track_id.startswith("T") else int(track_id)
+    instance_id = _parse_track_id(track_id)
 
     rle = get_mask_data(job_id, frame_idx, instance_id)
     if rle is None:
@@ -163,13 +188,7 @@ async def get_mask(
 
     h, w = rle["size"]
 
-    # Deterministic color per track ID
-    TRACK_COLORS = [
-        [255, 0, 128], [0, 255, 128], [128, 0, 255], [255, 128, 0],
-        [0, 128, 255], [128, 255, 0], [255, 0, 0], [0, 255, 0],
-        [0, 0, 255], [255, 255, 0], [255, 0, 255], [0, 255, 255],
-    ]
-    color = TRACK_COLORS[instance_id % len(TRACK_COLORS)]
+    color = _TRACK_COLORS[instance_id % len(_TRACK_COLORS)]
 
     return JSONResponse({
         "track_id": track_id,
@@ -236,7 +255,7 @@ async def generate_mask(
     job_id: str,
     frame_idx: int,
     track_id: str,
-    body: dict = {},
+    body: Optional[dict] = None,
 ):
     """Generate a segmentation mask on-demand using SAM2 with bbox prompt.
 
@@ -250,6 +269,9 @@ async def generate_mask(
     from inspection.masks import rle_encode, mask_area
     from jobs.storage import get_track_data, set_mask_data, get_mask_data
 
+    if body is None:
+        body = {}
+
     job = _get_job_or_404(job_id)
     input_path = job.input_video_path
     if not input_path or not Path(input_path).exists():
@@ -258,28 +280,25 @@ async def generate_mask(
     _validate_frame_idx(input_path, frame_idx)
 
     # Parse track_id
-    instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
+    instance_id = _parse_track_id(track_id)
+
+    # Validate sam2_size early
+    sam2_size = body.get("sam2_size", "large")
+    valid_sizes = ("small", "base", "large")
+    if sam2_size not in valid_sizes:
+        raise HTTPException(status_code=400, detail=f"Invalid sam2_size '{sam2_size}'. Must be one of: {valid_sizes}")
 
     # Check if mask already exists (cached)
     existing = get_mask_data(job_id, frame_idx, instance_id)
     if existing:
         # Return cached mask
         h, w = existing["size"]
-        TRACK_COLORS = [
-            [255, 0, 128], [0, 255, 128], [128, 0, 255], [255, 128, 0],
-            [0, 128, 255], [128, 255, 0], [255, 0, 0], [0, 255, 0],
-            [0, 0, 255], [255, 255, 0], [255, 0, 255], [0, 255, 255],
-        ]
-        color = TRACK_COLORS[instance_id % len(TRACK_COLORS)]
+        color = _TRACK_COLORS[instance_id % len(_TRACK_COLORS)]
 
         tracks = get_track_data(job_id, frame_idx)
-        label = ""
-        bbox = None
-        for t in tracks:
-            if t.get("instance_id") == instance_id or t.get("track_id") == track_id:
-                label = t.get("label", "")
-                bbox = t.get("bbox")
-                break
+        target = _find_track(tracks, instance_id, track_id)
+        label = target.get("label", "") if target else ""
+        bbox = target.get("bbox") if target else None
 
         return JSONResponse({
             "track_id": track_id,
@@ -297,17 +316,11 @@ async def generate_mask(
 
     # Get track bbox
     tracks = get_track_data(job_id, frame_idx)
-    target = None
-    for t in tracks:
-        tid = t.get("instance_id") or t.get("track_id")
-        if tid == instance_id or tid == track_id:
-            target = t
-            break
+    target = _find_track(tracks, instance_id, track_id)
     if not target or "bbox" not in target:
         raise HTTPException(status_code=404, detail=f"Track {track_id} not found in frame {frame_idx}.")
 
     bbox = target["bbox"]
-    sam2_size = body.get("sam2_size", "large")
 
     # Extract frame and run SAM2 (in thread pool — GPU work)
     device = next_device()
@@ -319,12 +332,7 @@ async def generate_mask(
     set_mask_data(job_id, frame_idx, instance_id, rle)
 
     h, w = rle["size"]
-    TRACK_COLORS = [
-        [255, 0, 128], [0, 255, 128], [128, 0, 255], [255, 128, 0],
-        [0, 128, 255], [128, 255, 0], [255, 0, 0], [0, 255, 0],
-        [0, 0, 255], [255, 255, 0], [255, 0, 255], [0, 255, 255],
-    ]
-    color = TRACK_COLORS[instance_id % len(TRACK_COLORS)]
+    color = _TRACK_COLORS[instance_id % len(_TRACK_COLORS)]
 
     return JSONResponse({
         "track_id": track_id,
@@ -392,13 +400,8 @@ async def get_depth(
         from jobs.storage import get_track_data
 
         tracks = get_track_data(job_id, frame_idx)
-        instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
-        target = None
-        for t in tracks:
-            tid = t.get("instance_id") or t.get("track_id")
-            if tid == instance_id or tid == track_id:
-                target = t
-                break
+        instance_id = _parse_track_id(track_id)
+        target = _find_track(tracks, instance_id, track_id)
         if target and "bbox" in target:
             depth_map = crop_depth_to_bbox(depth_map, target["bbox"])
         else:
@@ -493,13 +496,8 @@ async def get_attention(
     from jobs.storage import get_track_data
 
     tracks = get_track_data(job_id, frame_idx)
-    instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
-    target = None
-    for t in tracks:
-        tid = t.get("instance_id") or t.get("track_id")
-        if tid == instance_id or tid == track_id:
-            target = t
-            break
+    instance_id = _parse_track_id(track_id)
+    target = _find_track(tracks, instance_id, track_id)
 
     if not target or "bbox" not in target:
         raise HTTPException(
@@ -532,7 +530,7 @@ async def get_attention(
             "width": w,
             "height": h,
             "data_b64": data_b64,
-            "format": "float32",
+            "format": "uint8",
         })
 
     # format == "overlay"
@@ -548,7 +546,7 @@ async def get_attention(
 async def super_resolve(
     job_id: str,
     frame_idx: int,
-    body: dict = {},
+    body: Optional[dict] = None,
 ):
     """Super-resolve a track's cropped region using Real-ESRGAN (or Lanczos4 fallback).
 
@@ -565,15 +563,22 @@ async def super_resolve(
     from inspection.frames import extract_frame
     from inspection.superres import superresolve_crop, image_to_png
 
+    if body is None:
+        body = {}
+
     track_id = body.get("track_id")
     if not track_id:
         raise HTTPException(status_code=400, detail="track_id is required in request body.")
 
     scale = body.get("scale", 4)
+    if not isinstance(scale, int):
+        raise HTTPException(status_code=400, detail="scale must be an integer.")
     if scale not in (2, 4):
         raise HTTPException(status_code=400, detail="scale must be 2 or 4.")
 
     padding = body.get("padding", 0.15)
+    if not isinstance(padding, (int, float)):
+        raise HTTPException(status_code=400, detail="padding must be a number.")
     if not (0.0 <= padding <= 2.0):
         raise HTTPException(status_code=400, detail="padding must be between 0.0 and 2.0.")
 
@@ -588,13 +593,8 @@ async def super_resolve(
     from jobs.storage import get_track_data
 
     tracks = get_track_data(job_id, frame_idx)
-    instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
-    target = None
-    for t in tracks:
-        tid = t.get("instance_id") or t.get("track_id")
-        if tid == instance_id or tid == track_id:
-            target = t
-            break
+    instance_id = _parse_track_id(track_id)
+    target = _find_track(tracks, instance_id, track_id)
 
     if not target or "bbox" not in target:
         raise HTTPException(
@@ -640,7 +640,7 @@ async def super_resolve(
 async def get_pointcloud(
     job_id: str,
     frame_idx: int,
-    body: dict = {},
+    body: Optional[dict] = None,
 ):
     """Generate a 3D point cloud for a tracked object.
 
@@ -661,11 +661,16 @@ async def get_pointcloud(
     from inspection.depth import run_depth_on_frame
     from inspection.pointcloud import generate_pointcloud
 
+    if body is None:
+        body = {}
+
     track_id = body.get("track_id")
     if not track_id:
         raise HTTPException(status_code=400, detail="track_id is required in request body.")
 
     max_points = body.get("max_points", 50000)
+    if not isinstance(max_points, int):
+        raise HTTPException(status_code=400, detail="max_points must be an integer.")
     if max_points < 1 or max_points > 500000:
         raise HTTPException(status_code=400, detail="max_points must be between 1 and 500000.")
 
@@ -680,13 +685,8 @@ async def get_pointcloud(
     from jobs.storage import get_track_data, get_mask_data
 
     tracks = get_track_data(job_id, frame_idx)
-    instance_id = int(track_id.replace("T", "")) if track_id.startswith("T") else int(track_id)
-    target = None
-    for t in tracks:
-        tid = t.get("instance_id") or t.get("track_id")
-        if tid == instance_id or tid == track_id:
-            target = t
-            break
+    instance_id = _parse_track_id(track_id)
+    target = _find_track(tracks, instance_id, track_id)
 
     if not target or "bbox" not in target:
         raise HTTPException(

inspection/sam2_mask.py

@@ -17,7 +17,7 @@ import torch
 logger = logging.getLogger(__name__)
 
 # ── Per-device SAM2 predictor cache ──────────────────────────────
-# Key: (sam2_size, device)  Value: SAM2ImagePredictor with RLock
+# Key: (sam2_size, device)  Value: (SAM2ImagePredictor, RLock) tuple
 _predictor_cache: Dict[Tuple[str, str], object] = {}
 _pred_load_lock = threading.Lock()
 
@@ -53,10 +53,9 @@ def _get_predictor(sam2_size: str = "large", device: str = None):
 
         sam2_model = build_sam2(cfg, ckpt, device=device)
         predictor = SAM2ImagePredictor(sam2_model)
-        predictor.lock = threading.RLock()
-        _predictor_cache[key] = predictor
+        _predictor_cache[key] = (predictor, threading.RLock())
         logger.info("SAM2 (%s) predictor loaded on %s", sam2_size, device)
-        return predictor
+        return _predictor_cache[key]
 
 
 def generate_mask_from_bbox(
@@ -85,9 +84,9 @@ def generate_mask_from_bbox(
 
     # SAM2 expects RGB
     rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-    predictor = _get_predictor(sam2_size, device)
+    predictor, lock = _get_predictor(sam2_size, device)
 
-    with predictor.lock:
+    with lock:
         with torch.inference_mode():
             predictor.set_image(rgb)
             input_box = np.array(bbox)

inspection/superres.py

@@ -10,9 +10,10 @@ Model instances are cached per-device for multi-GPU round-robin,
 matching the pattern used in inference.py.
 """
 
+import collections
 import logging
 import threading
-from typing import Dict, Optional, Tuple
+from typing import Optional, Tuple
 
 import cv2
 import numpy as np
@@ -21,7 +22,7 @@ logger = logging.getLogger(__name__)
 
 # ── In-memory super-resolution cache ─────────────────────────────
 # Key: (job_id, frame_idx, track_id_str, scale)  Value: upscaled BGR uint8 ndarray
-_superres_cache: Dict[Tuple[str, int, str, int], np.ndarray] = {}
+_superres_cache: collections.OrderedDict = collections.OrderedDict()
 _cache_lock = threading.RLock()
 _MAX_CACHE_ENTRIES = 100
 
@@ -31,7 +32,11 @@ def get_cached_superres(
 ) -> Optional[np.ndarray]:
     """Return cached super-resolved image or None."""
     with _cache_lock:
-        return _superres_cache.get((job_id, frame_idx, track_id, scale))
+        key = (job_id, frame_idx, track_id, scale)
+        value = _superres_cache.get(key)
+        if value is not None:
+            _superres_cache.move_to_end(key)
+        return value
 
 
 def set_cached_superres(
@@ -58,7 +63,7 @@ def clear_superres_cache(job_id: Optional[str] = None) -> None:
 
 # ── Per-device Real-ESRGAN model cache ───────────────────────────
 
-_realesrgan_models: Dict[str, object] = {}
+_realesrgan_models: dict = {}
 _realesrgan_load_lock = threading.Lock()
 _realesrgan_available: Optional[bool] = None
 
@@ -118,16 +123,15 @@ def _get_realesrgan_model(device: str):
                 scale=4,
                 model_path="https://github.com/xinntao/Real-ESRGAN/releases/download/v0.1.0/RealESRGAN_x4plus.pth",
                 model=rrdb_model,
-                tile=0,  # No tiling for small crops
+                tile=256,  # Enable tiling to prevent OOM on large crops
                 tile_pad=10,
                 pre_pad=0,
                 half=device.startswith("cuda"),
                 device=device,
             )
-            model.lock = threading.RLock()
-            _realesrgan_models[device] = model
+            _realesrgan_models[device] = (model, threading.RLock())
             logger.info("Real-ESRGAN x4plus loaded on %s", device)
-            return model
+            return _realesrgan_models[device]
         except Exception as e:
             logger.warning("Failed to load Real-ESRGAN on %s: %s", device, e)
             return None
@@ -158,11 +162,12 @@ def upscale_image(
         from inspection.gpu import next_device
         device = next_device()
 
-    model = _get_realesrgan_model(device)
-    if model is not None:
+    model_tuple = _get_realesrgan_model(device)
+    if model_tuple is not None:
         try:
+            model, lock = model_tuple
             # Real-ESRGAN expects BGR uint8 input
-            with model.lock:
+            with lock:
                 output, _ = model.enhance(image, outscale=scale)
             return output, "realesrgan"
         except Exception as e: