ocr: make managed Surya-2 vLLM reclaimable by VRAM eviction

The Surya-2 engine serves its VLM via a managed vLLM subprocess
(vllm_worker.ensure_service). That subprocess is not a manager-tracked
model, and the OCR VRAM releaser only tore down the lightweight worker
pools (killing the HTTP workers) — it never stopped the vLLM holding the
bulk VRAM (gpu_memory_utilization × card). So on-request eviction saw the
VRAM as used, found nothing it could free, and (with no CPU fallback) a
docTR/Paddle load on the same card could OOM.

- vllm_worker: add stop_service_for(cfg, model_path, served_name) (stop ONE
  service, return estimated GB freed) + is_running() helper.
- OcrManager._release_vram now also stops the managed Surya-2 vLLM
  (_stop_surya_vllm) when surya_serve == "vllm", so the registered external
  releaser actually reclaims its VRAM. Next OCR request re-boots it via
  ensure_service. Targeted by (model_path, served_name) so an LLM vLLM
  instance is left alone.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Mw2KQiswmD69T45fTfjKwW

codai/api/vllm_worker.py

@@ -243,5 +243,43 @@ def stop_all() -> None:
         stop_service(k)
 
 
+def is_running(cfg, model_path: Optional[str] = None, served_name: Optional[str] = None) -> bool:
+    """True if a vLLM service for this (model_path, served_name) is currently up."""
+    _resolved, svc_key = resolve_service_key(cfg, model_path)
+    if served_name:
+        svc_key = f"{svc_key}|{served_name}"
+    with _lock:
+        svc = _services.get(svc_key)
+        return bool(svc and svc["proc"].poll() is None)
+
+
+def stop_service_for(cfg, model_path: Optional[str] = None,
+                     served_name: Optional[str] = None) -> float:
+    """Stop the vLLM service for ONE specific (model_path, served_name), if running.
+
+    Used by the OCR subsystem as a VRAM releaser: the managed Surya-2 vLLM subprocess
+    isn't a manager-tracked model, so this is what lets on-request eviction reclaim its
+    VRAM. Returns an estimate of the GB it was holding (gpu_memory_utilization × total
+    VRAM), 0.0 if nothing was running."""
+    _resolved, svc_key = resolve_service_key(cfg, model_path)
+    if served_name:
+        svc_key = f"{svc_key}|{served_name}"
+    with _lock:
+        svc = _services.get(svc_key)
+        running = bool(svc and svc["proc"].poll() is None)
+    if not running:
+        return 0.0
+    gmu = float(getattr(cfg, "gpu_memory_utilization", 0) or 0)
+    stop_service(svc_key)
+    if gmu > 0:
+        try:
+            from codai.models.manager import multi_model_manager
+            total = multi_model_manager._total_vram_gb()
+        except Exception:
+            total = 24.0
+        return max(0.5, gmu * total)
+    return 4.6   # small VLM fallback estimate
+
+
 import atexit as _atexit
 _atexit.register(stop_all)

codai/ocr/manager.py

@@ -207,10 +207,33 @@ class OcrManager:
                 freed += pool.release_sync()
             except Exception:
                 pass
+        # The managed Surya-2 vLLM subprocess isn't a manager-tracked model and its VRAM
+        # (gpu_memory_utilization × card) isn't reclaimed by tearing down the worker pool
+        # above — it must be stopped explicitly. Do it here so on-request eviction can
+        # reclaim it; the next OCR request re-boots it via ensure_service.
+        freed += self._stop_surya_vllm()
         if freed:
-            print(f"[ocr] released ~{freed:.1f} GB (pools torn down for VRAM eviction)")
+            print(f"[ocr] released ~{freed:.1f} GB (pools + Surya vLLM torn down for VRAM eviction)")
         return freed
 
+    def _stop_surya_vllm(self) -> float:
+        """Stop the managed Surya-2 vLLM subprocess (if this build serves Surya via vLLM).
+        Returns estimated GB freed."""
+        cfg = self._cfg
+        if cfg is None or (getattr(cfg, "surya_serve", "") or "").strip().lower() != "vllm":
+            return 0.0
+        try:
+            from codai.api import vllm_worker
+            from codai.models.manager import get_active_vllm_config
+            vcfg = get_active_vllm_config()
+            if vcfg is None:
+                return 0.0
+            model = (getattr(cfg, "surya_model", "") or "datalab-to/surya-ocr-2").strip()
+            return vllm_worker.stop_service_for(vcfg, model_path=model, served_name=model)
+        except Exception as e:
+            print(f"[ocr] Surya vLLM stop skipped: {e}")
+            return 0.0
+
     @staticmethod
     def _pool_params(cfg) -> tuple:
         return (