Add Human demo videos (robot → human, fal MiniMax-H3 480P) as a Video generation option

README.md

@@ -105,6 +105,40 @@ combined. Camera names load automatically from the dataset's `meta/info.json`
 before estimating, including private datasets after login. Depth and non-video
 features are excluded. Visual questions have a configurable cadence.
 
+## Human demo videos
+
+A separate **Video generation** category on the Annotations step, next to the text
+annotations. For every subtask (or up to 1–8 per episode), the job turns the robot view into
+a first-person human demonstration, following LeRobot's `--human_video` annotation module:
+
+1. the selected vision model reads the robot segment and writes the human task and an
+   image-edit prompt;
+2. `Qwen/Qwen-Image-Edit-2509` removes the robot from the segment's first full-resolution
+   frame and adds two resting human hands;
+3. the vision model writes a step-by-step video prompt from the edited frame;
+4. `MiniMaxAI/MiniMax-H3` animates it at 480P (5 s).
+
+Both media models run on fal through HF Inference Providers with the signed-in user's
+token. Selecting videos also selects Subtasks, since there is one video per subtask.
+
+The estimate follows the selected episodes' lengths: it expects about one subtask per 5 s of
+each episode (at least one per episode), allows twice that in the upper bound, and applies the
+per-episode cap. **Parallel generations** (1–32, default 8) limits simultaneous media
+requests across the whole job. The review warns when the expected videos at that parallelism
+would outlast the job timeout (about 3 min per video).
+
+Videos start only after an episode's
+annotations pass validation, and each request reserves its quoted price
+(`model_catalog.HUMAN_VIDEO`) from the spending target first. A failed video is recorded
+and the others continue. Reaching the target or the runtime limit stops further videos but
+keeps the episode's annotations. The launch checks that both fal mappings are live.
+
+The output holds `annotation_studio/human_videos/episode_XXXXXX/segment_XXX{.mp4,
+_first_frame.png,_robot_frame.png}` (named by source episode) and `index.json`, which lists
+each segment's subtask, timestamps, prompts, status and materialized `episode_index`.
+These are synthetic media, not language columns. The Jobs page previews completed videos
+next to the robot frame, served through the Space so private outputs stay private.
+
 ## Prompts and examples
 
 Every annotation card has an **Edit prompt** editor with the actual model

frontend/src/JobsView.tsx

