fix: stop active pipeline workers before unload to free VRAM

src/scope/server/frame_processor.py

@@ -292,6 +292,11 @@ def stop(self, error_message: str = None):
         for processor in self.pipeline_processors:
             processor.stop()
 
+        # Remove the pre-unload hooks registered in graph_executor so the
+        # pipeline_manager doesn't retain stale closures over stopped processors.
+        for node_id in list(self._processors_by_node_id.keys()):
+            self.pipeline_manager.unregister_pre_unload_hook(node_id)
+
         # Clear pipeline processors
         self.pipeline_processors.clear()
 

src/scope/server/graph_executor.py

@@ -137,6 +137,15 @@ def build_graph(
         node_processors[node.id] = processor
         pipeline_ids.append(node.pipeline_id)
 
+        # If this pipeline gets unloaded mid-session (e.g. a user-triggered
+        # pipeline swap), the worker thread must exit before the manager drops
+        # its reference — otherwise the worker keeps allocating GPU memory
+        # right through the "unload" and OOMs the next load. Capture processor
+        # by default-arg to avoid the late-binding closure pitfall.
+        pipeline_manager.register_pre_unload_hook(
+            node.id, lambda _key, p=processor: p.stop()
+        )
+
         for e in graph.edges_to(node.id):
             if e.kind != "stream":
                 continue

src/scope/server/pipeline_manager.py

@@ -5,6 +5,7 @@
 import logging
 import threading
 import time
+from collections.abc import Callable
 from enum import Enum
 from typing import Any
 
@@ -64,11 +65,33 @@ def __init__(self):
         # Loading stage for frontend display (e.g., "Loading diffusion model...")
         self._loading_stage: str | None = None
 
+        # Pre-unload hooks keyed by instance_key. Invoked before a pipeline is
+        # removed so callers (e.g. PipelineProcessor workers) can release their
+        # pipeline reference and stop allocating GPU memory. Without this, a
+        # swap-while-active leaves worker threads holding live CUDA tensors and
+        # the next load OOMs.
+        self._pre_unload_hooks: dict[str, Callable[[str], None]] = {}
+
     def set_loading_stage(self, stage: str | None) -> None:
         """Set the current loading stage (thread-safe)."""
         with self._lock:
             self._loading_stage = stage
 
+    def register_pre_unload_hook(self, key: str, hook: Callable[[str], None]) -> None:
+        """Register a callback invoked before *key* is unloaded.
+
+        Used by PipelineProcessor to ensure its worker thread is stopped — and
+        therefore releases its pipeline reference and GPU tensors — before the
+        next load begins. The hook MUST block until its worker has exited.
+        """
+        with self._lock:
+            self._pre_unload_hooks[key] = hook
+
+    def unregister_pre_unload_hook(self, key: str) -> None:
+        """Remove a previously registered hook (idempotent)."""
+        with self._lock:
+            self._pre_unload_hooks.pop(key, None)
+
     @property
     def status(self) -> PipelineStatus:
         """Get current pipeline status."""
@@ -824,6 +847,18 @@ def _unload_pipeline_by_id_unsafe(
 
         logger.info(f"Unloading pipeline: {pipeline_id}")
 
+        # Stop any active worker (e.g. PipelineProcessor thread) bound to this
+        # key BEFORE dropping our reference. If we skip this, the worker keeps
+        # running during the subsequent gc/empty_cache, holds the pipeline alive
+        # via its closure, and continues allocating CUDA memory — causing the
+        # next load to OOM despite our "cleanup".
+        hook = self._pre_unload_hooks.pop(pipeline_id, None)
+        if hook is not None:
+            try:
+                hook(pipeline_id)
+            except Exception as e:
+                logger.warning(f"Pre-unload hook for {pipeline_id} raised: {e}")
+
         # Remove from tracked pipelines
         if pipeline_id in self._pipelines:
             del self._pipelines[pipeline_id]