studio: single resident generator + engine log tail; fix ray PG leak

reload-after-unload failed because RayDistributedExecutor.shutdown() killed
the actors but never removed the placement group — the GPUs stayed reserved
forever. remove it on shutdown.

studio generator management simplified to a single slot: one VideoGenerator
instance at a time, one load in flight at a time (409 otherwise), loading a
new config always releases the old instance, unload shuts it down and
deletes it. jobs wait out an in-flight preload and reuse it on config match.

GET /api/engine/logs: incremental tail of the server's stdout/stderr (ray
relays worker output to the driver, so this captures every rank).

apps/fastvideo_studio/job_runner.py

@@ -259,12 +259,14 @@ class JobRunner:
         self._jobs_lock = threading.Lock()
         self._load_jobs()
 
-        # Cache of loaded generators keyed by model config so that we only pay
-        # the model-loading cost once per model configuration.
-        self._generators: dict[tuple, Any] = {}
-        self._generators_lock = threading.Lock()
-        # Explicit preloads in flight/failed, keyed like _generators.
-        self._preloads: dict[tuple, dict[str, Any]] = {}
+        # Exactly one generator lives in memory at a time. Loading a new
+        # config always releases the old instance first (shutdown + placement
+        # group teardown); unload deletes it outright.
+        self._generator: Any | None = None
+        self._generator_config: dict[str, Any] | None = None
+        self._generator_state: str = "empty"  # empty | loading | ready | failed
+        self._generator_error: str | None = None
+        self._generator_lock = threading.Lock()
 
         # Shared Manager for log queues (avoids spawning a new process per job)
         self._mp_manager = get_mp_context().Manager()
@@ -620,7 +622,7 @@ class JobRunner:
         }
 
     @staticmethod
-    def _generator_cache_key(
+    def _generator_config_dict(
         model_id: str,
         workload_type: str,
         num_gpus: int,
@@ -633,88 +635,146 @@ class JobRunner:
         vsa_sparsity: float = 0.0,
         tp_size: int = -1,
         sp_size: int = -1,
-    ) -> tuple:
-        return (
-            model_id,
-            workload_type,
-            num_gpus,
-            dit_cpu_offload,
-            text_encoder_cpu_offload,
-            vae_cpu_offload,
-            image_encoder_cpu_offload,
-            use_fsdp_inference,
-            enable_torch_compile,
-            vsa_sparsity,
-            tp_size,
-            sp_size,
-        )
+    ) -> dict[str, Any]:
+        """Canonical engine-config dict; equality here == same generator."""
+        return {
+            "model_id": model_id,
+            "workload_type": workload_type,
+            "num_gpus": num_gpus,
+            "dit_cpu_offload": dit_cpu_offload,
+            "text_encoder_cpu_offload": text_encoder_cpu_offload,
+            "vae_cpu_offload": vae_cpu_offload,
+            "image_encoder_cpu_offload": image_encoder_cpu_offload,
+            "use_fsdp_inference": use_fsdp_inference,
+            "enable_torch_compile": enable_torch_compile,
+            "vsa_sparsity": vsa_sparsity,
+            "tp_size": tp_size,
+            "sp_size": sp_size,
+        }
 
-    @staticmethod
-    def _generator_key_dict(key: tuple) -> dict[str, Any]:
-        names = ("model_id", "workload_type", "num_gpus", "dit_cpu_offload",
-                 "text_encoder_cpu_offload", "vae_cpu_offload",
-                 "image_encoder_cpu_offload", "use_fsdp_inference",
-                 "enable_torch_compile", "vsa_sparsity", "tp_size", "sp_size")
-        return dict(zip(names, key, strict=True))
+    def _slot_entry(self) -> dict[str, Any]:
+        return {
+            "state": self._generator_state,
+            "error": self._generator_error,
+            **(self._generator_config or {}),
+        }
+
+    def _running_inference_jobs(self) -> list[str]:
+        with self._jobs_lock:
+            return [j.id for j in self._jobs.values()
+                    if j.status == JobStatus.RUNNING and j.job_type == "inference"]
+
+    def _release_generator(self) -> None:
+        """Shut down and delete the resident generator (caller holds no lock)."""
+        with self._generator_lock:
+            gen = self._generator
+            self._generator = None
+            self._generator_state = "empty"
+            self._generator_config = None
+            self._generator_error = None
+        if gen is not None:
+            logger.info("Releasing resident generator")
+            gen.shutdown()
+            del gen
 
     def preload_generator(self, **params: Any) -> dict[str, Any]:
