video: a LongCat generation can be cancelled (v0.2.64)

Asked to stop an offloaded render, I killed the service process by hand — and had
to, because nothing else could. Every diffusers video path calls
task_registry.raise_if_cancelled() between denoise steps and shows up on the Tasks
page; LongCat was never registered at all. So the LONGEST-running video job in the
system was the only one with no stop button: a 93-frame bf16 render offloading
across PCIe can go for a very long time.

Its poll loop claimed otherwise. The docstring said it would "carry cancellation
and the thermal gate across the process boundary" and the body did neither — it
only mirrored progress. A comment describing intent rather than behaviour.

It is now registered like the others, reports step progress to the task, and polls
for cancellation every 2s. Because the work happens in ANOTHER PROCESS, cancelling
cannot raise inside it: the service is stopped instead, which is exactly what
eviction already does to it, so that path is well exercised. The request then ends
499 rather than returning a partial video as success, and the task is closed on
every exit — cancelled, error or done.

I also owe a correction: I told the user there was no cancel for
/v1/video/generations. There is, for every other model — I searched for a
/v1/video/* endpoint and missed the task registry. The gap was LongCat
specifically, which is code I wrote.

7 tests.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent d8a62966
......@@ -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.63"
__version__ = "0.2.64"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
......@@ -4241,12 +4241,43 @@ async def _generate_longcat(request: VideoGenerationRequest, model_name: str,
_stop = threading.Event()
_done_steps = {"base": 0}
# Register like every other video path, so this generation appears on the Tasks
# page and can be cancelled there. It could not be before: the poll loop below
# claimed in its own docstring to "carry cancellation across the process
# boundary" and never did, so a LongCat render — the longest-running video job
# there is — was the only one nothing could stop.
_lc_steps = sum(lc.STAGE_DEFAULTS.get(_s, (0, 0))[0] for _s in stages) or 1
_tid = task_registry.register(
"video", title=(payload.get("prompt") or mode or "")[:80],
model=model_name or "", total=_lc_steps)
task_registry.start(_tid)
_cancelled = {"flag": False}
def _poll():
"""Mirror the service's progress onto the engine's bar, and carry cancellation
and the thermal gate across the process boundary."""
across the process boundary.
The generation runs in another process, so cancelling cannot raise inside it:
stopping the SERVICE is what ends it, and the blocking generate() call then
returns with an error. That is also exactly what eviction does to it, so the
path is already exercised."""
_seen_stage = ""
while not _stop.wait(2.0):
try:
if not _cancelled["flag"]:
try:
task_registry.raise_if_cancelled(_tid)
except TaskCancelled:
_cancelled["flag"] = True
print(f" [longcat] cancelled — stopping the service",
flush=True)
try:
longcat_worker.stop_service(
longcat_worker.service_key(model_path, model_cfg))
except Exception as exc: # noqa: BLE001
print(f" [longcat] could not stop the service: {exc}",
flush=True)
continue
p = longcat_worker.progress(model_path, model_cfg)
if not p or not p.get("active"):
continue
......@@ -4256,7 +4287,9 @@ async def _generate_longcat(request: VideoGenerationRequest, model_name: str,
_done_steps["base"] += lc.STAGE_DEFAULTS.get(
_seen_stage, (0, 0))[0]
_seen_stage = st
_vid_progress_step(_done_steps["base"] + int(p.get("step") or 0))
_step = _done_steps["base"] + int(p.get("step") or 0)
_vid_progress_step(_step)
task_registry.step(_tid, _step)
except Exception:
pass
......@@ -4265,9 +4298,20 @@ async def _generate_longcat(request: VideoGenerationRequest, model_name: str,
try:
data = await asyncio.to_thread(longcat_worker.generate, model_name, payload,
model_cfg)
except Exception:
if _cancelled["flag"]:
task_registry.finish(_tid, "cancelled", "stopped on request")
raise HTTPException(status_code=499,
detail="LongCat generation cancelled") from None
task_registry.finish(_tid, "error", "generation failed")
raise
finally:
_stop.set()
_vid_progress_done()
if _cancelled["flag"]:
task_registry.finish(_tid, "cancelled", "stopped on request")
raise HTTPException(status_code=499, detail="LongCat generation cancelled")
task_registry.finish(_tid, "done")
mp4_bytes = base64.b64decode(data.get("mp4_b64") or "")
frames_made = data.get("num_frames")
......
"""A LongCat generation must be cancellable, like every other video path.
Every diffusers-based video path calls task_registry.raise_if_cancelled() between
denoise steps, so the Tasks page can stop it. LongCat did not: it was never
registered at all. Its poll loop even claimed, in its own docstring, to "carry
cancellation across the process boundary" — and only mirrored progress.
That left the LONGEST-running video job in the system as the only one nothing
could stop. A 93-frame bf16 render offloading across PCIe can run for a very long
time, and the only way out was killing the service by hand.
"""
import pathlib
import re
ROOT = pathlib.Path(__file__).resolve().parents[1]
VIDEO = ROOT / "codai" / "api" / "video.py"
def _longcat_body() -> str:
src = VIDEO.read_text(encoding="utf-8")
start = src.index("async def _generate_longcat")
nxt = re.search(r"\n(async def |def )", src[start + 10:])
return src[start:start + 10 + nxt.start()] if nxt else src[start:]
def test_the_generation_is_registered_as_a_task():
body = _longcat_body()
assert "task_registry.register(" in body, "not on the Tasks page at all"
assert "task_registry.start(" in body
def test_it_checks_for_cancellation():
assert "task_registry.raise_if_cancelled(" in _longcat_body()
def test_cancelling_stops_the_service():
"""It runs in another process, so cancelling cannot raise inside it — stopping
the service is what actually ends the work."""
body = _longcat_body()
assert "longcat_worker.stop_service(" in body
def test_progress_reaches_the_task_not_only_the_bar():
body = _longcat_body()
assert "task_registry.step(" in body, "the Tasks page would show no progress"
def test_the_task_is_closed_on_every_exit():
body = _longcat_body()
for status in ('"cancelled"', '"error"', '"done"'):
assert f"task_registry.finish(_tid, {status}" in body, status
def test_a_cancelled_generation_does_not_report_success():
body = _longcat_body()
assert "499" in body, "a cancelled request should not return a video or a 200"
def test_the_docstring_no_longer_claims_what_the_code_does_not_do():
body = _longcat_body()
if "carry cancellation across the process boundary" in body:
assert "raise_if_cancelled" in body, \
"the docstring promises cancellation; the code must deliver it"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment