compose: talking presenter (lip-synced character) and /v1/video/talking-head

Motus request 4. The most common reel format is a person talking to camera, and
only this server can build it: composition synthesises the narration, so it holds
the exact audio of each scene, and the lip-sync engines and identity-locked video
generation are here too.

- codai/compose/presenter.py: per scene — a face (a character profile's front
  reference, or a portrait), something to move in (a still, the caller's clip, or
  a short identity-locked shot generated through the ordinary video endpoint so
  its model config/acceleration/keyframe identity all apply), then an engine
  driven by THAT scene's narration, held to the scene's exact length and the
  canvas size. wav2lip and sadtalker are known by name; anything else installed
  under its own name can be selected by that name, so a newer engine needs no API
  change. engine=auto prefers a video-driven engine when there is a clip and a
  portrait engine otherwise.
- Layouts in render.py: full (replaces the scene's visuals — the scene may then
  have none), pip (a masked circle/rounded/rect bubble with an optional border,
  inside the caption safe area) and split (stacked halves). With a presenter a
  scene's visuals are joined without an internal transition and composited as one
  clip; without one nothing changes.
- Bottom captions move above a bottom bubble: ASS takes a per-event MarginV, so
  only the scenes that need it move (captions.py carries margin_v through).
- Failure never loses a render: no engine, no face, a crash or a failed composite
  falls back to the scene's visuals (a gradient when there are none) and warns,
  naming the scene. New stage "presenter"; result.scenes[i].presenter says which
  engine and layout, or null.
- Lip-sync is the only GPU step in a composition, so it takes a lease from the
  central scheduler instead of fighting a generation for the card.
- POST /v1/video/talking-head (+ /{id}, /{id}/cancel) is the same pipeline as a
  one-scene full-screen composition, for a character-page preview, capped at 15 s;
  GET /v1/video/presenter/engines tells a client what is installed so it can grey
  the feature out instead of meeting a 501. Both stay local in the capability
  gateway (a video pod has the shims but not this install's characters).
- 400 for an unknown character / invalid layout, position, size, shape, motion, or
  a pip scene with no visuals; 501 with what to install when no engine is present.
- docs/video-compose.md, README; tests/test_compose.py 32 -> 43 (engines stubbed,
  compositing verified on real pixels).
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B7EnDW6RSd2nhEPZMzMkqi

README.md

@@ -205,6 +205,11 @@ Full guide: [`docs/cluster.md`](docs/cluster.md).
 - **Character Profiles**: Apply up to 6 saved character profiles for visual consistency. Pipelines with IP-Adapter (AnimateDiff and other SD-based video models) are conditioned directly. For the rest — Wan, CogVideoX, LTX, SVD, i.e. most video models — CoderAI renders an **identity keyframe** on the image endpoint (which does have real IP-Adapter) and runs the video as `ti2v` from that frame, so `character_profiles` locks a face on every model. Controlled by `keyframe_identity` (`auto` / `always` / `never`) with optional `keyframe_model` and `keyframe_steps`; combine with a per-character LoRA for the strongest consistency
 - **Environment Profiles**: Apply up to 6 saved environment profiles to condition the scene
 - **Multi-Character Dialog**: Assign spoken lines to individual characters — each line picks a character profile, voice profile or TTS voice ID, and text; lines are **voice-cloned with F5-TTS** from the named profile, mixed with correct timing (sequential or manual), and lip-synced via Wav2Lip
+- **Talking presenter**: a scene can be a character talking to camera, lip-synced to that
+  scene's narration — full screen, a bubble over the B-roll (captions move above it), or a
+  split screen; the face comes from a character profile, the motion from a short
+  identity-locked clip, a still or footage you supply. `POST /v1/video/talking-head` renders
+  one line for a preview. Any installed lip-sync engine (wav2lip, sadtalker, …) can be named
 - **Composition (reels)**: `POST /v1/video/compose` turns a list of **scenes** — narration plus visuals — into one finished, uploadable video: the scene's length comes from its narration, visuals are Ken-Burns'd or trimmed to fit, music is looped and ducked under the voice, captions are burned in with word timings (karaoke presets) inside the platforms' safe area, and a thumbnail, SRT/VTT and the narration track come back with it. Async jobs with stage/progress/cancel; a visual that fails to download becomes a gradient and a warning instead of a failed render. See [`docs/video-compose.md`](docs/video-compose.md)
 - **Lip Sync**: Wav2Lip and SadTalker. The Docker/OCI image ships both (launcher shims plus a dedicated lip-sync venv; checkpoints download on first use). A from-source install has no shims, so CoderAI falls back to a self-managed Wav2Lip that fetches and patches itself on first use. Either way, if lip sync can't run the response carries a `warnings` entry saying the audio was muxed without mouth sync — rather than passing off an unsynced video as a success
 
@@ -831,6 +836,8 @@ persisted per `config_id`, so the configs don't overwrite each other.
 | `POST /v1/video/generations` | Generate video (t2v/i2v/v2v/ti2v/interp) — supports `character_profiles`, `environment_profiles`, `dialogs` |
 | `POST /v1/video/compose` | Compose scenes into a finished reel (narration timing, visuals, ducked music, burned captions, thumbnail) — async job |
 | `GET /v1/video/compose/{id}` | Composition progress / result; `POST …/{id}/cancel` stops it |
+| `POST /v1/video/talking-head` | Lip-sync one line onto a character's face (preview); same job store |
+| `GET /v1/video/presenter/engines` | Which lip-sync engines this install has |
 | `POST /v1/video/upscale` | Upscale video |
 | `POST /v1/video/subtitle` | Generate/burn subtitles |
 | `POST /v1/video/interpolate` | Frame interpolation |

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.22"
+__version__ = "0.2.23"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/compose.py

@@ -68,7 +68,8 @@ def set_global_file_path(path: str) -> None:
 _MAX_SCENES = 200
 _MAX_VISUALS_PER_SCENE = 12
 _JOB_HISTORY = 100
-_STAGES = ("queued", "narration", "captions", "visuals", "music", "render", "thumbnail")
+_STAGES = ("queued", "narration", "captions", "presenter", "visuals", "music",
+           "render", "thumbnail")
 
 
 # =============================================================================
@@ -111,6 +112,27 @@ class Scene(BaseModel):
     padding: float = 0.15
     visuals: List[Visual] = Field(default_factory=list)
     voice: Optional[VoiceSpec] = None       # per-scene override (a second speaker)
+    # A presenter block for this scene, or false to opt out of the request's one.
+    presenter: Optional[Any] = None
+    model_config = ConfigDict(extra="allow")
+
+
+class PresenterSpec(BaseModel):
+    """A character talking to camera, lip-synced to the scene's narration."""
+    character: Optional[str] = None          # a saved character profile
+    image: Optional[str] = None              # or an explicit portrait
+    engine: str = "auto"                     # auto | sadtalker | wav2lip | <installed name>
+    motion: str = "generate"                 # generate | still | src
+    base_src: Optional[str] = None           # clip to drive (motion=src)
+    prompt: Optional[str] = None             # for motion=generate
+    video_model: Optional[str] = None        # model for motion=generate (blank = default)
+    layout: str = "full"                     # full | pip | split
+    position: str = "bottom-right"           # pip
+    size: float = 0.38                       # pip: width as a fraction of the canvas
+    shape: str = "circle"                    # pip: circle | rounded | rect
+    split: str = "bottom"                    # split: which half the presenter takes
+    remove_background: bool = False
+    border: Optional[Dict[str, Any]] = None  # pip: {"color": …, "width": …}
     model_config = ConfigDict(extra="allow")
 
 
@@ -180,6 +202,7 @@ class ComposeRequest(BaseModel):
     transition: Dict[str, Any] = Field(default_factory=lambda: {"type": "none",
                                                                "duration": 0.3})
     music: Optional[MusicSpec] = None
+    presenter: Optional[PresenterSpec] = None
     captions: Optional[CaptionSpec] = None
     overlays: List[Overlay] = Field(default_factory=list)
     thumbnail: Optional[ThumbnailSpec] = None
@@ -361,9 +384,13 @@ def _validate(req: ComposeRequest) -> None:
             raise HTTPException(status_code=400, detail=(
                 f"scenes[{i}]: a scene with no text and no audio needs an explicit "
                 f"\"duration\""))
-        if not sc.visuals:
-            raise HTTPException(status_code=400,
-                                detail=f"scenes[{i}].visuals: at least one visual is required")
+        pres = presenter_for(sc, req)
+        if pres is not None:
+            _validate_presenter(pres, f"scenes[{i}].presenter" if sc.presenter else "presenter")
+        if not sc.visuals and not (pres and (pres.get("layout") or "full").lower() == "full"):
+            raise HTTPException(status_code=400, detail=(
+                f"scenes[{i}].visuals: at least one visual is required "
+                f"(a scene with no visuals needs a full-screen presenter)"))
         if len(sc.visuals) > _MAX_VISUALS_PER_SCENE:
             raise HTTPException(status_code=400, detail=(
                 f"scenes[{i}].visuals: at most {_MAX_VISUALS_PER_SCENE} per scene"))
@@ -414,6 +441,86 @@ def _validate(req: ComposeRequest) -> None:
             "(engine=clone) on this install; pass per-scene \"audio\" instead"))
 
 
