longcat: i2v, continuation, long video — against the real pipeline API (v0.2.33)

Step 4. It also CORRECTS step 2, whose generate() was written against an assumed
interface. The real one, read from upstream's demo scripts, differs in four ways that all
mattered:

- there is no single callable: generate_t2v(height, width, …), generate_i2v(image,
  resolution, …), generate_vc(video, num_cond_frames, use_kv_cache, offload_kv_cache, …)
  and generate_refine(stage1_video, spatial_refine_only, …) are separate methods, and
  t2v takes height/width where i2v and vc take a resolution NAME;
- each returns [0]-indexed frames normalised to [0,1], not an object with .frames —
  upstream converts with (tensor * 255).clamp(0,255).to(uint8);
- the distilled stage is a use_distill=True FLAG on the same method, not a LoRA to fuse
  by hand, so _apply_stage_lora is gone;
- the 50-step passes take a negative_prompt and the distilled one does not (its guidance
  is 1.0, so it would do nothing).

On top of that:
- The segment loop is upstream's: a call emits num_frames but its first num_cond_frames
  re-render the previous tail, so only the remainder is new and the fresh frames become
  the next call's conditioning. torch_gc() between segments, as upstream does.
- Modes map onto the tasks: t2v, i2v/ti2v -> i2v, extend/vc/v2v -> vc. Continuation
  reuses the EXISTING cond_frames field, so a client that already chains clips through
  the VACE path needs no new vocabulary — and gets native pretrained continuation
  instead, which is what stops the colour drift chained conditioning suffers.
- Long video: num_segments or duration_seconds, and the SERVER owns the loop. The
  response carries the tail a caller needs to continue by hand if it prefers to.
- The segment boundary is now a real handover point. POST /yield asks a generation to
  stop at the next boundary; the releaser calls it before waiting, so an eviction
  arriving mid-request costs one segment instead of the whole generation, and the request
  still returns — shorter, with a warning that reaches the caller — rather than failing.

Unsupported modes (the audio-driven avatar tasks) are still refused with a 400 rather
than downgraded to t2v, which would read as a quality bug rather than a gap.

23 new tests pinning the upstream call contract so it cannot drift back, plus the segment
arithmetic and the yield behaviour. Still no real generation: that needs the venv, the
repo checkout and the ~83 GB checkpoint.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.2.32"
+__version__ = "0.2.33"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/longcat_worker.py

@@ -556,6 +556,20 @@ def generate(model_name: str, payload: dict, config: dict = None,
 
 # ── eviction: giving the card back ───────────────────────────────────────────
 
+def _request_yield(key: str) -> bool:
+    """Ask a service to stop at its next segment boundary. Best-effort."""
+    import requests
+    with _lock:
+        svc = _services.get(key)
+        url = svc["url"] if svc else ""
+    if not url:
+        return False
+    try:
+        return bool(requests.post(url + "/yield", json={}, timeout=5).ok)
+    except Exception:
+        return False
+
+
 def _drain_timeout_s() -> float:
     sec = _cfg_section()
     try:
@@ -585,8 +599,14 @@ def release_vram(needed_gb: float = 999.0) -> float:
         return 0.0
     if busy:
         budget = _drain_timeout_s()
-        print(f"[longcat] {busy} generation(s) in flight — waiting up to {budget:.0f}s "
-              f"for a clean handover before releasing the card", flush=True)
+        # Ask the services to stop at their next SEGMENT boundary. Upstream already calls
+        # torch_gc() there, so it is a genuinely safe point, and it turns the wait from
+        # "however long the whole request takes" into "one more segment" — the in-flight
+        # request still returns, shorter, with a warning, instead of being lost.
+        for key in keys:
+            _request_yield(key)
+        print(f"[longcat] {busy} generation(s) in flight — asked them to yield at the "
+              f"next segment boundary, waiting up to {budget:.0f}s", flush=True)
         _idle.clear()
         if not _idle.wait(budget):
             print(f"[longcat] still busy after {budget:.0f}s — releasing anyway; the "

codai/api/video.py

@@ -4121,23 +4121,57 @@ async def _generate_longcat(request: VideoGenerationRequest, model_name: str,
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc))
 
+    # coderai's request modes map onto upstream's three entry points. 'extend' is the
+    # existing VACE vocabulary for "continue from these frames", and LongCat serves it
+    # natively — generate_vc is pretrained for continuation, which is why it does not
+    # drift in colour or quality the way chained frame-tail conditioning does.
     mode = (request.mode or "t2v").lower()
-    if mode not in ("t2v", ""):
-        # Honest refusal beats a silent downgrade: the payload for these is designed but
-        # the worker does not implement them yet.
+    _task = {"t2v": "t2v", "": "t2v",
+             "i2v": "i2v", "ti2v": "i2v",
+             "extend": "vc", "vc": "vc", "v2v": "vc"}.get(mode)
+    if _task is None:
         raise HTTPException(status_code=400, detail=(
-            f"LongCat-Video mode '{mode}' is not implemented yet — this build supports "
-            f"text-to-video. Image-to-video, continuation and the avatar tasks are the "
-            f"next increment."))
+            f"LongCat-Video does not support mode '{mode}'. Use 't2v', 'i2v' or "
+            f"'extend' (continuation); the audio-driven avatar tasks are a separate "
+            f"increment."))
 
     payload = {
         "prompt": request.prompt or "",
         "quality": quality,
         "stage": stage,
         "model": model_path,
+        "task": _task,
         "num_frames": int(request.num_frames) if request.num_frames
                       else lc.DEFAULT_NUM_FRAMES,
     }
+    if request.negative_prompt:
+        payload["negative_prompt"] = request.negative_prompt
+
+    # i2v takes one conditioning image; coderai accepts it as init_image or image.
+    if _task == "i2v":
+        _first = request.init_image or request.image
+        if not _first:
+            raise HTTPException(status_code=400, detail=(
+                "mode 'i2v' needs init_image (or image)"))
+        payload["image"] = _as_b64(_first)
+
+    # Continuation takes the previous clip's tail — the same `cond_frames` field the VACE
+    # path uses, so a client that already chains clips needs no new vocabulary.
+    if _task == "vc":
+        if not request.cond_frames:
+            raise HTTPException(status_code=400, detail=(
+                "mode 'extend' needs cond_frames — the previous clip's last frames"))
+        payload["cond_frames_b64"] = [_as_b64(x) for x in request.cond_frames]
+
+    # Long video: ask for a duration or a segment count and the server owns the loop,
+    # rather than the caller chaining requests and tracking tails itself.
+    _segments = getattr(request, "num_segments", None)
+    if _segments:
+        payload["num_segments"] = int(_segments)
+    elif getattr(request, "duration_seconds", None):
+        payload["total_frames"] = int(float(request.duration_seconds) * lc.DEFAULT_FPS)
+    if getattr(request, "num_cond_frames", None):
+        payload["num_cond_frames"] = int(request.num_cond_frames)
     if not payload["prompt"]:
         raise HTTPException(status_code=400, detail="prompt is required")
     if request.width and request.height:
@@ -4195,6 +4229,10 @@ async def _generate_longcat(request: VideoGenerationRequest, model_name: str,
     frames_made = data.get("num_frames")
     fps_used = int(data.get("fps") or lc.DEFAULT_FPS)
     stages_run = data.get("stages") or list(stages)
+    if data.get("warning"):
+        # e.g. it yielded the GPU at a segment boundary and the video is shorter than
+        # asked for. The caller gets the video AND the reason, not a silent surprise.
+        warnings.append(str(data["warning"]))
     if not mp4_bytes:
         raise HTTPException(status_code=500, detail="LongCat-Video returned no video")
     print(f"  [longcat] {'+'.join(stages_run)}: {frames_made} frames "

codai/pydantic/videorequest.py

@@ -87,6 +87,21 @@ class VideoGenerationRequest(BaseModel):
     # 480p, a distilled 16-step pass, then a 720p refinement that takes the earlier
     # frames as input — so a caller either picks a preset and lets the server run the
     # stages, or drives one stage per request and keeps the intermediate frames.
+    # Long video. LongCat generates in segments whose first `num_cond_frames` re-render
+    # the previous segment's tail, so asking for a duration or a segment count lets the
+    # SERVER own the loop — a client no longer has to chain requests and track tails,
+    # which is what the VACE 'extend' path requires.
+    num_segments: Optional[int] = Field(None, description=(
+        "LongCat-Video: how many segments to generate. Each adds "
+        "(num_frames - num_cond_frames) new frames; ~11 segments is a minute. "
+        "Overrides `duration_seconds`."))
+    duration_seconds: Optional[float] = Field(None, description=(
+        "LongCat-Video: target length in seconds, converted to a segment count. "
+        "Ignored when `num_segments` is given."))
+    num_cond_frames: Optional[int] = Field(None, description=(
+        "LongCat-Video: how many frames of the previous segment each new segment "
+        "conditions on (default 13 of 93). More = stronger motion continuity, fewer "
+        "new frames per segment."))
     quality: Optional[str] = Field(None, description=(
         "LongCat-Video quality preset: 'draft' (base stage only, fastest), 'fast' "
         "(distilled, 16 steps — the default), 'best' (base + 720p refinement). Ignored "

tests/test_longcat_tasks.py

+"""Step 4: image-to-video, continuation, long video, and the staging.
+
+Step 2's generate() was written against an ASSUMED pipeline interface and was wrong. The
+real one, read from upstream's demos, is four separate methods —
+``generate_t2v(height, width, …)``, ``generate_i2v(image, resolution, …)``,
+``generate_vc(video, num_cond_frames, use_kv_cache, …)`` and
+``generate_refine(stage1_video, spatial_refine_only, …)`` — each returning ``[0]``-indexed
+frames normalised to [0, 1] rather than an object with ``.frames``, and the distilled pass
+is a ``use_distill=True`` FLAG, not a LoRA we fuse. These tests pin that contract so it
+cannot drift back.
+
+Still no GPU, venv or weights here: the pipeline is a recording stub.
+"""
+
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+
+def _service():
+    """Import the service module without its heavy deps resolved (they load lazily)."""
+    spec = importlib.util.spec_from_file_location(
+        "longcat_service", ROOT / "tools" / "longcat_service.py")
+    m = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(m)
+    return m
+
+
+SVC = _service()
+LC = SVC.LC
+
+
+class _Pipe:
+    """Records which upstream method was called, with what."""
+
+    def __init__(self, frames=4):
+        self.calls = []
+        self._n = frames
+
+    def _out(self):
+        import numpy as np
+        # [0]-indexed, normalised [0, 1], shape (frames, h, w, 3) — as upstream returns.
+        return [np.zeros((self._n, 8, 8, 3), dtype="float32")]
+
+    def generate_t2v(self, **kw):
+        self.calls.append(("t2v", kw))
+        return self._out()
+
+    def generate_i2v(self, **kw):
+        self.calls.append(("i2v", kw))
+        return self._out()
+
+    def generate_vc(self, **kw):
+        self.calls.append(("vc", kw))
+        return self._out()
+
+    def generate_refine(self, **kw):
+        self.calls.append(("refine", kw))
+        return self._out()
+
+
+def _ctx(**kw):
+    base = {"prompt": "a cat", "negative_prompt": "blurry", "height": 480, "width": 832,
+            "num_frames": 93, "num_cond_frames": 13, "num_segments": 1,
+            "resolution": "480p", "num_inference_steps": None, "guidance_scale": None,
+            "offload_kv_cache": False, "image": None, "cond_video": None,
+            "generator": None}
+    base.update(kw)
+    return base
+
+
+# ------------------------------------------------------------ the real API
+def test_t2v_uses_height_and_width():
+    p = _Pipe()
+    SVC._one_pass(p, "t2v", "base", _ctx())
+    name, kw = p.calls[0]
+    assert name == "t2v"
+    assert kw["height"] == 480 and kw["width"] == 832
+    assert "resolution" not in kw           # that is the i2v/vc parameter
+
+
+def test_i2v_uses_a_resolution_name():
+    p = _Pipe()
+    SVC._one_pass(p, "i2v", "base", _ctx(image="IMG"))
+    name, kw = p.calls[0]
+    assert name == "i2v"
+    assert kw["image"] == "IMG" and kw["resolution"] == "480p"
+    assert "height" not in kw
+
+
+def test_continuation_passes_the_conditioning_video():
+    p = _Pipe()
+    SVC._one_pass(p, "vc", "base", _ctx(cond_video=["f1", "f2"]))
+    name, kw = p.calls[0]
+    assert name == "vc"
+    assert kw["video"] == ["f1", "f2"]
+    assert kw["num_cond_frames"] == 13
+    assert kw["use_kv_cache"] is True
+
+
+def test_the_distilled_stage_is_a_flag_not_a_lora():
+    """Upstream passes use_distill=True to the SAME method; fusing cfg_step_lora by hand
+    is not the interface."""
+    p = _Pipe()
+    SVC._one_pass(p, "t2v", "distill", _ctx())
+    _name, kw = p.calls[0]
+    assert kw["use_distill"] is True
+    assert kw["num_inference_steps"] == 16 and kw["guidance_scale"] == 1.0
+    assert not hasattr(SVC, "_apply_stage_lora")
+
+
+def test_the_distilled_stage_passes_no_negative_prompt():
+    """Its guidance is 1.0, so a negative prompt does nothing — upstream omits it."""
+    p = _Pipe()
+    SVC._one_pass(p, "t2v", "distill", _ctx())
+    assert "negative_prompt" not in p.calls[0][1]
+    p2 = _Pipe()
+    SVC._one_pass(p2, "t2v", "base", _ctx())
+    assert p2.calls[0][1]["negative_prompt"] == "blurry"
+
+
+def test_frames_come_back_as_pil_from_normalised_floats():
+    import numpy as np
+    out = SVC._frames_from_output(np.zeros((3, 4, 4, 3), dtype="float32"))
+    assert len(out) == 3
+    assert out[0].mode == "RGB" and out[0].size == (4, 4)
+
+
+def test_already_uint8_output_is_not_rescaled():
+    import numpy as np
+    arr = np.full((2, 4, 4, 3), 255, dtype="uint8")
+    out = SVC._frames_from_output(arr)
+    assert out[0].getpixel((0, 0)) == (255, 255, 255)
+
+
+def test_the_resolution_name_follows_the_size():
+    assert SVC._resolution_name(480, 832) == "480p"
+    assert SVC._resolution_name(720, 1280) == "720p"
+
+
+# ------------------------------------------------------------ the segment loop
+def test_a_single_segment_keeps_every_frame():
+    p = _Pipe(frames=10)
+    frames, yielded = SVC._run_segments(p, "t2v", "base", _ctx(num_segments=1),
+                                        log=lambda *_: None)
+    assert len(frames) == 10 and yielded == 0
+    assert [c[0] for c in p.calls] == ["t2v"]
+
+
+def test_later_segments_drop_the_re_rendered_tail():
+    """A call emits num_frames but its first num_cond_frames re-render the previous
+    tail, so only the remainder is new. Keeping them would stutter the video."""
+    p = _Pipe(frames=20)
+    frames, _ = SVC._run_segments(p, "t2v", "base",
+                                  _ctx(num_segments=3, num_cond_frames=5),
+                                  log=lambda *_: None)
+    assert len(frames) == 20 + 15 + 15
+    # The first call starts the video; every later one continues it.
+    assert [c[0] for c in p.calls] == ["t2v", "vc", "vc"]
+
+
+def test_a_continuation_request_continues_from_the_first_call():
+    p = _Pipe(frames=20)
+    SVC._run_segments(p, "vc", "base",
+                      _ctx(num_segments=2, cond_video=["prev"]), log=lambda *_: None)
+    assert [c[0] for c in p.calls] == ["vc", "vc"]
+    assert p.calls[0][1]["video"] == ["prev"]
+
+
+def test_each_segment_conditions_on_the_previous_output():
+    p = _Pipe(frames=20)
+    SVC._run_segments(p, "t2v", "base", _ctx(num_segments=2, num_cond_frames=5),
+                      log=lambda *_: None)
+    # The second call's conditioning is the first call's frames, not the accumulator.
+    assert len(p.calls[1][1]["video"]) == 20
+
+
+def test_progress_tracks_the_segment():
+    p = _Pipe(frames=20)
+    SVC._run_segments(p, "t2v", "base", _ctx(num_segments=3), log=lambda *_: None)
+    assert SVC._progress["segments"] == 3
+
+
+# ------------------------------------------------------------ yielding the GPU
+def test_a_yield_request_stops_at_a_segment_boundary():
+    """A release arriving mid-request costs ONE segment instead of the whole generation,
+    and the request still returns rather than being lost."""
+    p = _Pipe(frames=20)
+    SVC._yield_flag.set()
+    try:
+        frames, yielded = SVC._run_segments(p, "t2v", "base",
+                                            _ctx(num_segments=5, num_cond_frames=5),
+                                            log=lambda *_: None)
+    finally:
+        SVC._yield_flag.clear()
+    assert len(p.calls) == 1            # stopped after the first segment
+    assert yielded == 4 and len(frames) == 20
+
+
+def test_a_single_segment_request_is_not_interrupted():
+    p = _Pipe(frames=20)
+    SVC._yield_flag.set()
+    try:
+        _frames, yielded = SVC._run_segments(p, "t2v", "base", _ctx(num_segments=1),
+                                             log=lambda *_: None)
+    finally:
+        SVC._yield_flag.clear()
+    assert yielded == 0 and len(p.calls) == 1
+
+
+def test_the_worker_asks_for_a_yield_before_waiting():
+    from codai.api import longcat_worker as W
+    src = (ROOT / "codai/api/longcat_worker.py").read_text()
+    rel = src[src.index("def release_vram"):]
+    assert "_request_yield(key)" in rel
+    assert callable(W._request_yield)
+
+
+# ------------------------------------------------------------ refinement
+def test_refinement_takes_the_earlier_frames_as_stage1_video():
+    src = (ROOT / "tools/longcat_service.py").read_text()
+    body = src[src.index('if stage == "refinement":'):]
+    assert "stage1_video" in body
+    assert "generate_refine" in body
+
+
+# ------------------------------------------------------------ dispatch mapping
+def test_the_modes_map_onto_the_upstream_tasks():
+    src = (ROOT / "codai/api/video.py").read_text()
+    seg = src[src.index('_task = {"t2v"'):src.index("if _task is None")]
+    for mode, task in (("i2v", "i2v"), ("ti2v", "i2v"), ("extend", "vc"), ("v2v", "vc")):
+        assert f'"{mode}": "{task}"' in seg, mode
+
+
+def test_an_unsupported_mode_is_still_refused():
+    src = (ROOT / "codai/api/video.py").read_text()
+    fn = src[src.index("async def _generate_longcat"):]
+    fn = fn[:fn.index("\nasync def ")]
+    seg = fn[fn.index("if _task is None"):fn.index('payload = {')]
+    assert "status_code=400" in seg
+
+
+def test_continuation_reuses_the_existing_cond_frames_vocabulary():
+    """A client that already chains clips through the VACE path needs no new field."""
+    src = (ROOT / "codai/api/video.py").read_text()
+    fn = src[src.index("async def _generate_longcat"):]
+    fn = fn[:fn.index("\nasync def ")]
+    assert "request.cond_frames" in fn and "cond_frames_b64" in fn
+
+
+def test_the_long_video_fields_exist_and_the_server_owns_the_loop():
+    from codai.pydantic.videorequest import VideoGenerationRequest as R
+    r = R(model="m", prompt="p", num_segments=11, num_cond_frames=13)
+    assert (r.num_segments, r.num_cond_frames) == (11, 13)
+    src = (ROOT / "codai/api/video.py").read_text()
+    fn = src[src.index("async def _generate_longcat"):]
+    assert "num_segments" in fn and "total_frames" in fn
+
+
+def test_a_shortened_video_is_reported_to_the_caller():
+    """Yielding produces a shorter video; the caller must learn that from the response,
+    not discover it."""
+    src = (ROOT / "codai/api/video.py").read_text()
+    fn = src[src.index("async def _generate_longcat"):]
+    assert 'data.get("warning")' in fn and "warnings.append" in fn
+
+
+def test_the_service_returns_a_tail_for_chaining():
+    src = (ROOT / "tools/longcat_service.py").read_text()
+    assert "tail_b64" in src

tests/test_longcat_worker.py

@@ -273,10 +273,12 @@ def test_the_request_model_accepts_the_staging_fields():
     assert VideoGenerationRequest(model="m", prompt="p").quality is None
 
 
-def test_unimplemented_modes_are_refused_not_downgraded():
-    """Step 2 is t2v only. A silent downgrade would look like a quality bug."""
+def test_unsupported_modes_are_refused_not_downgraded():
+    """t2v, i2v and continuation are implemented; the audio-driven avatar tasks are not.
+    A silent downgrade to t2v would look like a quality bug rather than a gap."""
     src = (ROOT / "codai/api/video.py").read_text()
     fn = src[src.index("async def _generate_longcat"):]
     fn = fn[:fn.index("\nasync def ")]
-    assert "is not implemented yet" in fn
+    assert "does not support mode" in fn
     assert "status_code=400" in fn
+    assert "avatar" in fn.lower()          # named as the separate increment it is

tools/longcat_service.py

@@ -62,6 +62,9 @@ _state = {
 }
 _progress = {"active": False, "stage": "", "segment": 0, "segments": 0,
              "step": 0, "steps": 0}
+# Set by POST /yield: stop at the next SEGMENT boundary and return what has been
+# generated, so another model can have the card without the request being lost.
+_yield_flag = threading.Event()
 
 
 def log(msg):
@@ -150,41 +153,145 @@ def unload(reason: str = ""):
         log(f"unloaded{f' ({reason})' if reason else ''}")
 
 
-def _apply_stage_lora(pipe, stage: str, checkpoint_dir: str):
-    """Fuse the LoRA a stage needs, or drop it for the base stage.
-
-    Stages 2 and 3 ARE their LoRAs — the distilled stage is 16 steps because of
-    cfg_step_lora, and refinement is 720p because of refinement_lora. Switching stage
-    without switching the adapter silently produces the wrong thing."""
-    want = {"distill": "distill", "refinement": "refinement"}.get(stage)
-    if _state["loaded_lora"] == want:
-        return
-    unload_fn = getattr(pipe, "unload_lora_weights", None)
-    if _state["loaded_lora"] is not None and callable(unload_fn):
-        unload_fn()
-    if want:
-        rel = LC.LORA_FILES[want]
-        path = os.path.join(os.path.expanduser(checkpoint_dir), rel)
-        load_fn = getattr(pipe, "load_lora_weights", None)
-        if not callable(load_fn):
-            raise RuntimeError(
-                f"stage '{stage}' needs {rel} but this pipeline exposes no "
-                f"load_lora_weights()")
-        load_fn(path)
-        log(f"stage '{stage}': fused {rel}")
-    _state["loaded_lora"] = want
-
-
-# ── generation ────────────────────────────────────────────────────────────────
+def torch_gc():
+    """What upstream calls between segments. The allocator holds onto a segment's
+    activations otherwise, and the next one starts that much closer to the ceiling."""
+    try:
+        import gc
+        import torch
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+            torch.cuda.ipc_collect()
+    except Exception:
+        pass
+
+
+def _frames_from_output(output):
+    """The pipeline returns ``method(...)[0]`` — an array of frames normalised to [0, 1],
+    NOT an object with a ``.frames`` attribute. Upstream converts with
+    ``(tensor * 255).clamp(0, 255).to(uint8)``; we go on to PIL because the refinement
+    stage takes PIL frames as its input and because that is what the muxer wants."""
+    import numpy as np
+    from PIL import Image
+    arr = np.asarray(output)
+    if arr.dtype != np.uint8:
+        arr = (arr * 255.0).clip(0, 255).astype(np.uint8)
+    return [Image.fromarray(f).convert("RGB") for f in arr]
+
+
+def _resolution_name(height: int, width: int) -> str:
+    """generate_i2v and generate_vc take a resolution NAME, where generate_t2v takes
+    height/width. Mapping it here keeps that asymmetry out of the request payload."""
+    return "720p" if max(int(height or 0), int(width or 0)) > 1000 else "480p"
+
+
+def _one_pass(pipe, task: str, stage: str, ctx: dict, cond=None, log=print):
+    """One pipeline call: the first segment of a task, or a continuation of it.
+
+    ``cond`` is the previous segment's frames (PIL). When present the call is always
+    generate_vc, whatever the task started as — that is how a continuation continues."""
+    import torch
+
+    params = LC.stage_params(stage, ctx.get("num_inference_steps"),
+                             ctx.get("guidance_scale"))
+    common = {
+        "prompt": ctx["prompt"],
+        "num_frames": ctx["num_frames"],
+        "num_inference_steps": params["num_inference_steps"],
+        "guidance_scale": params["guidance_scale"],
+    }
+    if ctx.get("generator") is not None:
+        common["generator"] = ctx["generator"]
+    # The distilled stage is this FLAG, not a LoRA we fuse ourselves — upstream passes
+    # use_distill=True to the same method and the pipeline applies cfg_step_lora.
+    if stage == "distill":
+        common["use_distill"] = True
+    elif ctx.get("negative_prompt"):
+        # The distilled demos pass no negative prompt (guidance is 1.0, so it does
+        # nothing); the 50-step passes do.
+        common["negative_prompt"] = ctx["negative_prompt"]
+
+    if cond is not None:
+        kw = dict(common)
+        kw.update(video=cond, num_cond_frames=ctx["num_cond_frames"],
+                  resolution=ctx["resolution"], use_kv_cache=True,
+                  offload_kv_cache=bool(ctx.get("offload_kv_cache")))
+        if stage == "distill":
+            kw["enhance_hf"] = False
+        return pipe.generate_vc(**kw)[0]
+
+    if task == "i2v":
+        return pipe.generate_i2v(image=ctx["image"], resolution=ctx["resolution"],
+                                 **common)[0]
+    if task == "vc":
+        kw = dict(common)
+        kw.update(video=ctx["cond_video"], num_cond_frames=ctx["num_cond_frames"],
+                  resolution=ctx["resolution"], use_kv_cache=True,
+                  offload_kv_cache=bool(ctx.get("offload_kv_cache")))
+        if stage == "distill":
+            kw["enhance_hf"] = False
+        return pipe.generate_vc(**kw)[0]
+    return pipe.generate_t2v(height=ctx["height"], width=ctx["width"], **common)[0]
+
+
+def _run_segments(pipe, task: str, stage: str, ctx: dict, log=print):
+    """Generate ``num_segments`` segments, feeding each one's tail into the next.
+
+    This is upstream's loop: a call emits ``num_frames`` but its first
+    ``num_cond_frames`` re-render the previous segment's tail, so only the remainder is
+    new — ``all_generated_frames.extend(new_video[num_cond_frames:])`` — and the fresh
+    frames become the next call's conditioning. Native pretrained continuation is why
+    this does not drift in colour or quality the way chained frame-tail conditioning on
+    other models does.
+
+    Between segments is also the one safe point to hand the GPU over, so the yield flag
+    is checked here: a release that arrives mid-request costs at most one segment instead
+    of the whole generation."""
+    segments = max(1, int(ctx.get("num_segments") or 1))
+    k = int(ctx["num_cond_frames"])
+    acc: list = []
+    cur = ctx.get("cond_video")
+    first_is_continuation = cur is not None
+    yielded = 0
+
+    for index in range(segments):
+        with _lock:
+            _progress.update(segment=index + 1, segments=segments, step=0,
+                             steps=LC.stage_params(
+                                 stage, ctx.get("num_inference_steps"),
+                                 ctx.get("guidance_scale"))["num_inference_steps"])
+        out = _one_pass(pipe, task, stage, ctx,
+                        cond=cur if (index or first_is_continuation) else None, log=log)
+        new = _frames_from_output(out)
+        if index == 0 and not first_is_continuation:
+            acc.extend(new)
+        else:
+            acc.extend(new[k:])
+        cur = new
+        log(f"stage '{stage}' segment {index + 1}/{segments}: "
+            f"{len(acc)} frames so far")
+        torch_gc()
+        if index + 1 < segments and _yield_flag.is_set():
+            # Another model is waiting for the card. Stop at this boundary and return
+            # what exists: a shorter video plus a warning beats losing the request, and
+            # beats blocking the swap for the remaining segments.
+            yielded = segments - (index + 1)
+            log(f"yielding the GPU after segment {index + 1}/{segments} "
+                f"({yielded} not generated)")
+            break
+    return acc, yielded
+
 
 def _mux(frames, fps: int) -> bytes:
-    """Frames (PIL images) → mp4 bytes, via imageio-ffmpeg in the venv."""
+    """Frames (PIL) → mp4 bytes. Upstream writes with libx264 at crf 18 for the 480p
+    stages and crf 10 for refinement; this keeps 18 and lets the engine re-encode if a
+    model entry sets output_crf."""
     import imageio
     import numpy as np
     buf = io.BytesIO()
     writer = imageio.get_writer(buf, format="mp4", fps=int(fps), codec="libx264",
-                               quality=None, ffmpeg_params=["-crf", "18",
-                                                            "-pix_fmt", "yuv420p"])
+                                ffmpeg_params=["-crf", "18", "-pix_fmt", "yuv420p"])
     try:
         for f in frames:
             writer.append_data(np.asarray(f.convert("RGB")))
@@ -193,8 +300,43 @@ def _mux(frames, fps: int) -> bytes:
     return buf.getvalue()
 
 
+def _decode_image(data: str):
+    from PIL import Image
+    raw = data.split(",", 1)[1] if str(data).startswith("data:") else data
+    return Image.open(io.BytesIO(base64.b64decode(raw))).convert("RGB")
+
+
+def _encode_image(img) -> str:
+    buf = io.BytesIO()
+    img.save(buf, format="PNG")
+    return base64.b64encode(buf.getvalue()).decode()
+
+
+def _peak_vram_gb() -> float:
+    """The peak VRAM this process reached, in GB, or 0.0.
+
+    Reported back so the engine sizes future reservations from a MEASUREMENT instead of
+    an estimate. It has to come from in here: the pipeline loads lazily on the first
+    request, so a host-side free-VRAM delta would also be counting whatever else was
+    loading or evicting on the card. torch's own counter is this process's allocation —
+    reserved rather than allocated, because the caching allocator's arenas are what the
+    card is actually unable to give to anything else."""
+    try:
+        import torch
+        if not torch.cuda.is_available():
+            return 0.0
+        return round(torch.cuda.max_memory_reserved() / (1024 ** 3), 2)
+    except Exception:
+        return 0.0
+
+
 def generate(body: dict) -> dict:
-    """One generation. Step 2 scope: text-to-video, one segment, single GPU."""
+    """One request: text-to-video, image-to-video or continuation, over N segments.
+
+    Tasks map to upstream's entry points — generate_t2v (height/width), generate_i2v
+    (image + resolution name) and generate_vc (conditioning video) — plus generate_refine
+    for the 720p stage, which takes the earlier stage's frames as ``stage1_video``.
+    """
     prompt = (body.get("prompt") or "").strip()
     if not prompt:
         raise ValueError("prompt is required")
@@ -205,6 +347,10 @@ def generate(body: dict) -> dict:
     if problems:
         raise ValueError("; ".join(problems))
 
+    task = (body.get("task") or "t2v").strip().lower()
+    if task not in ("t2v", "i2v", "vc"):
+        raise ValueError(f"unknown task {task!r}; expected t2v, i2v or vc")
+
     pipe = load_pipeline(checkpoint, body.get("source") or _state["source"],
                          body.get("dtype") or _state["dtype"],
                          body.get("offload") or "")
@@ -212,78 +358,100 @@ def generate(body: dict) -> dict:
     height = int(body.get("height") or LC.DEFAULT_BASE_SIZE[0])
     width = int(body.get("width") or LC.DEFAULT_BASE_SIZE[1])
     num_frames = int(body.get("num_frames") or LC.DEFAULT_NUM_FRAMES)
+    cond_frames = int(body.get("num_cond_frames") or LC.DEFAULT_COND_FRAMES)
+    if cond_frames >= num_frames:
+        raise ValueError(f"num_cond_frames ({cond_frames}) must be smaller than "
+                         f"num_frames ({num_frames}) — a segment would add nothing")
     fps = int(body.get("fps") or LC.DEFAULT_FPS)
-    seed = body.get("seed")
 
+    segments = body.get("num_segments")
+    if not segments and body.get("total_frames"):
+        segments = LC.segments_for(int(body["total_frames"]), num_frames, cond_frames)
+    segments = max(1, int(segments or 1))
+
+    image = None
+    if task == "i2v":
+        if not body.get("image"):
+            raise ValueError("task 'i2v' needs an image")
+        image = _decode_image(body["image"])
+    cond_video = None
+    if task == "vc":
+        if not body.get("cond_frames_b64"):
+            raise ValueError("task 'vc' needs cond_frames_b64 (the previous tail)")
+        cond_video = [_decode_image(x) for x in body["cond_frames_b64"]]
+        if len(cond_video) < cond_frames:
+            raise ValueError(f"task 'vc' needs at least {cond_frames} conditioning "
+                             f"frames, got {len(cond_video)}")
+
+    generator = None
+    if body.get("seed") is not None:
+        import torch
+        generator = torch.Generator(
+            device="cuda" if torch.cuda.is_available() else "cpu"
+        ).manual_seed(int(body["seed"]))
+
+    ctx = {
+        "prompt": prompt, "negative_prompt": body.get("negative_prompt") or "",
+        "height": height, "width": width, "num_frames": num_frames,
+        "num_cond_frames": cond_frames, "num_segments": segments,
+        "resolution": _resolution_name(height, width),
+        "num_inference_steps": body.get("num_inference_steps"),
+        "guidance_scale": body.get("guidance_scale"),
+        "offload_kv_cache": body.get("offload_kv_cache"),
+        "image": image, "cond_video": cond_video, "generator": generator,
+    }
+
+    _yield_flag.clear()
     frames = None
+    yielded = 0
     with _lock:
-        _progress.update(active=True, stage="", segment=0, segments=1, step=0, steps=0)
+        _progress.update(active=True, stage="", segment=0, segments=segments,
+                         step=0, steps=0)
     try:
         for stage in stages:
-            params = LC.stage_params(stage, body.get("num_inference_steps"),
-                                     body.get("guidance_scale"))
-            _apply_stage_lora(pipe, stage, checkpoint)
             with _lock:
-                _progress.update(stage=stage, step=0,
-                                 steps=params["num_inference_steps"])
-
-            def _cb(_pipe, step_index, _t, kw):
-                with _lock:
-                    _progress["step"] = int(step_index) + 1
-                return kw
-
-            call = {"prompt": prompt, "height": height, "width": width,
-                    "num_frames": num_frames, **params}
-            if seed is not None:
-                import torch
-                call["generator"] = torch.Generator(
-                    device="cuda" if torch.cuda.is_available() else "cpu"
-                ).manual_seed(int(seed))
-            # The refinement stage takes the PREVIOUS stage's frames as its input — that
-            # is what makes it a refinement and not a second generation.
+                _progress["stage"] = stage
             if stage == "refinement":
                 if frames is None:
                     raise ValueError(
-                        "the refinement stage needs frames from an earlier stage; run "
-                        "quality='best' or pass stage='base' first")
-                call["video"] = frames
-            try:
-                out = pipe(callback_on_step_end=_cb, **call)
-            except TypeError:
-                out = pipe(**call)          # a build without the callback hook
-            frames = getattr(out, "frames", out)
-            if frames and isinstance(frames[0], (list, tuple)):
-                frames = frames[0]
+                        "the refinement stage needs frames from an earlier stage — run "
+                        "quality='best', or pass stage='base' first and feed its frames "
+                        "back with stage='refinement'")
+                params = LC.stage_params(stage, body.get("num_inference_steps"),
+                                         body.get("guidance_scale"))
+                kw = {"prompt": prompt, "stage1_video": frames,
+                      "num_inference_steps": params["num_inference_steps"]}
+                if generator is not None:
+                    kw["generator"] = generator
+                if body.get("spatial_refine_only") is not None:
+                    kw["spatial_refine_only"] = bool(body["spatial_refine_only"])
+                if image is not None:
+                    kw["image"] = image
+                    kw["num_cond_frames"] = 1
+                log(f"stage 'refinement': refining {len(frames)} frames")
+                frames = _frames_from_output(pipe.generate_refine(**kw)[0])
+            else:
+                frames, yielded = _run_segments(pipe, task, stage, ctx, log=log)
             log(f"stage '{stage}': {len(frames)} frames")
     finally:
+        _yield_flag.clear()
         with _lock:
             _progress.update(active=False, stage="", step=0)
 
     if stages[-1] == "refinement":
         fps = int(body.get("fps") or 30)
     mp4 = _mux(frames, fps)
-    return {"mp4_b64": base64.b64encode(mp4).decode(), "num_frames": len(frames),
-            "fps": fps, "stages": list(stages), "segments": 1,
-            "vram_gb": _peak_vram_gb()}
-
-
-def _peak_vram_gb() -> float:
-    """The peak VRAM this process reached, in GB, or 0.0.
-
-    Reported back so the engine can size future reservations from a MEASUREMENT instead
-    of an estimate. It has to come from in here: the pipeline loads lazily on the first
-    request, and a host-side free-VRAM delta would also be counting whatever else was
-    loading or evicting on the card at the same time. torch's own peak counter is
-    exactly this process's allocation."""
-    try:
-        import torch
-        if not torch.cuda.is_available():
-            return 0.0
-        # reserved, not allocated: the caching allocator's arenas are what the card is
-        # actually unable to give to anything else.
-        return round(torch.cuda.max_memory_reserved() / (1024 ** 3), 2)
-    except Exception:
-        return 0.0
+    out = {"mp4_b64": base64.b64encode(mp4).decode(), "num_frames": len(frames),
+           "fps": fps, "stages": list(stages), "task": task,
+           "segments": max(1, segments - yielded), "vram_gb": _peak_vram_gb()}
+    if yielded:
+        out["warning"] = (f"stopped after {segments - yielded} of {segments} segments to "
+                          f"hand the GPU to another model; the video is shorter than "
+                          f"requested")
+    # The tail a caller needs to continue from here, so chaining requests does not mean
+    # re-deriving which frames overlap.
+    out["tail_b64"] = [_encode_image(f) for f in frames[-cond_frames:]]
+    return out
 
 
 # ── HTTP ──────────────────────────────────────────────────────────────────────
@@ -324,6 +492,12 @@ def make_handler():
             try:
                 if self.path.startswith("/generate"):
                     self._json(generate(self._read_json()))
+                elif self.path.startswith("/yield"):
+                    # Stop at the next segment boundary. The in-flight request still
+                    # returns — shorter, with a warning — so a swap costs one segment
+                    # rather than the whole generation.
+                    _yield_flag.set()
+                    self._json({"ok": True})
                 elif self.path.startswith("/unload"):
                     unload("requested")
                     self._json({"ok": True})