-        """Load a generator into the cache without running a job, so the first
-        generation doesn't pay the model load. Returns the preload state dict
-        (state: loading | ready | failed)."""
-        key = self._generator_cache_key(**params)
-        with self._generators_lock:
-            if key in self._generators:
-                return {"state": "ready", **self._generator_key_dict(key)}
-            entry = self._preloads.get(key)
-            if entry is not None and entry["state"] == "loading":
-                return entry
-            entry = {
-                "state": "loading",
-                "started_at": time.time(),
-                "error": None,
-                **self._generator_key_dict(key),
-            }
-            self._preloads[key] = entry
+        """Load a model into memory ahead of time. One load at a time; loading
+        a different config always releases the current instance first."""
+        config = self._generator_config_dict(**params)
+        with self._generator_lock:
+            if self._generator_state == "loading":
+                raise RuntimeError("a model load is already in progress")
+            if self._generator_state == "ready" and self._generator_config == config:
+                return self._slot_entry()
+            if self._running_inference_jobs():
+                raise RuntimeError("cannot swap models while inference jobs are running")
+            self._generator_state = "loading"
+            self._generator_config = config
+            self._generator_error = None
+            entry = self._slot_entry()
 
         def _load() -> None:
             try:
-                self._get_or_create_generator(**params)
-                entry["state"] = "ready"
+                self._create_generator_into_slot(config)
             except Exception as exc:  # noqa: BLE001 -- surfaced via the API
-                logger.exception("Preload failed for %s", params.get("model_id"))
-                entry["state"] = "failed"
-                entry["error"] = str(exc)
+                logger.exception("Preload failed for %s", config["model_id"])
+                with self._generator_lock:
+                    self._generator_state = "failed"
+                    self._generator_error = str(exc)
 
         threading.Thread(target=_load, daemon=True, name="generator-preload").start()
         return entry
 
-    def list_generators(self) -> list[dict[str, Any]]:
-        """Resident generators plus preloads in flight or failed."""
-        out: list[dict[str, Any]] = []
-        with self._generators_lock:
-            resident = set(self._generators)
-            for key in resident:
-                out.append({"state": "ready", **self._generator_key_dict(key)})
-            for key, entry in self._preloads.items():
-                if key not in resident and entry["state"] != "ready":
-                    out.append(entry)
-        return out
+    def _create_generator_into_slot(self, config: dict[str, Any]) -> Any:
+        """Release whatever is resident, load ``config``, make it THE instance."""
+        with self._generator_lock:
+            gen = self._generator
+            self._generator = None
+        if gen is not None:
+            logger.info("Releasing resident generator before loading a new one")
+            gen.shutdown()
+            del gen
 
-    def unload_generator(self, **params: Any) -> bool:
-        """Shut down a cached generator and free its GPU memory. Refuses while
-        any inference job is running (the job may be using it)."""
-        with self._jobs_lock:
-            running = [j.id for j in self._jobs.values()
-                       if j.status == JobStatus.RUNNING and j.job_type == "inference"]
+        # Import lazily so starting the server is fast even without a GPU.
+        from fastvideo import VideoGenerator
+
+        # Deployment-level knobs (set where the API server is launched):
+        # FASTVIDEO_STUDIO_MODEL_PATHS="id=/local/dir,..." serves a registered
+        # model id from local weights instead of the HF hub;
+        # FASTVIDEO_STUDIO_EXECUTOR_BACKEND=ray runs workers on an existing
+        # Ray cluster (the multi-node path — "mp" spawns local processes only).
+        model_path = config["model_id"]
+        for pair in os.environ.get("FASTVIDEO_STUDIO_MODEL_PATHS", "").split(","):
+            mid, sep, path = pair.partition("=")
+            if sep and mid.strip() == config["model_id"]:
+                model_path = path.strip()
+        executor_kwargs: dict[str, Any] = {}
+        backend = os.environ.get("FASTVIDEO_STUDIO_EXECUTOR_BACKEND")
+        if backend:
+            executor_kwargs["distributed_executor_backend"] = backend
+
+        logger.info("Loading model %s (%s)", config["model_id"],
+                    ", ".join(f"{k}={v}" for k, v in config.items() if k != "model_id"))
+        new_gen = VideoGenerator.from_pretrained(
+            model_path,
+            workload_type=config["workload_type"],
+            num_gpus=config["num_gpus"],
+            dit_cpu_offload=config["dit_cpu_offload"],
+            text_encoder_cpu_offload=config["text_encoder_cpu_offload"],
+            vae_cpu_offload=config["vae_cpu_offload"],
+            image_encoder_cpu_offload=config["image_encoder_cpu_offload"],
+            use_fsdp_inference=config["use_fsdp_inference"],
+            enable_torch_compile=config["enable_torch_compile"],
+            VSA_sparsity=config["vsa_sparsity"],
+            tp_size=config["tp_size"],
+            sp_size=config["sp_size"],
+            **executor_kwargs,
+        )
+        with self._generator_lock:
+            self._generator = new_gen
+            self._generator_config = config
+            self._generator_state = "ready"
+            self._generator_error = None
+        return new_gen
+
+    def list_generators(self) -> list[dict[str, Any]]:
+        """The resident slot, or empty when nothing is loaded."""
+        with self._generator_lock:
+            if self._generator_state == "empty":
+                return []
+            return [self._slot_entry()]
+
+    def unload_generator(self, **_ignored: Any) -> bool:
+        """Shut down and delete the resident generator, freeing GPU memory."""
+        with self._generator_lock:
+            if self._generator_state == "loading":
+                raise RuntimeError("cannot unload while a model load is in progress")
+            if self._generator_state == "empty":
+                return False
+        running = self._running_inference_jobs()
         if running:
             raise RuntimeError(f"cannot unload while inference jobs are running: {running}")
-        key = self._generator_cache_key(**params)
-        with self._generators_lock:
-            gen = self._generators.pop(key, None)
-            self._preloads.pop(key, None)
-        if gen is None:
-            return False
-        gen.shutdown()
+        self._release_generator()
         return True
 
     def _get_or_create_generator(
@@ -733,7 +793,10 @@ class JobRunner:
         sp_size: int = -1,
         log_queue: mp.Queue | None = None,
     ) -> Any:
-        cache_key = self._generator_cache_key(
+        """Return the resident generator if the config matches; otherwise
+        replace the slot (waiting out an in-flight preload first)."""
+        del log_queue  # single-slot generators log via the engine tee
+        config = self._generator_config_dict(
             model_id=model_id,
             workload_type=workload_type,
             num_gpus=num_gpus,
@@ -747,72 +810,21 @@ class JobRunner:
             tp_size=tp_size,
             sp_size=sp_size,
         )
+        # A preload may be in flight (possibly for exactly this config): wait
+        # for it rather than racing it for the GPUs.
+        deadline = time.time() + 3600
+        while time.time() < deadline:
+            with self._generator_lock:
+                state = self._generator_state
+                if state != "loading":
+                    if state == "ready" and self._generator_config == config:
+                        return self._generator
+                    break
+            time.sleep(1.0)
+        else:
+            raise RuntimeError("timed out waiting for the in-flight model load")
 
-        # Generators are cached by model_id and configuration parameters
-        with self._generators_lock:
-            if cache_key in self._generators:
-                return self._generators[cache_key]
-
-        # Import lazily so starting the server is fast even without a GPU.
-        from fastvideo import VideoGenerator
-
-        # Deployment-level knobs (set where the API server is launched):
-        # FASTVIDEO_STUDIO_MODEL_PATHS="id=/local/dir,..." serves a registered
-        # model id from local weights instead of the HF hub;
-        # FASTVIDEO_STUDIO_EXECUTOR_BACKEND=ray runs workers on an existing
-        # Ray cluster (the multi-node path — "mp" spawns local processes only).
-        model_path = model_id
-        for pair in os.environ.get("FASTVIDEO_STUDIO_MODEL_PATHS", "").split(","):
-            mid, sep, path = pair.partition("=")
-            if sep and mid.strip() == model_id:
-                model_path = path.strip()
-        executor_kwargs: dict[str, Any] = {}
-        backend = os.environ.get("FASTVIDEO_STUDIO_EXECUTOR_BACKEND")
-        if backend:
-            executor_kwargs["distributed_executor_backend"] = backend
-
-        logger.info(
-            "Loading model %s (workload=%s, num_gpus=%d, offloads: "
-            "dit=%s text_encoder=%s vae=%s image_encoder=%s, fsdp=%s, "
-            "torch_compile=%s, vsa_sparsity=%.2f, tp=%d sp=%d) …",
-            model_id,
-            workload_type,
-            num_gpus,
-            dit_cpu_offload,
-            text_encoder_cpu_offload,
-            vae_cpu_offload,
-            image_encoder_cpu_offload,
-            use_fsdp_inference,
-            enable_torch_compile,
-            vsa_sparsity,
-            tp_size,
-            sp_size,
-        )
-
-        gen = VideoGenerator.from_pretrained(
-            model_path,
-            workload_type=workload_type,
-            num_gpus=num_gpus,
-            dit_cpu_offload=dit_cpu_offload,
-            text_encoder_cpu_offload=text_encoder_cpu_offload,
-            vae_cpu_offload=vae_cpu_offload,
-            image_encoder_cpu_offload=image_encoder_cpu_offload,
-            use_fsdp_inference=use_fsdp_inference,
-            enable_torch_compile=enable_torch_compile,
-            VSA_sparsity=vsa_sparsity,
-            tp_size=tp_size,
-            sp_size=sp_size,
-            log_queue=log_queue,
-            **executor_kwargs,
-        )
-
-        with self._generators_lock:
-            if cache_key not in self._generators:
-                self._generators[cache_key] = gen
-            else:  # Another thread may have created it while we were loading.
-                gen.shutdown()
-                gen = self._generators[cache_key]
-        return gen
+        return self._create_generator_into_slot(config)
 
     def _run_job(self, job: Job):
         if job.job_type == "inference":

apps/fastvideo_studio/server.py

@@ -34,8 +34,8 @@ from fastvideo.registry import (get_registered_model_paths, get_registered_model
 from fastvideo_studio.database import Database, _get_db_path
 from fastvideo_studio.gpu import get_gpu_snapshot
 from fastvideo_studio.job_runner import JobRunner, JobStatus
-from fastvideo_studio.models import (CreateDatasetRequest, CreateJobRequest, GeneratorRequest, SettingsUpdate, UpdateCaptionRequest,
-                                     model_label)
+from fastvideo_studio.models import (CreateDatasetRequest, CreateJobRequest, GeneratorRequest, SettingsUpdate,
+                                     UpdateCaptionRequest, model_label)
 
 logging.basicConfig(
     level=logging.INFO,
@@ -45,6 +45,60 @@ logger = logging.getLogger("fastvideo.studio.api")
 
 DEFAULT_OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "..", "outputs", "ui_jobs")
 
+class _EngineLogBuffer:
+    """Thread-safe ring buffer over the server process's stdout/stderr.
+
+    Because ray relays worker output to the driver (log_to_driver), teeing the
+    server's own streams captures engine output from every rank on every node;
+    under the mp executor, worker logs arrive via the logging handlers which
+    also write to stderr.
+    """
+
+    def __init__(self, maxlen: int = 5000) -> None:
+        import collections
+        import threading
+        self._lines: collections.deque[str] = collections.deque(maxlen=maxlen)
+        self._dropped = 0
+        self._lock = threading.Lock()
+        self._partial = ""
+
+    def write(self, text: str) -> None:
+        with self._lock:
+            buf = self._partial + text
+            *complete, self._partial = buf.split("\n")
+            for line in complete:
+                if len(self._lines) == self._lines.maxlen:
+                    self._dropped += 1
+                self._lines.append(line)
+
+    def get_lines(self, after: int = 0) -> tuple[list[str], int]:
+        with self._lock:
+            total = self._dropped + len(self._lines)
+            start = max(0, after - self._dropped)
+            return list(self._lines)[start:], total
+
+
+class _Tee:
+    """File-like that forwards to the original stream and the ring buffer."""
+
+    def __init__(self, orig: Any, buffer: _EngineLogBuffer) -> None:
+        self._orig = orig
+        self._buffer = buffer
+
+    def write(self, text: str) -> int:
+        self._buffer.write(text)
+        return self._orig.write(text)
+
+    def flush(self) -> None:
+        self._orig.flush()
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._orig, name)
+
+
+engine_log = _EngineLogBuffer()
+
+
 _available_models: list[dict[str, str]] = [{
     "id": path,
     "label": model_label(path)
@@ -261,8 +315,8 @@ def list_generators() -> list[dict[str, Any]]:
 
 @app.post("/api/generators/preload", status_code=202)
 def preload_generator(req: GeneratorRequest) -> dict[str, Any]:
-    """Load a model into memory ahead of time so the first generation with
-    this configuration skips the model-load wait. Idempotent."""
+    """Load a model into memory ahead of time (replacing whatever is
+    resident). One load at a time — 409 while another load is in flight."""
     valid_ids = {m["id"] for m in _available_models}
     if req.model_id not in valid_ids and not os.path.isdir(req.model_id):
         raise HTTPException(
@@ -270,21 +324,32 @@ def preload_generator(req: GeneratorRequest) -> dict[str, Any]:
             detail=(f"Unknown model_id '{req.model_id}'. "
                     f"Valid options: {sorted(valid_ids)}"),
         )
-    return job_runner.preload_generator(**req.model_dump())
+    try:
+        return job_runner.preload_generator(**req.model_dump())
+    except RuntimeError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
 
 
 @app.post("/api/generators/unload")
-def unload_generator(req: GeneratorRequest) -> dict[str, Any]:
-    """Shut down a resident generator and free its GPU memory."""
+def unload_generator() -> dict[str, Any]:
+    """Shut down and delete the resident generator, freeing GPU memory."""
     try:
-        unloaded = job_runner.unload_generator(**req.model_dump())
+        unloaded = job_runner.unload_generator()
     except RuntimeError as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     if not unloaded:
-        raise HTTPException(status_code=404, detail="No such resident generator")
+        raise HTTPException(status_code=404, detail="No model is loaded")
     return {"unloaded": True}
 
 
+@app.get("/api/engine/logs")
+def engine_logs(after: int = 0) -> dict[str, Any]:
+    """Incremental tail of the engine's stdout/stderr (driver + relayed
+    worker output). Poll with ?after=<total from the previous response>."""
+    lines, total = engine_log.get_lines(after=after)
+    return {"lines": lines, "total": total}
+
+
 @app.post("/api/jobs", status_code=201)
 def create_job(req: CreateJobRequest) -> dict[str, Any]:
     """Create a new job (does **not** start it automatically)."""
@@ -656,6 +721,14 @@ def create_local_env(host: str, port: int) -> None:
 
 
 def main() -> None:
+    import sys
+    sys.stdout = _Tee(sys.stdout, engine_log)
+    sys.stderr = _Tee(sys.stderr, engine_log)
+    # handlers created before the tee (module-level basicConfig) hold the
+    # original stream objects — re-point them or their output bypasses the buffer
+    for h in logging.getLogger().handlers:
+        if isinstance(h, logging.StreamHandler) and h.stream in (sys.__stderr__, sys.__stdout__):
+            h.setStream(sys.stderr)
     global job_runner, database, upload_dir, verbose, datasets_upload_dir  # noqa: PLW0603
 
     # Set up signal handlers to prevent worker crashes from killing the server

apps/fastvideo_studio/tests/test_generator_cache.py

@@ -1,9 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
-"""Tests for the resident-generator API: preload, list, unload.
+"""Tests for the single-slot resident generator: preload, list, unload.
 
-The generator itself is faked at the JobRunner boundary — these tests pin the
-cache/lifecycle contract (idempotent preload, state transitions, unload
-refusal while a job runs), not model loading.
+Exactly one VideoGenerator lives in memory. One load at a time; loading a new
+config always releases the old instance; unload deletes it. The generator is
+faked at the ``_create_generator_into_slot``/VideoGenerator boundary.
 """
 
 from __future__ import annotations
@@ -26,99 +26,187 @@ class _FakeGenerator:
 
 
 @pytest.fixture()
-def runner(tmp_path, monkeypatch):
+def runner():
     r = JobRunner.__new__(JobRunner)  # skip __init__: no DB/Manager needed
     r._jobs = {}
     r._jobs_lock = threading.Lock()
-    r._generators = {}
-    r._generators_lock = threading.Lock()
-    r._preloads = {}
+    r._generator = None
+    r._generator_config = None
+    r._generator_state = "empty"
+    r._generator_error = None
+    r._generator_lock = threading.Lock()
     return r
 
 
-def _wait_state(runner, model_id, state, timeout=5.0):
+def _install_fake_loader(runner, monkeypatch, made: list | None = None,
+                         gate: threading.Event | None = None,
+                         fail: str | None = None):
+    """Replace the VideoGenerator load inside _create_generator_into_slot."""
+    import fastvideo_studio.job_runner as jr
+
+    class _FakeVG:
+        @staticmethod
+        def from_pretrained(model_path, **kwargs):
+            if gate is not None:
+                assert gate.wait(5), "test gate never opened"
+            if fail is not None:
+                raise RuntimeError(fail)
+            gen = _FakeGenerator()
+            if made is not None:
+                made.append((model_path, gen))
+            return gen
+
+    monkeypatch.setitem(__import__("sys").modules, "fastvideo",
+                        SimpleNamespace(VideoGenerator=_FakeVG))
+    return jr
+
+
+def _wait_state(runner, state, timeout=5.0):
     deadline = time.time() + timeout
     while time.time() < deadline:
-        entries = [g for g in runner.list_generators() if g["model_id"] == model_id]
-        if entries and entries[0]["state"] == state:
-            return entries[0]
+        with runner._generator_lock:
+            if runner._generator_state == state:
+                return runner._slot_entry()
         time.sleep(0.02)
-    raise AssertionError(f"{model_id} never reached state {state}: {runner.list_generators()}")
+    raise AssertionError(f"never reached state {state}: {runner._generator_state}")
 
 
-def test_preload_loads_once_and_lists_ready(runner, monkeypatch):
-    created = []
+def test_preload_then_ready_then_idempotent(runner, monkeypatch):
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
 
-    def fake_create(self, **params):
-        gen = _FakeGenerator()
-        created.append(params["model_id"])
-        key = self._generator_cache_key(**params)
-        with self._generators_lock:
-            self._generators[key] = gen
-        return gen
+    entry = runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    assert entry["state"] in ("loading", "ready")
+    _wait_state(runner, "ready")
 
-    monkeypatch.setattr(JobRunner, "_get_or_create_generator", fake_create)
-
-    first = runner.preload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8)
-    # the loader thread may finish before we return; either state is legal here
-    assert first["state"] in ("loading", "ready")
-    ready = _wait_state(runner, "org/model-a", "ready")
-    assert ready["num_gpus"] == 8
-
-    # idempotent: preloading a resident config is an immediate ready, no reload
-    again = runner.preload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8)
+    again = runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
     assert again["state"] == "ready"
-    assert created == ["org/model-a"]
+    assert len(made) == 1  # same config never reloads
 
 
-def test_preload_failure_is_reported_not_raised(runner, monkeypatch):
-    def fake_create(self, **params):
-        raise RuntimeError("no CUDA on this box")
+def test_only_one_load_at_a_time(runner, monkeypatch):
+    gate = threading.Event()
+    _install_fake_loader(runner, monkeypatch, gate=gate)
 
-    monkeypatch.setattr(JobRunner, "_get_or_create_generator", fake_create)
-    entry = runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1)
-    assert entry["state"] in ("loading", "failed")
-    failed = _wait_state(runner, "org/broken", "failed")
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    with pytest.raises(RuntimeError, match="already in progress"):
+        runner.preload_generator(model_id="org/b", workload_type="t2v", num_gpus=8)
+    gate.set()
+    _wait_state(runner, "ready")
+
+
+def test_new_config_always_releases_old_instance(runner, monkeypatch):
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
+
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    _wait_state(runner, "ready")
+    first = made[0][1]
+
+    runner.preload_generator(model_id="org/b", workload_type="t2v", num_gpus=8)
+    _wait_state(runner, "ready")
+    assert first.shutdown_calls == 1  # old instance released, not stacked
+    assert [m[0] for m in made] == ["org/a", "org/b"]
+    assert len(runner.list_generators()) == 1
+    assert runner.list_generators()[0]["model_id"] == "org/b"
+
+
+def test_failed_load_reports_error_and_allows_retry(runner, monkeypatch):
+    _install_fake_loader(runner, monkeypatch, fail="no CUDA on this box")
+    runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1)
+    failed = _wait_state(runner, "failed")
     assert "no CUDA" in failed["error"]
-    # a failed preload must not leave a resident entry
-    assert all(g["state"] != "ready" for g in runner.list_generators())
+
+    # retry after failure must be accepted (this was the reported bug)
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
+    runner.preload_generator(model_id="org/broken", workload_type="t2v", num_gpus=1)
+    _wait_state(runner, "ready")
+    assert len(made) == 1
 
 
-def test_unload_shuts_down_and_removes(runner):
-    gen = _FakeGenerator()
-    key = runner._generator_cache_key(model_id="org/model-a", workload_type="t2v", num_gpus=8)
-    runner._generators[key] = gen
+def test_unload_deletes_instance(runner, monkeypatch):
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    _wait_state(runner, "ready")
 
-    assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is True
-    assert gen.shutdown_calls == 1
+    assert runner.unload_generator() is True
+    assert made[0][1].shutdown_calls == 1
     assert runner.list_generators() == []
-    # second unload: nothing resident
-    assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is False
+    assert runner._generator is None
+    assert runner.unload_generator() is False  # nothing resident
 
 
-def test_unload_refuses_while_inference_job_runs(runner):
-    key = runner._generator_cache_key(model_id="org/model-a", workload_type="t2v", num_gpus=8)
-    gen = _FakeGenerator()
-    runner._generators[key] = gen
+def test_unload_refuses_while_inference_job_runs(runner, monkeypatch):
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    _wait_state(runner, "ready")
     runner._jobs["j1"] = SimpleNamespace(id="j1", status=JobStatus.RUNNING, job_type="inference")
 
     with pytest.raises(RuntimeError, match="j1"):
-        runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8)
-    assert gen.shutdown_calls == 0  # still resident
+        runner.unload_generator()
+    assert made[0][1].shutdown_calls == 0
 
-    # training jobs don't block unload
-    runner._jobs["j1"].job_type = "finetune"
-    assert runner.unload_generator(model_id="org/model-a", workload_type="t2v", num_gpus=8) is True
+    runner._jobs["j1"].job_type = "finetune"  # training doesn't block
+    assert runner.unload_generator() is True
 
 
-def test_cache_key_matches_job_lookup_defaults(runner):
-    """A preload with GeneratorRequest defaults must hit the same cache slot a
-    job with CreateJobRequest defaults resolves to — else preloading is a lie."""
+def test_job_waits_for_matching_preload(runner, monkeypatch):
+    gate = threading.Event()
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made, gate=gate)
+
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    got = []
+    t = threading.Thread(
+        target=lambda: got.append(
+            runner._get_or_create_generator("org/a", "t2v", 8)),
+        daemon=True)
+    t.start()
+    time.sleep(0.2)
+    assert not got  # job blocked on the in-flight load
+    gate.set()
+    t.join(5)
+    assert got and got[0] is made[0][1]
+    assert len(made) == 1  # the job reused the preloaded instance
+
+
+def test_job_with_different_config_replaces_slot(runner, monkeypatch):
+    made = []
+    _install_fake_loader(runner, monkeypatch, made=made)
+    runner.preload_generator(model_id="org/a", workload_type="t2v", num_gpus=8)
+    _wait_state(runner, "ready")
+
+    gen = runner._get_or_create_generator("org/b", "t2v", 4)
+    assert gen is made[1][1]
+    assert made[0][1].shutdown_calls == 1  # old released before new load
+    assert runner.list_generators()[0]["model_id"] == "org/b"
+
+
+def test_config_dict_matches_request_defaults(runner):
+    """GeneratorRequest defaults and CreateJobRequest defaults must resolve to
+    the same slot config — else preloading never matches the job."""
     from fastvideo_studio.models import CreateJobRequest, GeneratorRequest
 
     job = CreateJobRequest(model_id="m", prompt="p").model_dump()
     pre = GeneratorRequest(model_id="m").model_dump()
-    job_key = runner._generator_cache_key(
+    assert runner._generator_config_dict(**pre) == runner._generator_config_dict(
         **{k: job[k] for k in pre})
-    pre_key = runner._generator_cache_key(**pre)
-    assert job_key == pre_key
+
+
+def test_engine_log_buffer_incremental_tail():
+    from fastvideo_studio.server import _EngineLogBuffer
+
+    buf = _EngineLogBuffer(maxlen=3)
+    buf.write("a\nb\n")
+    buf.write("c")          # partial line: not visible yet
+    lines, total = buf.get_lines(0)
+    assert lines == ["a", "b"] and total == 2
+    buf.write("!\nd\ne\n")  # completes "c!", then overflows the ring
+    lines, total = buf.get_lines(total)
+    assert lines == ["c!", "d", "e"] and total == 5
+    # reader far behind: dropped lines are skipped, no crash
+    lines, _ = buf.get_lines(0)
+    assert lines == ["c!", "d", "e"]

fastvideo/worker/ray_distributed_executor.py

@@ -385,6 +385,16 @@ class RayDistributedExecutor(Executor):
             ray.kill(worker)
 
         self.workers = []
+        # Killing the actors does not release the placement group — without
+        # this, the GPUs stay reserved and no new engine can ever schedule
+        # in the same ray cluster.
+        pg = getattr(self.fastvideo_args, "ray_placement_group", None)
+        if pg is not None:
+            try:
+                ray.util.remove_placement_group(pg)
+            except Exception:  # noqa: BLE001 -- already-removed / cluster gone
+                logger.warning("Failed to remove ray placement group", exc_info=True)
+            self.fastvideo_args.ray_placement_group = None
 
     def __del__(self):
         self.shutdown()