13, 13b and A4: progress a client can actually use, and a still that is not 70s

13. A request may name itself (`request_id` in the body, or an X-Request-Id
header) and poll /v1/{images,video,audio}/progress?request_id=... for its OWN
steps. One global slot per media type meant two callers could not tell their
progress apart, so a client stopped reading them altogether and showed no step
progress at all while the server was drawing. The name is used as the TASK id,
so this is a VIEW over the existing task registry -- which already tracks
step/total/status per task under a lock and prunes finished ones -- rather than
a second place progress is written. Claimed once, because a request registers
a short "loading" task as well and both adopting the id would serve the client
whichever happened to be last. Unknown id = 404, so a client stops polling on
a definite answer instead of reading somebody else's progress.

13b. The encode reports frames written (`encoding 1200/2400 frames`) across the
band it owns, so the longest step of a render stops looking like a stall; and
the job's own log, kept all along but unreachable, is served as a tail in the
status document plus GET /v1/video/compose/{id}/log?since=N for the whole
thing incrementally. The record holds the live list, not a snapshot.

A4. 72s per 768x1344 still -> 21.4s, measured on an idle 3090. The card carried
`offload_strategy: "model"`, which is the model page's DEFAULT, so
enable_model_cpu_offload() ran on a model that fits a 23.56 GB card whole and
paid PCIe for it on every request. `auto` loads on GPU and falls back to
offload only on OOM. The doc's other hypothesis is ruled out rather than
assumed: the live progress endpoint reports total=10, so the card's step count
IS applied, and an explicit steps=10 takes the same 21.2s.

Also: `models.civitai_cache_dir`, configured like hf_cache_dir and
gguf_cache_dir because it is the same kind of thing. Unset, these
multi-gigabyte checkpoints land beside the other caches -- on this host a
140 GB volume, which a 13.5 GB checkpoint took to 99% until the next model
could not write even its config fetch ("No space left on device"). That is
what was actually stopping Z-Image-Turbo-IntoRealism from loading at all.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014T3rZnLhydX9QqUcfBbqLu

codai/api/audio_gen.py

@@ -26,6 +26,7 @@ import io
 import os
 import time
 import uuid
+from typing import Optional
 
 from fastapi import APIRouter, HTTPException, Request
 
@@ -448,8 +449,25 @@ def _decode_b64_or_url(data: str) -> bytes:
 
 
 @router.get("/v1/audio/progress", summary="Audio generation progress")
-async def get_audio_progress():
-    """Return current audio generation progress including speed."""
+async def get_audio_progress(request_id: Optional[str] = None):
+    """Audio generation progress — this request's own, or the global slot.
+
+    With ``request_id`` (the id the caller passed as ``request_id`` in the
+    generation body or an ``X-Request-Id`` header) this answers for THAT
+    request only, so two callers generating at once can each follow their own.
+    A 404 means that id is finished and pruned, or was never used.
+
+    Without it, the single global slot below, exactly as before.
+    """
+    if request_id:
+        from codai.api import request_progress as _rp
+        _rid = _rp.sanitize(request_id)
+        _view = _rp.view(_rid) if _rid else None
+        if _view is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"no generation in progress for request_id {request_id!r}")
+        return _view
     elapsed = time.monotonic() - _aud_progress["started_at"] if _aud_progress["active"] else 0.0
     total = _aud_progress["total"]
     current = _aud_progress["current"]
@@ -471,7 +489,13 @@ async def audio_generate(request: AudioGenerationRequest, http_request: Request
     """
     Generate music, sound effects, or ambient audio.
     Compatible models: MusicGen, AudioGen, AudioLDM2, StableAudio.
+
+    A caller may name this request (``request_id`` in the body, or an
+    ``X-Request-Id`` header) and then poll
+    ``GET /v1/audio/progress?request_id=...`` for its OWN progress.
     """
+    from codai.api import request_progress as _rp
+    _rp.set_current(_rp.from_request(request, http_request))
     _aud_progress_loading(request.model or "audio")
     model_info = await asyncio.to_thread(
         multi_model_manager.request_model, request.model, model_type="audio_gen")
@@ -513,8 +537,11 @@ async def audio_generate(request: AudioGenerationRequest, http_request: Request
         multi_model_manager.models[model_key] = pipe
         multi_model_manager.current_model_key = model_key
 
+    # Claims the caller's own id when they named one, so
+    # /v1/audio/progress?request_id=... follows THIS request.
     _tid = task_registry.register(
-        "audio", title=(request.prompt or "")[:80], model=model_name or "")
+        "audio", title=(request.prompt or "")[:80], model=model_name or "",
+        task_id=_rp.take())
     task_registry.start(_tid)
     try:
         audio_bytes, ext = await asyncio.get_event_loop().run_in_executor(

codai/api/compose.py

@@ -344,12 +344,28 @@ def _get(job_id: str) -> Optional[dict]:
         return dict(rec) if rec else None
 
 
+#: How many of the job's own log lines the polled document carries. A client
+#: logs every change of stage/message and wants the detail behind a long step,
+#: but a growing log re-sent on every 2-3 s poll is waste; the tail covers what
+#: just happened and ``/log?since=N`` serves the whole thing incrementally.
+_LOG_TAIL = 40
+
+
 def _public(rec: dict) -> dict:
     """The document a client polls: the spec's fields, in the spec's order."""
+    _log = rec.get("log") or []
     return {"id": rec["id"], "status": rec["status"], "stage": rec.get("stage"),
             "progress": round(float(rec.get("progress") or 0.0), 1),
             "message": rec.get("message") or "", "warnings": rec.get("warnings") or [],
-            "error": rec.get("error"), "result": rec.get("result")}
+            "error": rec.get("error"), "result": rec.get("result"),
+            # The job's own log. Some steps are long with no change of stage at
+            # all (an encode, a presenter being generated, a model loading), so
+            # a client watching only stage/message has nothing to show for
+            # minutes. `log_total` is the full length, so a client can tell it
+            # is seeing a tail and fetch the rest from
+            # GET /v1/video/compose/{id}/log?since=N
+            "log": list(_log[-_LOG_TAIL:]),
+            "log_total": len(_log)}
 
 
 def cancel_job(job_id: str) -> bool:
@@ -1110,11 +1126,18 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
     tdur = float((req.transition or {}).get("duration") or 0.3)
     caps = req.captions
     log_lines: List[str] = []
+    # The SAME list object on the record, so a poll sees lines as they are
+    # appended rather than a copy taken at some earlier moment. Trimming
+    # below mutates it in place, which keeps that true.
+    _update(job_id, log=log_lines)
 
     def log(msg: str) -> None:
         log_lines.append(msg)
-        if len(log_lines) > 200:
-            del log_lines[:100]
+        if len(log_lines) > 2000:
+            # Deep enough to cover a whole render for a client reading
+            # incrementally; trimmed from the front so the tail is the recent
+            # work. `log_total` in the status document reports the real length.
+            del log_lines[:500]
 
     def stage(name: str, progress: float, message: str) -> None:
         if cancel.is_set():
@@ -1734,11 +1757,30 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
 
         fmt = (req.output.format or "mp4").lower()
         final = os.path.join(work, f"{job_id}.{fmt}")
+
+        # The encode is the longest single step and has no sub-stages, so its
+        # own frame counter drives progress across the band this stage owns
+        # (76 -> 95). Without it a client's bar sat still for minutes with the
+        # message unchanged, which reads as a stall.
+        _enc_frames = max(1, int(round(float(total) * float(canvas.fps or 30))))
+        _enc_last = [0.0]
+
+        def _on_frame(n: int) -> None:
+            pct = 76.0 + 19.0 * min(1.0, n / _enc_frames)
+            # Throttled: ffmpeg reports several times a second and each update
+            # takes the jobs lock, which every poll also needs.
+            if pct - _enc_last[0] < 0.5 and n < _enc_frames:
+                return
+            _enc_last[0] = pct
+            _update(job_id, stage="render", progress=round(pct, 1),
+                    message=f"encoding {n}/{_enc_frames} frames", status="running")
+
         rnd.final_render(silent, mixed, final, width=canvas.width, height=canvas.height,
                          fps=canvas.fps, total=total, ass_file=ass_path,
                          overlays=overlays, crf=req.output.crf,
                          audio_bitrate=req.output.audio_bitrate,
-                         video_codec=req.output.video_codec, cancel=cancel, log=log)
+                         video_codec=req.output.video_codec, cancel=cancel, log=log,
+                         on_frame=_on_frame)
 
         # ---------------------------------------------------- 6. artefacts
         stage("thumbnail", 95.0, "artefacts")
@@ -1889,6 +1931,34 @@ async def compose_status(job_id: str, _auth=Depends(_require_api_auth)):
     return _public(rec)
 
 
+@router.get("/v1/video/compose/{job_id}/log", summary="A composition job's own log",
+            tags=["Video"])
+async def compose_log(job_id: str, since: int = 0, _auth=Depends(_require_api_auth)):
+    """The job's log from line ``since`` onwards.
+
+    Polled alongside the status document when a client wants the detail behind
+    a long step: an encode, a presenter being generated, a model loading. Pass
+    the ``next`` from the previous answer as ``since`` to get only what is new.
+
+    ``since`` beyond the end returns nothing rather than an error, so a client
+    that polls once more after the job finished gets an empty tail. A very old
+    ``since`` may have been trimmed away; ``dropped`` says so, and the lines
+    start from the oldest line still held.
+    """
+    rec = _get(job_id)
+    if rec is None:
+        raise HTTPException(status_code=404, detail=f"unknown composition job {job_id!r}")
+    lines = list(rec.get("log") or [])
+    total = len(lines)
+    start = max(0, int(since or 0))
+    dropped = start > total                  # asked past the end
+    if dropped:
+        start = total
+    return {"id": job_id, "status": rec["status"], "since": start,
+            "next": total, "total": total, "dropped": dropped,
+            "log": lines[start:]}
+
+
 @router.post("/v1/video/compose/{job_id}/cancel", summary="Cancel a composition job",
              tags=["Video"])
 async def compose_cancel(job_id: str, _auth=Depends(_require_api_auth)):

codai/api/images.py

@@ -1575,9 +1575,15 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
 
     # Register this generation as a cancellable task (live view + cooperative
     # cancel via the step callback below).
+    # The caller's own id, when they named one, so /v1/images/progress can be
+    # asked about THIS request instead of a global slot shared with everyone
+    # else generating at the same time. Claimed once: the short "loading" task
+    # registered earlier keeps its own id (see codai.api.request_progress).
+    from codai.api import request_progress as _rp
     _tid = task_registry.register(
         "image", title=(request.prompt or "")[:80],
-        model=getattr(request, 'model', '') or '', total=num_steps)
+        model=getattr(request, 'model', '') or '', total=num_steps,
+        task_id=_rp.take())
     task_registry.start(_tid)
 
     # ------------------------------------------------------------------
@@ -2204,8 +2210,27 @@ router = APIRouter()
 
 
 @router.get("/v1/images/progress", summary="Image generation progress")
-async def get_image_progress():
-    """Return current image generation step progress including speed."""
+async def get_image_progress(request_id: Optional[str] = None):
+    """Image generation progress — this request's own, or the global slot.
+
+    With ``request_id`` (the id the caller passed as ``request_id`` in the
+    generation body or an ``X-Request-Id`` header) this answers for THAT
+    request only: `phase` is "queued" while it waits, "loading" while the model
+    loads and "generating" once steps are counted, and a 404 means the id is
+    finished and pruned or was never used. Two callers generating at once can
+    each follow their own.
+
+    Without it, the single global slot below, exactly as before.
+    """
+    if request_id:
+        from codai.api import request_progress as _rp
+        _rid = _rp.sanitize(request_id)
+        _view = _rp.view(_rid) if _rid else None
+        if _view is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"no generation in progress for request_id {request_id!r}")
+        return _view
     elapsed = _time.monotonic() - _gen_progress["started_at"] if _gen_progress["active"] else 0.0
     return {
         "current":  _gen_progress["current"],
@@ -2228,8 +2253,15 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
     Supports:
     - Stable Diffusion via stable-diffusion-cpp-python (sd.cpp)
     - Stable Diffusion XL (via local inference with diffusers)
+
+    A caller may name this request (``request_id`` in the body, or an
+    ``X-Request-Id`` header) and then poll
+    ``GET /v1/images/progress?request_id=...`` for its OWN steps rather than a
+    slot shared with everyone else generating at the same time.
     """
     global global_args, global_file_path, model_semaphores, queue_flags
+    from codai.api import request_progress as _rp
+    _rp_token = _rp.set_current(_rp.from_request(request, http_request))
     
     # Get or create semaphore for this model
     model_key = f"image:{request.model}" if request.model else "image"

codai/api/request_progress.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Progress for ONE request, instead of one global slot per media type.
+
+``/v1/images/progress`` and its video/audio siblings each reported a single
+module-level dict. With two people generating at once, neither can tell their
+own request's steps from the other's, so a client that cares about its own
+progress has to stop reading them altogether -- which is what happened: a
+caller rendering video footage and AI edits showed no step progress at all
+while the server was drawing.
+
+A request can therefore name itself, with ``request_id`` in the body or an
+``X-Request-Id`` header, and ask for only its own progress. That name is used
+as the TASK id, so this is a view over :mod:`codai.tasks.registry` rather than
+a second place where progress is written: the task registry already tracks
+step/total/status per task under a lock, prunes finished ones, and is already
+updated from inside every step callback. A parallel registry would be a second
+thing to keep correct.
+
+Without an id the endpoints answer exactly as before, so nothing that does not
+ask for this changes.
+"""
+
+from typing import Any, Dict, Optional
+
+#: The header a client may use instead of a body field. Same name HTTP proxies
+#: and tracing tools already use, so a caller often has one to hand.
+HEADER = "X-Request-Id"
+
+#: How a caller's own id is accepted in a request body.
+BODY_FIELD = "request_id"
+
+#: Cap on a client-supplied id. It becomes a task id and is echoed back, so it
+#: is bounded and stripped of anything that is not safe to put in a log line.
+MAX_LEN = 128
+
+
+def sanitize(raw: Any) -> Optional[str]:
+    """A client id reduced to something safe to use as a task id, or None.
+
+    Client-supplied, so it is treated as untrusted: control characters cannot
+    reach a log line, and the length is bounded. An id that survives none of
+    its characters is no id at all.
+    """
+    if raw is None:
+        return None
+    s = str(raw).strip()
+    if not s:
+        return None
+    keep = [c for c in s if c.isalnum() or c in "-_.:"]
+    s = "".join(keep)[:MAX_LEN]
+    return s or None
+
+
+def from_request(body: Any = None, http_request: Any = None) -> Optional[str]:
+    """The caller's own id for this request: body field, else header, else None.
+
+    The body wins, because a caller that put the id in the payload meant that
+    one; a header may have been added by a proxy on the way.
+    """
+    rid = None
+    if body is not None:
+        got = getattr(body, BODY_FIELD, None)
+        if got is None and isinstance(body, dict):
+            got = body.get(BODY_FIELD)
+        rid = sanitize(got)
+    if rid:
+        return rid
+    if http_request is not None:
+        try:
+            rid = sanitize(http_request.headers.get(HEADER))
+        except Exception:
+            rid = None
+    return rid
+
+
+def _phase(task: Dict[str, Any]) -> str:
+    """The task's state in the words a progress bar needs.
+
+    ``queued`` while it waits its turn, ``loading`` while the model is being
+    loaded (running, but no step budget yet -- a caller holds the bar there
+    rather than animating a total it does not have), ``generating`` once steps
+    are being counted, and the terminal state otherwise.
+    """
+    status = str(task.get("status") or "")
+    if status == "queued":
+        return "queued"
+    if status == "running":
+        return "generating" if int(task.get("total") or 0) > 0 else "loading"
+    return status or "idle"
+
+
+def view(request_id: str) -> Optional[Dict[str, Any]]:
+    """``{phase, current, total, pct, active, ...}`` for one request, or None.
+
+    None means "nothing known under that id" -- either it was never used or it
+    has been pruned -- and the endpoints turn that into a 404, so a client can
+    stop polling on a definite answer instead of a stale global one.
+    """
+    from codai.tasks import task_registry
+    task = task_registry.get(request_id)
+    if not task:
+        return None
+    current = int(task.get("step") or 0)
+    total = int(task.get("total") or 0)
+    return {
+        "request_id": request_id,
+        "phase": _phase(task),
+        "current": current,
+        "total": total,
+        "pct": int(current / total * 100) if total > 0 else 0,
+        "active": bool(task.get("active")),
+        "model": task.get("model") or "",
+        "message": task.get("message") or "",
+        "status": task.get("status") or "",
+        "rate": task.get("rate") or 0.0,
+        "paused": bool(task.get("paused")),
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Carrying the id from the endpoint to the task that does the work
+# --------------------------------------------------------------------------- #
+#
+# The id is resolved at the endpoint, but the task that counts steps is
+# registered deeper in, past several call layers and inside a worker thread.
+# A ContextVar carries it without threading an argument through all of them:
+# `asyncio.to_thread` copies the context, which is how every generation path
+# here offloads, so the value is visible in the step callback.
+#
+# `take` is one-shot ON PURPOSE. One request registers more than one task -- a
+# short "loading" task and then the generation itself -- and if both adopted
+# the caller's id the second would overwrite the first under the same key and
+# the client would be served whichever happened to be last. The generation
+# task claims the id; anything else keeps its own generated one.
+
+import contextvars as _contextvars
+
+_current: "_contextvars.ContextVar[Optional[str]]" = _contextvars.ContextVar(
+    "coderai_request_progress_id", default=None)
+
+
+def set_current(request_id: Optional[str]):
+    """Make ``request_id`` the id the next task may claim. Returns a token."""
+    return _current.set(request_id or None)
+
+
+def reset_current(token) -> None:
+    try:
+        _current.reset(token)
+    except Exception:
+        pass
+
+
+def take() -> Optional[str]:
+    """The caller's id, once. Returns None on every later call."""
+    rid = _current.get()
+    if rid:
+        _current.set(None)
+    return rid
+
+
+def peek() -> Optional[str]:
+    """The caller's id without consuming it (for logs and messages)."""
+    return _current.get()

codai/api/video.py

@@ -2731,9 +2731,13 @@ def _generate_video(pipe, request: VideoGenerationRequest):
 
     _vid_progress_reset(kw['num_inference_steps'])
 
+    # Claims the caller's own id when they named one, so
+    # /v1/video/progress?request_id=... follows THIS request.
+    from codai.api import request_progress as _rp
     _tid = task_registry.register(
         "video", title=(request.prompt or mode or "")[:80],
-        model=getattr(request, 'model', '') or '', total=kw['num_inference_steps'])
+        model=getattr(request, 'model', '') or '', total=kw['num_inference_steps'],
+        task_id=_rp.take())
     task_registry.start(_tid)
 
     def _vid_step_cb(pipe, step_index, timestep, callback_kwargs):
@@ -4000,8 +4004,25 @@ def _translate_srt(srt_path: str, target_lang: str, temps: list) -> str:
 # =============================================================================
 
 @router.get("/v1/video/progress", summary="Video generation progress")
-async def get_video_progress():
-    """Return current video generation step progress including speed."""
+async def get_video_progress(request_id: Optional[str] = None):
+    """Video generation progress — this request's own, or the global slot.
+
+    With ``request_id`` (the id the caller passed as ``request_id`` in the
+    generation body or an ``X-Request-Id`` header) this answers for THAT
+    request only, so two callers generating at once can each follow their own.
+    A 404 means that id is finished and pruned, or was never used.
+
+    Without it, the single global slot below, exactly as before.
+    """
+    if request_id:
+        from codai.api import request_progress as _rp
+        _rid = _rp.sanitize(request_id)
+        _view = _rp.view(_rid) if _rid else None
+        if _view is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"no generation in progress for request_id {request_id!r}")
+        return _view
     elapsed = time.monotonic() - _vid_progress["started_at"] if _vid_progress["active"] else 0.0
     return {
         "current":  _vid_progress["current"],
@@ -4763,9 +4784,16 @@ async def video_generations(request: VideoGenerationRequest,
       v2v   – video-to-video (video required)
       ti2v  – text + image → video (prompt is primary driver)
       interp – frame interpolation (init_image + end_image required)
+
+    A caller may name this request (``request_id`` in the body, or an
+    ``X-Request-Id`` header) and then poll
+    ``GET /v1/video/progress?request_id=...`` for its OWN steps rather than a
+    slot shared with everyone else generating at the same time.
     """
     if not request.model:
         raise HTTPException(status_code=400, detail="model is required")
+    from codai.api import request_progress as _rp
+    _rp.set_current(_rp.from_request(request, http_request))
     _vid_progress_loading(request.model)
 
     # Infer mode from inputs if not set

codai/compose/render.py

@@ -88,23 +88,65 @@ def _die_with_parent() -> None:
 
 
 def run(args: List[str], *, cancel: Optional[threading.Event] = None,
-        timeout: float = 7200.0, log: Optional[Callable[[str], None]] = None) -> None:
+        timeout: float = 7200.0, log: Optional[Callable[[str], None]] = None,
+        on_frame: Optional[Callable[[int], None]] = None) -> None:
     """Run one ffmpeg command, honouring cancellation while it runs.
 
     The default ``timeout`` is a backstop for a long video encode, not a
     sensible bound for every call -- an audio mix of a few seconds that is
     still running after a minute is wedged, not slow. Callers that know the
     shape of their work pass their own.
+
+    ``on_frame`` is called with the count of frames written so far, as ffmpeg
+    reports it. The final encode of a reel is minutes of work behind a single
+    unchanging stage, so without this a client's progress bar sits still
+    through the longest part of the job with nothing to say.
     """
     if cancel is not None and cancel.is_set():
         raise Cancelled()
     cmd = [media.ffmpeg_bin(), "-hide_banner", "-nostdin", "-y",
-           "-threads", _thread_cap()] + list(args)
+           "-threads", _thread_cap()]
+    if on_frame is not None:
+        # `-progress pipe:1` writes machine-readable key=value lines to stdout,
+        # which is parsed below. Asked for only when somebody is listening, so
+        # a command's stdout keeps its usual shape otherwise.
+        cmd += ["-progress", "pipe:1", "-nostats"]
+    cmd += list(args)
     if log:
         log("ffmpeg " + " ".join(shlex.quote(a) for a in args[:40])
             + (" …" if len(args) > 40 else ""))
-    proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
+    _stdout = subprocess.PIPE if on_frame is not None else subprocess.DEVNULL
+    proc = subprocess.Popen(cmd, stdout=_stdout, stderr=subprocess.PIPE,
                             preexec_fn=_die_with_parent)
+
+    _reader = None
+    if on_frame is not None:
+        def _pump():
+            # A frame count that only ever goes up: ffmpeg repeats the key on
+            # every progress block, and a caller turning this into a
+            # percentage should never see it jump backwards.
+            seen = 0
+            try:
+                for raw in proc.stdout:
+                    line = raw.decode("utf-8", "replace").strip()
+                    if not line.startswith("frame="):
+                        continue
+                    try:
+                        n = int(line.split("=", 1)[1])
+                    except Exception:
+                        continue
+                    if n > seen:
+                        seen = n
+                        try:
+                            on_frame(n)
+                        except Exception:
+                            # A reporting callback must never break the encode.
+                            pass
+            except Exception:
+                pass
+        _reader = threading.Thread(target=_pump, daemon=True)
+        _reader.start()
+
     waited = 0.0
     while True:
         try:
@@ -122,6 +164,8 @@ def run(args: List[str], *, cancel: Optional[threading.Event] = None,
                 proc.kill()
                 proc.communicate()
                 raise RenderError(f"ffmpeg timed out after {timeout:.0f}s")
+    if _reader is not None:
+        _reader.join(timeout=2.0)
     if proc.returncode != 0:
         tail = (err or b"").decode(errors="replace").strip().splitlines()[-12:]
         raise RenderError("ffmpeg failed (rc=%d): %s" % (proc.returncode, " | ".join(tail)))
@@ -931,8 +975,14 @@ def final_render(video: str, audio: str, out_path: str, *, width: int, height: i
                  fps: int, total: float, ass_file: Optional[str] = None,
                  overlays: Optional[List[dict]] = None, crf: int = 20,
                  audio_bitrate: str = "192k", video_codec: str = "h264",
-                 fonts_dir: Optional[str] = None, cancel=None, log=None) -> None:
-    """Burn captions and overlays and encode the deliverable."""
+                 fonts_dir: Optional[str] = None, cancel=None, log=None,
+                 on_frame: Optional[Callable[[int], None]] = None) -> None:
+    """Burn captions and overlays and encode the deliverable.
+
+    ``on_frame`` reports frames written as the encode runs. This is the longest
+    single step of a render and it has no sub-stages, so a client polling
+    stage/message sees nothing change for minutes without it.
+    """
     inputs = ["-i", video, "-i", audio]
     img_overlays = [o for o in (overlays or [])
                     if (o.get("type") or "image").lower() == "image" and o.get("_path")]
@@ -1037,7 +1087,7 @@ def final_render(video: str, audio: str, out_path: str, *, width: int, height: i
     run(inputs + ["-filter_complex", ";".join(graph), "-map", "[vout]", "-map", "1:a"]
         + enc + ["-c:a", "aac", "-b:a", str(audio_bitrate or "192k"),
                  "-ar", "48000", "-ac", "2", "-shortest",
-                 "-t", f"{max(0.1, total):.3f}", out_path],
+                 "-t", f"{max(0.1, total):.3f}", out_path], on_frame=on_frame,
         cancel=cancel, log=log)
 
 

codai/config.py

@@ -178,6 +178,15 @@ class ModelsConfig:
     default_load_mode: str = "ondemand"
     hf_cache_dir: Optional[str] = None
     gguf_cache_dir: Optional[str] = None
+    #: Where Civitai single-file checkpoints are kept. Configured like the two
+    #: above because they are the same KIND of thing -- multi-gigabyte weights
+    #: that belong on whichever disk holds the weights, not on the partition
+    #: the server runs from. Left unset they land beside the other coderai
+    #: caches, which on a host whose caches are a small volume fills it: seen
+    #: on 2026-10-28, a 13.5 GB checkpoint took the cache volume to 99% and the
+    #: next model could not even write its config fetch ("No space left on
+    #: device"). CODERAI_CIVITAI_CACHE still wins over this.
+    civitai_cache_dir: Optional[str] = None
     max_model_instances: int = 1  # max concurrent instances per model (global default; overridable per-model via "max_instances")
     # Per-engine overrides for max_model_instances, keyed by engine name
     # (e.g. {"nvidia": 2, "radeon": 1}). Applied per-engine process; blank = default.

codai/models/hf_loading.py

@@ -469,6 +469,13 @@ def civitai_cache_root() -> Path:
     explicit = (os.environ.get("CODERAI_CIVITAI_CACHE") or "").strip()
     if explicit:
         return Path(os.path.expanduser(explicit))
+    # `models.civitai_cache_dir`, read from config.json the same way
+    # `models.hf_cache_dir` is. An operator configuring where the weights live
+    # should not have to set a container environment variable to move these,
+    # and env is not somewhere a running server's settings page can reach.
+    configured = _configured_civitai_cache_dir()
+    if configured:
+        return Path(configured)
     try:
         from codai.platform_paths import legacy_style_cache_root
         return legacy_style_cache_root() / "civitai"
@@ -476,6 +483,24 @@ def civitai_cache_root() -> Path:
         return Path(os.path.expanduser("~/.cache/coderai/civitai"))
 
 
+def _configured_civitai_cache_dir() -> str:
+    """``models.civitai_cache_dir`` from config.json, or "".
+
+    Read from the file rather than through the config object: this is called
+    from loader code that runs in engine processes and in isolated workers,
+    some of which never build a Config at all.
+    """
+    import json as _json
+    try:
+        _dir = os.path.join(
+            os.environ.get("CODERAI_CONFIG_DIR", "/config"), "coderai")
+        with open(os.path.join(_dir, "config.json")) as fh:
+            raw = (_json.load(fh).get("models") or {}).get("civitai_cache_dir")
+    except Exception:
+        return ""
+    return os.path.expanduser(str(raw).strip()) if raw and str(raw).strip() else ""
+
+
 def resolve_civitai(ref: str) -> str:
     """``civitai:<modelVersionId>[:<fileId>]`` -> a local checkpoint, fetched once.
 

codai/pydantic/audiogenrequest.py

@@ -21,6 +21,12 @@ from pydantic import BaseModel, ConfigDict
 
 
 class AudioGenerationRequest(BaseModel):
+    #: The caller's own name for this request, so it can poll
+    #: ``/v1/audio/progress?request_id=...`` for its OWN steps instead of a
+    #: single global slot shared with everyone generating at the same time.
+    #: An ``X-Request-Id`` header does the same; the body wins when both are
+    #: given. Echoed back in the progress document.
+    request_id: Optional[str] = None
     model: str
     prompt: str
     duration: Optional[float] = 10.0       # seconds

codai/pydantic/imagerequest.py

@@ -38,6 +38,12 @@ class LoraConfig(BaseModel):
 
 
 class ImageGenerationRequest(BaseModel):
+    #: The caller's own name for this request, so it can poll
+    #: ``/v1/images/progress?request_id=...`` for its OWN steps instead of a
+    #: single global slot shared with everyone generating at the same time.
+    #: An ``X-Request-Id`` header does the same; the body wins when both are
+    #: given. Echoed back in the progress document.
+    request_id: Optional[str] = None
     model: str = Field(..., description="Model id to generate with (must be a configured image model).")
     prompt: str = Field(..., description="Text prompt describing the image.")
     n: int = Field(1, description="Number of images to generate.")

codai/pydantic/videorequest.py

@@ -52,6 +52,12 @@ class CharacterDialogLine(BaseModel):
 
 
 class VideoGenerationRequest(BaseModel):
+    #: The caller's own name for this request, so it can poll
+    #: ``/v1/video/progress?request_id=...`` for its OWN steps instead of a
+    #: single global slot shared with everyone generating at the same time.
+    #: An ``X-Request-Id`` header does the same; the body wins when both are
+    #: given. Echoed back in the progress document.
+    request_id: Optional[str] = None
     model: str = Field(..., description="Model id to generate with (must be a configured video model).")
     prompt: str = Field("", description="Text prompt describing the video. Optional for pure i2v.")
     negative_prompt: Optional[str] = Field(None, description="What to avoid in the output.")

tests/test_compose_job_log_and_encode_progress.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""13b: finer compose progress, and the job's own log.
+
+A client logs every change of `stage`/`message` and moves its bar on
+`progress`. Some steps are long with no change at all -- the encode from 76 to
+95 %, a presenter being generated, a model loading -- so the bar sat still for
+minutes, which reads as a stall. The encode now reports frames written, and the
+job's log (already kept, previously unreachable) is served both as a tail in
+the status document and incrementally.
+"""
+
+import subprocess
+
+import pytest
+
+from codai.api import compose as C
+from codai.compose import render as rnd
+
+
+def _ffmpeg_ok():
+    try:
+        from codai.compose import media
+        subprocess.run([media.ffmpeg_bin(), "-version"],
+                       capture_output=True, timeout=30)
+        return True
+    except Exception:
+        return False
+
+
+needs_ffmpeg = pytest.mark.skipif(not _ffmpeg_ok(), reason="ffmpeg not available")
+
+
+# ------------------------------------------------------------------ the log
+
+def test_the_status_document_carries_a_log_tail_and_the_real_length():
+    rec = {"id": "j1", "status": "running", "stage": "render", "progress": 80.0,
+           "log": [f"line {i}" for i in range(200)]}
+    doc = C._public(rec)
+    assert doc["log_total"] == 200, "a client cannot tell it is seeing a tail"
+    assert len(doc["log"]) == C._LOG_TAIL
+    assert doc["log"][-1] == "line 199", "the tail must be the RECENT lines"
+
+
+def test_a_job_with_no_log_still_answers():
+    doc = C._public({"id": "j2", "status": "queued"})
+    assert doc["log"] == [] and doc["log_total"] == 0
+
+
+def test_the_log_endpoint_serves_only_what_is_new():
+    """A client passes back the previous `next`, so each poll is incremental."""
+    import asyncio
+    jid = "test-log-job"
+    C._jobs[jid] = {"id": jid, "status": "running",
+                    "log": ["a", "b", "c", "d"]}
+    try:
+        first = asyncio.run(C.compose_log(jid, since=0, _auth=None))
+        assert first["log"] == ["a", "b", "c", "d"]
+        assert first["next"] == 4 and first["dropped"] is False
+
+        C._jobs[jid]["log"].append("e")
+        second = asyncio.run(C.compose_log(jid, since=first["next"], _auth=None))
+        assert second["log"] == ["e"], second
+        assert second["next"] == 5
+    finally:
+        C._jobs.pop(jid, None)
+
+
+def test_polling_past_the_end_is_empty_not_an_error():
+    """A client that polls once more after the job finished gets a clean tail."""
+    import asyncio
+    jid = "test-log-job-2"
+    C._jobs[jid] = {"id": jid, "status": "done", "log": ["x"]}
+    try:
+        got = asyncio.run(C.compose_log(jid, since=99, _auth=None))
+        assert got["log"] == []
+        assert got["dropped"] is True, "a since past the end should say so"
+    finally:
+        C._jobs.pop(jid, None)
+
+
+def test_an_unknown_job_log_is_a_404():
+    import asyncio
+    from fastapi import HTTPException
+    with pytest.raises(HTTPException) as ei:
+        asyncio.run(C.compose_log("no-such-job", since=0, _auth=None))
+    assert ei.value.status_code == 404
+
+
+def test_the_log_is_the_live_list_not_a_snapshot():
+    """The record holds the SAME list the job appends to, so a poll sees lines
+    as they are written rather than as they were at some earlier moment."""
+    import inspect
+    src = inspect.getsource(C._run_job)
+    assert "_update(job_id, log=log_lines)" in src, (
+        "the job's log list is no longer published on the record")
+
+
+# -------------------------------------------------------- encode progress
+
+@needs_ffmpeg
+def test_run_reports_frames_written(tmp_path):
+    """The hook that makes the encode band move."""
+    out = str(tmp_path / "out.mp4")
+    seen = []
+    rnd.run(["-f", "lavfi", "-i", "testsrc=size=160x120:rate=10:duration=2",
+             "-pix_fmt", "yuv420p", out],
+            on_frame=seen.append, timeout=120)
+    assert seen, "ffmpeg wrote frames but none were reported"
+    assert seen == sorted(seen), f"the frame count went backwards: {seen}"
+    assert seen[-1] >= 10, f"only reached {seen[-1]} frames of ~20"
+
+
+@needs_ffmpeg
+def test_a_command_without_the_hook_is_unchanged(tmp_path):
+    """-progress is asked for only when somebody is listening."""
+    out = str(tmp_path / "plain.mp4")
+    rnd.run(["-f", "lavfi", "-i", "testsrc=size=160x120:rate=10:duration=1",
+             "-pix_fmt", "yuv420p", out], timeout=120)
+    import os
+    assert os.path.getsize(out) > 0
+
+
+@needs_ffmpeg
+def test_a_raising_callback_does_not_break_the_encode(tmp_path):
+    """Reporting progress must never cost the render."""
+    out = str(tmp_path / "boom.mp4")
+
+    def _boom(n):
+        raise RuntimeError("callback exploded")
+
+    rnd.run(["-f", "lavfi", "-i", "testsrc=size=160x120:rate=10:duration=1",
+             "-pix_fmt", "yuv420p", out], on_frame=_boom, timeout=120)
+    import os
+    assert os.path.getsize(out) > 0, "a broken callback lost the output"
+
+
+def test_final_render_accepts_the_hook():
+    import inspect
+    assert "on_frame" in inspect.signature(rnd.final_render).parameters
+
+
+def test_compose_drives_its_encode_band_from_frames():
+    import inspect
+    src = inspect.getsource(C._run_job)
+    assert "encoding {n}/{_enc_frames} frames" in src or \
+           'f"encoding {n}/{_enc_frames} frames"' in src, (
+        "the encode no longer reports frames written")
+    assert "on_frame=_on_frame" in src, "the hook is not passed to final_render"

tests/test_per_request_progress.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""13: progress for ONE request, not a single global slot.
+
+`/v1/images/progress` and its video/audio siblings reported one module-level
+dict, so with two people generating at once neither could tell their own
+request's steps from the other's -- and a client that cares about its own
+progress has to stop reading them, which is what happened.
+
+A request may name itself (`request_id` in the body, or an `X-Request-Id`
+header) and ask only about itself. The name is used as the TASK id, so this is
+a view over the existing task registry rather than a second place progress is
+written.
+"""
+
+import pytest
+
+from codai.api import request_progress as rp
+from codai.tasks import task_registry
+
+
+# ------------------------------------------------------------------ the id
+
+@pytest.mark.parametrize("raw,expect", [
+    ("abc-123", "abc-123"),
+    ("  spaced  ", "spaced"),
+    ("a/b\\c", "abc"),                      # path separators dropped
+    ("tab\there", "tabhere"),               # control characters dropped
+    ("req.1:2-3_4", "req.1:2-3_4"),         # the safe punctuation survives
+    ("", None),
+    ("   ", None),
+    (None, None),
+    ("///", None),                          # nothing safe left == no id
+])
+def test_a_client_id_is_reduced_to_something_safe(raw, expect):
+    assert rp.sanitize(raw) == expect
+
+
+def test_a_very_long_id_is_bounded():
+    got = rp.sanitize("x" * 5000)
+    assert got is not None and len(got) == rp.MAX_LEN
+
+
+def test_the_body_field_wins_over_the_header():
+    """A caller that put the id in the payload meant that one; a header may
+    have been added by a proxy on the way."""
+    class _Body:
+        request_id = "from-body"
+
+    class _Http:
+        headers = {"X-Request-Id": "from-header"}
+
+    assert rp.from_request(_Body(), _Http()) == "from-body"
+
+
+def test_the_header_is_used_when_the_body_has_no_id():
+    class _Body:
+        request_id = None
+
+    class _Http:
+        headers = {"X-Request-Id": "from-header"}
+
+    assert rp.from_request(_Body(), _Http()) == "from-header"
+
+
+def test_no_id_anywhere_is_none():
+    class _Body:
+        request_id = None
+
+    class _Http:
+        headers = {}
+
+    assert rp.from_request(_Body(), _Http()) is None
+    assert rp.from_request(None, None) is None
+
+
+def test_a_dict_body_works_too():
+    assert rp.from_request({"request_id": "d1"}, None) == "d1"
+
+
+# ------------------------------------------------- carrying it to the task
+
+def test_the_id_is_claimed_exactly_once():
+    """One request registers more than one task.
+
+    A short "loading" task is registered before the generation itself. If both
+    adopted the caller's id the second would overwrite the first under the same
+    key and the client would be served whichever happened to be last.
+    """
+    rp.set_current("only-once")
+    assert rp.take() == "only-once"
+    assert rp.take() is None, "a second task also claimed the caller's id"
+
+
+def test_peek_does_not_consume():
+    rp.set_current("peeked")
+    assert rp.peek() == "peeked"
+    assert rp.peek() == "peeked"
+    assert rp.take() == "peeked"
+
+
+def test_no_id_set_means_tasks_keep_their_own():
+    rp.set_current(None)
+    assert rp.take() is None
+
+
+# --------------------------------------------------------------- the view
+
+@pytest.fixture
+def a_task():
+    tid = "test-req-" + "abc123"
+    task_registry.remove(tid) if hasattr(task_registry, "remove") else None
+    yield tid
+    try:
+        task_registry.remove(tid)
+    except Exception:
+        pass
+
+
+def test_an_unknown_id_has_no_view(a_task):
+    """So the endpoint can 404 and the client stops polling on a definite
+    answer rather than reading a stale global slot."""
+    assert rp.view("nothing-was-ever-registered-here") is None
+
+
+def test_a_queued_request_reports_queued(a_task):
+    task_registry.register("image", total=0, status="queued", task_id=a_task)
+    v = rp.view(a_task)
+    assert v["phase"] == "queued"
+    assert v["active"] is True
+    assert v["pct"] == 0
+
+
+def test_a_running_request_with_no_step_budget_reports_loading(a_task):
+    """A client holds the bar during a load rather than animating a total it
+    does not have yet."""
+    task_registry.register("image", total=0, task_id=a_task)
+    task_registry.start(a_task)
+    v = rp.view(a_task)
+    assert v["phase"] == "loading", v
+    assert v["total"] == 0
+
+
+def test_a_running_request_with_steps_reports_generating_and_a_pct(a_task):
+    task_registry.register("image", total=20, model="m", task_id=a_task)
+    task_registry.start(a_task)
+    task_registry.step(a_task, 5)
+    v = rp.view(a_task)
+    assert v["phase"] == "generating"
+    assert (v["current"], v["total"], v["pct"]) == (5, 20, 25)
+    assert v["model"] == "m"
+    assert v["request_id"] == a_task
+
+
+def test_a_finished_request_is_no_longer_active(a_task):
+    task_registry.register("image", total=4, task_id=a_task)
+    task_registry.start(a_task)
+    task_registry.step(a_task, 4)
+    task_registry.finish(a_task, "done")
+    v = rp.view(a_task)
+    assert v is None or v["active"] is False, v
+
+
+def test_two_requests_do_not_see_each_others_steps():
+    """The whole point of 13."""
+    a, b = "test-req-aaa", "test-req-bbb"
+    try:
+        task_registry.register("image", total=10, task_id=a)
+        task_registry.register("image", total=100, task_id=b)
+        task_registry.start(a)
+        task_registry.start(b)
+        task_registry.step(a, 1)
+        task_registry.step(b, 50)
+        va, vb = rp.view(a), rp.view(b)
+        assert (va["current"], va["total"], va["pct"]) == (1, 10, 10)
+        assert (vb["current"], vb["total"], vb["pct"]) == (50, 100, 50)
+    finally:
+        for t in (a, b):
+            try:
+                task_registry.remove(t)
+            except Exception:
+                pass
+
+
+def test_the_generation_models_declare_the_field():
+    """So it is discoverable in /openapi.json, not only accepted as an extra."""
+    from codai.pydantic.imagerequest import ImageGenerationRequest
+    from codai.pydantic.videorequest import VideoGenerationRequest
+    from codai.pydantic.audiogenrequest import AudioGenerationRequest
+    for model in (ImageGenerationRequest, VideoGenerationRequest,
+                  AudioGenerationRequest):
+        assert "request_id" in model.model_fields, model.__name__