perf(agent): batch SigLIP crop embeds per image + load truncated images

Two issues surfaced by the live logs (GPU pegged at ~0% util, 0.5 jobs/s,
truncated-image failures):

- BATCH the SigLIP embeds: collect all of an image's crops (figure + booru_yolo
  components + panels) and embed them in ONE forward pass instead of one
  forward+lock per crop. The per-crop path serialised every crop through the
  inference lock and starved the GPU (≈0% util, autoscaler stuck oscillating);
  batching gives a real GPU-bound workload + far higher throughput. CCIP still
  runs per figure inline.
- LOAD_TRUNCATED_IMAGES in the agent (matches the server embedder): slightly-
  truncated scraped images now load instead of failing the job 3× then erroring
  ("image file is truncated (N bytes not processed)").

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

agent/fc_agent/embedder.py

@@ -58,12 +58,20 @@ class CropEmbedder:
 
     def embed(self, image: Image.Image) -> list[float]:
         """A crop → its embedding as a plain float list, ready to POST."""
+        return self.embed_batch([image])[0]
+
+    def embed_batch(self, images: list) -> list[list[float]]:
+        """Embed many crops in ONE forward pass — far better GPU utilisation +
+        only one lock acquisition than embedding each crop separately (which
+        starved the GPU and serialised the whole pool)."""
+        if not images:
+            return []
         self.load()
         torch = self._torch
-        enc = self._processor(images=image, return_tensors="pt")
+        enc = self._processor(images=images, return_tensors="pt")
         pixel_values = enc["pixel_values"].to(self._device, self._dt)
         with self._infer_lock, torch.no_grad():
             out = self._model.get_image_features(pixel_values=pixel_values)
         pooled = out.pooler_output if hasattr(out, "pooler_output") else out
-        vec = pooled[0].float().cpu().numpy().astype(np.float32).reshape(-1)
-        return vec.tolist()
+        arr = pooled.float().cpu().numpy().astype(np.float32)
+        return [row.reshape(-1).tolist() for row in arr]

agent/fc_agent/media.py

@@ -6,7 +6,12 @@ import os
 import subprocess
 import tempfile
 
-from PIL import Image
+from PIL import Image, ImageFile
+
+# Load slightly-truncated images (a few missing trailing bytes) instead of
+# raising — matches the server embedder. These are common in scraped libraries
+# and would otherwise fail the job 3× then error (operator-flagged 2026-06-30).
+ImageFile.LOAD_TRUNCATED_IMAGES = True
 
 
 def is_video(mime: str) -> bool:

agent/fc_agent/worker.py

@@ -348,17 +348,6 @@ class Worker:
             ccip_ev = self.cfg.ccip_model or "ccip-default"
             dv = f"person-{self.cfg.detector_level}"
 
-            def _concept(frame, bbox, t, score, detver, kind="concept"):
-                """A SigLIP region for one crop (None if below the size floor)."""
-                crop = crop_region(frame, bbox)
-                if crop is None:
-                    return None
-                return {
-                    "kind": kind, "bbox": list(bbox), "frame_time": t,
-                    "score": score, "siglip_embedding": embedder.embed(crop),
-                    "embedding_version": embed_version, "detector_version": detver,
-                }
-
             for t, frame in frames:
                 # FIGURE boxes: imgutils detect_person ∪ general COCO person,
                 # NMS-merged → CCIP identity (+ a concept crop). Covers anime +
@@ -367,6 +356,11 @@ class Worker:
                 figs = proposers.figures(frame, base)
                 if not figs:
                     figs = [((0.0, 0.0, 1.0, 1.0), 1.0, "whole")]  # whole-frame fallback
+
+                # Collect every crop that needs a SigLIP embedding, then embed
+                # them in ONE batched forward pass (huge GPU-util + throughput
+                # win vs one forward per crop). CCIP runs per figure inline.
+                pending = []   # (crop, region-template-without-embedding)
                 for bbox, score, _label in figs:
                     crop = crop_region(frame, bbox)
                     if crop is None:
@@ -381,25 +375,33 @@ class Worker:
                             "embedding_version": ccip_ev, "detector_version": dv,
                         })
                     if want_siglip:
-                        regions.append({
+                        pending.append((crop, {
                             "kind": "concept", "bbox": list(bbox), "frame_time": t,
-                            "score": score,
-                            "siglip_embedding": embedder.embed(crop),
-                            "embedding_version": embed_version, "detector_version": dv,
-                        })
+                            "score": score, "detector_version": dv,
+                        }))
                 if not want_siglip:
                     continue
-                # ANATOMY components (booru_yolo: head/cat-head/anatomy/…) →
-                # concept crops only (not full characters, so no CCIP).
+                # ANATOMY components (booru_yolo) + PANELS → concept/panel crops.
                 for bbox, score, label in proposers.components(frame):
-                    r = _concept(frame, bbox, t, score, f"booru:{label}")
-                    if r is not None:
-                        regions.append(r)
-                # PANEL crops (comic page → panels) → kind='panel' (still SigLIP).
+                    crop = crop_region(frame, bbox)
+                    if crop is not None:
+                        pending.append((crop, {
+                            "kind": "concept", "bbox": list(bbox), "frame_time": t,
+                            "score": score, "detector_version": f"booru:{label}",
+                        }))
                 for bbox, score, _label in proposers.panels(frame):
-                    r = _concept(frame, bbox, t, score, "panel", kind="panel")
-                    if r is not None:
-                        regions.append(r)
+                    crop = crop_region(frame, bbox)
+                    if crop is not None:
+                        pending.append((crop, {
+                            "kind": "panel", "bbox": list(bbox), "frame_time": t,
+                            "score": score, "detector_version": "panel",
+                        }))
+                if pending:
+                    vecs = embedder.embed_batch([c for c, _ in pending])
+                    for (_c, tmpl), vec in zip(pending, vecs):
+                        tmpl["siglip_embedding"] = vec
+                        tmpl["embedding_version"] = embed_version
+                        regions.append(tmpl)
             self.client.submit(job["job_id"], regions, replace_kinds)
             self._bump(processed=1)
             return True