fix: serialize gallery inference so Metal cannot be entered twice

The server segfaulted a few seconds into every live scan that used the crop
lane. The crash report put the fault in libtorch's Metal shader library
(`MetalShaderLibrary::exec_unary_kernel` inside an unordered_map insert),
reached from `copy_cast_kernel_mps` under `clone`/`reshape`.

Reproduced deterministically: eight threads calling `embed()` on a fresh
process crash it, while the same eight threads after one warm forward are
fine. Metal's shader library is process-global and is not thread-safe while
kernels are still being compiled into it, which is exactly what the first
forward does. The detector already serializes its predict for this reason;
the gallery did not.

Two changes:

* A module-level `_MPS_INFER_LOCK` covers the whole MPS region of a forward -
  model load, `.to(device)`, the forward, and the normalize that follows it.
  Normalize dispatches its own Metal kernels, so leaving it outside the lock
  left the same first-compile race open. Module-level because the state being
  protected is the process's, not one gallery's.
* A cache hit skips the embedding forward, so the model used to be built and
  the kernels compiled by whichever classify worker asked first - up to eight
  racing into one process-wide cache. The load path now warms the backbone
  single-threaded under the existing lock, on the cache-hit branch too. The
  readiness path loads the gallery before the socket accepts, so the warmup
  is free. It is Metal-only and best-effort: a CPU gallery has no shader
  library to compile, and a cache hit is served without a usable backbone by
  contract.

The regression test records overlap rather than timing: a fake backbone that
sleeps and counts concurrent entries, run from four threads, asserting the
count never exceeds one. It fails on the previous gallery.py and needs no MPS.

recognization/code/gallery.py

@@ -56,6 +56,17 @@ DINOV2_EMBED_DIM = 384
 CACHE_VERSION = 1
 
 
+# One thread at a time runs the gallery backbone. Metal's shader library is
+# process-global and is not thread-safe while kernels are still being compiled
+# into it, which is what the first forward on a fresh process does; two threads
+# reaching it together segfault the whole server. Module-level because the state
+# being protected is the process's, not one gallery's.
+# ponytail: serializes every embed. Audit C5's batching embedding worker is the
+# upgrade path, and a second MPS consumer (--device mps puts YOLO on Metal too)
+# would need to take this same lock rather than its own.
+_MPS_INFER_LOCK = threading.Lock()
+
+
 def dinov2_weights_path() -> str:
     override = os.environ.get("DORAYAKI_DINOV2_WEIGHTS")
     if override:
@@ -444,11 +455,12 @@ class ProductGallery:
         if not kept:
             return None
         try:
-            model = self._load_model()
-            tensor = torch.from_numpy(np.stack(kept)).to(self.device)
-            with torch.no_grad():
-                features = model(tensor)
-            features = torch.nn.functional.normalize(features, dim=1).cpu().numpy()
+            with _MPS_INFER_LOCK:
+                model = self._load_model()
+                tensor = torch.from_numpy(np.stack(kept)).to(self.device)
+                with torch.no_grad():
+                    features = model(tensor)
+                features = torch.nn.functional.normalize(features, dim=1).cpu().numpy()
         except Exception:
             return None
         return rows, features.astype(np.float32, copy=False)
@@ -492,6 +504,16 @@ class ProductGallery:
                 self._cache_write(key, rows, features)
         else:
             rows, features = cached
+            # A cache hit skips the embedding forward, so without this the model
+            # is built and Metal's kernels are compiled by whichever classify
+            # worker asks first - up to eight racing into one process-wide cache.
+            # The readiness path loads the gallery before the socket accepts, so
+            # the warmup is free and no worker meets a cold model.
+            #
+            # Best-effort on purpose: a cache hit is served without a usable
+            # backbone by contract, and _MPS_INFER_LOCK already makes the lazy
+            # path safe. This only spares the first worker the compile.
+            self._warm_backbone()
         self._entries = [
             GalleryEntry(sku_id=sku_id, view=path, embedding=features[index])
             for index, (sku_id, path) in enumerate(rows)
@@ -508,13 +530,45 @@ class ProductGallery:
         import numpy as np
         import torch
 
-        model = self._load_model()
-        tensor = torch.from_numpy(np.stack([_prepare(crop, self.image_size)])).to(self.device)
-        with torch.no_grad():
-            features = model(tensor)
-        features = torch.nn.functional.normalize(features, dim=1).cpu().numpy()
+        with _MPS_INFER_LOCK:
+            model = self._load_model()
+            tensor = torch.from_numpy(np.stack([_prepare(crop, self.image_size)])).to(self.device)
+            with torch.no_grad():
+                features = model(tensor)
+            features = torch.nn.functional.normalize(features, dim=1).cpu().numpy()
         return features[0]
 
+    def _warm_backbone(self) -> None:
+        """Build the backbone and compile its kernels, single-threaded.
+
+        Called under self._lock from the load path, so the lock order is
+        self._lock then _MPS_INFER_LOCK - the same order _embed_views takes, and
+        no path takes them reversed. The forward is a dummy canvas: what is
+        being warmed is the model and the shader library, not any one crop.
+
+        Metal only. A CPU gallery has no shader library to compile and a cache
+        hit is served without touching the backbone by contract, so warming one
+        would spend a model load to protect against nothing.
+        """
+        import numpy as np
+
+        if self._resolved_device() != "mps":
+            return
+        try:
+            self.embed(np.zeros((self.image_size, self.image_size, 3), dtype=np.uint8))
+        except Exception:
+            pass
+
+    def _resolved_device(self) -> str:
+        """The device the backbone will run on, without building it."""
+        if self.device is not None:
+            return self.device
+        import torch
+
+        if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available():
+            return "mps"
+        return "cpu"
+
     def _rank_index(self) -> Tuple[List[str], Any, Any]:
         """Segment bounds for the per-SKU maximum, in first-appearance order.