feat: add UI-less prompt_to_video_assets endpoint (LLM->image->LLM)

Chains Qwen (first-frame image prompt) -> image model -> R2 upload ->
Qwen (video prompt grounded on the frame) in one streaming call. Exposed
via gr.api so it adds no Gradio UI. Returns video_prompt + first_frame_url
(presigned GET) plus r2 filekey/bucket and the intermediate frame prompt.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app.py

@@ -676,6 +676,174 @@ def assistant_chat(
             yield text, _progress("done", 1.0, label="Done")
 
 
+# =============================================================================
+# Combined endpoint: text -> first-frame image + video prompt (no UI)
+# =============================================================================
+# One call chains LLM + image model + LLM to turn a raw idea into the two assets
+# an image-to-video pipeline needs:
+#   1. LLM (Qwen) writes a first-frame *image* prompt from
+#      `image_instruction` + `original_input`.
+#   2. The image model renders that first frame.
+#   3. The frame is uploaded to R2 (same path as the Generate tab).
+#   4. LLM (Qwen) writes a *video* prompt from `video_instruction` +
+#      `original_input`, grounded on the rendered first frame image.
+# It returns the video prompt and the first-frame URL (+ R2 filekey/bucket and
+# the intermediate image prompt). Exposed via `gr.api` so it has no Gradio UI.
+#
+# Like the other streaming endpoints it is a generator yielding one structured
+# dict per frame (so progress crosses the /call SSE stream, see _progress); the
+# same dict carries the results, filled in as each stage completes, and the final
+# yield has `done=True`. Stage weights below sum to 1.0.
+_P2V_W_FRAME_PROMPT = 0.30  # LLM writes the first-frame image prompt
+_P2V_W_IMAGE = 0.30         # image model renders the first frame
+_P2V_W_UPLOAD = 0.05        # R2 upload
+_P2V_W_VIDEO_PROMPT = 0.35  # LLM writes the video prompt
+
+
+def _compose_instruction(instruction, original_input):
+    """Join an instruction with the raw idea into a single LLM message."""
+    instruction = (instruction or "").strip()
+    original_input = (original_input or "").strip()
+    if instruction and original_input:
+        return f"{instruction}\n\nInput:\n{original_input}"
+    return instruction or original_input
+
+
+def prompt_to_video_assets(
+    original_input: str,
+    image_instruction: str,
+    video_instruction: str,
+    model_name: str = MODEL_ZIMAGE,
+    height: int = 1024,
+    width: int = 1024,
+    num_inference_steps: int = 9,
+    guidance_scale: float = 0.0,
+    seed: int = 42,
+    randomize_seed: bool = True,
+    reasoning: str = "Off",
+    max_new_tokens: int = 512,
+    request: gr.Request = None,
+) -> dict:
+    """Text -> first-frame image (uploaded to R2) + video generation prompt.
+
+    Streams progress and returns a dict with ``video_prompt``, ``first_frame_url``
+    (a presigned GET, usable directly), ``r2_filekey``/``r2_bucket`` (for callers
+    that resolve their own public URL), ``first_frame_prompt`` (the intermediate
+    image prompt) and ``seed``. Has no UI (registered via ``gr.api``).
+    """
+    state = {
+        "stage": "frame_prompt", "p": 0.0, "step": 0, "total": 0, "label": "",
+        "first_frame_prompt": None,
+        "first_frame_url": None,
+        "r2_filekey": None,
+        "r2_bucket": None,
+        "video_prompt": None,
+        "seed": None,
+        "done": False,
+        "error": None,
+    }
+
+    def frame(stage, base, span, frac=1.0, step=0, total=0, label=""):
+        state.update(
+            stage=stage,
+            p=max(0.0, min(1.0, base + span * max(0.0, min(1.0, frac)))),
+            step=int(step), total=int(total), label=label,
+        )
+        return dict(state)
+
+    if not (original_input or "").strip():
+        state["error"] = "original_input is required"
+        state["done"] = True
+        yield dict(state)
+        return
+
+    # --- Stage 1: LLM writes the first-frame image prompt ---------------------
+    base = 0.0
+    frame_prompt = ""
+    for ev in vlm_chat(
+        _compose_instruction(image_instruction, original_input),
+        None, reasoning, max_new_tokens,
+    ):
+        if ev[0] == "progress":
+            _, produced, budget, partial = ev
+            frame_prompt = partial
+            yield frame("frame_prompt", base, _P2V_W_FRAME_PROMPT,
+                        frac=produced / max(budget, 1), step=produced, total=budget,
+                        label="Writing first-frame prompt")
+        else:
+            frame_prompt = ev[1]
+    frame_prompt = (frame_prompt or "").strip()
+    state["first_frame_prompt"] = frame_prompt
+    yield frame("frame_prompt", base, _P2V_W_FRAME_PROMPT, label="First-frame prompt ready")
+
+    # --- Stage 2: image model renders the first frame -------------------------
+    base += _P2V_W_FRAME_PROMPT
+    image, used_seed = None, seed
+    use_negative_prompt = (model_name == MODEL_NOOBXL)
+    for ev in generate_image(
+        model_name, frame_prompt, NOOBXL_NEGATIVE, use_negative_prompt,
+        height, width, num_inference_steps, guidance_scale, seed, randomize_seed,
+    ):
+        if ev[0] == "progress":
+            _, step, total = ev
+            yield frame("image", base, _P2V_W_IMAGE,
+                        frac=step / max(total, 1), step=step, total=total,
+                        label=f"Rendering first frame {step}/{total}")
+        else:
+            _, image, used_seed = ev
+    state["seed"] = int(used_seed)
+
+    # --- Stage 3: upload the first frame to R2 --------------------------------
+    base += _P2V_W_IMAGE
+    yield frame("upload", base, _P2V_W_UPLOAD, frac=0.1, label="Uploading first frame")
+    uid = r2_uploader.uid_from_request(request)
+    buf = io.BytesIO()
+    image.save(buf, format="PNG")
+    params = {
+        "model": model_name,
+        "prompt": frame_prompt,
+        "negative_prompt": (NOOBXL_NEGATIVE if use_negative_prompt else ""),
+        "height": int(height),
+        "width": int(width),
+        "num_inference_steps": int(num_inference_steps),
+        "guidance_scale": float(guidance_scale),
+        "seed": int(used_seed),
+        "uid": uid,
+        "source": "prompt_to_video_assets",
+        "original_input": original_input,
+    }
+    up = r2_uploader.upload_asset(
+        namespace=R2_NAMESPACE, prompt=frame_prompt, params=params,
+        data=buf.getvalue(), ext=".png", content_type="image/png", uid=uid,
+    )
+    if up.get("ok"):
+        state["r2_filekey"] = up["filekey"]
+        state["r2_bucket"] = up["bucket"]
+        state["first_frame_url"] = r2_uploader.presign_get_url(up["filekey"], up["bucket"])
+    else:
+        state["error"] = up.get("error", "R2 upload failed")
+
+    # --- Stage 4: LLM writes the video prompt (grounded on the frame) ---------
+    base += _P2V_W_UPLOAD
+    video_prompt = ""
+    for ev in vlm_chat(
+        _compose_instruction(video_instruction, original_input),
+        image, reasoning, max_new_tokens,
+    ):
+        if ev[0] == "progress":
+            _, produced, budget, partial = ev
+            video_prompt = partial
+            yield frame("video_prompt", base, _P2V_W_VIDEO_PROMPT,
+                        frac=produced / max(budget, 1), step=produced, total=budget,
+                        label="Writing video prompt")
+        else:
+            video_prompt = ev[1]
+    state["video_prompt"] = (video_prompt or "").strip()
+
+    state["done"] = True
+    yield frame("done", 1.0, 0.0, label="Done")
+
+
 # Recommended defaults per model: (steps, guidance, height, width)
 MODEL_DEFAULTS = {
     MODEL_ZIMAGE: dict(steps=9, guidance=0.0, height=1024, width=1024),
@@ -951,6 +1119,11 @@ with gr.Blocks(fill_height=True) as demo:
         fn=assistant_chat, inputs=vlm_inputs, outputs=[vlm_output, vlm_progress],
     )
 
+    # UI-less combined endpoint: text -> first-frame image (R2) + video prompt.
+    # `gr.api` derives its schema from the function's type hints and registers no
+    # components, so it adds an API route without touching the visible UI.
+    gr.api(prompt_to_video_assets, api_name="prompt_to_video_assets")
+
 if __name__ == "__main__":
     demo.launch(
         theme=custom_theme,

r2_uploader.py

@@ -126,6 +126,27 @@ def _ensure_bucket(s3, bucket: str) -> None:
         _ensured_buckets.add(bucket)
 
 
+def presign_get_url(filekey: str, bucket: str | None = None, expires: int = 604800) -> str | None:
+    """Return a presigned GET URL for an uploaded object, or None on failure.
+
+    ``expires`` defaults to 7 days (the SigV4 maximum). This lets callers hand
+    back a directly-usable URL even when no public R2 domain is bound; downstream
+    consumers that have a public base can still rebuild a clean URL from the
+    ``filekey``/``bucket`` reported alongside it. Never raises.
+    """
+    try:
+        cfg = _load_cfg()
+        bucket = bucket or cfg.get("bucket") or DEFAULT_BUCKET
+        s3 = _client(cfg)
+        return s3.generate_presigned_url(
+            "get_object",
+            Params={"Bucket": bucket, "Key": filekey},
+            ExpiresIn=int(expires),
+        )
+    except Exception:  # noqa: BLE001 - URL is best-effort, never fatal
+        return None
+
+
 def upload_asset(
     *,
     namespace: str,