+def presenter_for(scene: Scene, req: ComposeRequest) -> Optional[dict]:
+    """This scene's presenter block, or None.
+
+    A scene may override the request's presenter, or opt out with
+    ``"presenter": false`` — the sign-off is full screen, the middle is B-roll."""
+    own = scene.presenter
+    if own is False or (isinstance(own, str) and own.strip().lower() in ("false", "none", "off")):
+        return None
+    base = req.presenter.model_dump() if req.presenter else None
+    if own is None or own is True:
+        return dict(base) if base else None
+    if isinstance(own, dict):
+        merged = dict(base or {})
+        merged.update({k: v for k, v in own.items() if v is not None})
+        # A scene-level block with neither a character nor an image inherits the
+        # request's face rather than failing.
+        if not (merged.get("character") or merged.get("image")):
+            return None
+        return merged
+    return dict(base) if base else None
+
+
+def _validate_presenter(spec: dict, where: str) -> None:
+    layout = (spec.get("layout") or "full").lower()
+    if layout not in ("full", "pip", "split"):
+        raise HTTPException(status_code=400, detail=(
+            f"{where}.layout: expected full, pip or split (got {spec.get('layout')!r})"))
+    if layout == "pip":
+        pos = (spec.get("position") or "bottom-right").lower().replace("_", "-")
+        if pos not in ("top-left", "top-right", "bottom-left", "bottom-right",
+                       "bottom-center", "bottom-centre", "top-center", "top-centre",
+                       "center", "centre"):
+            raise HTTPException(status_code=400, detail=(
+                f"{where}.position: expected one of top-left, top-right, bottom-left, "
+                f"bottom-right, bottom-center (got {spec.get('position')!r})"))
+        try:
+            size = float(spec.get("size") if spec.get("size") is not None else 0.38)
+        except (TypeError, ValueError):
+            size = -1.0
+        if not (0.1 <= size <= 0.95):
+            raise HTTPException(status_code=400,
+                                detail=f"{where}.size: expected 0.1..0.95 (got {spec.get('size')!r})")
+        if (spec.get("shape") or "circle").lower() not in ("circle", "rounded", "round", "rect"):
+            raise HTTPException(status_code=400, detail=(
+                f"{where}.shape: expected circle, rounded or rect (got {spec.get('shape')!r})"))
+    if layout == "split" and (spec.get("split") or "bottom").lower() not in ("top", "bottom"):
+        raise HTTPException(status_code=400,
+                            detail=f"{where}.split: expected top or bottom")
+    if (spec.get("motion") or "generate").lower() not in ("generate", "still", "src"):
+        raise HTTPException(status_code=400, detail=(
+            f"{where}.motion: expected generate, still or src (got {spec.get('motion')!r})"))
+    if (spec.get("motion") or "generate").lower() == "src" and not spec.get("base_src"):
+        raise HTTPException(status_code=400,
+                            detail=f'{where}.base_src: required when motion is "src"')
+    char = (spec.get("character") or "").strip()
+    if not char and not (spec.get("image") or "").strip():
+        raise HTTPException(status_code=400,
+                            detail=f"{where}: needs a character or an image")
+    if char:
+        try:
+            from codai.api.characters import _load_character_meta
+            known = bool(_load_character_meta(char))
+        except Exception:
+            known = True          # cannot check → let the render warn instead
+        if not known:
+            raise HTTPException(status_code=400,
+                                detail=f"{where}.character: no character profile named {char!r}")
+    from codai.compose import presenter as _pres
+    want = (spec.get("engine") or "auto").strip()
+    if not _pres.engine_installed(want):
+        have = _pres.available_engines()
+        raise HTTPException(status_code=501, detail=(
+            f"{where}: lip-sync engine "
+            + (f"{want!r} is not installed" if want.lower() != "auto"
+               else "is not installed")
+            + (f" (installed: {', '.join(have)})" if have else
+               "; install the wav2lip or sadtalker shim (the published images ship both), "
+               "or drop the presenter block")))
+
+
 def _model_ids(mtype: str) -> List[str]:
     try:
         from codai.models.manager import multi_model_manager
@@ -485,6 +592,38 @@ def _await(loop, coro, timeout: float = 1800.0):
     return fut.result(timeout=timeout)
 
 
+class _Lease:
+    """A scheduler lease held across a blocking GPU step, released on close().
+
+    Lip-sync is the only GPU work in a composition; everything else is ffmpeg.
+    Taking a lease from the central scheduler means it queues with ordinary model
+    requests instead of fighting a generation for the card. A scheduler that
+    refuses (or is not there) is not a reason to fail: the step just runs."""
+
+    def __init__(self, loop, lease):
+        self._loop, self._lease = loop, lease
+
+    def close(self) -> None:
+        if self._lease is None:
+            return
+        try:
+            from codai.queue.manager import queue_manager
+            _await(self._loop, queue_manager.release(self._lease), timeout=60)
+        except Exception:
+            pass
+        self._lease = None
+
+
+def _gpu_lease(loop, model_key: str) -> _Lease:
+    try:
+        from codai.queue.manager import queue_manager
+        lease = _await(loop, queue_manager.acquire(
+            f"compose-{model_key}-{uuid.uuid4().hex[:8]}", model_key), timeout=3600)
+    except Exception:
+        lease = None
+    return _Lease(loop, lease)
+
+
 def _b64_to_file(b64: str, path: str) -> str:
     with open(path, "wb") as f:
         f.write(base64.b64decode(b64))
@@ -653,6 +792,9 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
             starts.append(round(t, 3))
             t += d
 
+        # Clips are built one transition longer so the overlap is absorbed.
+        extra_t = tdur if ttype != "none" else 0.0
+
         # ---------------------------------------------------- 2. captions
         events: List[dict] = []
         if caps and caps.enabled:
@@ -676,22 +818,74 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
                     except Exception as exc:
                         warn(f"scene {i}: word timestamps failed ({exc}); estimated instead")
                 span = max(0.2, scene_dur[i] - max(0.0, float(sc.padding or 0.0)))
-                words += (cap.align_words(text, stt, starts[i], span) if stt
-                          else cap.estimate_words(text, starts[i], span))
+                got = (cap.align_words(text, stt, starts[i], span) if stt
+                       else cap.estimate_words(text, starts[i], span))
+                # A presenter bubble at the bottom would sit on the captions:
+                # push this scene's lines above it (per-event ASS margin).
+                bump = _caption_margin_bump(presenter_for(sc, req), caps, canvas)
+                if bump:
+                    for w in got:
+                        w["margin_v"] = bump
+                words += got
             events = cap.break_lines(words, caps.max_words_per_line, caps.max_lines)
             if caps.font and not media.font_installed(caps.font):
                 warn(f"font {caps.font!r} is not installed; captions use the "
                      f"closest available family")
 
-        # ---------------------------------------------------- 3. visuals
+        # ------------------------------------------- 3. presenter (lip-sync)
+        # Built before the visuals because a full-screen presenter replaces them,
+        # and because it is the GPU part: it should not wait behind ffmpeg.
+        presenters: List[Optional[dict]] = [None] * n_scenes
+        want_pres = [presenter_for(sc, req) for sc in req.scenes]
+        if any(p is not None for p in want_pres):
+            from codai.compose import presenter as _pres
+            for i, sc in enumerate(req.scenes):
+                spec = want_pres[i]
+                if spec is None:
+                    continue
+                stage("presenter", 30.0 + 9.0 * i / max(1, n_scenes),
+                      f"scene {i + 1}/{n_scenes}: lip-sync")
+                if spec.get("remove_background"):
+                    warn(f"scene {i}: presenter.remove_background is not available on "
+                         f"this install; the presenter is composited as filmed")
+                try:
+                    presenters[i] = _pres.build_scene_presenter(
+                        spec, narration=scene_audio[i], duration=scene_dur[i] + extra_t,
+                        width=canvas.width, height=canvas.height, fps=canvas.fps,
+                        workdir=work, index=i, loop=loop, cancel=cancel, log=log,
+                        gpu_lease=lambda: _gpu_lease(loop, "lipsync"))
+                    presenters[i]["layout"] = (spec.get("layout") or "full").lower()
+                    presenters[i]["spec"] = spec
+                except rnd.Cancelled:
+                    raise
+                except Exception as exc:
+                    warn(f"scene {i}: presenter skipped ({exc}); the scene's visuals "
+                         f"are used instead")
+                    if not sc.visuals:
+                        # Nothing left to show: a gradient is the documented floor.
+                        sc.visuals = [Visual(type="gradient")]
+
+        # ---------------------------------------------------- 4. visuals
         clips: List[str] = []
         clip_durs: List[float] = []
         made = 0
         total_visuals = sum(len(sc.visuals) for sc in req.scenes)
-        extra = tdur if ttype != "none" else 0.0
         for i, sc in enumerate(req.scenes):
+            pres = presenters[i]
+            # A full-screen presenter IS the scene: no visuals are rendered.
+            if pres and pres["layout"] == "full":
+                stage("visuals", 40.0 + 28.0 * made / max(1, total_visuals or 1),
+                      f"scene {i + 1}/{n_scenes}: presenter (full screen)")
+                clips.append(pres["path"])
+                clip_durs.append(scene_dur[i] + extra_t)
+                continue
+            scene_clips: List[str] = []
             n = max(1, len(sc.visuals))
-            slot = scene_dur[i] / n
+            # With a presenter the scene is composited as ONE clip, so its visuals
+            # share the scene without a transition between them; without one they
+            # stay separate clips and transitions apply between every visual, as
+            # they always have.
+            slot = scene_dur[i] / n if pres else scene_dur[i] / n
             for j, v in enumerate(sc.visuals):
                 stage("visuals", 40.0 + 28.0 * made / max(1, total_visuals),
                       f"scene {i + 1}/{n_scenes}: visual {j + 1}/{n}")
@@ -710,7 +904,7 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
                 out = os.path.join(work, f"clip-{i}-{j}.mp4")
                 try:
                     rnd.build_visual(spec, out, width=canvas.width, height=canvas.height,
-                                     fps=canvas.fps, duration=slot + extra, fit=fit,
+                                     fps=canvas.fps, duration=slot + extra_t, fit=fit,
                                      index=made, workdir=work, cancel=cancel, log=log)
                 except rnd.Cancelled:
                     raise
@@ -718,11 +912,46 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
                     warn(f"scene {i} visual {j}: render failed ({exc}); used gradient")
                     rnd.build_visual({"type": "gradient"}, out, width=canvas.width,
                                      height=canvas.height, fps=canvas.fps,
-                                     duration=slot + extra, fit=fit, index=made,
+                                     duration=slot + extra_t, fit=fit, index=made,
                                      workdir=work, cancel=cancel, log=log)
-                clips.append(out)
-                clip_durs.append(slot + extra)
+                scene_clips.append(out)
                 made += 1
+            if pres:
+                # One clip for the scene, then the bubble or the split over it.
+                joined = os.path.join(work, f"scene-{i}.mp4")
+                if len(scene_clips) == 1:
+                    joined = scene_clips[0]
+                else:
+                    rnd.concat_clips(scene_clips, joined, work, cancel=cancel, log=log)
+                comp = os.path.join(work, f"scene-{i}-pres.mp4")
+                spec = pres["spec"]
+                try:
+                    if pres["layout"] == "pip":
+                        rnd.composite_pip(joined, pres["path"], comp,
+                                          width=canvas.width, height=canvas.height,
+                                          fps=canvas.fps, size=spec.get("size", 0.38),
+                                          position=spec.get("position") or "bottom-right",
+                                          shape=spec.get("shape") or "circle",
+                                          border=spec.get("border"), workdir=work,
+                                          cancel=cancel, log=log)
+                    else:
+                        rnd.composite_split(joined, pres["path"], comp,
+                                            width=canvas.width, height=canvas.height,
+                                            fps=canvas.fps,
+                                            presenter_half=spec.get("split") or "bottom",
+                                            cancel=cancel, log=log)
+                    clips.append(comp)
+                except rnd.Cancelled:
+                    raise
+                except Exception as exc:
+                    warn(f"scene {i}: presenter could not be composited ({exc}); "
+                         f"the scene's visuals are used as they are")
+                    presenters[i] = None
+                    clips.append(joined)
+                clip_durs.append(scene_dur[i] + extra_t)
+            else:
+                clips.extend(scene_clips)
+                clip_durs.extend([slot + extra_t] * len(scene_clips))
 
         stage("visuals", 69.0, "joining the clips")
         silent = os.path.join(work, "video.mp4")
@@ -863,9 +1092,13 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
                     result[kind] = publish(paths[kind], f"{job_id}.{kind}")
         if "narration" in want:
             result["narration"] = publish(narration, f"{job_id}_voice.wav")
-        result["scenes"] = [{"index": i, "start": starts[i],
-                             "end": round(starts[i] + scene_dur[i], 3)}
-                            for i in range(n_scenes)]
+        result["scenes"] = [
+            {"index": i, "start": starts[i],
+             "end": round(starts[i] + scene_dur[i], 3),
+             "presenter": ({"engine": presenters[i]["engine"],
+                            "layout": presenters[i]["layout"]}
+                           if presenters[i] else None)}
+            for i in range(n_scenes)]
         result["duration"] = total
 
         _update(job_id, status="done", stage="thumbnail", progress=100.0,
@@ -876,6 +1109,25 @@ def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
         shutil.rmtree(work, ignore_errors=True)
 
 
+def _caption_margin_bump(pres: Optional[dict], caps, canvas) -> int:
+    """How far up this scene's captions move, in canvas pixels.
+
+    Only a bottom bubble collides with bottom captions; a full-screen presenter or
+    a split layout does not, and centred captions are already above a bubble."""
+    if not pres or (pres.get("layout") or "full").lower() != "pip":
+        return 0
+    if (caps.position or "center").lower() != "bottom":
+        return 0
+    pos = (pres.get("position") or "bottom-right").lower()
+    if not pos.startswith("bottom"):
+        return 0
+    _, bh, _, y = rnd.bubble_geometry(canvas.width, canvas.height,
+                                      pres.get("size", 0.38), pos,
+                                      pres.get("shape") or "circle")
+    # Clear the bubble's top edge plus a little air.
+    return max(0, int(canvas.height - y + canvas.height * 0.02))
+
+
 def _output_dir() -> str:
     d = media.files_dir()
     if not d:
@@ -953,6 +1205,124 @@ async def compose_cancel(job_id: str, _auth=Depends(_require_api_auth)):
     return {"id": job_id, "status": "cancelled"}
 
 
+# =============================================================================
+# Talking head — the presenter on its own
+# =============================================================================
+
+class TalkingHeadRequest(BaseModel):
+    """A character saying one thing, for a preview on a character page."""
+    character: Optional[str] = None
+    image: Optional[str] = None
+    text: Optional[str] = None
+    audio: Optional[str] = None               # pre-rendered instead of text
+    voice: VoiceSpec = Field(default_factory=VoiceSpec)
+    engine: str = "auto"
+    motion: str = "still"                     # a preview does not need a generated shot
+    base_src: Optional[str] = None
+    prompt: Optional[str] = None
+    video_model: Optional[str] = None
+    canvas: Canvas = Field(default_factory=lambda: Canvas(width=720, height=720, fps=25))
+    output: OutputSpec = Field(default_factory=OutputSpec)
+    is_async: bool = Field(default=True, alias="async")
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+
+#: A preview is a preview: a quarter-minute is plenty and keeps the GPU free.
+TALKING_HEAD_MAX_SECONDS = 15.0
+
+
+def _talking_head_spec(req: TalkingHeadRequest) -> ComposeRequest:
+    """A talking-head request IS a one-scene composition with a full-screen
+    presenter, no captions and no music — so it runs on exactly the same code."""
+    pres = {"character": req.character, "image": req.image, "engine": req.engine,
+            "motion": req.motion, "base_src": req.base_src, "prompt": req.prompt,
+            "video_model": req.video_model, "layout": "full"}
+    scene = {"text": req.text or "", "audio": req.audio,
+             "visuals": [], "padding": 0.05}
+    return ComposeRequest(canvas=req.canvas, voice=req.voice,
+                          presenter=PresenterSpec(**pres), scenes=[Scene(**scene)],
+                          captions=CaptionSpec(enabled=False), output=req.output,
+                          outputs=["video"], **{"async": req.is_async})
+
+
+@router.post("/v1/video/talking-head", summary="A character saying one line",
+             tags=["Video"])
+async def talking_head(req: TalkingHeadRequest, http_request: Request,
+                       _auth=Depends(_require_api_auth)):
+    """Lip-sync one line onto a character's face — the preview a character page
+    shows before anyone pays for a render.
+
+    Same job semantics as ``/v1/video/compose`` (and the same job store, so
+    ``GET /v1/video/compose/{id}`` works too): ``202 {id, status}`` then poll, or
+    ``"async": false`` to block. Text is synthesised with ``voice``; supply
+    ``audio`` instead to skip synthesis. Capped at 15 s of audio."""
+    if not (req.text or "").strip() and not req.audio:
+        raise HTTPException(status_code=400, detail="talking-head needs text or audio")
+    spec = _talking_head_spec(req)
+    _validate(spec)
+    if req.audio:
+        # A caller's clip is the one thing we can measure before rendering.
+        try:
+            import tempfile as _tf
+            with _tf.TemporaryDirectory() as td:
+                p = media.resolve(req.audio, td, "th-audio")
+                dur = media.duration_of(p)
+            if dur > TALKING_HEAD_MAX_SECONDS + 0.5:
+                raise HTTPException(status_code=400, detail=(
+                    f"audio is {dur:.1f}s; talking-head is capped at "
+                    f"{TALKING_HEAD_MAX_SECONDS:.0f}s (use /v1/video/compose for a full render)"))
+        except media.MediaError as exc:
+            raise HTTPException(status_code=400, detail=f"audio: {exc}")
+    try:
+        from codai.api.urlutils import get_base_url
+        base_url = get_base_url(http_request)
+    except Exception:
+        base_url = ""
+    rec = submit(spec, base_url)
+    if req.is_async:
+        from fastapi.responses import JSONResponse
+        return JSONResponse(status_code=202,
+                            content={"id": rec["id"], "status": "queued"})
+    job_id = rec["id"]
+    while True:
+        await asyncio.sleep(0.5)
+        cur = _get(job_id)
+        if cur is None:
+            raise HTTPException(status_code=500, detail="job disappeared")
+        if cur["status"] in ("done", "failed", "cancelled"):
+            if cur["status"] == "failed":
+                raise HTTPException(status_code=500,
+                                    detail=cur.get("error") or "talking-head failed")
+            return _public(cur)
+
+
+@router.get("/v1/video/talking-head/{job_id}", summary="Talking-head job status",
+            tags=["Video"])
+async def talking_head_status(job_id: str, _auth=Depends(_require_api_auth)):
+    """The same document as ``GET /v1/video/compose/{id}`` — one job store."""
+    return await compose_status(job_id, _auth)
+
+
+@router.post("/v1/video/talking-head/{job_id}/cancel",
+             summary="Cancel a talking-head job", tags=["Video"])
+async def talking_head_cancel(job_id: str, _auth=Depends(_require_api_auth)):
+    return await compose_cancel(job_id, _auth)
+
+
+@router.get("/v1/video/presenter/engines", summary="Installed lip-sync engines",
+            tags=["Video"])
+async def presenter_engines(_auth=Depends(_require_api_auth)):
+    """What a presenter can be driven with here — so a client can grey out the
+    feature instead of discovering a 501 at submission."""
+    from codai.compose import presenter as _pres
+    have = _pres.available_engines()
+    return {"engines": have, "available": bool(have),
+            "video_engines": [e for e in have if e in _pres.VIDEO_ENGINES],
+            "portrait_engines": [e for e in have if e in _pres.PORTRAIT_ENGINES],
+            "layouts": ["full", "pip", "split"],
+            "talking_head_max_seconds": TALKING_HEAD_MAX_SECONDS}
+
+
 @router.get("/v1/video/compose", summary="List composition jobs", tags=["Video"])
 async def compose_list(_auth=Depends(_require_api_auth)):
     """Every job this front still remembers, newest first — for an admin page or

codai/api/remote_gateway.py

@@ -136,7 +136,8 @@ _LOCAL_PREFIXES = ("/v1/files/upload", "/v1/files/blob")
 #: endpoints itself, and the visuals it was given may be files only this install
 #: holds. Forwarded to a video pod it would find neither the models nor the files.
 _ORCHESTRATION_PREFIXES = ("/v1/pipelines", "/v1/characters", "/v1/environments",
-                           "/v1/video/compose")
+                           "/v1/video/compose", "/v1/video/talking-head",
+                           "/v1/video/presenter")
 
 #: `model` field in a multipart body, without paying for a full form parse.
 _MULTIPART_MODEL = re.compile(

codai/broker/capabilities.py

@@ -46,6 +46,10 @@ DEFAULT_STUDIO_ENDPOINTS = [
     "v1/video/compose/{id}/cancel",
     "v1/files/upload",
     "v1/files/blob/{hash}",
+    # The talking presenter (codai/compose/presenter.py).
+    "v1/video/talking-head",
+    "v1/video/talking-head/{id}",
+    "v1/video/presenter/engines",
 ]
 
 

codai/compose/captions.py

@@ -194,13 +194,19 @@ def break_lines(words: List[dict], max_words_per_line: int = 4,
         if not chunk:
             continue
         rows = [chunk[k:k + per_line] for k in range(0, len(chunk), per_line)]
-        out.append({
+        ev = {
             "start": chunk[0]["start"],
             "end": chunk[-1]["end"],
             "words": chunk,
             "rows": [[w["word"] for w in row] for row in rows],
             "text": " ".join(w["word"] for w in chunk),
-        })
+        }
+        # Words carry the margin their scene needs (a bottom presenter bubble);
+        # an event takes the largest of the ones it groups.
+        mv = max((int(w.get("margin_v") or 0) for w in chunk), default=0)
+        if mv:
+            ev["margin_v"] = mv
+        out.append(ev)
     return out
 
 
@@ -273,11 +279,14 @@ def build_ass(events: List[dict], *, width: int, height: int, font: str = "Monts
     body = []
     for ev in events:
         rows = ev.get("rows") or [[w["word"] for w in ev.get("words", [])]]
+        # A scene with a presenter bubble at the bottom pushes its captions up:
+        # ASS takes a per-event MarginV, so only those events move.
+        mv = int(ev.get("margin_v") or 0)
         if not p["karaoke"]:
             text = "\\N".join(" ".join(_ass_escape(w.upper() if upper else w) for w in row)
                               for row in rows)
             body.append(f"Dialogue: 0,{_ass_time(ev['start'])},{_ass_time(ev['end'])},"
-                        f"cap,,0,0,0,,{text}")
+                        f"cap,,0,0,{mv},,{text}")
             continue
         # Karaoke: one event per word, the whole line drawn each time with the
         # active word in the highlight colour. Simpler than \k timings and it
@@ -299,7 +308,7 @@ def build_ass(events: List[dict], *, width: int, height: int, font: str = "Monts
             if end <= start:
                 end = start + 0.04
             body.append(f"Dialogue: 0,{_ass_time(start)},{_ass_time(end)},"
-                        f"cap,,0,0,0,,{{\\c{primary}}}" + "\\N".join(parts))
+                        f"cap,,0,0,{mv},,{{\\c{primary}}}" + "\\N".join(parts))
     return "\n".join(head + body) + "\n"
 
 

codai/compose/presenter.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.
+
+"""The talking presenter: a character whose mouth moves with the narration.
+
+The most common reel format is a person talking to camera — full screen, or as a
+bubble over B-roll. Only this server can build it: composition synthesises the
+narration, so it alone holds the exact audio of each scene, and the lip-sync
+engines and identity-locked video generation are here too.
+
+Per scene the work is: get a face (a saved character profile's reference, or a
+portrait the caller supplied), get something for it to move in (a still, a clip
+the caller gave, or a short identity-locked clip generated for this scene), then
+drive an engine with **that scene's narration** and hold the result to the
+scene's exact length. The layouts (`full`, `pip`, `split`) are compositing and
+live in :mod:`codai.compose.render`.
+
+Engines are whatever the install has. ``wav2lip`` and ``sadtalker`` are known by
+name; anything else on PATH under its own name can be selected by that name, so a
+newer engine (LatentSync, MuseTalk, Hallo…) needs no API change. ``auto`` picks a
+video-driven engine when there is a clip to drive and a portrait engine
+otherwise.
+
+Nothing here ever fails a render: a missing engine, a face the engine cannot
+find, a crash — each falls back to the scene's ordinary visuals and leaves a line
+in the job's warnings. That is the same rule a broken stock clip follows.
+"""
+
+import glob
+import os
+import shutil
+import subprocess
+import threading
+from typing import List, Optional
+
+from codai.compose import media
+from codai.compose import render as rnd
+
+#: Engines that drive a VIDEO (they need a base clip; the mouth is replaced).
+VIDEO_ENGINES = ("wav2lip",)
+#: Engines that animate a PORTRAIT (they make their own motion from a still).
+PORTRAIT_ENGINES = ("sadtalker",)
+#: Preference order for ``engine: "auto"`` — best mouth first for a clip, best
+#: motion first for a still.
+_AUTO_WITH_BASE = ("wav2lip", "sadtalker")
+_AUTO_STILL = ("sadtalker", "wav2lip")
+
+
+class PresenterError(Exception):
+    """This scene gets no presenter; the caller falls back and warns."""
+
+
+# ------------------------------------------------------------------ engines
+def available_engines() -> List[str]:
+    """Which lip-sync engines this install can actually run, in a stable order.
+
+    ``wav2lip`` counts when the packaged shim is on PATH *or* the managed install
+    already has its code and weights — not when they would merely be
+    downloadable, because a compose job is the wrong place to fetch 500 MB."""
+    out = []
+    if shutil.which("wav2lip") or shutil.which("Wav2Lip"):
+        out.append("wav2lip")
+    else:
+        try:
+            from codai.api import lipsync
+            if lipsync.is_available():
+                out.append("wav2lip")
+        except Exception:
+            pass
+    if shutil.which("sadtalker"):
+        out.append("sadtalker")
+    return out
+
+
+def engine_installed(name: str) -> bool:
+    n = (name or "").strip().lower()
+    if not n or n == "auto":
+        return bool(available_engines())
+    if n in available_engines():
+        return True
+    # An engine this module has never heard of, installed under its own name.
+    return bool(shutil.which(n))
+
+
+def pick_engine(requested: str, has_base: bool) -> str:
+    """The engine to use, or raise when there is none."""
+    req = (requested or "auto").strip().lower()
+    have = available_engines()
+    if req and req != "auto":
+        if req in have or shutil.which(req):
+            return req
+        raise PresenterError(
+            f"lip-sync engine {req!r} is not installed"
+            + (f" (installed: {', '.join(have)})" if have else " and neither is any other"))
+    for cand in (_AUTO_WITH_BASE if has_base else _AUTO_STILL):
+        if cand in have:
+            return cand
+    if have:
+        return have[0]
+    raise PresenterError("no lip-sync engine is installed (install the wav2lip or "
+                         "sadtalker shim, or the managed wav2lip assets)")
+
+
+def run_engine(engine: str, face_path: str, audio_path: str, out_path: str,
+               cancel: Optional[threading.Event] = None, timeout: int = 3600,
+               log=None) -> str:
+    """Drive one engine: ``face`` (a clip or a portrait) + audio → a talking clip.
+
+    Kept as one function so a test can replace it, and so a new engine is a
+    couple of lines rather than a new code path."""
+    if cancel is not None and cancel.is_set():
+        raise rnd.Cancelled()
+    eng = (engine or "").lower()
+    workdir = os.path.dirname(out_path) or "."
+
+    if eng == "wav2lip":
+        binary = shutil.which("wav2lip") or shutil.which("Wav2Lip")
+        if binary:
+            cmd = [binary, "--face", face_path, "--audio", audio_path,
+                   "--outfile", out_path]
+            _run(cmd, cancel=cancel, timeout=timeout, log=log)
+            if os.path.isfile(out_path):
+                return out_path
+            raise PresenterError("wav2lip produced no output")
+        from codai.api.lipsync import run_wav2lip
+        return run_wav2lip(face_path, audio_path, out_path, timeout=timeout)
+
+    if eng == "sadtalker":
+        binary = shutil.which("sadtalker")
+        if not binary:
+            raise PresenterError("sadtalker is not on PATH")
+        res_dir = os.path.join(workdir, "sadtalker-out")
+        os.makedirs(res_dir, exist_ok=True)
+        still = media.kind_of(media.sniff_mime(open(face_path, "rb").read(32))) == "image"
+        cmd = [binary, "--driven_audio", audio_path,
+               "--source_image" if still else "--source_video", face_path,
+               "--result_dir", res_dir]
+        _run(cmd, cancel=cancel, timeout=timeout, log=log)
+        made = sorted(glob.glob(os.path.join(res_dir, "**", "*.mp4"), recursive=True),
+                      key=os.path.getmtime)
+        if not made:
+            raise PresenterError("sadtalker produced no video")
+        shutil.move(made[-1], out_path)
+        return out_path
+
+    # An engine installed under its own name: the wav2lip-style CLI is the de
+    # facto shape (face + audio + outfile). If it wants something else, the
+    # operator can wrap it in a shim of that name.
+    binary = shutil.which(eng)
+    if not binary:
+        raise PresenterError(f"lip-sync engine {engine!r} is not installed")
+    _run([binary, "--face", face_path, "--audio", audio_path, "--outfile", out_path],
+         cancel=cancel, timeout=timeout, log=log)
+    if not os.path.isfile(out_path):
+        raise PresenterError(f"{engine}: produced no output")
+    return out_path
+
+
+def _run(cmd: List[str], cancel=None, timeout: int = 3600, log=None) -> None:
+    if log:
+        log("presenter: " + " ".join(cmd[:8]) + (" …" if len(cmd) > 8 else ""))
+    proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    waited = 0.0
+    while True:
+        try:
+            _, err = proc.communicate(timeout=0.5)
+            break
+        except subprocess.TimeoutExpired:
+            waited += 0.5
+            if cancel is not None and cancel.is_set():
+                proc.kill()
+                proc.communicate()
+                raise rnd.Cancelled()
+            if waited > timeout:
+                proc.kill()
+                proc.communicate()
+                raise PresenterError(f"{os.path.basename(cmd[0])} timed out")
+    if proc.returncode != 0:
+        tail = (err or b"").decode(errors="replace").strip().splitlines()[-6:]
+        raise PresenterError(f"{os.path.basename(cmd[0])} failed "
+                             f"(rc={proc.returncode}): {' | '.join(tail)}")
+
+
+# -------------------------------------------------------------------- face
+def portrait_path(spec: dict, workdir: str, name_hint: str = "presenter") -> str:
+    """The face to use: an explicit portrait, or a character profile's best
+    reference (a front view when one is labelled, else the first)."""
+    ref = (spec.get("image") or "").strip() if isinstance(spec.get("image"), str) else ""
+    if ref:
+        try:
+            return media.resolve(ref, workdir, name_hint)
+        except media.MediaError as exc:
+            raise PresenterError(f"presenter image: {exc}")
+    char = (spec.get("character") or "").strip()
+    if not char:
+        raise PresenterError("presenter needs a character or an image")
+    try:
+        from codai.api.characters import _load_character_images, _load_character_meta
+    except Exception as exc:
+        raise PresenterError(f"character profiles unavailable: {exc}")
+    if not _load_character_meta(char):
+        raise PresenterError(f"no character profile named {char!r}")
+    images = _load_character_images(char)
+    if not images:
+        raise PresenterError(f"character {char!r} has no reference images")
+    best = next((i for i in images if "front" in (i.label or "").lower()), images[0])
+    try:
+        data = media.fetch_bytes(best.data)
+    except media.MediaError as exc:
+        raise PresenterError(f"character {char!r}: {exc}")
+    os.makedirs(workdir, exist_ok=True)
+    out = os.path.join(workdir, f"{name_hint}-face.png")
+    with open(out, "wb") as f:
+        f.write(data)
+    return out
+
+
+# ------------------------------------------------------------------- clips
+def base_clip(spec: dict, portrait: str, out_path: str, *, duration: float, width: int,
+              height: int, fps: int, workdir: str, loop=None, cancel=None,
+              log=None) -> Optional[str]:
+    """What the engine drives, or None when the engine animates a portrait.
+
+    ``motion``: ``still`` holds the portrait, ``src`` uses the caller's clip,
+    ``generate`` asks the video model for a short identity-locked "talking to
+    camera" shot of this character."""
+    motion = (spec.get("motion") or "generate").strip().lower()
+    if motion == "src":
+        ref = (spec.get("base_src") or "").strip()
+        if not ref:
+            raise PresenterError('motion="src" needs base_src')
+        src = media.resolve(ref, workdir, "presenter-base")
+        rnd.build_visual({"type": "video", "_path": src, "_probe": media.probe(src)},
+                         out_path, width=width, height=height, fps=fps,
+                         duration=duration, fit="cover", workdir=workdir,
+                         cancel=cancel, log=log)
+        return out_path
+    if motion == "generate":
+        gen = _generate_base(spec, portrait, out_path, duration=duration, width=width,
+                             height=height, fps=fps, workdir=workdir, loop=loop,
+                             cancel=cancel, log=log)
+        if gen:
+            return gen
+        # Falling back to a still is better than losing the presenter; the caller
+        # turns the reason into a warning.
+        if log:
+            log("presenter: video generation unavailable; using a still")
+    rnd.still_video(portrait, out_path, duration=duration, width=width, height=height,
+                    fps=fps, cancel=cancel, log=log)
+    return out_path
+
+
+def _generate_base(spec: dict, portrait: str, out_path: str, *, duration: float,
+                   width: int, height: int, fps: int, workdir: str, loop=None,
+                   cancel=None, log=None) -> Optional[str]:
+    """One short identity-locked clip of the character speaking to camera, made
+    with the ordinary video endpoint (so its model config, acceleration, LoRAs and
+    keyframe identity bridge all apply). None when it cannot be done."""
+    from codai.api.compose import _await          # the module's own loop
+    prompt = (spec.get("prompt")
+              or "talking to the camera, natural head movement, soft studio light")
+    model = (spec.get("video_model") or spec.get("model") or "").strip()
+    char = (spec.get("character") or "").strip()
+    try:
+        from codai.api.video import VideoGenerationRequest, generate_video
+    except Exception as exc:
+        if log:
+            log(f"presenter: video endpoint unavailable ({exc})")
+        return None
+    body = {"prompt": prompt, "mode": "i2v", "image": _data_uri(portrait),
+            "width": width, "height": height, "num_frames": max(9, int(duration * fps)),
+            "fps": fps, "response_format": "b64_json"}
+    if model:
+        body["model"] = model
+    if char:
+        body["character_profiles"] = [char]
+    try:
+        req = VideoGenerationRequest(**body)
+    except Exception as exc:
+        if log:
+            log(f"presenter: video request rejected ({exc})")
+        return None
+    try:
+        res = _await(loop, generate_video(req, None), timeout=3600)
+    except Exception as exc:
+        if log:
+            log(f"presenter: video generation failed ({exc})")
+        return None
+    item = ((res or {}).get("data") or [{}])[0] if isinstance(res, dict) else {}
+    blob = item.get("b64_json") or item.get("b64_mp4")
+    raw = None
+    if blob:
+        import base64
+        raw = base64.b64decode(blob)
+    elif item.get("url"):
+        try:
+            raw = media.fetch_bytes(item["url"])
+        except media.MediaError:
+            raw = None
+    if not raw:
+        return None
+    gen = os.path.join(workdir, "presenter-gen.mp4")
+    with open(gen, "wb") as f:
+        f.write(raw)
+    # Normalise to the canvas and hold to the clip's length: a generated shot is
+    # usually a couple of seconds, the scene may be longer.
+    rnd.build_visual({"type": "video", "_path": gen, "_probe": media.probe(gen)},
+                     out_path, width=width, height=height, fps=fps, duration=duration,
+                     fit="cover", workdir=workdir, cancel=cancel, log=log)
+    return out_path
+
+
+def _data_uri(path: str) -> str:
+    import base64
+    with open(path, "rb") as f:
+        raw = f.read()
+    return f"data:{media.sniff_mime(raw)};base64," + base64.b64encode(raw).decode()
+
+
+# --------------------------------------------------------------- the scene
+def build_scene_presenter(spec: dict, *, narration: Optional[str], duration: float,
+                          width: int, height: int, fps: int, workdir: str,
+                          index: int = 0, loop=None, cancel=None, log=None,
+                          gpu_lease=None) -> dict:
+    """The presenter clip for one scene.
+
+    Returns ``{"path": …, "engine": …}``; raises :class:`PresenterError` when this
+    scene cannot have one (the caller warns and falls back). ``narration`` is the
+    scene's audio — without it there is nothing to sync to, and a presenter that
+    just stands there is what the scene's visuals already are."""
+    if not narration or not os.path.isfile(narration):
+        raise PresenterError("the scene has no narration to lip-sync to")
+    face = portrait_path(spec, workdir, f"pres{index}")
+    motion = (spec.get("motion") or "generate").strip().lower()
+    if motion not in ("generate", "still", "src"):
+        raise PresenterError(f'motion must be generate, still or src (got {motion!r})')
+
+    # The engine drives the mouth for exactly as long as the narration sounds;
+    # the scene's padding is held afterwards.
+    speech = media.duration_of(narration)
+    engine = pick_engine(spec.get("engine") or "auto",
+                         has_base=(motion in ("generate", "src")))
+    # Portrait engines make their own motion — a base clip would throw it away.
+    base = None
+    if engine in VIDEO_ENGINES or motion == "src":
+        base = base_clip(spec, face, os.path.join(workdir, f"pres{index}-base.mp4"),
+                         duration=speech + 0.35, width=width, height=height, fps=fps,
+                         workdir=workdir, loop=loop, cancel=cancel, log=log)
+    raw = os.path.join(workdir, f"pres{index}-sync.mp4")
+    lease = gpu_lease() if callable(gpu_lease) else None
+    try:
+        run_engine(engine, base or face, narration, raw, cancel=cancel, log=log)
+    finally:
+        if lease is not None:
+            try:
+                lease.close()
+            except Exception:
+                pass
+    if not os.path.isfile(raw):
+        raise PresenterError(f"{engine}: produced no clip")
+    # Exactly the scene's length, at the canvas fps, whatever the engine returned.
+    out = os.path.join(workdir, f"pres{index}.mp4")
+    rnd.hold_to(raw, out, duration=duration, fps=fps, width=width, height=height,
+                cancel=cancel, log=log)
+    return {"path": out, "engine": engine}

codai/compose/render.py

@@ -248,6 +248,148 @@ def build_visual(visual: dict, out_path: str, *, width: int, height: int, fps: i
     run(pre + ["-i", src, "-vf", vf] + common_out, cancel=cancel, log=log)
 
 
+def still_video(image: str, out_path: str, *, duration: float, width: int, height: int,
+                fps: int, cancel=None, log=None) -> str:
+    """A motionless clip of a portrait — what a video-driven lip-sync engine needs
+    when there is no footage to drive (and what a generated shot falls back to).
+    No Ken Burns: a presenter's head must not drift while it talks."""
+    run(["-loop", "1", "-i", image, "-t", f"{max(0.08, duration):.3f}",
+         "-vf", fit_filter(width, height, "cover") + f",fps={fps},format=yuv420p",
+         "-r", str(fps), "-an"] + _INTER_ARGS + [out_path], cancel=cancel, log=log)
+    return out_path
+
+
+def hold_to(src: str, out_path: str, *, duration: float, fps: int,
+            width: Optional[int] = None, height: Optional[int] = None,
+            fit: str = "cover", cancel=None, log=None) -> str:
+    """Make a clip exactly ``duration`` long, and the canvas's size when asked.
+
+    A lip-sync engine returns the length of the audio it was given and whatever
+    frame size it likes (SadTalker squares its output), while the scene is the
+    narration plus its padding at the canvas size — so: hold the final frame (what
+    a presenter does when they stop talking), trim if longer, and scale/crop to
+    the canvas."""
+    chain = [f"fps={fps}"]
+    if width and height:
+        chain.append(fit_filter(int(width), int(height), fit))
+    chain += [f"tpad=stop_mode=clone:stop_duration={max(0.05, duration):.3f}",
+              f"trim=0:{max(0.08, duration):.3f}", "setpts=PTS-STARTPTS",
+              "format=yuv420p"]
+    run(["-i", src, "-vf", ",".join(chain), "-an", "-r", str(fps)]
+        + _INTER_ARGS + [out_path], cancel=cancel, log=log)
+    return out_path
+
+
+# ------------------------------------------------------------ presenter layout
+#: The presenter bubble keeps the captions' safe area: clear of the platform UI.
+SAFE_X = 0.06
+SAFE_Y = 0.10
+
+
+def bubble_mask(path: str, w: int, h: int, shape: str = "circle",
+                border_width: int = 0, border_colour: str = "#FFFFFF") -> tuple:
+    """(mask, ring) PNGs for a picture-in-picture bubble.
+
+    The mask is greyscale — white where the presenter shows — and becomes the
+    clip's alpha; the ring is an RGBA outline drawn on top afterwards, so the
+    border is not eaten by the same mask that rounds the corners."""
+    from PIL import Image, ImageDraw
+    w, h = max(2, int(w)), max(2, int(h))
+    shape = (shape or "circle").lower()
+    radius = int(min(w, h) * 0.18)
+    mask = Image.new("L", (w, h), 0)
+    md = ImageDraw.Draw(mask)
+    if shape == "circle":
+        md.ellipse((0, 0, w - 1, h - 1), fill=255)
+    elif shape in ("rounded", "round"):
+        md.rounded_rectangle((0, 0, w - 1, h - 1), radius=radius, fill=255)
+    else:
+        md.rectangle((0, 0, w - 1, h - 1), fill=255)
+    mask_path = path + ".mask.png"
+    mask.save(mask_path)
+
+    ring_path = None
+    if border_width and border_width > 0:
+        col = (border_colour or "#FFFFFF").lstrip("#")
+        if len(col) == 3:
+            col = "".join(c * 2 for c in col)
+        rgb = tuple(int(col[i:i + 2], 16) for i in (0, 2, 4)) if len(col) == 6 else (255, 255, 255)
+        ring = Image.new("RGBA", (w, h), (0, 0, 0, 0))
+        rd = ImageDraw.Draw(ring)
+        bw = max(1, int(border_width))
+        box = (bw // 2, bw // 2, w - 1 - bw // 2, h - 1 - bw // 2)
+        if shape == "circle":
+            rd.ellipse(box, outline=rgb + (255,), width=bw)
+        elif shape in ("rounded", "round"):
+            rd.rounded_rectangle(box, radius=radius, outline=rgb + (255,), width=bw)
+        else:
+            rd.rectangle(box, outline=rgb + (255,), width=bw)
+        ring_path = path + ".ring.png"
+        ring.save(ring_path)
+    return mask_path, ring_path
+
+
+def bubble_geometry(width: int, height: int, size: float, position: str,
+                    shape: str = "circle") -> tuple:
+    """(w, h, x, y) of the bubble inside the safe area."""
+    size = max(0.1, min(0.95, float(size or 0.38)))
+    bw = int(round(width * size / 2) * 2)
+    bh = bw if (shape or "circle").lower() in ("circle", "rounded", "round")         else int(round(bw * height / max(1, width) / 2) * 2)
+    bh = min(bh, int(height * 0.8))
+    mx, my = int(width * SAFE_X), int(height * SAFE_Y)
+    pos = (position or "bottom-right").lower().replace("_", "-")
+    vert, _, horiz = pos.partition("-") if "-" in pos else (pos, "", "center")
+    x = {"left": mx, "center": (width - bw) // 2, "centre": (width - bw) // 2,
+         "right": width - bw - mx}.get(horiz, width - bw - mx)
+    y = {"top": my, "center": (height - bh) // 2, "centre": (height - bh) // 2,
+         "middle": (height - bh) // 2,
+         "bottom": height - bh - my}.get(vert, height - bh - my)
+    return bw, bh, max(0, x), max(0, y)
+
+
+def composite_pip(base: str, presenter: str, out_path: str, *, width: int, height: int,
+                  fps: int, size: float = 0.38, position: str = "bottom-right",
+                  shape: str = "circle", border: Optional[dict] = None,
+                  workdir: str = "", cancel=None, log=None) -> str:
+    """The presenter as a bubble over the scene's visuals."""
+    bw, bh, x, y = bubble_geometry(width, height, size, position, shape)
+    border = border or {}
+    mask, ring = bubble_mask(os.path.join(workdir or ".", f"bubble-{bw}x{bh}"), bw, bh,
+                             shape, int(border.get("width") or 0),
+                             border.get("color") or border.get("colour") or "#FFFFFF")
+    inputs = ["-i", base, "-i", presenter, "-i", mask]
+    graph = [f"[1:v]scale={bw}:{bh}:force_original_aspect_ratio=increase,"
+             f"crop={bw}:{bh},format=rgba[pv]",
+             "[2:v]format=gray[pm]",
+             "[pv][pm]alphamerge[pa]"]
+    top = "pa"
+    if ring:
+        inputs += ["-i", ring]
+        graph.append("[pa][3:v]overlay=0:0[pb]")
+        top = "pb"
+    graph.append(f"[0:v][{top}]overlay={x}:{y}:format=auto,format=yuv420p,fps={fps}[vout]")
+    run(inputs + ["-filter_complex", ";".join(graph), "-map", "[vout]", "-an"]
+        + _INTER_ARGS + [out_path], cancel=cancel, log=log)
+    return out_path
+
+
+def composite_split(base: str, presenter: str, out_path: str, *, width: int, height: int,
+                    fps: int, presenter_half: str = "bottom", cancel=None,
+                    log=None) -> str:
+    """Presenter and B-roll stacked, each filling its half."""
+    half = int(height / 2 / 2) * 2
+    fill = (f"scale={width}:{half}:force_original_aspect_ratio=increase,"
+            f"crop={width}:{half},setsar=1")
+    top_in, bottom_in = ((1, 0) if (presenter_half or "bottom").lower() == "top"
+                         else (0, 1))
+    graph = [f"[{top_in}:v]{fill}[t]", f"[{bottom_in}:v]{fill}[b]",
+             f"[t][b]vstack=inputs=2,pad={width}:{height}:0:0,"
+             f"format=yuv420p,fps={fps}[vout]"]
+    run(["-i", base, "-i", presenter, "-filter_complex", ";".join(graph),
+         "-map", "[vout]", "-an"] + _INTER_ARGS + [out_path], cancel=cancel, log=log)
+    return out_path
+
+
 # --------------------------------------------------------------------- join
 def concat_clips(clips: List[str], out_path: str, workdir: str,
                  cancel: Optional[threading.Event] = None,

docs/video-compose.md

@@ -182,6 +182,78 @@ The body may also be JSON (`{"file": "<base64 or data: URI>"}`) or the raw bytes
 with a `Content-Type`. Video, image and audio only; the type is decided by the
 content, not by the client's claim.
 
+## The talking presenter
+
+A person talking to camera is the format most short videos use, and it is
+something only this server can assemble: composition synthesises the narration,
+so it holds the exact audio of every scene, and the lip-sync engines and
+identity-locked video generation are here too.
+
+```jsonc
+{
+  "presenter": {
+    "character": "nova",            // a saved character profile (its front reference is used)
+    "image": null,                  // or an explicit portrait (URL / /v1/files / data URI / base64)
+    "engine": "auto",               // auto | wav2lip | sadtalker | <any installed engine name>
+    "motion": "generate",           // generate: a short identity-locked clip per scene
+                                    // still: animate the portrait   src: use base_src
+    "base_src": null,
+    "prompt": "talking to the camera, natural head movement, soft studio light",
+    "layout": "full",               // full | pip | split
+    "position": "bottom-right",     // pip: top/bottom + left/right/center
+    "size": 0.38,                   // pip: bubble width as a fraction of the canvas
+    "shape": "circle",              // pip: circle | rounded | rect
+    "split": "bottom",              // split: which half the presenter takes
+    "border": {"color": "#FFFFFF", "width": 6},
+    "remove_background": false      // ignored (with a warning) when no matting is installed
+  },
+  "scenes": [
+    {"text": "Hi! Today: three hearts.", "visuals": [...]},                      // the default presenter
+    {"text": "Here's how they work.",    "visuals": [...], "presenter": false},  // B-roll only
+    {"text": "Follow for more!",         "visuals": [], "presenter": {"layout": "full"}}
+  ]
+}
+```
+
+* **Timing** is the scene's: the engine is driven by *that scene's* narration —
+  synthesised or supplied — and the clip is held to narration + padding (or
+  `min_duration`). Captions, music and ducking are unchanged.
+* **`engine: "auto"`** takes a video-driven engine (wav2lip) when there is a clip
+  to drive and a portrait engine (sadtalker) otherwise. Any other engine installed
+  under its own name can be named instead — no API change for a new one. Ask
+  `GET /v1/video/presenter/engines` what this install has.
+* **`layout`**: `full` replaces the scene's visuals (a scene may then have none);
+  `pip` composites a bubble over them inside the caption safe area — and captions
+  at the bottom move above it automatically; `split` stacks presenter and B-roll,
+  each filling its half. With a presenter, a scene's visuals are joined without a
+  transition *between them* and composited as one clip; without one, nothing
+  changes.
+* **Nothing fails the render**: no engine, no face, a crash, a failed composite —
+  each falls back to the scene's visuals (or a gradient when there are none) and
+  adds a warning naming the scene. `stage` is `presenter`, with
+  `message: "scene 2/6: lip-sync"`, and `result.scenes[i].presenter` is
+  `{"engine": …, "layout": …}` or `null`.
+* **Errors at submission**: `400` for an unknown character, an invalid layout,
+  position, size, shape, motion or a `pip` scene with no visuals; `501` when a
+  presenter is asked for and no lip-sync engine is installed, naming what to
+  install.
+* **GPU**: lip-sync is the only GPU step in a composition (the rest is ffmpeg), so
+  it takes a lease from the same scheduler ordinary model requests use instead of
+  fighting a generation for the card.
+
+### `POST /v1/video/talking-head`
+
+The presenter on its own, for a preview on a character page: a character (or a
+portrait), `text` synthesised with `voice` — or `audio` you already have — and the
+same job semantics as compose (same job store, so either status URL works). Capped
+at 15 s of audio. `result.video` as usual.
+
+```bash
+curl -H "Authorization: Bearer $TOKEN" -X POST …/v1/video/talking-head \
+  -d '{"character":"nova","text":"Hi, I am Nova.","canvas":{"width":720,"height":720}}'
+# {"id":"cmp_…","status":"queued"}
+```
+
 ## The same references elsewhere
 
 The extractors that build characters and voice profiles from a user's own media
@@ -231,6 +303,7 @@ claimed.
 |---|---|
 | Endpoints, validation, job queue, orchestration | `codai/api/compose.py` |
 | Uploads (`/v1/files/upload`, `/v1/files/blob/{hash}`) | `codai/api/uploads.py` |
+| The talking presenter (engines, face, base clip) | `codai/compose/presenter.py` |
 | Reference resolving, probing, fonts | `codai/compose/media.py` |
 | Word timings, line breaking, ASS/SRT/VTT | `codai/compose/captions.py` |
 | The ffmpeg pipeline | `codai/compose/render.py` |

tests/test_compose.py

@@ -636,7 +636,11 @@ def test_compose_and_uploads_are_never_forwarded_to_a_remote():
     assert capability_for("/v1/video/compose") == "video"     # by prefix…
     for path in ("/v1/video/compose", "/v1/video/compose/cmp_1",
                  "/v1/video/compose/cmp_1/cancel", "/v1/files/upload",
-                 "/v1/files/blob/" + "a" * 64):
+                 "/v1/files/blob/" + "a" * 64,
+                 # A video pod has the lip-sync shims but not this install's
+                 # character profiles, so the presenter stays here too.
+                 "/v1/video/talking-head", "/v1/video/talking-head/cmp_1",
+                 "/v1/video/presenter/engines"):
         assert resolve_target(path, "POST", "", b"", "application/json") is None, path
 
 
@@ -725,3 +729,307 @@ def test_a_character_is_saved_from_an_uploaded_image(tmp_path, files_dir, monkey
     assert r.status_code == 200, r.text
     saved = tmp_path / "chars" / "ada" / "ref00.png"
     assert saved.is_file() and saved.read_bytes() == png
+
+
+# ============================================== request 4: talking presenter
+def _portrait(tmp_path, name="face.png", w=120, h=160):
+    """A portrait with something face-shaped in it (engines are stubbed, but the
+    compositing is real, so the pixels should differ from the background)."""
+    from PIL import Image, ImageDraw
+    img = Image.new("RGB", (w, h), (30, 30, 40))
+    d = ImageDraw.Draw(img)
+    d.ellipse((w * 0.2, h * 0.1, w * 0.8, h * 0.7), fill=(230, 200, 180))
+    d.rectangle((w * 0.3, h * 0.7, w * 0.7, h), fill=(40, 90, 160))
+    p = tmp_path / name
+    img.save(str(p))
+    return str(p)
+
+
+def _stub_engine(monkeypatch, mod, engines=("wav2lip",), fail=False):
+    """Stand in for the GPU: a 'lip-synced' clip as long as the audio it is given."""
+    calls = []
+    monkeypatch.setattr(mod, "available_engines", lambda: list(engines))
+    monkeypatch.setattr(mod, "engine_installed", lambda name: bool(engines))
+
+    def fake_run(engine, face, audio, out, cancel=None, timeout=3600, log=None):
+        if fail:
+            calls.append({"engine": engine, "face": face, "audio_seconds": None})
+            raise mod.PresenterError("stub engine says no")
+        dur = media.duration_of(audio)
+        # The job's workdir is gone by the time a test looks, so record the fact
+        # that matters here: how long the audio it was driven with was.
+        calls.append({"engine": engine, "face": face, "audio_seconds": dur})
+        subprocess.run([media.ffmpeg_bin(), "-y", "-hide_banner", "-loglevel", "error",
+                        "-f", "lavfi", "-i", f"color=c=#8844aa:s=240x320:r=25:d={dur:.3f}",
+                        "-c:v", "libx264", "-pix_fmt", "yuv420p", out], check=True)
+        return out
+
+    monkeypatch.setattr(mod, "run_engine", fake_run)
+    return calls
+
+
+def test_presenter_engines_are_reported_for_the_client(files_dir):
+    client, _ = _client(files_dir)
+    r = client.get("/v1/video/presenter/engines")
+    assert r.status_code == 200
+    d = r.json()
+    assert d["layouts"] == ["full", "pip", "split"]
+    assert isinstance(d["engines"], list) and d["available"] == bool(d["engines"])
+    assert d["talking_head_max_seconds"] == 15.0
+
+
+def test_a_full_screen_presenter_replaces_the_scenes_visuals(tmp_path, files_dir,
+                                                             monkeypatch):
+    from codai.compose import presenter as pres
+    client, compose_mod = _client(files_dir)
+    calls = _stub_engine(monkeypatch, pres)
+    voice = _uri(_tone(tmp_path / "pv.wav", 1.0))
+    spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
+            "presenter": {"image": _uri_file(_portrait(tmp_path)), "motion": "still",
+                          "layout": "full"},
+            "captions": {"enabled": False}, "outputs": ["video"],
+            "scenes": [{"text": "hi there", "audio": voice, "padding": 0.2,
+                        "visuals": []}]}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc
+    assert doc["warnings"] == []
+    assert calls and calls[0]["engine"] == "wav2lip"
+    # It was driven by THIS scene's narration, and the clip is the scene's length.
+    assert calls[0]["audio_seconds"] == pytest.approx(1.0, abs=0.1)
+    assert doc["result"]["scenes"][0]["presenter"] == {"engine": "wav2lip", "layout": "full"}
+    vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
+    info = media.probe(str(vid))
+    assert (info["width"], info["height"]) == (160, 284)
+    assert info["duration"] == pytest.approx(1.2, abs=0.2)
+
+
+def _uri_file(path):
+    return _uri(Path(path))
+
+
+def test_a_pip_bubble_sits_in_the_safe_area_over_the_b_roll(tmp_path, files_dir,
+                                                            monkeypatch):
+    from codai.compose import presenter as pres
+    client, _ = _client(files_dir)
+    _stub_engine(monkeypatch, pres)
+    spec = {"canvas": {"width": 240, "height": 426, "fps": 24},
+            "presenter": {"image": _uri_file(_portrait(tmp_path)), "motion": "still",
+                          "layout": "pip", "position": "bottom-right", "size": 0.4,
+                          "shape": "circle", "border": {"color": "#FFFFFF", "width": 4}},
+            "captions": {"enabled": False}, "outputs": ["video"],
+            "scenes": [{"text": "over b-roll", "audio": _uri(_tone(tmp_path / "pb.wav", 0.8)),
+                        "visuals": [{"type": "color", "color": "#000000"}]}]}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc
+    assert doc["result"]["scenes"][0]["presenter"]["layout"] == "pip"
+    vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
+    # The bubble is where the geometry says, and the corners of the canvas are not.
+    frame = tmp_path / "f.png"
+    subprocess.run([media.ffmpeg_bin(), "-y", "-hide_banner", "-loglevel", "error",
+                    "-ss", "0.4", "-i", str(vid), "-frames:v", "1", str(frame)], check=True)
+    from PIL import Image
+    img = Image.open(frame).convert("RGB")
+    bw, bh, x, y = rnd.bubble_geometry(240, 426, 0.4, "bottom-right", "circle")
+    centre = img.getpixel((x + bw // 2, y + bh // 2))
+    assert sum(centre) > 60, centre                     # the presenter is drawn there
+    assert img.getpixel((4, 4)) == (0, 0, 0)            # the B-roll is untouched
+    # Safe area: the bubble never touches the edges.
+    assert x >= int(240 * 0.05) and (y + bh) <= 426 - int(426 * 0.09)
+
+
+def test_split_layout_stacks_presenter_and_b_roll(tmp_path, files_dir, monkeypatch):
+    from codai.compose import presenter as pres
+    client, _ = _client(files_dir)
+    _stub_engine(monkeypatch, pres)
+    spec = {"canvas": {"width": 240, "height": 426, "fps": 24},
+            "presenter": {"image": _uri_file(_portrait(tmp_path)), "motion": "still",
+                          "layout": "split", "split": "bottom"},
+            "captions": {"enabled": False}, "outputs": ["video"],
+            "scenes": [{"text": "half and half",
+                        "audio": _uri(_tone(tmp_path / "ps.wav", 0.8)),
+                        "visuals": [{"type": "color", "color": "#00FF00"}]}]}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc
+    vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
+    frame = tmp_path / "fs.png"
+    subprocess.run([media.ffmpeg_bin(), "-y", "-hide_banner", "-loglevel", "error",
+                    "-ss", "0.4", "-i", str(vid), "-frames:v", "1", str(frame)], check=True)
+    from PIL import Image
+    img = Image.open(frame).convert("RGB")
+    top = img.getpixel((120, 80))
+    bottom = img.getpixel((120, 340))
+    assert top[1] > 150 and top[0] < 100, top           # B-roll green on top
+    assert bottom != top                                # presenter below
+
+
+def test_a_scene_can_opt_out_and_another_can_override(tmp_path, files_dir, monkeypatch):
+    from codai.compose import presenter as pres
+    client, _ = _client(files_dir)
+    calls = _stub_engine(monkeypatch, pres)
+    face = _uri_file(_portrait(tmp_path))
+    spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
+            "presenter": {"image": face, "motion": "still", "layout": "pip"},
+            "captions": {"enabled": False}, "outputs": ["video"],
+            "scenes": [
+                {"text": "hook", "audio": _uri(_tone(tmp_path / "a1.wav", 0.5)),
+                 "visuals": [{"type": "color", "color": "#111827"}]},
+                {"text": "b-roll only", "audio": _uri(_tone(tmp_path / "a2.wav", 0.5)),
+                 "visuals": [{"type": "gradient"}], "presenter": False},
+                {"text": "sign off", "audio": _uri(_tone(tmp_path / "a3.wav", 0.5)),
+                 "visuals": [], "presenter": {"layout": "full"}},
+            ]}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc
+    layouts = [(s["presenter"] or {}).get("layout") for s in doc["result"]["scenes"]]
+    assert layouts == ["pip", None, "full"]
+    assert len(calls) == 2                       # the opted-out scene ran no engine
+
+
+def test_a_failing_engine_falls_back_to_the_visuals_with_a_warning(tmp_path, files_dir,
+                                                                   monkeypatch):
+    from codai.compose import presenter as pres
+    client, _ = _client(files_dir)
+    _stub_engine(monkeypatch, pres, fail=True)
+    spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
+            "presenter": {"image": _uri_file(_portrait(tmp_path)), "motion": "still",
+                          "layout": "full"},
+            "captions": {"enabled": False}, "outputs": ["video"],
+            "scenes": [{"text": "still shipped",
+                        "audio": _uri(_tone(tmp_path / "pf.wav", 0.6)),
+                        "visuals": []}]}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc          # a render is never lost to this
+    assert any("presenter skipped" in w for w in doc["warnings"]), doc["warnings"]
+    assert doc["result"]["scenes"][0]["presenter"] is None
+    vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
+    assert media.probe(str(vid))["has_video"]
+
+
+def test_bottom_captions_move_above_a_bottom_bubble(tmp_path, files_dir, monkeypatch):
+    from codai.compose import presenter as pres
+    client, compose_mod = _client(files_dir)
+    _stub_engine(monkeypatch, pres)
+    base = {"canvas": {"width": 240, "height": 426, "fps": 24},
+            "captions": {"enabled": True, "timing": "estimate", "position": "bottom",
+                         "font": "DejaVu Sans", "preset": "classic"},
+            "outputs": ["video", "srt"],
+            "scenes": [{"text": "words under the face",
+                        "audio": _uri(_tone(tmp_path / "pc.wav", 1.0)),
+                        "visuals": [{"type": "color", "color": "#101010"}]}]}
+    # The margin is computed from the presenter geometry; check the helper directly
+    # (the burned frame is pixels, the intent is a margin).
+    from codai.api.compose import _caption_margin_bump, CaptionSpec, Canvas
+    caps = CaptionSpec(position="bottom")
+    canvas = Canvas(width=240, height=426, fps=24)
+    bump = _caption_margin_bump({"layout": "pip", "position": "bottom-right",
+                                 "size": 0.4, "shape": "circle"}, caps, canvas)
+    assert bump > int(426 * 0.10)                 # higher than the normal safe margin
+    assert _caption_margin_bump({"layout": "pip", "position": "top-right"}, caps, canvas) == 0
+    assert _caption_margin_bump({"layout": "full"}, caps, canvas) == 0
+    assert _caption_margin_bump({"layout": "pip", "position": "bottom-right"},
+                                CaptionSpec(position="center"), canvas) == 0
+    # And the render still succeeds with the bump applied.
+    spec = dict(base)
+    spec["presenter"] = {"image": _uri_file(_portrait(tmp_path)), "motion": "still",
+                         "layout": "pip", "position": "bottom-right", "size": 0.4}
+    doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
+    assert doc["status"] == "done", doc
+
+
+def test_presenter_specs_are_validated_and_a_missing_engine_is_501(tmp_path, files_dir,
+                                                                   monkeypatch):
+    from codai.compose import presenter as pres
+    client, _ = _client(files_dir)
+    _stub_engine(monkeypatch, pres)
+    face = _uri_file(_portrait(tmp_path))
+    scene = {"text": "x", "audio": _uri(_tone(tmp_path / "pv2.wav", 0.4)),
+             "visuals": [{"type": "color"}]}
+
+    def err(presenter, code=400):
+        r = client.post("/v1/video/compose",
+                        json={"scenes": [scene], "presenter": presenter,
+                              "captions": {"enabled": False}})
+        assert r.status_code == code, r.text
+        return r.json()["detail"]
+
+    assert "layout" in err({"image": face, "layout": "hologram"})
+    assert "position" in err({"image": face, "layout": "pip", "position": "middle-left"})
+    assert "size" in err({"image": face, "layout": "pip", "size": 2.0})
+    assert "shape" in err({"image": face, "layout": "pip", "shape": "star"})
+    assert "split" in err({"image": face, "layout": "split", "split": "left"})
+    assert "motion" in err({"image": face, "motion": "teleport"})
+    assert "base_src" in err({"image": face, "motion": "src"})
+    assert "needs a character or an image" in err({"layout": "full"})
+    assert "no character profile" in err({"character": "nobody-here"})
+    # A scene with no visuals and a pip presenter has nothing to composite onto.
+    r = client.post("/v1/video/compose",
+                    json={"scenes": [{"text": "x", "audio": scene["audio"], "visuals": []}],
+                          "presenter": {"image": face, "layout": "pip"},
+                          "captions": {"enabled": False}})
+    assert r.status_code == 400 and "full-screen presenter" in r.json()["detail"]
+    # No engine installed at all → 501 naming what to install.
+    monkeypatch.setattr(pres, "available_engines", lambda: [])
+    monkeypatch.setattr(pres, "engine_installed", lambda name: False)
+    d = err({"image": face, "layout": "full"}, code=501)
+    assert "wav2lip" in d and "sadtalker" in d
+
+
+def test_talking_head_is_a_one_scene_composition(tmp_path, files_dir, monkeypatch):
+    from codai.compose import presenter as pres
+    client, compose_mod = _client(files_dir)
+    calls = _stub_engine(monkeypatch, pres)
+    face = _uri_file(_portrait(tmp_path))
+    r = client.post("/v1/video/talking-head",
+                    json={"image": face, "audio": _uri(_tone(tmp_path / "th.wav", 1.0)),
+                          "canvas": {"width": 128, "height": 128, "fps": 24},
+                          "motion": "still"})
+    assert r.status_code == 202, r.text
+    job = r.json()["id"]
+    doc = _wait(client, job)
+    assert doc["status"] == "done", doc
+    assert len(calls) == 1
+    vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
+    info = media.probe(str(vid))
+    assert (info["width"], info["height"]) == (128, 128)
+    assert info["duration"] == pytest.approx(1.05, abs=0.2)
+    # The compose job store is the same store.
+    assert client.get(f"/v1/video/compose/{job}").json()["status"] == "done"
+    assert client.get(f"/v1/video/talking-head/{job}").json()["status"] == "done"
+    # Neither text nor audio is an error; too much audio is refused.
+    assert client.post("/v1/video/talking-head", json={"image": face}).status_code == 400
+    long_audio = _uri(_tone(tmp_path / "long_th.wav", 16.0))
+    r = client.post("/v1/video/talking-head", json={"image": face, "audio": long_audio})
+    assert r.status_code == 400 and "capped at 15s" in r.json()["detail"]
+
+
+def test_a_character_profile_supplies_the_face(tmp_path, files_dir, monkeypatch):
+    from codai.compose import presenter as pres
+    import codai.api.characters as ch
+    monkeypatch.setattr(ch, "_chars_dir", lambda: str(tmp_path / "chars"))
+    # Two references: the one labelled "front" is the one a presenter should use.
+    ch._save_character("nova", "host", [
+        ch.CharacterImage(label="side", data=_uri_file(_portrait(tmp_path, "side.png"))),
+        ch.CharacterImage(label="front", data=_uri_file(_portrait(tmp_path, "front.png",
+                                                                 130, 170))),
+    ])
+    work = str(tmp_path / "w")
+    face = pres.portrait_path({"character": "nova"}, work)
+    from PIL import Image
+    assert Image.open(face).size == (130, 170)      # the front reference
+    with pytest.raises(pres.PresenterError):
+        pres.portrait_path({"character": "ghost"}, work)
+
+
+def test_engine_selection_prefers_video_engines_for_a_clip(monkeypatch):
+    from codai.compose import presenter as pres
+    monkeypatch.setattr(pres, "available_engines", lambda: ["wav2lip", "sadtalker"])
+    assert pres.pick_engine("auto", has_base=True) == "wav2lip"
+    assert pres.pick_engine("auto", has_base=False) == "sadtalker"
+    assert pres.pick_engine("sadtalker", has_base=True) == "sadtalker"
+    monkeypatch.setattr(pres, "available_engines", lambda: ["sadtalker"])
+    assert pres.pick_engine("auto", has_base=True) == "sadtalker"
+    with pytest.raises(pres.PresenterError):
+        pres.pick_engine("wav2lip", has_base=True)
+    monkeypatch.setattr(pres, "available_engines", lambda: [])
+    with pytest.raises(pres.PresenterError):
+        pres.pick_engine("auto", has_base=False)