@@ -1,4 +1,4 @@
-import type { ReactNode } from "react";
+import { useState, type ReactNode } from "react";
 import {
   ArrowUpRight,
   ChevronDown,
@@ -30,6 +30,8 @@ export type Progress = {
     input_tokens?: number;
     output_tokens?: number;
     rate_limited_requests?: number;
+    media_usd?: number;
+    completed_media_requests?: number;
   };
   cost?: {
     inference_usd: number | null;
@@ -42,7 +44,81 @@ export type Progress = {
   };
   annotation_url?: string | null;
   output_mode?: string;
+  human_videos?: Record<string, number> | null;
 };
+type HumanVideo = {
+  episode_index?: number | null;
+  source_episode_index: number;
+  segment_index?: number;
+  subtask?: string;
+  caption?: string | null;
+  status: string;
+  robot_frame?: string;
+  first_frame?: string;
+  video?: string;
+};
+
+function HumanVideos({ jobId }: { jobId: string }) {
+  const [videos, setVideos] = useState<HumanVideo[] | null>(null);
+  const [error, setError] = useState("");
+  const [loading, setLoading] = useState(false);
+  async function load() {
+    setLoading(true);
+    setError("");
+    try {
+      const response = await fetch(`/api/jobs/${jobId}/human-videos`, {
+        credentials: "same-origin",
+      });
+      if (!response.ok) throw new Error("No human demo videos saved yet.");
+      setVideos(await response.json());
+    } catch (e) {
+      setError((e as Error).message);
+    } finally {
+      setLoading(false);
+    }
+  }
+  if (!videos)
+    return (
+      <div className="human-videos">
+        <button className="text-action" onClick={load} disabled={loading}>
+          {loading ? "Loading videos…" : "Preview human demo videos"}
+        </button>
+        {error && <p className="job-cost-note">{error}</p>}
+      </div>
+    );
+  const ready = videos.filter((v) => v.status === "completed" && v.video);
+  return (
+    <div className="human-videos">
+      {ready.length === 0 && <p>No completed videos yet.</p>}
+      <div className="human-video-grid">
+        {ready.map((v) => (
+          <figure key={`${v.source_episode_index}-${v.segment_index}`}>
+            <div className="human-video-pair">
+              {v.robot_frame && (
+                <img src={v.robot_frame} alt="Robot view" loading="lazy" />
+              )}
+              <video
+                src={v.video}
+                poster={v.first_frame}
+                controls
+                muted
+                loop
+                playsInline
+                preload="none"
+              />
+            </div>
+            <figcaption>
+              <strong>
+                Episode {v.episode_index ?? v.source_episode_index}
+              </strong>{" "}
+              · {v.subtask}
+            </figcaption>
+          </figure>
+        ))}
+      </div>
+    </div>
+  );
+}
 const terminal = new Set(["COMPLETED", "ERROR", "CANCELED", "DELETED"]);
 const usd = (value: number | null | undefined) =>
   value == null
@@ -252,7 +328,24 @@ export default function JobsView({
                         {p?.usage.successful_calls?.toLocaleString() ?? "—"}
                       </dd>
                     </div>
+                    {p?.human_videos && (
+                      <div>
+                        <dt>Human videos</dt>
+                        <dd>
+                          {(p.human_videos.completed || 0).toLocaleString()}
+                          {p.human_videos.failed
+                            ? ` · ${p.human_videos.failed} failed`
+                            : ""}
+                          {p.usage.media_usd
+                            ? ` · ${usd(p.usage.media_usd)}`
+                            : ""}
+                        </dd>
+                      </div>
+                    )}
                   </dl>
+                  {!!p?.human_videos?.completed && (
+                    <HumanVideos jobId={job.id} />
+                  )}
                   {p &&
                     (p.failed > 0 ||
                       (p.stopped || 0) > 0 ||

frontend/src/jobs.css

@@ -312,3 +312,39 @@
   flex-direction: row;
   align-items: center;
 }
+
+.human-videos {
+  margin-top: 14px;
+  font-size: 12px;
+}
+.human-video-grid {
+  display: grid;
+  grid-template-columns: repeat(auto-fill, minmax(320px, 1fr));
+  gap: 14px;
+}
+.human-video-pair {
+  display: grid;
+  grid-template-columns: 1fr 1fr;
+  gap: 6px;
+}
+.human-video-pair img,
+.human-video-pair video {
+  width: 100%;
+  aspect-ratio: 4 / 3;
+  object-fit: cover;
+  border-radius: 6px;
+  background: var(--line-subtle);
+}
+.human-videos figure {
+  margin: 0;
+}
+.human-videos figcaption {
+  margin-top: 6px;
+  color: var(--fg-muted);
+  line-height: 1.5;
+}
+@media (max-width: 480px) {
+  .human-video-grid {
+    grid-template-columns: 1fr;
+  }
+}

frontend/src/main.tsx

@@ -45,7 +45,9 @@ type Feature =
   | "Memory"
   | "Task rephrasings"
   | "Interjections & speech"
-  | "Visual questions";
+  | "Visual questions"
+  | "Human demo videos";
+const HUMAN_VIDEOS: Feature = "Human demo videos";
 type Options = {
   model: string;
   dataset: string;
@@ -57,6 +59,8 @@ type Options = {
   hours: number;
   mode: string;
   prompts: Record<string, string>;
+  human_video_segments: number;
+  human_video_parallel: number;
 };
 type PromptTemplate = {
   key: string;
@@ -84,10 +88,26 @@ type ModelProfile = {
   pricing_url: string;
   pricing_note?: string;
 };
+type HumanVideoProfile = {
+  provider_name: string;
+  edit_model: string;
+  video_model: string;
+  resolution: string;
+  duration_s: number;
+  edit_usd: number;
+  video_usd: number;
+  checked_at: string;
+  pricing_url: string;
+  segment_choices: number[];
+  parallel_choices: number[];
+  seconds_per_subtask: number;
+  minutes_per_video: number;
+};
 type ModelCatalog = {
   models: ModelProfile[];
   default: string;
   unavailable: { name: string; reason: string }[];
+  human_video?: HumanVideoProfile;
 };
 type Quote = {
   model_profile: ModelProfile;
@@ -96,6 +116,13 @@ type Quote = {
     low: number;
     high: number;
     inference: number;
+    media?: number;
+    media_max?: number;
+    videos?: number;
+    videos_max?: number;
+    video_minutes?: number;
+    video_parallel?: number;
+    segment_usd?: number;
     cpu_max: number;
     custom_prompts: number;
     calls: number;
@@ -133,6 +160,8 @@ const defaults: Options = {
   hours: 4,
   mode: COPY,
   prompts: {},
+  human_video_segments: 0,
+  human_video_parallel: 8,
 };
 const definitions: {
   name: Feature;
@@ -519,11 +548,16 @@ function App() {
     setNotice("");
   }
   function toggle(name: Feature) {
-    change({
-      features: options.features.includes(name)
-        ? options.features.filter((x) => x !== name)
-        : [...options.features, name],
-    });
+    const on = !options.features.includes(name);
+    let features = on
+      ? [...options.features, name]
+      : options.features.filter((x) => x !== name);
+    // Videos are generated per subtask, so they need the subtask annotations.
+    if (on && name === HUMAN_VIDEOS && !features.includes("Subtasks"))
+      features = ["Subtasks", ...features];
+    if (!on && name === "Subtasks")
+      features = features.filter((x) => x !== HUMAN_VIDEOS);
+    change({ features });
   }
   async function estimate(advance = false) {
     setLoading(true);
@@ -936,6 +970,12 @@ function App() {
                   className="card step-card"
                   aria-label="Annotation settings"
                 >
+                  <div className="category-heading">
+                    <h3>Text annotations</h3>
+                    <p className="field-help">
+                      Language written into the dataset's Parquet files.
+                    </p>
+                  </div>
                   <div className="annotation-choices">
                     {definitions.map(({ name, title, description }) => (
                       <div
@@ -1005,6 +1045,111 @@ function App() {
                   </details>
                 </section>
               )}
+              {step === 1 && (
+                <section
+                  className={`card step-card video-category ${options.features.includes(HUMAN_VIDEOS) ? "selected" : ""}`}
+                  aria-label="Video generation settings"
+                >
+                  <div className="category-heading">
+                    <h3>
+                      <Video size={16} /> Video generation
+                    </h3>
+                    <p className="field-help">
+                      New media generated from your robot videos, saved next to
+                      the dataset.
+                    </p>
+                  </div>
+                  <div className="annotation-choice">
+                    <label>
+                      <input
+                        type="checkbox"
+                        checked={options.features.includes(HUMAN_VIDEOS)}
+                        onChange={() => toggle(HUMAN_VIDEOS)}
+                      />
+                      <span>
+                        <strong>Human demo videos</strong>
+                        <small>
+                          A first-person video of human hands doing each
+                          subtask: the robot is edited out of the first frame,
+                          then the scene is animated.
+                        </small>
+                      </span>
+                    </label>
+                  </div>
+                  {options.features.includes(HUMAN_VIDEOS) && (
+                    <div className="fields-row video-settings">
+                      <label>
+                        Videos per episode
+                        <select
+                          value={options.human_video_segments}
+                          onChange={(e) =>
+                            change({
+                              human_video_segments: Number(e.target.value),
+                            })
+                          }
+                        >
+                          {(
+                            modelCatalog?.human_video?.segment_choices ?? [
+                              0, 1, 2, 3, 5, 8,
+                            ]
+                          ).map((n) => (
+                            <option key={n} value={n}>
+                              {n === 0 ? "Every subtask" : `Up to ${n}`}
+                            </option>
+                          ))}
+                        </select>
+                      </label>
+                      <label>
+                        Parallel generations
+                        <select
+                          value={options.human_video_parallel}
+                          onChange={(e) =>
+                            change({
+                              human_video_parallel: Number(e.target.value),
+                            })
+                          }
+                        >
+                          {(
+                            modelCatalog?.human_video?.parallel_choices ?? [
+                              1, 4, 8, 16, 32,
+                            ]
+                          ).map((n) => (
+                            <option key={n} value={n}>
+                              Up to {n} at a time
+                            </option>
+                          ))}
+                        </select>
+                      </label>
+                    </div>
+                  )}
+                  {modelCatalog?.human_video && (
+                    <p className="field-help dependency-hint">
+                      {modelCatalog.human_video.edit_model} →{" "}
+                      {modelCatalog.human_video.video_model}{" "}
+                      {modelCatalog.human_video.resolution} via{" "}
+                      {modelCatalog.human_video.provider_name} on Inference
+                      Providers · about{" "}
+                      {money(
+                        modelCatalog.human_video.edit_usd +
+                          modelCatalog.human_video.video_usd,
+                      )}{" "}
+                      per video, billed to your account. The estimate counts
+                      about one subtask per{" "}
+                      {modelCatalog.human_video.seconds_per_subtask} s of each
+                      episode.
+                      {options.features.includes(HUMAN_VIDEOS) &&
+                        " One video per subtask, so Subtasks are included."}{" "}
+                      <a
+                        href={modelCatalog.human_video.pricing_url}
+                        target="_blank"
+                        rel="noreferrer"
+                      >
+                        Pricing
+                      </a>
+                    </p>
+                  )}
+                </section>
+              )}
               {step === 2 && (
                 <section
                   className="card step-card model-picker"
@@ -1263,6 +1408,19 @@ function App() {
                           <span>–</span>
                           {money(quote.costs.high)}
                         </div>
+                        {!!quote.costs.video_minutes &&
+                          quote.costs.video_minutes >
+                            quote.costs.hours * 60 && (
+                            <p className="field-help video-time-warning">
+                              ~{num(quote.costs.videos ?? 0)} videos at{" "}
+                              {quote.costs.video_parallel} at a time may need
+                              about {Math.ceil(quote.costs.video_minutes / 60)}{" "}
+                              h, longer than the {quote.costs.hours} h timeout.
+                              Later videos would be skipped; raise parallel
+                              generations or the timeout, or cap videos per
+                              episode.
+                            </p>
+                          )}
                         <details className="progressive-details cost-breakdown">
                           <summary>
                             Cost breakdown <ChevronDown size={15} />
@@ -1276,6 +1434,24 @@ function App() {
                                   : "—"}
                               </strong>
                             </div>
+                            {!!quote.costs.videos && (
+                              <div>
+                                <span>
+                                  Human demo videos (~{num(quote.costs.videos)},
+                                  up to {num(quote.costs.videos_max ?? 0)}){" "}
+                                  <span
+                                    className="info-tooltip"
+                                    title="One video per subtask, estimated from each episode's length. Each video reserves its quoted price before it starts; the spending target stops further videos."
+                                  >
+                                    <HelpCircle size={12} />
+                                  </span>
+                                </span>
+                                <strong>
+                                  {money(quote.costs.media ?? 0)}–
+                                  {money(quote.costs.media_max ?? 0)}
+                                </strong>
+                              </div>
+                            )}
                             <div>
                               <span>
                                 CPU job reservation{" "}

frontend/src/wizard.css

@@ -201,6 +201,34 @@
 .dependency-hint {
   margin-top: 20px;
 }
+.category-heading {
+  margin-bottom: 18px;
+}
+.category-heading h3 {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+  font-size: 14px;
+  font-weight: 600;
+  margin: 0 0 4px;
+}
+.category-heading .field-help {
+  margin: 0;
+}
+.video-category {
+  margin-top: 16px;
+}
+.video-category .annotation-choice {
+  padding: 0;
+  border: 0;
+}
+.video-settings {
+  margin-top: 16px;
+}
+.video-time-warning {
+  color: var(--warning, #b45309);
+  margin-top: 8px;
+}
 .text-action {
   display: inline-block;
   color: var(--fg-muted);

human-video-plan.txt

@@ -0,0 +1,14 @@
+You are preparing a human demonstration of one step of a robot manipulation video.
+
+The {n} images are frames sampled uniformly from the robot's segment ({duration:.1f} s); image 1 is its first frame.
+Subtask performed in this segment (authoritative): "{subtask}"
+Episode instruction (context only): "{task}"
+
+Return one JSON object with exactly these keys:
+- "human_task": the subtask phrased for a person, naming the hand that acts (the hand on the same image side as the robot arm that acts), e.g. "Pick up the red cup with the right hand and place it on the tray."
+- "object": short name and appearance of the manipulated object.
+- "target": short name and appearance of where it goes (or "" if there is none).
+- "hand": "left" or "right".
+- "steps": list of 2 to 4 short steps describing how the objects move during the segment.
+- "final_state": one short phrase describing the final object state.
+- "edit_prompt": an instruction for an image-editing model that turns image 1 into a first-person view of a person doing the task. It must: (1) completely remove every robot arm, gripper, cable and robot base, filling those areas with the continuing table and background; (2) add exactly two realistic human hands and forearms (the viewer's own) entering from the bottom edge and resting relaxed on the near edge of the table, touching nothing; (3) list every object that must stay exactly as it is, with colour and position; (4) keep the camera viewpoint, table, background, lighting and layout unchanged and photorealistic. It must not add any other people, arms or objects.

human-video-prompt.txt

@@ -0,0 +1,14 @@
+Image 1 is the first frame of a new first-person video of a person doing a task; an image-to-video model will animate it.
+Images 2 onwards show how a robot did the same task; use them only for the order of events and object motion. No robot may appear in the new video.
+
+Task: "{human_task}"
+Object: {object}
+Target: {target}
+Steps: {steps}
+Final state: {final_state}
+
+Write "video_prompt": one paragraph of 100 to 170 words for the image-to-video model. Start with "From a fixed first-person camera perspective, based strictly on the initial image, the viewer's two human hands are resting on the near edge of the table." Describe the object and the target with their colours and positions as in image 1. Include verbatim: "Generate only these two human arms, with no extra limbs and no robotic arms. Every object stays still unless the hand holds it." Then describe with "First, ... Next, ... Finally, ..." how the {hand} hand reaches the object, grasps it, moves it and releases it at the target while the other hand stays still, then returns to rest without touching anything. The camera never moves. End with "The final frame shows {final_state}, with both hands resting at the near edge." The motion is realistic and smooth and completes within about {duration_s} seconds.
+
+Also write "caption": two sentences describing the human video.
+
+Return one JSON object with keys "video_prompt" and "caption".

human_video.py

@@ -0,0 +1,237 @@
+"""Robot → human demonstration videos, one per annotated subtask.
+
+Mirrors LeRobot's ``human_video`` annotation module (``--human_video.enabled``), self-contained so
+that it runs on the pinned pipeline bundle:
+
+1. the VLM reads frames of the robot segment and writes the human task and an image-edit prompt;
+2. Qwen-Image-Edit turns the segment's first full-resolution frame into a first-person human view;
+3. the VLM writes a step-by-step video prompt from the edited frame;
+4. MiniMax-H3 (480P) animates it.
+
+Media requests go through Hugging Face Inference Providers (fal) with the signed-in user's token and
+reserve their quoted price from the job's spending target before they start.
+"""
+
+import io
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from urllib.parse import urlparse
+
+from model_catalog import human_video_profile
+from PIL import Image
+from provider import BudgetExceededError, DeadlineExceededError
+
+from lerobot.annotations.steerable_pipeline.frames import to_image_blocks
+
+PLAN_PROMPT = Path(__file__).with_name("human-video-plan.txt").read_text()
+VIDEO_PROMPT = Path(__file__).with_name("human-video-prompt.txt").read_text()
+# A video takes about three minutes; do not start one the runtime limit would cut off.
+VIDEO_SECONDS_NEEDED = 420
+
+
+def png_bytes(image):
+    buffer = io.BytesIO()
+    image.convert("RGB").save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+def subtask_spans(record, staging):
+    """``{text, start, end}`` from the staged subtask rows; each span ends at the next start."""
+    rows = [r for r in staging.read("plan") if r.get("style") == "subtask"] if staging.has("plan") else []
+    rows = sorted(rows, key=lambda r: float(r["timestamp"]))
+    end_of_episode = float(record.frame_timestamps[-1])
+    spans = []
+    for i, row in enumerate(rows):
+        start = float(row["timestamp"])
+        end = float(rows[i + 1]["timestamp"]) if i + 1 < len(rows) else end_of_episode
+        if end > start and row.get("content"):
+            spans.append({"text": str(row["content"]), "start": start, "end": end})
+    return spans
+
+
+class FalMedia:
+    """Qwen-Image-Edit + MiniMax-H3 on fal through the HF router, charged to ``budget``."""
+
+    def __init__(self, token, budget, remaining_time, profile=None, concurrency=None, poll_interval=2.0):
+        self.token, self.budget, self.remaining_time = token, budget, remaining_time
+        self.profile = profile or human_video_profile()
+        self.poll_interval = poll_interval
+        # Job-wide limit on simultaneous media requests, shared by every episode thread.
+        self.concurrency = max(1, min(int(concurrency or self.profile["concurrency"]), 32))
+        self.semaphore = threading.BoundedSemaphore(self.concurrency)
+
+    def edit_image(self, image, prompt):
+        from huggingface_hub import InferenceClient
+
+        self.remaining_time()
+        with self.semaphore:
+            reservation = self.budget.reserve_media(self.profile["edit_usd"])
+            client = InferenceClient(provider=self.profile["provider"], api_key=self.token)
+            edited = client.image_to_image(
+                png_bytes(image),
+                prompt=prompt,
+                model=self.profile["edit_model"],
+                guidance_scale=self.profile["edit_guidance_scale"],
+            )
+            self.budget.record_media(reservation)
+        # fal returns a square image; keep the camera's size and aspect ratio.
+        return edited.convert("RGB").resize(image.size, Image.LANCZOS)
+
+    def image_to_video(self, image, prompt):
+        from huggingface_hub.inference._providers.fal_ai import FalAIImageToVideoTask
+        from huggingface_hub.utils import get_session, hf_raise_for_status
+
+        if self.remaining_time() < VIDEO_SECONDS_NEEDED:
+            raise DeadlineExceededError("Not enough runtime left for another video.")
+        # InferenceClient has no image-text-to-video method yet; fal's image-to-video helper speaks
+        # the same queue protocol once its task matches the model's mapping.
+        helper = FalAIImageToVideoTask()
+        helper.task = self.profile["video_task"]
+        params = {
+            "prompt": prompt,
+            "resolution": self.profile["resolution"],
+            "duration": self.profile["duration_s"],
+        }
+        session = get_session()
+        with self.semaphore:
+            request = helper.prepare_request(
+                inputs=png_bytes(image),
+                parameters=params,
+                headers={},
+                model=self.profile["video_model"],
+                api_key=self.token,
+            )
+            reservation = self.budget.reserve_media(self.profile["video_usd"])
+            response = session.post(request.url, json=request.json, headers=request.headers, timeout=60)
+            hf_raise_for_status(response)
+            queued = response.json()
+            if not queued.get("request_id") or not queued.get("response_url"):
+                raise RuntimeError("The video provider did not queue the request.")
+            parsed = urlparse(request.url)
+            base = f"{parsed.scheme}://{parsed.netloc}"
+            if parsed.netloc == "router.huggingface.co":
+                base += "/fal-ai"
+            query = f"?{parsed.query}" if parsed.query else ""
+            path = urlparse(queued["response_url"]).path
+            status = queued.get("status")
+            while status != "COMPLETED":
+                if status in ("FAILED", "ERROR", "CANCELLED"):
+                    raise RuntimeError(f"The video provider reported {status}.")
+                self.remaining_time()
+                time.sleep(self.poll_interval)
+                polled = session.get(f"{base}{path}/status{query}", headers=request.headers, timeout=30)
+                hf_raise_for_status(polled)
+                status = polled.json().get("status")
+            result = session.get(f"{base}{path}{query}", headers=request.headers, timeout=60)
+            hf_raise_for_status(result)
+            url = result.json()["video"]["url"]
+            # The result is a public CDN link; the token is never sent there.
+            video = session.get(url, timeout=120)
+            hf_raise_for_status(video)
+            self.budget.record_media(reservation)
+        return video.content
+
+
+def _nearest(timestamps, t):
+    return min(timestamps, key=lambda ts: abs(ts - t))
+
+
+def generate_human_videos(record, staging, vlm, frames, full_frames, media, out_dir, max_segments):
+    """Generate the episode's videos (``max_segments`` 0 = every subtask); returns entries with local ``files``.
+
+    A failed segment is recorded and the others continue. Reaching the spending target or runtime
+    limit stops the remaining segments without failing the (already validated) annotations.
+    """
+    spans = subtask_spans(record, staging)
+    if max_segments:
+        spans = spans[:max_segments]
+    if not spans:
+        return []
+    out_dir.mkdir(parents=True, exist_ok=True)
+    ts = record.frame_timestamps
+    contexts = []
+    for span in spans:
+        samples = [span["start"] + (span["end"] - span["start"]) * i / 5 for i in range(6)]
+        contexts.append(frames.frames_at(record, [_nearest(ts, t) for t in samples]))
+    plans = vlm.generate_json(
+        [
+            [
+                {
+                    "role": "user",
+                    "content": [
+                        *to_image_blocks(ctx),
+                        {
+                            "type": "text",
+                            "text": PLAN_PROMPT.format(
+                                n=len(ctx),
+                                duration=span["end"] - span["start"],
+                                subtask=span["text"],
+                                task=record.episode_task,
+                            ),
+                        },
+                    ],
+                }
+            ]
+            for span, ctx in zip(spans, contexts, strict=True)
+        ]
+    )
+    stop = threading.Event()
+    profile = media.profile
+
+    def segment(k):
+        span, ctx, plan = spans[k], contexts[k], plans[k]
+        stem = out_dir / f"segment_{k:03d}"
+        entry = {
+            "source_episode_index": record.episode_index,
+            "segment_index": k,
+            "subtask": span["text"],
+            "start_timestamp": span["start"],
+            "end_timestamp": span["end"],
+            "provider": profile["provider"],
+            "edit_model": profile["edit_model"],
+            "video_model": profile["video_model"],
+            "resolution": profile["resolution"],
+            "files": {},
+        }
+        if stop.is_set():
+            entry["status"] = "skipped"
+            return entry
+        try:
+            if not isinstance(plan, dict) or not plan.get("edit_prompt"):
+                raise ValueError("The model returned no human-video plan.")
+            robot = full_frames.frames_at(record, [_nearest(ts, span["start"])])[0].convert("RGB")
+            robot.save(f"{stem}_robot_frame.png")
+            entry["files"]["robot_frame"] = f"{stem}_robot_frame.png"
+            entry.update(human_task=plan.get("human_task"), edit_prompt=plan["edit_prompt"])
+            edited = media.edit_image(robot, str(plan["edit_prompt"]))
+            edited.save(f"{stem}_first_frame.png")
+            entry["files"]["first_frame"] = f"{stem}_first_frame.png"
+            text = VIDEO_PROMPT.format(
+                human_task=plan.get("human_task", ""),
+                object=plan.get("object", ""),
+                target=plan.get("target", ""),
+                steps="; ".join(map(str, plan.get("steps") or [])),
+                final_state=plan.get("final_state", "the task completed"),
+                hand=plan.get("hand", "right"),
+                duration_s=profile["duration_s"],
+            )
+            content = [{"type": "image", "image": edited}, *to_image_blocks(ctx)]
+            messages = [{"role": "user", "content": [*content, {"type": "text", "text": text}]}]
+            written = vlm.generate_json([messages])[0]
+            if not isinstance(written, dict) or not written.get("video_prompt"):
+                raise ValueError("The model returned no video prompt.")
+            entry.update(video_prompt=written["video_prompt"], caption=written.get("caption"))
+            Path(f"{stem}.mp4").write_bytes(media.image_to_video(edited, str(written["video_prompt"])))
+            entry["files"]["video"] = f"{stem}.mp4"
+            entry["status"] = "completed"
+        except BudgetExceededError as exc:
+            stop.set()
+            entry["status"] = "time_stopped" if isinstance(exc, DeadlineExceededError) else "budget_stopped"
+        except Exception as exc:
+            entry.update(status="failed", error_type=type(exc).__name__)
+        return entry
+
+    with ThreadPoolExecutor(max_workers=min(len(spans), getattr(media, "concurrency", 1))) as pool:
+        return list(pool.map(segment, range(len(spans))))

model_catalog.py

@@ -206,6 +206,36 @@ MODELS = {
 
 UNAVAILABLE = []
 
+# Robot → human demonstration videos. Both media models run on fal through HF Inference Providers
+# and are billed per request to the signed-in account, so the Space reserves a fixed quoted price.
+HUMAN_VIDEO = {
+    "provider": "fal-ai",
+    "provider_name": "fal",
+    "edit_model": "Qwen/Qwen-Image-Edit-2509",
+    # Lower guidance keeps the scene's colours; fal's default oversaturates.
+    "edit_guidance_scale": 1.5,
+    "video_model": "MiniMaxAI/MiniMax-H3",
+    # MiniMax-H3 is mapped under the image-text-to-video task on fal.
+    "video_task": "image-text-to-video",
+    "resolution": "480P",
+    "duration_s": 5,
+    # $0.03 per megapixel; fal returns about 1 MP, reserved as 1.33 MP.
+    "edit_usd": 0.04,
+    # $0.05 per second at 480p; the clip is 5 s, reserved as 6 s.
+    "video_usd": 0.30,
+    # Default simultaneous media requests per job (user-selectable up to 32).
+    "concurrency": 8,
+    # Estimation only: the action-phase prompt yields about one subtask per 5 s of episode
+    # (the same rate the memory estimate uses), and a video takes up to ~3 min on fal.
+    "seconds_per_subtask": 5,
+    "minutes_per_video": 3,
+    "checked_at": "2026-09-30",
+    "pricing_url": "https://fal.ai/models/minimax/h3/image-to-video",
+}
+# 0 = one video for every subtask.
+HUMAN_VIDEO_SEGMENT_CHOICES = (0, 1, 2, 3, 5, 8)
+HUMAN_VIDEO_PARALLEL_CHOICES = (1, 4, 8, 16, 32)
+
 
 def model_profile(key=DEFAULT_MODEL):
     if key not in MODELS:
@@ -213,9 +243,23 @@ def model_profile(key=DEFAULT_MODEL):
     return deepcopy(MODELS[key])
 
 
+def human_video_profile():
+    return deepcopy(HUMAN_VIDEO)
+
+
+def human_video_segment_cost(profile=None):
+    profile = profile or HUMAN_VIDEO
+    return profile["edit_usd"] + profile["video_usd"]
+
+
 def public_models():
     return {
         "default": DEFAULT_MODEL,
         "models": [model_profile(k) for k in MODELS],
         "unavailable": deepcopy(UNAVAILABLE),
+        "human_video": {
+            **human_video_profile(),
+            "segment_choices": list(HUMAN_VIDEO_SEGMENT_CHOICES),
+            "parallel_choices": list(HUMAN_VIDEO_PARALLEL_CHOICES),
+        },
     }

provider.py

@@ -37,6 +37,8 @@ class Budget:
         self.context_length = context_length
         self.pricing_tiers = sorted(pricing_tiers or [], key=lambda t: t["min_input_tokens"])
         self.input_tokens = self.output_tokens = self.cached_tokens = self.attempts = self.rate_limits = 0
+        self.media_attempts = self.media_completed = 0
+        self.media_usd = 0.0
         self.lock = threading.Lock()
 
     def reserve(self, max_tokens):
@@ -54,6 +56,22 @@ class Budget:
         output_rate = max([self.output_rate] + [t["output_rate"] for t in self.pricing_tiers])
         return (self.context_length * input_rate + max_tokens * output_rate) / 1e6
 
+    def reserve_media(self, amount):
+        """Reserve a fixed-price media request (image edit or video). Never refunded."""
+        with self.lock:
+            if self.reserved + amount > self.limit:
+                raise BudgetExceededError("Spending target reached; completed results are saved.")
+            self.reserved += amount
+            self.media_attempts += 1
+        return amount
+
+    def record_media(self, amount):
+        """A media request returned its output; count its quoted price as measured."""
+        with self.lock:
+            self.measured += amount
+            self.media_usd += amount
+            self.media_completed += 1
+
     def reconcile(self, reservation, usage):
         if not isinstance(usage, dict):
             return  # Unknown charges keep the complete reservation.
@@ -94,6 +112,9 @@ class Budget:
                 "attempted_requests": self.attempts,
                 "requests_without_usage": self.attempts - self.calls,
                 "rate_limited_requests": self.rate_limits,
+                "media_requests": self.media_attempts,
+                "completed_media_requests": self.media_completed,
+                "media_usd": self.media_usd,
             }
 
 

server.py

@@ -4,6 +4,7 @@ import base64
 import hashlib
 import json
 import os
+import re
 import secrets
 import tempfile
 import threading
@@ -14,7 +15,7 @@ from urllib.parse import urlencode
 
 import requests
 from fastapi import FastAPI, HTTPException, Request
-from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
+from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
 from fastapi.staticfiles import StaticFiles
 from huggingface_hub import HfApi, hf_hub_download
 from model_catalog import DEFAULT_MODEL, public_models
@@ -43,12 +44,15 @@ class EstimateInput(DatasetInput):
     model: str = Field(default=DEFAULT_MODEL, min_length=1, max_length=80)
     start: int = Field(default=0, ge=0)
     count: int = Field(default=100, ge=0)
-    features: list[str] = Field(min_length=1, max_length=6)
+    features: list[str] = Field(min_length=1, max_length=7)
     camera: str | None = Field(default=None, max_length=200)
     hz: float = 0.1
     hours: int = 4
     mode: Literal["Annotated dataset copy"] = COPY
     prompts: dict[str, str] = Field(default_factory=dict, max_length=7)
+    # 0 = every subtask.
+    human_video_segments: int | None = Field(default=None, ge=0, le=8)
+    human_video_parallel: int = Field(default=8, ge=1, le=32)
 
 
 class PromptInput(BaseModel):
@@ -265,6 +269,8 @@ def calculate(options: EstimateInput, request: Request):
             current.get("token", False),
             prompts=options.prompts,
             model_key=options.model,
+            human_video_segments=options.human_video_segments,
+            human_video_parallel=options.human_video_parallel,
         )
     except Exception as exc:
         raise HTTPException(400, safe_error(exc)) from None
@@ -387,27 +393,40 @@ def cancel(job_id: str, request: Request):
     return {"ok": True}
 
 
+def studio_output(job_id, current):
+    """The job and its output dataset, only for Annotation Studio jobs in the user's namespace."""
+    api = HfApi(token=current["token"])
+    job = api.inspect_job(job_id=job_id, namespace=current["user"]["name"])
+    output = output_repo(job, current["user"]["name"])
+    if (job.labels or {}).get("app") != "lerobot-annotation-studio" or not output.startswith(
+        current["user"]["name"] + "/"
+    ):
+        raise ValueError("This job has no Annotation Studio result.")
+    return job, output
+
+
+def read_output_file(output, path, token):
+    with tempfile.TemporaryDirectory() as temp:
+        local = hf_hub_download(output, path, repo_type="dataset", token=token, cache_dir=temp)
+        return Path(local).read_bytes()
+
+
+HUMAN_VIDEO_FILE = re.compile(
+    r"annotation_studio/human_videos/episode_\d{6}/segment_\d{3}(_robot_frame\.png|_first_frame\.png|\.mp4)"
+)
+
+
 @app.get("/api/jobs/{job_id}/progress")
 def progress(job_id: str, request: Request):
     current = session(request, required=True)
     try:
-        api = HfApi(token=current["token"])
-        job = api.inspect_job(job_id=job_id, namespace=current["user"]["name"])
-        output = output_repo(job, current["user"]["name"])
-        if (job.labels or {}).get("app") != "lerobot-annotation-studio" or not output.startswith(
-            current["user"]["name"] + "/"
-        ):
-            raise ValueError("This job has no Annotation Studio result.")
-        with tempfile.TemporaryDirectory() as temp:
-            path = hf_hub_download(
-                output,
-                "annotation_studio/run.json",
-                repo_type="dataset",
-                token=current["token"],
-                cache_dir=temp,
-            )
-            report = json.loads(Path(path).read_text())
+        job, output = studio_output(job_id, current)
+        report = json.loads(read_output_file(output, "annotation_studio/run.json", current["token"]))
         episodes = report.get("episodes", [])
+        videos = {}
+        for episode in episodes:
+            for status, count in (episode.get("human_videos") or {}).items():
+                videos[status] = videos.get(status, 0) + count
         return {
             "status": "interrupted"
             if job.status.stage in {"COMPLETED", "ERROR", "CANCELED", "DELETED"}
@@ -423,11 +442,61 @@ def progress(job_id: str, request: Request):
             "annotation_url": annotation_url(report),
             "output_mode": report.get("manifest", {}).get("mode"),
             "model": report.get("manifest", {}).get("model"),
+            "human_videos": videos if (report.get("manifest", {}).get("human_video")) else None,
         }
     except Exception as exc:
         raise HTTPException(400, "No saved progress yet. The job may still be starting.") from exc
 
 
+@app.get("/api/jobs/{job_id}/human-videos")
+def human_videos(job_id: str, request: Request):
+    current = session(request, required=True)
+    try:
+        _, output = studio_output(job_id, current)
+        index = json.loads(
+            read_output_file(output, "annotation_studio/human_videos/index.json", current["token"])
+        )
+    except Exception as exc:
+        raise HTTPException(400, "No human demo videos saved yet.") from exc
+    media = f"/api/jobs/{job_id}/media?path="
+    return [
+        {
+            key: entry.get(key)
+            for key in (
+                "episode_index",
+                "source_episode_index",
+                "segment_index",
+                "subtask",
+                "human_task",
+                "caption",
+                "start_timestamp",
+                "end_timestamp",
+                "status",
+            )
+        }
+        | {
+            kind: media + entry[f"{kind}_path"]
+            for kind in ("robot_frame", "first_frame", "video")
+            if HUMAN_VIDEO_FILE.fullmatch(entry.get(f"{kind}_path") or "")
+        }
+        for entry in index[:60]
+    ]
+
+
+@app.get("/api/jobs/{job_id}/media")
+def human_video_media(job_id: str, path: str, request: Request):
+    """Stream a generated video or frame from the user's (possibly private) output dataset."""
+    current = session(request, required=True)
+    if not HUMAN_VIDEO_FILE.fullmatch(path):
+        raise HTTPException(400, "Unsupported media path.")
+    try:
+        _, output = studio_output(job_id, current)
+        content = read_output_file(output, path, current["token"])
+    except Exception as exc:
+        raise HTTPException(404, "Media not found.") from exc
+    return Response(content, media_type="video/mp4" if path.endswith(".mp4") else "image/png")
+
+
 @app.get("/health")
 def health():
     return {"status": "ok", "service": "lerobot-annotation-studio"}

studio.py

@@ -14,7 +14,14 @@ from pathlib import Path, PurePosixPath
 import pyarrow.parquet as pq
 from huggingface_hub import HfApi, hf_hub_download
 from huggingface_hub.utils import validate_repo_id
-from model_catalog import DEFAULT_MODEL, model_profile
+from model_catalog import (
+    DEFAULT_MODEL,
+    HUMAN_VIDEO_PARALLEL_CHOICES,
+    HUMAN_VIDEO_SEGMENT_CHOICES,
+    human_video_profile,
+    human_video_segment_cost,
+    model_profile,
+)
 from prompt_config import TEMPLATES, validate_prompts
 
 MODEL = model_profile()["route"]
@@ -26,7 +33,10 @@ FEATURES = {
     "Task rephrasings": "task_aug",
     "Interjections & speech": "interjection",
     "Visual questions": "vqa",
+    # Not a language style: videos are saved under annotation_studio/human_videos/.
+    "Human demo videos": "human_video",
 }
+HUMAN_VIDEOS = "Human demo videos"
 COPY = "Annotated dataset copy"
 EXPORT = "Annotation files only"  # Historical reports only; new runs always write Parquet.
 
@@ -151,11 +161,28 @@ def read_selection(dataset, start, count, token=False, revision=None, cache_dir=
         }
 
 
-def estimate(selection, features, vqa_hz, hours, cpu_rate, prompts=None, model_key=DEFAULT_MODEL):
+def estimate(
+    selection,
+    features,
+    vqa_hz,
+    hours,
+    cpu_rate,
+    prompts=None,
+    model_key=DEFAULT_MODEL,
+    human_video_segments=None,
+    human_video_parallel=None,
+):
     profile = model_profile(model_key)
     prompts = validate_prompts(prompts or {}, features)
     if not features or any(f not in FEATURES for f in features):
         raise ValueError("Choose at least one annotation type.")
+    if HUMAN_VIDEOS in features:
+        if "Subtasks" not in features:
+            raise ValueError("Human demo videos are generated per subtask; also select Subtasks.")
+        if human_video_segments not in HUMAN_VIDEO_SEGMENT_CHOICES:
+            raise ValueError("Choose a supported number of videos per episode.")
+        if human_video_parallel not in HUMAN_VIDEO_PARALLEL_CHOICES:
+            raise ValueError("Choose a supported number of parallel videos.")
     if hours not in (1, 4, 12) or vqa_hz not in (0.1, 0.5, 1.0):
         raise ValueError("Choose a supported timeout and visual-question frequency.")
     n, seconds = len(selection["episodes"]), selection["seconds"]
@@ -185,6 +212,24 @@ def estimate(selection, features, vqa_hz, hours, cpu_rate, prompts=None, model_k
         inputs += ticks * 1500
         outputs += ticks * 200
         calls += ticks
+    videos = videos_max = media = media_max = video_minutes = 0
+    if HUMAN_VIDEOS in features:
+        video_profile = human_video_profile()
+        # One video per subtask, and subtask counts scale with episode length: expect one per
+        # `seconds_per_subtask`, allow twice that in the upper bound, and apply the per-episode cap.
+        cap = human_video_segments or math.inf
+        per = video_profile["seconds_per_subtask"]
+        fps = selection["info"]["fps"]
+        lengths = [r["length"] / fps for r in selection["episodes"]]
+        videos = sum(min(cap, max(1.0, s / per)) for s in lengths)
+        videos_max = sum(min(cap, max(1, math.ceil(2 * s / per))) for s in lengths)
+        media = videos * human_video_segment_cost()
+        media_max = videos_max * human_video_segment_cost()
+        video_minutes = math.ceil(videos / human_video_parallel) * video_profile["minutes_per_video"]
+        # Two VLM calls per video: plan + edit prompt (6 frames), video prompt (edited frame + 6).
+        inputs += videos * (2600 + 3500)
+        outputs += videos * (300 + 350)
+        calls += videos * 2
     if "plan_render" in prompts:
         plan_calls = max(n, seconds / 5) + (n * 3 if "Interjections & speech" in features else 0)
         inputs += plan_calls * 1500
@@ -207,9 +252,16 @@ def estimate(selection, features, vqa_hz, hours, cpu_rate, prompts=None, model_k
     cpu_max = math.ceil(hours * 60 + 1) * cpu_rate
     # Range includes retries, grounding, variable segment counts and recovery.
     return {
-        "low": inference * 0.6 + cpu_max,
-        "high": inference * 2.5 + cpu_max,
+        "low": inference * 0.6 + media * 0.6 + cpu_max,
+        "high": inference * 2.5 + media_max + cpu_max,
         "inference": inference,
+        "media": media,
+        "media_max": media_max,
+        "videos": math.ceil(videos),
+        "videos_max": videos_max,
+        "video_minutes": video_minutes,
+        "video_parallel": human_video_parallel if videos else 0,
+        "segment_usd": human_video_segment_cost() if videos else 0,
         "cpu_max": cpu_max,
         "cpu_rate": cpu_rate,
         "calls": math.ceil(calls),
@@ -226,10 +278,34 @@ def estimate(selection, features, vqa_hz, hours, cpu_rate, prompts=None, model_k
     }
 
 
-def fingerprint(dataset, start, count, features, camera, vqa_hz, hours, mode, model_key=DEFAULT_MODEL):
+def fingerprint(
+    dataset,
+    start,
+    count,
+    features,
+    camera,
+    vqa_hz,
+    hours,
+    mode,
+    model_key=DEFAULT_MODEL,
+    human_video_segments=None,
+    human_video_parallel=None,
+):
     return hashlib.sha256(
         json.dumps(
-            [dataset.strip(), start, count, sorted(features), camera, vqa_hz, hours, mode, model_key],
+            [
+                dataset.strip(),
+                start,
+                count,
+                sorted(features),
+                camera,
+                vqa_hz,
+                hours,
+                mode,
+                model_key,
+                human_video_segments,
+                human_video_parallel,
+            ],
             sort_keys=True,
         ).encode()
     ).hexdigest()
@@ -253,9 +329,13 @@ def prepare(
     token=False,
     prompts=None,
     model_key=DEFAULT_MODEL,
+    human_video_segments=None,
+    human_video_parallel=None,
 ):
     profile = model_profile(model_key)
     prompts = validate_prompts(prompts or {}, features)
+    if HUMAN_VIDEOS not in features:
+        human_video_segments = human_video_parallel = None
     if mode != COPY:
         raise ValueError("Annotations must be saved in a LeRobot Parquet dataset. Refresh the Space.")
     selection = read_selection(dataset, start, count, token)
@@ -266,7 +346,17 @@ def prepare(
     hardware = next(h for h in HfApi(token=token or False).list_jobs_hardware() if h.name == "cpu-upgrade")
     if hardware.unit_label != "minute":
         raise ValueError("HF changed its billing units; refresh the Space before launching.")
-    costs = estimate(selection, features, vqa_hz, hours, hardware.unit_cost_usd, prompts, model_key)
+    costs = estimate(
+        selection,
+        features,
+        vqa_hz,
+        hours,
+        hardware.unit_cost_usd,
+        prompts,
+        model_key,
+        human_video_segments,
+        human_video_parallel,
+    )
     return {
         "selection": selection,
         "features": list(features),
@@ -276,12 +366,42 @@ def prepare(
         "prompts": prompts,
         "costs": costs,
         "model_profile": profile,
+        "human_video": {
+            # 0 = every subtask.
+            "segments_per_episode": human_video_segments,
+            "parallel": human_video_parallel,
+            "profile": human_video_profile(),
+        }
+        if HUMAN_VIDEOS in features
+        else None,
         "created": time.time(),
         "nonce": uuid.uuid4().hex,
-        "fingerprint": fingerprint(dataset, start, count, features, camera, vqa_hz, hours, mode, model_key),
+        "fingerprint": fingerprint(
+            dataset,
+            start,
+            count,
+            features,
+            camera,
+            vqa_hz,
+            hours,
+            mode,
+            model_key,
+            human_video_segments,
+            human_video_parallel,
+        ),
     }
 
 
+def provider_live(api, model_id, provider):
+    mapping = api.model_info(model_id, expand=["inferenceProviderMapping"]).inference_provider_mapping
+    hosted = (
+        mapping.get(provider)
+        if isinstance(mapping, dict)
+        else next((item for item in mapping or [] if item.provider == provider), None)
+    )
+    return hosted is not None and hosted.status == "live"
+
+
 def job_command(space, revision, timeout_minutes=None):
     # Only trusted deployment values enter shell syntax. User options travel in JSON env.
     validate_repo_id(space)
@@ -382,20 +502,23 @@ def launch(
         * max([profile["input_rate"]] + [t["input_rate"] for t in profile.get("pricing_tiers", [])])
         + 4096 * max([profile["output_rate"]] + [t["output_rate"] for t in profile.get("pricing_tiers", [])])
     ) / 1e6
+    human_video = plan.get("human_video")
+    if human_video:
+        if human_video["profile"] != human_video_profile():
+            raise ValueError("Video pricing changed. Calculate a fresh estimate.")
+        # One video must fit next to one conservative text reservation.
+        minimum_reservation += human_video_segment_cost()
     if budget - plan["costs"]["cpu_max"] < minimum_reservation:
         raise ValueError(
             f"This model needs at least ${minimum_reservation + plan['costs']['cpu_max']:.2f} for one conservative request reservation. Increase the target or choose a lower-cost model."
         )
-    mapping = api.model_info(
-        profile["model_id"], expand=["inferenceProviderMapping"]
-    ).inference_provider_mapping
-    hosted = (
-        mapping.get(profile["provider"])
-        if isinstance(mapping, dict)
-        else next((item for item in mapping or [] if item.provider == profile["provider"]), None)
-    )
-    if hosted is None or hosted.status != "live":
+    if not provider_live(api, profile["model_id"], profile["provider"]):
         raise ValueError("The selected model is temporarily unavailable from this provider.")
+    if human_video and not all(
+        provider_live(api, human_video["profile"][key], human_video["profile"]["provider"])
+        for key in ("edit_model", "video_model")
+    ):
+        raise ValueError("The video models are temporarily unavailable from fal.")
     space = space or os.environ.get("SPACE_ID")
     if not space:
         raise ValueError("Job launch is enabled only on the deployed Hugging Face Space.")

test_human_video.py

@@ -0,0 +1,485 @@
+"""Human demo videos: pricing, launch checks, generation and media access, without paid requests."""
+
+import json
+import sys
+import time
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).parent))
+import human_video  # noqa: E402
+import studio  # noqa: E402
+from model_catalog import human_video_profile, human_video_segment_cost  # noqa: E402
+from provider import Budget, BudgetExceededError, DeadlineExceededError  # noqa: E402
+
+from lerobot.annotations.steerable_pipeline.reader import EpisodeRecord
+from lerobot.annotations.steerable_pipeline.staging import EpisodeStaging
+from lerobot.annotations.steerable_pipeline.vlm_client import StubVlmClient
+
+CAMERA = "observation.images.top"
+PLAN = {
+    "human_task": "Pick up the cup with the right hand and place it on the table.",
+    "object": "white cup",
+    "target": "table",
+    "hand": "right",
+    "steps": ["grasp the cup", "place it"],
+    "final_state": "the cup on the table",
+    "edit_prompt": "Remove the robot and add two human hands resting at the bottom edge.",
+}
+
+
+def selection():
+    return {
+        "episodes": [{"episode_index": 0, "length": 300}, {"episode_index": 1, "length": 600}],
+        "seconds": 30,
+        "info": {"fps": 30},
+        "dataset": "source/robot",
+        "revision": "a" * 40,
+        "start": 0,
+        "count": 2,
+    }
+
+
+def video_estimate(selection_, segments=0, parallel=8, hours=4):
+    return studio.estimate(
+        selection_,
+        ["Subtasks", studio.HUMAN_VIDEOS],
+        0.1,
+        hours,
+        0.0005,
+        human_video_segments=segments,
+        human_video_parallel=parallel,
+    )
+
+
+def episodes(*seconds, fps=30):
+    return {
+        **selection(),
+        "episodes": [{"episode_index": i, "length": int(s * fps)} for i, s in enumerate(seconds)],
+        "seconds": sum(seconds),
+    }
+
+
+def test_video_estimate_scales_with_episode_length_and_count():
+    base = studio.estimate(selection(), ["Subtasks"], 0.1, 4, 0.0005)
+    assert base["media"] == 0 and base["videos"] == 0
+    # 10 s and 20 s episodes: about one subtask per 5 s -> 2 + 4 videos, up to 4 + 8.
+    every = video_estimate(selection())
+    assert every["videos"] == 6 and every["videos_max"] == 12
+    assert every["media"] == pytest.approx(6 * human_video_segment_cost())
+    assert every["media_max"] == pytest.approx(12 * human_video_segment_cost())
+    assert every["high"] - base["high"] > every["media_max"]
+    assert every["calls"] == base["calls"] + 12
+    # Longer episodes and more episodes cost more; very short episodes still get one video.
+    assert video_estimate(episodes(40, 40))["media"] > every["media"]
+    assert video_estimate(episodes(10, 20, 10, 20))["videos"] == 12
+    assert video_estimate(episodes(1, 2))["videos"] == 2
+    # A per-episode cap bounds both the expected and maximum count.
+    capped = video_estimate(selection(), segments=3)
+    assert capped["videos"] == 5 and capped["videos_max"] == 6
+
+
+def test_parallel_generations_shorten_the_video_time_estimate():
+    long = episodes(*[60] * 20)  # 240 videos expected
+    slow, fast = video_estimate(long, parallel=1), video_estimate(long, parallel=32)
+    assert slow["video_minutes"] == 240 * 3 and fast["video_minutes"] == 8 * 3
+    assert slow["media"] == fast["media"]
+    with pytest.raises(ValueError, match="parallel"):
+        video_estimate(selection(), parallel=64)
+
+
+def test_videos_need_subtasks_and_supported_choices():
+    with pytest.raises(ValueError, match="Subtasks"):
+        studio.estimate(
+            selection(), [studio.HUMAN_VIDEOS], 0.1, 4, 0.0005, human_video_segments=0, human_video_parallel=8
+        )
+    with pytest.raises(ValueError, match="videos per episode"):
+        video_estimate(selection(), segments=4)
+
+
+def video_plan(profile=None):
+    features = ["Subtasks", studio.HUMAN_VIDEOS]
+    return {
+        "selection": selection(),
+        "features": features,
+        "camera": CAMERA,
+        "vqa_hz": 0.1,
+        "mode": studio.COPY,
+        "created": time.time(),
+        "nonce": "request123",
+        "human_video": {
+            "segments_per_episode": 2,
+            "parallel": 8,
+            "profile": profile or human_video_profile(),
+        },
+        "costs": studio.estimate(
+            selection(), features, 0.1, 4, 0.0005, human_video_segments=2, human_video_parallel=8
+        ),
+    }
+
+
+def api_mock(fal_status="live"):
+    api = Mock()
+    api.whoami.return_value = {"name": "alice"}
+    api.list_jobs.return_value = []
+    api.repo_exists.return_value = False
+    api.dataset_info.return_value = SimpleNamespace(id="source/robot", sha="a" * 40)
+    api.space_info.return_value = SimpleNamespace(sha="b" * 40)
+    api.list_jobs_hardware.return_value = [
+        SimpleNamespace(name="cpu-upgrade", unit_label="minute", unit_cost_usd=0.0005)
+    ]
+    api.model_info.return_value = SimpleNamespace(
+        inference_provider_mapping={
+            studio.PROVIDER: SimpleNamespace(status="live"),
+            "fal-ai": SimpleNamespace(status=fal_status),
+        }
+    )
+    api.run_job.return_value = SimpleNamespace(id="job123")
+    return api
+
+
+def launch(plan, api, budget=20):
+    return studio.launch(
+        plan,
+        "alice/out",
+        budget,
+        True,
+        "token",
+        space="lerobot/annotation-studio",
+        api_factory=lambda **_: api,
+    )
+
+
+def test_launch_checks_fal_models_and_pins_video_settings():
+    api = api_mock()
+    assert launch(video_plan(), api)["id"] == "job123"
+    checked = [call.args[0] for call in api.model_info.call_args_list]
+    assert {"MiniMaxAI/MiniMax-H3", "Qwen/Qwen-Image-Edit-2509"} <= set(checked)
+    manifest = json.loads(api.run_job.call_args.kwargs["env"]["ANNOTATION_MANIFEST"])
+    assert manifest["human_video"]["segments_per_episode"] == 2
+    assert manifest["human_video"]["profile"]["resolution"] == "480P"
+    # The job downloads the human-video prompt files with the other *.txt resources.
+    assert "*.txt" in str(api.run_job.call_args.kwargs["command"])
+
+
+def test_launch_refuses_offline_fal_models_or_stale_video_prices():
+    api = api_mock(fal_status="staging")
+    with pytest.raises(ValueError, match="fal"):
+        launch(video_plan(), api)
+    stale = {**human_video_profile(), "video_usd": 0.01}
+    with pytest.raises(ValueError, match="Video pricing"):
+        launch(video_plan(stale), api_mock())
+    api.run_job.assert_not_called()
+
+
+def test_media_budget_reserves_before_request_and_stops_at_target():
+    budget = Budget(0.5, 0.15, 0.5)
+    reservation = budget.reserve_media(0.3)
+    budget.record_media(reservation)
+    assert budget.summary()["media_usd"] == pytest.approx(0.3)
+    assert budget.summary()["measured_inference_usd"] == pytest.approx(0.3)
+    with pytest.raises(BudgetExceededError):
+        budget.reserve_media(0.3)
+    assert budget.summary()["media_requests"] == 1
+
+
+class FakeMedia:
+    concurrency = 4
+
+    def __init__(self, fail_first=False, budget_after=None):
+        self.profile = human_video_profile()
+        self.fail_first, self.budget_after = fail_first, budget_after
+        self.edits, self.videos = [], []
+
+    def edit_image(self, image, prompt):
+        self.edits.append((image.size, prompt))
+        return image.transpose(Image.FLIP_LEFT_RIGHT)
+
+    def image_to_video(self, image, prompt):
+        self.videos.append(prompt)
+        if self.fail_first and len(self.videos) == 1:
+            raise RuntimeError("provider error")
+        if self.budget_after is not None and len(self.videos) > self.budget_after:
+            raise BudgetExceededError("Spending target reached.")
+        return b"\x00\x00\x00\x18ftypmp42video"
+
+
+def respond(messages):
+    text = " ".join(p.get("text", "") for m in messages for p in m["content"] if isinstance(p, dict))
+    if "meaningful action phases" in text:
+        return {
+            "subtasks": [
+                {"text": "grasp the cup", "start": 0, "end": 2.5},
+                {"text": "place the cup on table", "start": 2.5, "end": 3.9},
+            ]
+        }
+    if "preparing a human demonstration" in text:
+        return PLAN
+    if "image-to-video model will animate it" in text:
+        return {"video_prompt": "From a fixed first-person camera perspective, ...", "caption": "A hand."}
+    raise AssertionError("Unexpected prompt")
+
+
+@pytest.fixture
+def episode(tmp_path):
+    import av
+    import numpy as np
+
+    video = tmp_path / "videos" / CAMERA / "chunk-000" / "file-000.mp4"
+    video.parent.mkdir(parents=True)
+    with av.open(str(video), "w") as container:
+        stream = container.add_stream("mpeg4", rate=10)
+        stream.width, stream.height, stream.pix_fmt = 320, 240, "yuv420p"
+        for i in range(40):
+            frame = av.VideoFrame.from_ndarray(np.full((240, 320, 3), i * 5, dtype=np.uint8), format="rgb24")
+            for packet in stream.encode(frame):
+                container.mux(packet)
+        for packet in stream.encode():
+            container.mux(packet)
+    info = {"fps": 10, "video_path": "videos/{video_key}/chunk-{chunk_index:03d}/file-{file_index:03d}.mp4"}
+    row = {
+        "length": 40,
+        f"videos/{CAMERA}/chunk_index": 0,
+        f"videos/{CAMERA}/file_index": 0,
+        f"videos/{CAMERA}/from_timestamp": 0.0,
+    }
+    record = EpisodeRecord(
+        0,
+        "move the cup to the table",
+        tuple(i / 10 for i in range(40)),
+        tuple(range(40)),
+        tmp_path / "x",
+        0,
+        40,
+    )
+    return record, row, info
+
+
+def annotate(tmp_path, episode, media, segments=2, staging="staging"):
+    from worker import annotate_episode
+
+    record, row, info = episode
+    manifest = {
+        "camera": CAMERA,
+        "features": ["Subtasks", studio.HUMAN_VIDEOS],
+        "vqa_hz": 0.1,
+        "prompts": {},
+        "human_video": {"segments_per_episode": segments, "profile": human_video_profile()},
+    }
+    return annotate_episode(
+        record, row, info, tmp_path, manifest, StubVlmClient(respond), tmp_path / staging, media
+    )
+
+
+def test_episode_generates_videos_after_annotations_from_full_resolution_frames(tmp_path, episode):
+    media = FakeMedia()
+    result = annotate(tmp_path, episode, media)
+    assert result["status"] == "completed" and result["human_videos"] == {"completed": 2}
+    entries = result["human_video_entries"]
+    assert [e["subtask"] for e in entries] == ["grasp the cup", "place the cup on table"]
+    assert entries[1]["start_timestamp"] == 2.5 and entries[1]["end_timestamp"] == 3.9
+    # The edit uses the camera's full resolution, not the 224 px VLM thumbnails.
+    assert {size for size, _ in media.edits} == {(320, 240)}
+    for entry in entries:
+        assert set(entry["files"]) == {"robot_frame", "first_frame", "video"}
+        assert Path(entry["files"]["video"]).read_bytes().startswith(b"\x00\x00\x00\x18ftyp")
+    # Only language styles are staged for the parquet writer.
+    staged = EpisodeStaging(tmp_path / "staging", 0).read_all()
+    assert {r.get("style") for rows in staged.values() for r in rows} == {"subtask"}
+
+
+def test_failed_video_is_recorded_and_budget_stop_keeps_annotations(tmp_path, episode):
+    result = annotate(tmp_path, episode, FakeMedia(fail_first=True))
+    assert sorted(e["status"] for e in result["human_video_entries"]) == ["completed", "failed"]
+
+    stopped = annotate(tmp_path, episode, FakeMedia(budget_after=0), segments=1, staging="second")
+    assert stopped["status"] == "completed"
+    assert stopped["human_videos"] == {"budget_stopped": 1}
+
+
+def test_segment_cap_limits_videos_and_zero_means_every_subtask(tmp_path, episode):
+    media = FakeMedia()
+    result = annotate(tmp_path, episode, media, segments=1)
+    assert len(result["human_video_entries"]) == 1 and len(media.videos) == 1
+    every = annotate(tmp_path, episode, FakeMedia(), segments=0, staging="every")
+    assert len(every["human_video_entries"]) == 2
+
+
+def test_parallel_setting_bounds_simultaneous_media_requests():
+    import threading
+
+    budget = Budget(5, 0.15, 0.5)
+    assert human_video.FalMedia("t", budget, lambda: 3600).concurrency == human_video_profile()["concurrency"]
+    assert human_video.FalMedia("t", budget, lambda: 3600, concurrency=64).concurrency == 32
+    media = human_video.FalMedia("t", budget, lambda: 3600, concurrency=3)
+    active, peak, lock = [0], [0], threading.Lock()
+
+    def request():
+        with media.semaphore:
+            with lock:
+                active[0] += 1
+                peak[0] = max(peak[0], active[0])
+            time.sleep(0.02)
+            with lock:
+                active[0] -= 1
+
+    threads = [threading.Thread(target=request) for _ in range(12)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert peak[0] == 3
+
+
+def test_operations_upload_media_and_index(tmp_path):
+    from worker import HUMAN_VIDEO_DIR, human_video_operations
+
+    video = tmp_path / "segment_000.mp4"
+    video.write_bytes(b"video")
+    entries = [
+        {"source_episode_index": 7, "segment_index": 0, "status": "completed", "files": {"video": str(video)}}
+    ]
+    index = []
+    operations = human_video_operations(entries, index)
+    assert [op.path_in_repo for op in operations] == [
+        f"{HUMAN_VIDEO_DIR}/episode_000007/segment_000.mp4",
+        f"{HUMAN_VIDEO_DIR}/index.json",
+    ]
+    assert index == [
+        {
+            "source_episode_index": 7,
+            "segment_index": 0,
+            "status": "completed",
+            "video_path": f"{HUMAN_VIDEO_DIR}/episode_000007/segment_000.mp4",
+        }
+    ]
+
+
+class FakeSession:
+    def __init__(self, statuses):
+        self.statuses, self.calls = list(statuses), []
+
+    def post(self, url, json=None, headers=None, timeout=None):
+        self.calls.append(("post", url, headers))
+        return SimpleNamespace(
+            status_code=200,
+            json=lambda: {
+                "request_id": "r1",
+                "status": "IN_QUEUE",
+                "response_url": "https://queue.fal.run/minimax/h3/requests/r1",
+            },
+        )
+
+    def get(self, url, headers=None, timeout=None):
+        self.calls.append(("get", url, headers))
+        if url.endswith("/status?_subdomain=queue"):
+            status = self.statuses.pop(0)
+            return SimpleNamespace(status_code=200, json=lambda: {"status": status})
+        if url.startswith("https://router.huggingface.co"):
+            return SimpleNamespace(
+                status_code=200, json=lambda: {"video": {"url": "https://cdn.fal.media/v.mp4"}}
+            )
+        return SimpleNamespace(status_code=200, content=b"mp4-bytes")
+
+
+@pytest.fixture
+def fal(monkeypatch):
+    import huggingface_hub.utils
+    from huggingface_hub.inference._providers.fal_ai import FalAIImageToVideoTask
+
+    seen = {}
+
+    def prepare_request(self, *, inputs, parameters, headers, model, api_key):
+        seen.update(task=self.task, parameters=parameters, model=model)
+        return SimpleNamespace(
+            url="https://router.huggingface.co/fal-ai/minimax/h3/image-to-video?_subdomain=queue",
+            json={"prompt": parameters["prompt"]},
+            headers={"authorization": f"Bearer {api_key}"},
+        )
+
+    monkeypatch.setattr(FalAIImageToVideoTask, "prepare_request", prepare_request)
+    monkeypatch.setattr(huggingface_hub.utils, "hf_raise_for_status", lambda response: None)
+
+    def use(session):
+        monkeypatch.setattr(huggingface_hub.utils, "get_session", lambda: session)
+
+    return seen, use
+
+
+def test_video_polls_router_charges_quoted_price_and_never_sends_token_to_cdn(fal):
+    seen, use = fal
+    session = FakeSession(["IN_PROGRESS", "COMPLETED"])
+    use(session)
+    budget = Budget(5, 0.15, 0.5)
+    media = human_video.FalMedia("hf_token", budget, lambda: 3600, poll_interval=0)
+    assert media.image_to_video(Image.new("RGB", (64, 48)), "a prompt") == b"mp4-bytes"
+    assert seen["task"] == "image-text-to-video" and seen["model"] == "MiniMaxAI/MiniMax-H3"
+    assert seen["parameters"] == {"prompt": "a prompt", "resolution": "480P", "duration": 5}
+    status_url = "https://router.huggingface.co/fal-ai/minimax/h3/requests/r1/status?_subdomain=queue"
+    assert [c[1] for c in session.calls if c[0] == "get"][:2] == [status_url, status_url]
+    cdn = [c for c in session.calls if c[1].startswith("https://cdn.fal.media")]
+    assert cdn and cdn[0][2] is None
+    assert budget.summary()["media_usd"] == pytest.approx(human_video_profile()["video_usd"])
+
+
+def test_video_is_not_started_without_runtime_for_it(fal):
+    _, use = fal
+    session = FakeSession([])
+    use(session)
+    budget = Budget(5, 0.15, 0.5)
+    media = human_video.FalMedia("hf_token", budget, lambda: 60, poll_interval=0)
+    with pytest.raises(DeadlineExceededError):
+        media.image_to_video(Image.new("RGB", (64, 48)), "a prompt")
+    assert session.calls == [] and budget.summary()["media_requests"] == 0
+
+
+def test_media_endpoint_serves_only_generated_files(monkeypatch):
+    from fastapi.testclient import TestClient
+
+    import server
+
+    server.SESSIONS.clear()
+    client = TestClient(server.app)
+    sid_response = client.get("/api/session")
+    sid = client.cookies.get(server.COOKIE)
+    server.SESSIONS[sid].update(user={"name": "alice"}, token="secret-token")
+    assert sid_response.status_code == 200
+    reads = []
+    monkeypatch.setattr(server, "studio_output", lambda job_id, current: (None, "alice/out"))
+
+    def read(output, path, token):
+        reads.append((output, path, token))
+        if path.endswith("index.json"):
+            return json.dumps(
+                [
+                    {
+                        "source_episode_index": 3,
+                        "episode_index": 0,
+                        "segment_index": 0,
+                        "status": "completed",
+                        "subtask": "grasp",
+                        "video_path": "annotation_studio/human_videos/episode_000003/segment_000.mp4",
+                        "edit_prompt": "not exposed in the preview",
+                    }
+                ]
+            ).encode()
+        return b"mp4"
+
+    monkeypatch.setattr(server, "read_output_file", read)
+    listed = client.get("/api/jobs/job1/human-videos").json()
+    assert listed[0]["video"] == (
+        "/api/jobs/job1/media?path=annotation_studio/human_videos/episode_000003/segment_000.mp4"
+    )
+    assert "edit_prompt" not in listed[0]
+    media = client.get(listed[0]["video"])
+    assert media.status_code == 200 and media.content == b"mp4"
+    assert media.headers["content-type"] == "video/mp4"
+    for bad in ("annotation_studio/run.json", "data/chunk-000/file-000.parquet", "../x.mp4"):
+        assert client.get(f"/api/jobs/job1/media?path={bad}").status_code == 400
+    assert "secret-token" not in json.dumps(listed)
+    anonymous = TestClient(server.app).get("/api/jobs/job1/media?path=" + listed[0]["video"].split("=")[1])
+    assert anonymous.status_code == 401

test_studio.py

@@ -111,7 +111,7 @@ def test_real_video_pipeline_and_checkbox_filtering(tmp_path, features, custom):
     )
     staged = EpisodeStaging(tmp_path / "staging", 0).read_all()
     actual_styles = {r.get("style") for rows in staged.values() for r in rows}
-    expected = {studio.FEATURES[f] for f in features}
+    expected = {studio.FEATURES[f] for f in features if f != studio.HUMAN_VIDEOS}
     if "Interjections & speech" in features:
         expected.add(None)
     assert actual_styles == expected

test_subset.py

@@ -268,7 +268,7 @@ def test_worker_publishes_only_successful_subset_without_hub_duplication(
     monkeypatch.setattr(worker, "hf_hub_download", download)
     monkeypatch.setattr(worker, "Provider", lambda *args, **kw: SimpleNamespace(concurrency=2))
 
-    def annotate(record, row, info, root, manifest, provider, staging):
+    def annotate(record, row, info, root, manifest, provider, staging, media=None):
         if record.episode_index == 1:
             raise ValueError("Simulated annotation validation failure")
         EpisodeStaging(staging, record.episode_index).write(

worker.py

@@ -15,9 +15,10 @@ from pathlib import Path
 import av
 import pyarrow.parquet as pq
 from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download, snapshot_download
+from human_video import FalMedia, generate_human_videos
 from prompt_config import validate_prompts
 from provider import Budget, BudgetExceededError, DeadlineExceededError, Provider
-from studio import COPY, FEATURES, read_selection, relative_path
+from studio import COPY, FEATURES, HUMAN_VIDEOS, read_selection, relative_path
 from subset import EpisodeSubset
 
 from lerobot.annotations.steerable_pipeline.config import InterjectionsConfig, PlanConfig, VqaConfig
@@ -127,9 +128,11 @@ class TunedPlan(plan_module.PlanSubtasksMemoryModule):
 class SourceFrames:
     """Exact timestamp seeks; a missing image fails the episode instead of generating from text."""
 
-    def __init__(self, row, info, root, camera):
+    def __init__(self, row, info, root, camera, max_size=(224, 224)):
         self.row, self.info, self.root = row, info, root
         self.camera_key, self.camera_keys, self.cache = camera, [camera], {}
+        # None keeps full resolution (the first frame of a human demo video).
+        self.max_size = max_size
 
     def frames_at(self, record, timestamps, camera_key=None):
         camera = camera_key or self.camera_key
@@ -159,7 +162,8 @@ class SourceFrames:
                     if closest is None or distance > 1 / fps + 1e-4:
                         raise RuntimeError("Source video is missing an expected frame.")
                     image = closest.to_image()
-                    image.thumbnail((224, 224))
+                    if self.max_size:
+                        image.thumbnail(self.max_size)
                     self.cache[idx] = image
         return [self.cache[i] for i in indices]
 
@@ -197,21 +201,29 @@ class PreservingWriter(LanguageColumnsWriter):
 
 
 def requested_styles(features):
-    styles = {FEATURES[f] for f in features}
+    # Human demo videos are media files, not language rows.
+    styles = {FEATURES[f] for f in features if f != HUMAN_VIDEOS}
     if "Interjections & speech" in features:
         styles.add(None)
     return styles
 
 
-def annotate_episode(record, row, info, root, manifest, provider, staging_dir):
+def annotate_episode(record, row, info, root, manifest, provider, staging_dir, media=None):
     context = CUSTOM_PROMPTS.set(validate_prompts(manifest.get("prompts", {}), manifest["features"]))
     try:
-        return _annotate_episode(record, row, info, root, manifest, provider, staging_dir)
+        return _annotate_episode(record, row, info, root, manifest, provider, staging_dir, media)
     finally:
         CUSTOM_PROMPTS.reset(context)
 
 
-def _annotate_episode(record, row, info, root, manifest, provider, staging_dir):
+def human_video_summary(entries):
+    counts = defaultdict(int)
+    for entry in entries:
+        counts[entry["status"]] += 1
+    return dict(counts)
+
+
+def _annotate_episode(record, row, info, root, manifest, provider, staging_dir, media=None):
     frames = SourceFrames(row, info, root, manifest["camera"])
     features = manifest["features"]
     staging = EpisodeStaging(staging_dir, record.episode_index)
@@ -252,15 +264,62 @@ def _annotate_episode(record, row, info, root, manifest, provider, staging_dir):
     report = StagingValidator(dataset_camera_keys=(manifest["camera"],)).validate([record], staging_dir)
     if not report.ok:
         raise ValueError("Generated annotations did not pass pipeline validation.")
+    videos = None
+    if HUMAN_VIDEOS in features and media is not None:
+        # After validation, so no video is paid for on an episode whose annotations fail.
+        full_frames = SourceFrames(row, info, root, manifest["camera"], max_size=None)
+        try:
+            videos = generate_human_videos(
+                record,
+                staging,
+                provider,
+                frames,
+                full_frames,
+                media,
+                staging_dir.parent / "human_videos" / f"episode_{record.episode_index:06d}",
+                manifest["human_video"]["segments_per_episode"],
+            )
+        except BudgetExceededError as exc:
+            # The annotations are complete; only the videos stop.
+            status = "time_stopped" if isinstance(exc, DeadlineExceededError) else "budget_stopped"
+            videos = [{"source_episode_index": record.episode_index, "status": status, "files": {}}]
     styles = requested_styles(features)
     for module_name, rows in staging.read_all().items():
         staging.write(module_name, [r for r in rows if r.get("style") in styles])
-    return {
+    result = {
         "episode_index": record.episode_index,
         "status": "completed",
         "recovery_used": getattr(plan, "used_fallback", False),
         "rows": sum(len(rows) for rows in staging.read_all().values()),
     }
+    if videos is not None:
+        result["human_videos"] = human_video_summary(videos)
+        result["human_video_entries"] = videos
+    return result
+
+
+HUMAN_VIDEO_DIR = "annotation_studio/human_videos"
+
+
+def human_video_operations(entries, index):
+    """Upload an episode's media and the updated index (source episode IDs until materialized)."""
+    operations = []
+    for entry in entries:
+        files = entry.pop("files", {})
+        folder = f"{HUMAN_VIDEO_DIR}/episode_{entry['source_episode_index']:06d}"
+        for kind, local in files.items():
+            name = f"{folder}/{Path(local).name}"
+            operations.append(CommitOperationAdd(path_in_repo=name, path_or_fileobj=local))
+            entry[f"{kind}_path"] = name
+        index.append(entry)
+    operations.append(human_video_index_operation(index))
+    return operations
+
+
+def human_video_index_operation(index):
+    return CommitOperationAdd(
+        path_in_repo=f"{HUMAN_VIDEO_DIR}/index.json", path_or_fileobj=json.dumps(index, indent=2).encode()
+    )
 
 
 def create_output_repo(api, manifest):
@@ -286,6 +345,21 @@ def create_output_repo(api, manifest):
         f"at `{manifest['dataset_revision']}`. {description} "
         "See `annotation_studio/run.json` for completeness.\n"
     )
+    if HUMAN_VIDEOS in manifest.get("features", ()):
+        profile = manifest["human_video"]["profile"]
+        card += (
+            "\n## Human demonstration videos\n\n"
+            "`annotation_studio/human_videos/` holds a generated first-person human video for "
+            + (
+                f"up to {manifest['human_video']['segments_per_episode']} subtasks"
+                if manifest["human_video"]["segments_per_episode"]
+                else "every subtask"
+            )
+            + " per episode: the robot frame, "
+            + f"the {profile['edit_model']} edit and a {profile['resolution']} {profile['video_model']} "
+            "video. `index.json` lists the subtask, timestamps, prompts and status of every segment. "
+            "These videos are synthetic.\n"
+        )
     api.upload_file(
         path_or_fileobj=card.encode(), path_in_repo="README.md", repo_id=output, repo_type="dataset"
     )
@@ -344,6 +418,18 @@ def run(manifest, work=Path("/workspace/run")):
         concurrency=manifest.get("request_concurrency", 2),
         generation_options=manifest.get("generation_options"),
     )
+    media = (
+        FalMedia(
+            token,
+            budget,
+            provider.remaining_time,
+            manifest["human_video"]["profile"],
+            concurrency=manifest["human_video"].get("parallel"),
+        )
+        if HUMAN_VIDEOS in manifest["features"] and manifest.get("human_video")
+        else None
+    )
+    human_index = []
     progress = {
         "manifest": manifest,
         "pipeline": json.loads(Path(__file__).with_name("pipeline-version.json").read_text()),
@@ -435,6 +521,7 @@ def run(manifest, work=Path("/workspace/run")):
                         manifest,
                         provider,
                         staging,
+                        media,
                     ): record
                     for record in records
                 }
@@ -461,8 +548,15 @@ def run(manifest, work=Path("/workspace/run")):
                             "status": "failed",
                             "error_type": type(exc).__name__,
                         }
+                    entries = result.pop("human_video_entries", None)
                     progress["episodes"].append(result)
-                    checkpoint()
+                    if entries:
+                        local_files = [f for e in entries for f in e.get("files", {}).values()]
+                        checkpoint(human_video_operations(entries, human_index))
+                        for local in local_files:
+                            Path(local).unlink(missing_ok=True)
+                    else:
+                        checkpoint()
                     print(f"Saved episode {record.episode_index}: {result['status']}", flush=True)
             if completed:
                 writer = PreservingWriter(
@@ -498,6 +592,11 @@ def run(manifest, work=Path("/workspace/run")):
                             local_dir=root,
                         )
                     new_id, written = subset.add(record.episode_index, annotated_table)
+                    extra_operations = []
+                    for entry in human_index:
+                        if entry["source_episode_index"] == record.episode_index:
+                            entry["episode_index"] = new_id
+                            extra_operations = [human_video_index_operation(human_index)]
                     with commit_lock:
                         previous_ids = list(progress["written_episode_ids"])
                         previous_map = list(progress["source_episode_map"])
@@ -511,6 +610,7 @@ def run(manifest, work=Path("/workspace/run")):
                                     )
                                     for p in [*written, *attribution]
                                 ]
+                                + extra_operations
                             )
                         except Exception:
                             progress["written_episode_ids"] = previous_ids