Add optional independent per-camera subtask timelines

README.md

@@ -449,3 +449,25 @@ Hand-annotated copies offer **View 3D hands** once an episode is materialized,
 opening synchronized wrists, hand skeletons and camera overlays with `?tab=hands`.
 Operators can set `HAND_VISUALIZER_URL` to a compatible visualizer deployment;
 the default is the official LeRobot dataset visualizer.
+
+## Per-camera subtasks
+
+In the dataset step, select **Separate subtasks per camera** and choose the
+camera views to annotate. The primary camera is always included. Each selected
+view is processed independently from its own images on the original episode
+clock. The estimate includes the additional subtask model calls; the same job
+budget still applies. Plans, memory, hand annotations and generated videos
+continue to use the primary view. Default jobs annotate the primary view only.
+
+The output preserves the compatible primary `language_persistent` timeline and
+also writes `annotation_studio/subtasks_by_camera/episode_XXXXXX.parquet`. Each
+row contains `episode_index`, `source_episode_index`, `camera`,
+`start_timestamp`, `end_timestamp` and `subtask`. These are the final validated
+pipeline spans, with no timestamp rounding in the sidecar. Canonical LeRobot
+subtask language rows remain camera-independent; camera-specific labels are
+kept in this explicit sidecar instead of changing that schema.
+
+In Jobs, **Preview camera subtask timelines** shows each view on a shared time
+slider, with an episode selector and a Parquet download. Only successfully
+exported episodes appear. This is camera-scoped labeling, not left/right-hand
+action attribution. Existing jobs and completed outputs are not reprocessed.

frontend/src/JobsView.tsx

@@ -7,6 +7,7 @@ import {
   RefreshCw,
 } from "lucide-react";
 import "./jobs.css";
+import SubtaskViews from "./SubtaskViews";
 
 export type Job = {
   id: string;
@@ -55,6 +56,8 @@ export type Progress = {
   annotation_url?: string | null;
   hand_visualizer_url?: string | null;
   output_mode?: string;
+  subtask_views?: "primary" | "per-camera";
+  subtask_cameras?: string[];
   human_videos?: Record<string, number> | null;
 };
 type HumanVideo = {
@@ -380,6 +383,9 @@ export default function JobsView({
                       </div>
                     )}
                   </dl>
+                  {p?.subtask_views === "per-camera" && !!p.completed && (
+                    <SubtaskViews jobId={job.id} />
+                  )}
                   {!!p?.human_videos?.completed && (
                     <HumanVideos jobId={job.id} />
                   )}

frontend/src/SubtaskViews.tsx

@@ -0,0 +1,132 @@
+import { useState } from "react";
+
+type Span = { start: number; end: number; text: string };
+type Episode = {
+  source_episode_index: number;
+  episode_index: number;
+  duration: number;
+  cameras: Record<string, Span[]>;
+};
+type Results = { output_repo: string; episodes: Episode[] };
+
+export default function SubtaskViews({ jobId }: { jobId: string }) {
+  const [results, setResults] = useState<Results | null>(null);
+  const [loading, setLoading] = useState(false);
+  const [error, setError] = useState("");
+  const [episodeIndex, setEpisodeIndex] = useState(0);
+  const [time, setTime] = useState(0);
+  async function load() {
+    setLoading(true);
+    setError("");
+    try {
+      const response = await fetch(
+        `/api/jobs/${encodeURIComponent(jobId)}/subtask-views`,
+        { credentials: "same-origin" },
+      );
+      if (!response.ok)
+        throw new Error("No exported camera timelines available yet.");
+      setResults(await response.json());
+    } catch (reason) {
+      setError((reason as Error).message);
+    } finally {
+      setLoading(false);
+    }
+  }
+  const episode = results?.episodes[episodeIndex];
+  return (
+    <section className="view-subtasks" aria-label="Camera subtask timelines">
+      <button className="text-action" onClick={load} disabled={loading}>
+        {loading
+          ? "Loading camera timelines…"
+          : results
+            ? "Refresh camera timelines"
+            : "Preview camera subtask timelines"}
+      </button>
+      {error && <p role="alert">{error}</p>}
+      {results && !episode && <p>No exported camera timelines yet.</p>}
+      {episode && (
+        <>
+          <div className="view-subtask-controls">
+            <label>
+              Episode{" "}
+              <select
+                value={episodeIndex}
+                onChange={(e) => {
+                  setEpisodeIndex(Number(e.target.value));
+                  setTime(0);
+                }}
+              >
+                {results!.episodes.map((ep, i) => (
+                  <option key={ep.episode_index} value={i}>
+                    {ep.episode_index} · source {ep.source_episode_index}
+                  </option>
+                ))}
+              </select>
+            </label>
+            <span>
+              {time.toFixed(2)} / {episode.duration.toFixed(2)} s
+            </span>
+            <a
+              href={`https://huggingface.co/datasets/${results!.output_repo}/resolve/main/annotation_studio/subtasks_by_camera/episode_${String(episode.episode_index).padStart(6, "0")}.parquet`}
+              target="_blank"
+              rel="noreferrer"
+            >
+              Download camera spans ↗
+            </a>
+          </div>
+          <input
+            aria-label="Camera timeline time"
+            type="range"
+            min={0}
+            max={episode.duration}
+            step={0.001}
+            value={time}
+            onChange={(e) => setTime(Number(e.target.value))}
+          />
+          {Object.entries(episode.cameras).map(([camera, spans], index) => {
+            const active = spans.find((s) => time >= s.start && time < s.end);
+            return (
+              <div
+                className="view-subtask-row"
+                key={`${episode.episode_index}-${camera}`}
+              >
+                <strong>{camera}</strong>
+                <p>{active?.text ?? "No subtask label at this timestamp"}</p>
+                <div
+                  className="view-subtask-lane"
+                  aria-label={`${camera} timeline`}
+                >
+                  {spans.map((span, i) => (
+                    <button
+                      key={i}
+                      className={span === active ? "active" : ""}
+                      style={{
+                        left: `${(100 * span.start) / episode.duration}%`,
+                        width: `${(100 * (span.end - span.start)) / episode.duration}%`,
+                        background: [
+                          "#f4b942",
+                          "#56b4d3",
+                          "#75b87a",
+                          "#b493d0",
+                        ][index % 4],
+                      }}
+                      title={`${span.start.toFixed(3)}–${span.end.toFixed(3)} s · ${span.text}`}
+                      aria-label={`${span.start.toFixed(3)} seconds: ${span.text}`}
+                      onClick={() => setTime(span.start)}
+                    />
+                  ))}
+                  <i style={{ left: `${(100 * time) / episode.duration}%` }} />
+                </div>
+              </div>
+            );
+          })}
+          <p className="job-cost-note">
+            Independent predictions from each camera's images, on the same
+            episode clock. Gaps and exact segment ends are retained. Existing
+            jobs are not re-annotated.
+          </p>
+        </>
+      )}
+    </section>
+  );
+}

frontend/src/jobs.css

@@ -353,3 +353,61 @@
     grid-template-columns: 1fr;
   }
 }
+
+.view-subtasks {
+  margin-top: 18px;
+  padding-top: 14px;
+  border-top: 1px solid var(--line-subtle);
+  font-size: 12px;
+}
+.view-subtask-controls {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  gap: 14px;
+  margin: 12px 0;
+}
+.view-subtask-controls select {
+  margin-left: 6px;
+  padding: 5px;
+  max-width: 240px;
+}
+.view-subtasks > input[type="range"] {
+  width: 100%;
+  margin: 8px 0;
+}
+.view-subtask-row {
+  margin: 14px 0;
+  overflow-wrap: anywhere;
+}
+.view-subtask-row p {
+  min-height: 1.5em;
+  margin: 5px 0;
+}
+.view-subtask-lane {
+  position: relative;
+  height: 24px;
+  background: var(--line-subtle);
+  border-radius: 4px;
+}
+.view-subtask-lane button {
+  position: absolute;
+  top: 2px;
+  bottom: 2px;
+  border: 1px solid var(--bg);
+  border-radius: 3px;
+  opacity: 0.65;
+  min-width: 1px;
+  padding: 0;
+}
+.view-subtask-lane button.active {
+  opacity: 1;
+  box-shadow: inset 0 0 0 2px var(--fg);
+}
+.view-subtask-lane i {
+  position: absolute;
+  top: -3px;
+  bottom: -3px;
+  border-left: 2px solid var(--fg);
+  pointer-events: none;
+}

frontend/src/main.tsx

@@ -68,6 +68,8 @@ type Options = {
   count: number;
   features: Feature[];
   camera: string | null;
+  subtask_views: "primary" | "per-camera";
+  subtask_cameras: string[];
   hz: number;
   hours: number;
   mode: string;
@@ -166,6 +168,8 @@ type Quote = {
   revision: string;
   cameras: string[];
   camera: string;
+  subtask_views?: "primary" | "per-camera";
+  subtask_cameras?: string[];
   private: boolean;
   first_episode: number;
   last_episode: number;
@@ -188,6 +192,8 @@ const defaults: Options = {
   count: 100,
   features: ["Subtasks"],
   camera: null,
+  subtask_views: "primary",
+  subtask_cameras: [],
   hz: 0.1,
   hours: 4,
   mode: COPY,
@@ -314,6 +320,14 @@ function initial(): Options {
           : safe.human_video_parallel,
       captioner: saved.captioner === "flora" ? "flora" : "selected-vlm",
       camera: typeof saved.camera === "string" ? saved.camera : null,
+      subtask_views:
+        saved.subtask_views === "per-camera" &&
+        safe.features.includes("Subtasks")
+          ? "per-camera"
+          : "primary",
+      subtask_cameras: Array.isArray(saved.subtask_cameras)
+        ? saved.subtask_cameras.filter((c: unknown) => typeof c === "string")
+        : [],
       slam: {
         compute: "rtx-pro-6000",
         output_fps: slam.output_fps === 30 ? 30 : 50,
@@ -532,6 +546,13 @@ function App() {
             old.camera && result.cameras.includes(old.camera)
               ? old.camera
               : result.cameras[0],
+          subtask_cameras:
+            old.subtask_views === "per-camera"
+              ? old.subtask_cameras.filter((c) => result.cameras.includes(c))
+                  .length
+                ? old.subtask_cameras.filter((c) => result.cameras.includes(c))
+                : result.cameras
+              : [],
         }));
         // Discovery also runs after a retry or login; require a fresh quote in case views changed.
         generation.current++;
@@ -657,6 +678,7 @@ function App() {
       setCameras([]);
       setCameraError("");
       patch.camera = null;
+      patch.subtask_cameras = [];
     }
     if ("dataset" in patch || "start" in patch || "count" in patch)
       setMetadata(null);
@@ -671,9 +693,16 @@ function App() {
     let features = on
       ? [...options.features, name]
       : options.features.filter((x) => x !== name);
+    if (!on && name === "Subtasks")
+      change({ subtask_views: "primary", subtask_cameras: [] });
     if (on && name === SLAM) {
       setVisibility("private");
-      change({ features: [SLAM], captioner: "selected-vlm" });
+      change({
+        features: [SLAM],
+        captioner: "selected-vlm",
+        subtask_views: "primary",
+        subtask_cameras: [],
+      });
       return;
     }
     if (on) features = features.filter((x) => x !== SLAM);
@@ -695,7 +724,19 @@ function App() {
     try {
       const q = await api<Quote>("/api/estimate", {
         method: "POST",
-        body: JSON.stringify(options),
+        body: JSON.stringify({
+          ...options,
+          subtask_cameras:
+            options.subtask_views === "per-camera"
+              ? [
+                  ...new Set(
+                    [options.camera, ...options.subtask_cameras].filter(
+                      (c): c is string => !!c && cameras.includes(c),
+                    ),
+                  ),
+                ]
+              : [],
+        }),
       });
       if (gen === generation.current) {
         setQuote(q);
@@ -1122,8 +1163,77 @@ function App() {
                     ) : null}
                   </div>
                   <p className="field-help" id="camera-help">
-                    One camera view per run.
+                    Primary view for the existing timeline, plans, memory and
+                    generated videos.
                   </p>
+                  {options.features.includes("Subtasks") && (
+                    <fieldset className="subtask-view-settings">
+                      <legend>Subtask camera views</legend>
+                      <label className="field-label" htmlFor="subtask-views">
+                        Output timelines
+                      </label>
+                      <div className="select-wrap">
+                        <Video size={16} />
+                        <select
+                          id="subtask-views"
+                          value={options.subtask_views}
+                          onChange={(e) =>
+                            change({
+                              subtask_views: e.target
+                                .value as Options["subtask_views"],
+                              subtask_cameras:
+                                e.target.value === "per-camera" ? cameras : [],
+                            })
+                          }
+                        >
+                          <option value="primary">Primary camera only</option>
+                          <option value="per-camera">
+                            Separate timeline per camera view
+                          </option>
+                        </select>
+                        <ChevronDown size={15} />
+                      </div>
+                      {options.subtask_views === "per-camera" && (
+                        <>
+                          <div className="subtask-camera-options">
+                            {cameras.map((camera) => (
+                              <label key={camera}>
+                                <input
+                                  type="checkbox"
+                                  checked={
+                                    camera === options.camera ||
+                                    options.subtask_cameras.includes(camera)
+                                  }
+                                  disabled={camera === options.camera}
+                                  onChange={(e) =>
+                                    change({
+                                      subtask_cameras: e.target.checked
+                                        ? [...options.subtask_cameras, camera]
+                                        : options.subtask_cameras.filter(
+                                            (c) => c !== camera,
+                                          ),
+                                    })
+                                  }
+                                />
+                                <span>
+                                  {camera}
+                                  {camera === options.camera
+                                    ? " (primary)"
+                                    : ""}
+                                </span>
+                              </label>
+                            ))}
+                          </div>
+                          <p className="field-help">
+                            Each checked view is annotated from its own images
+                            and keeps its own timestamped timeline. The estimate
+                            includes all checked views. Hands and generated
+                            videos still use the primary camera.
+                          </p>
+                        </>
+                      )}
+                    </fieldset>
+                  )}
                 </section>
               )}
               {step === 1 && !usesSlam && (
@@ -1647,6 +1757,14 @@ function App() {
                       <span>Camera</span>
                       <strong>{options.camera}</strong>
                     </div>
+                    <div>
+                      <span>Subtask views</span>
+                      <strong>
+                        {quote?.subtask_views === "per-camera"
+                          ? `${quote.subtask_cameras?.length} independent camera timelines per episode`
+                          : "Primary camera only"}
+                      </strong>
+                    </div>
                     <div>
                       <span>Annotations</span>
                       <strong>{options.features.join(", ")}</strong>
@@ -2265,9 +2383,17 @@ function App() {
               <dd>{quote && num(quote.episodes)}</dd>
             </div>
             <div>
-              <dt>Camera view</dt>
+              <dt>Primary camera</dt>
               <dd>{quote?.camera}</dd>
             </div>
+            <div>
+              <dt>Subtask camera views</dt>
+              <dd>
+                {quote?.subtask_views === "per-camera"
+                  ? quote.subtask_cameras?.join(", ")
+                  : "Primary camera only"}
+              </dd>
+            </div>
             <div>
               <dt>Annotations</dt>
               <dd>{options.features.join(", ")}</dd>

frontend/src/wizard.css

@@ -411,7 +411,9 @@
   gap: 6px;
 }
 @media (max-width: 480px) {
-  .slam-intrinsics-grid { grid-template-columns: 1fr; }
+  .slam-intrinsics-grid {
+    grid-template-columns: 1fr;
+  }
 }
 .visibility-options {
   display: grid;
@@ -573,3 +575,30 @@
 .hand-pipeline-steps li + li {
   margin-top: 12px;
 }
+
+.subtask-view-settings {
+  margin: 20px 0 0;
+  padding: 16px;
+  border: 1px solid var(--line-subtle);
+  border-radius: 10px;
+}
+.subtask-view-settings legend {
+  padding: 0 6px;
+  font-weight: 600;
+  font-size: 13px;
+}
+.subtask-camera-options {
+  display: grid;
+  gap: 8px;
+  margin: 12px 0;
+}
+.subtask-camera-options label {
+  display: flex;
+  align-items: flex-start;
+  gap: 8px;
+  font-size: 12px;
+  overflow-wrap: anywhere;
+}
+.subtask-camera-options input {
+  flex: 0 0 auto;
+}

server.py

@@ -27,6 +27,8 @@ from reporting import annotation_url, episode_progress, hand_visualizer_url, usa
 from slam_pipeline import catalog as slam_catalog
 from studio import COPY, FEATURES, launch, prepare, read_cameras
 
+from subtask_views import preview_timelines
+
 app = FastAPI(title="LeRobot Annotation Studio", docs_url=None, redoc_url=None, openapi_url=None)
 BASE = Path(__file__).parent
 HOST = os.environ.get("SPACE_HOST")
@@ -65,6 +67,8 @@ class EstimateInput(DatasetInput):
     hand_actions: HandActionInput | None = None
     slam: SlamInput | None = None
     camera: str | None = Field(default=None, max_length=200)
+    subtask_views: Literal["primary", "per-camera"] = "primary"
+    subtask_cameras: list[str] = Field(default_factory=list, max_length=32)
     hz: float = 0.1
     hours: int = 4
     mode: Literal["Annotated dataset copy"] = COPY
@@ -311,6 +315,8 @@ def calculate(options: EstimateInput, request: Request):
             captioner=options.captioner,
             hand_actions=options.hand_actions.model_dump() if options.hand_actions else None,
             slam=options.slam.model_dump() if options.slam else None,
+            subtask_views=options.subtask_views,
+            subtask_cameras=options.subtask_cameras,
         )
     except Exception as exc:
         raise HTTPException(400, safe_error(exc)) from None
@@ -336,6 +342,8 @@ def calculate(options: EstimateInput, request: Request):
         "model_profile": plan["model_profile"],
         "captioner": plan.get("captioner", "selected-vlm"),
         "camera": plan["camera"],
+        "subtask_views": plan.get("subtask_views", "primary"),
+        "subtask_cameras": plan.get("subtask_cameras", [plan["camera"]]),
         "cameras": s["cameras"],
         "episodes": len(s["episodes"]),
         "total_episodes": s["total_episodes"],
@@ -501,12 +509,25 @@ def progress(job_id: str, request: Request):
             "output_mode": report.get("manifest", {}).get("mode"),
             "model": report.get("manifest", {}).get("model"),
             "captioner": report.get("manifest", {}).get("captioner"),
+            "subtask_views": report.get("manifest", {}).get("subtask_views", "primary"),
+            "subtask_cameras": report.get("manifest", {}).get("subtask_cameras", []),
             "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}/subtask-views")
+def subtask_view_results(job_id: str, request: Request):
+    current = session(request, required=True)
+    try:
+        _, output = studio_output(job_id, current)
+        report = json.loads(read_output_file(output, "annotation_studio/run.json", current["token"]))
+        return {**preview_timelines(report), "output_repo": output}
+    except Exception as exc:
+        raise HTTPException(400, "No exported per-camera subtask timelines available yet.") from exc
+
+
 @app.get("/api/jobs/{job_id}/human-videos")
 def human_videos(job_id: str, request: Request):
     current = session(request, required=True)

studio.py

@@ -30,6 +30,8 @@ from model_catalog import (
 from prompt_config import TEMPLATES, validate_prompts
 from slam_pipeline import SLAM, SLAM_COMPUTE, prepare_slam, verify_release
 
+from subtask_views import select_subtask_cameras
+
 MODEL = model_profile()["route"]
 PROVIDER = model_profile()["provider"]
 FEATURES = {
@@ -193,6 +195,7 @@ def estimate(
     human_video_parallel=None,
     captioner="selected-vlm",
     human_video_model="minimax-h3",
+    subtask_camera_count=1,
 ):
     if captioner not in {"selected-vlm", "flora"}:
         raise ValueError("Choose a supported subtask captioner.")
@@ -224,10 +227,14 @@ def estimate(
         max(1, math.ceil(r["length"] / selection["info"]["fps"] / 600)) for r in selection["episodes"]
     )
     # Token estimates, not measured usage. Plans are deterministic and add no model calls.
+    if type(subtask_camera_count) is not int or not 1 <= subtask_camera_count <= 32:
+        raise ValueError("Choose between 1 and 32 subtask camera views.")
     provider_subtasks = needs_subtasks and captioner != "flora"
     inputs = (seconds / 10 * 1500 + windows * 1500) if provider_subtasks else 0
     outputs = (seconds / 5 * 55 + windows * 120) if provider_subtasks else 0
-    calls = windows if provider_subtasks else 0
+    inputs *= subtask_camera_count
+    outputs *= subtask_camera_count
+    calls = windows * subtask_camera_count if provider_subtasks else 0
     if "Memory" in features:
         memories = max(0, seconds / 5 - n)
         inputs += memories * 1800
@@ -273,7 +280,7 @@ def estimate(
         calls += plan_calls
     # Longer custom templates add input cost. Generation length remains uncertain.
     prompt_calls = {
-        "plan_subtasks": windows,
+        "plan_subtasks": windows * subtask_camera_count,
         "plan_memory": max(0, seconds / 5 - n),
         "plan_task_rephrasings": n,
         "interjections_initial_speech": n,
@@ -304,6 +311,8 @@ def estimate(
         "compute_name": "H200 GPU" if captioner == "flora" else "CPU",
         "helper_required": captioner == "flora" and needs_provider(features, prompts, captioner),
         "calls": math.ceil(calls),
+        "subtask_camera_count": subtask_camera_count,
+        "subtask_video_hours": seconds * subtask_camera_count / 3600,
         "custom_prompts": len(prompts),
         "hours": hours,
         "input_rate": profile["input_rate"],
@@ -331,6 +340,8 @@ def fingerprint(
     human_video_parallel=None,
     captioner="selected-vlm",
     human_video_model="minimax-h3",
+    subtask_views="primary",
+    subtask_cameras=None,
 ):
     return hashlib.sha256(
         json.dumps(
@@ -348,6 +359,8 @@ def fingerprint(
                 human_video_parallel,
                 captioner,
                 human_video_model,
+                subtask_views,
+                subtask_cameras,
             ],
             sort_keys=True,
         ).encode()
@@ -378,6 +391,8 @@ def prepare(
     hand_actions=None,
     slam=None,
     human_video_model="minimax-h3",
+    subtask_views="primary",
+    subtask_cameras=None,
 ):
     profile = model_profile(model_key)
     prompts = validate_prompts(prompts or {}, features)
@@ -396,6 +411,9 @@ def prepare(
     camera = camera or selection["cameras"][0]
     if camera not in selection["cameras"]:
         raise ValueError("Choose a camera in this dataset.")
+    views = select_subtask_cameras(camera, selection["cameras"], subtask_views, subtask_cameras)
+    if subtask_views == "per-camera" and "Subtasks" not in features:
+        raise ValueError("Select Subtasks to export per-camera timelines.")
     if captioner not in {"selected-vlm", "flora"}:
         raise ValueError("Choose a supported subtask captioner.")
     if SLAM in features and (set(features) != {SLAM} or captioner == "flora"):
@@ -423,6 +441,7 @@ def prepare(
         human_video_parallel,
         captioner,
         human_video_model,
+        subtask_camera_count=len(views),
     )
     if pika:
         costs.update(compute_flavor="a100-large", compute_name="A100 GPU")
@@ -436,6 +455,8 @@ def prepare(
         "selection": selection,
         "features": list(features),
         "camera": camera,
+        "subtask_views": subtask_views,
+        "subtask_cameras": views,
         "vqa_hz": vqa_hz,
         "mode": mode,
         "prompts": prompts,
@@ -469,6 +490,8 @@ def prepare(
             human_video_parallel,
             captioner,
             human_video_model,
+            subtask_views,
+            views,
         ),
     }
 

studio_runtime.py

@@ -272,6 +272,8 @@ def run_episode(record, row, info, root, manifest, provider, staging_dir, media=
             for key in (
                 "features",
                 "camera",
+                "subtask_views",
+                "subtask_cameras",
                 "vqa_hz",
                 "prompts",
                 "model",

subtask_views.py

@@ -0,0 +1,85 @@
+"""Camera-scoped subtask outputs alongside the compatible primary language timeline."""
+
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+
+VIEW_INSTRUCTIONS = """
+Annotate only the activity supported by this camera's images. This is one camera
+view, not a combined multi-camera observation. Do not infer hidden actions from
+the overall task or invent activity visible only in another camera. If the view
+does not show an action, explicitly describe the visible state or that the action
+is out of view. Keep the episode's original clock and the required JSON schema.
+"""
+
+
+def select_subtask_cameras(
+    primary: str, available: list[str], mode: str = "primary", cameras: list[str] | None = None,
+) -> list[str]:
+    if mode not in {"primary", "per-camera"}:
+        raise ValueError("Choose primary-camera or per-camera subtasks.")
+    if primary not in available:
+        raise ValueError("Choose a primary camera in this dataset.")
+    if mode == "primary":
+        return [primary]
+    chosen = list(available) if not cameras else cameras
+    if not isinstance(chosen, list) or not all(isinstance(c, str) for c in chosen):
+        raise ValueError("Subtask cameras must be a list of camera names.")
+    if len(chosen) + int(primary not in chosen) > 32 or len(set(chosen)) != len(chosen):
+        raise ValueError("Choose at most 32 distinct subtask cameras.")
+    if any(c not in available for c in chosen):
+        raise ValueError("A selected subtask camera is not in this dataset.")
+    # The primary timeline also feeds plans, memory and generated videos.
+    return [primary, *[c for c in chosen if c != primary]]
+
+
+def write_view_subtasks(
+    report: dict[str, Any], output_episode: int, root: Path, manifest: dict[str, Any],
+) -> Path:
+    """Save exact starts/ends and both episode identities without changing frame data."""
+    rows = []
+    for camera, spans in report["subtasks_by_camera"].items():
+        for span in spans:
+            start, end = span["start"], span["end"]
+            if not all(type(t) in (int, float) and math.isfinite(t) for t in (start, end)) or end <= start:
+                raise ValueError("Invalid camera subtask clock.")
+            rows.append({
+                "episode_index": output_episode,
+                "source_episode_index": report["episode_index"],
+                "camera": camera,
+                "start_timestamp": float(start),
+                "end_timestamp": float(end),
+                "subtask": span["text"],
+            })
+    schema = pa.schema([
+        ("episode_index", pa.int64()), ("source_episode_index", pa.int64()),
+        ("camera", pa.string()), ("start_timestamp", pa.float64()),
+        ("end_timestamp", pa.float64()), ("subtask", pa.string()),
+    ], metadata={b"annotation_schema": b"camera-subtasks-v1", b"source": json.dumps({
+        "repo": manifest.get("dataset"), "revision": manifest.get("dataset_revision"),
+        "model": manifest.get("model"), "captioner": manifest.get("captioner", "selected-vlm"),
+    }, sort_keys=True).encode()})
+    target = Path(root) / "annotation_studio" / "subtasks_by_camera" / f"episode_{output_episode:06d}.parquet"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    pq.write_table(pa.Table.from_pylist(rows, schema=schema), target, compression="zstd")
+    return target
+
+
+def preview_timelines(report: dict[str, Any]) -> dict[str, Any]:
+    mapping = {m["source_episode_index"]: m["episode_index"] for m in report.get("source_episode_map", [])}
+    episodes = []
+    for result in report.get("episodes", []):
+        source = result.get("episode_index")
+        if result.get("status") != "completed" or source not in mapping:
+            continue
+        views = result.get("subtasks_by_camera")
+        if not views:
+            continue
+        episodes.append({"source_episode_index": source, "episode_index": mapping[source],
+            "duration": result["subtask_view_duration_s"], "cameras": views})
+    return {"mode": report.get("manifest", {}).get("subtask_views", "primary"), "episodes": episodes}

test_studio.py

@@ -231,9 +231,9 @@ def test_launch_identity_secret_and_immutable_manifest(mapping_format):
     assert manifest["episode_ids_sha256"] == hashlib.sha256(json.dumps([0, 1]).encode()).hexdigest()
     assert manifest["space_revision"] == "b" * 40
     assert manifest["prompts"] == configured["prompts"]
-    assert manifest["model"] == "Qwen/Qwen3.8-27B:novita"
-    assert manifest["request_concurrency"] == manifest["episode_parallelism"] == 32
-    assert manifest["costs"]["context_length"] == 1_000_000
+    assert manifest["model"] == model_profile(studio.DEFAULT_MODEL)["route"]
+    assert manifest["request_concurrency"] == manifest["episode_parallelism"] == model_profile(studio.DEFAULT_MODEL)["concurrency"]
+    assert manifest["costs"]["context_length"] == model_profile(studio.DEFAULT_MODEL)["context_length"]
     assert "prompt-templates.json" in str(args["command"])
     assert manifest["inference_budget"] + plan()["costs"]["cpu_max"] == 5
     assert result["id"] == "job123"

test_subtask_views.py

@@ -0,0 +1,107 @@
+"""View isolation, explicit cost scaling and lossless camera/episode output identities."""
+import json
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import av
+import numpy as np
+import pyarrow.parquet as pq
+import pytest
+from fastapi.testclient import TestClient
+
+import server
+import studio
+import worker
+from subtask_views import preview_timelines, select_subtask_cameras, write_view_subtasks
+from test_server import login_client
+from test_studio import selection
+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
+
+CAMERAS = ["observation.images.top", "observation.images.wrist"]
+
+
+def test_views_are_explicit_distinct_and_primary_is_kept():
+    assert select_subtask_cameras(CAMERAS[0], CAMERAS) == CAMERAS[:1]
+    assert select_subtask_cameras(CAMERAS[0], CAMERAS, "per-camera") == CAMERAS
+    assert select_subtask_cameras(CAMERAS[0], CAMERAS, "per-camera", [CAMERAS[1]]) == CAMERAS
+    for names in [[CAMERAS[1], CAMERAS[1]], ["not-a-camera"], "not-a-list"]:
+        with pytest.raises(ValueError):
+            select_subtask_cameras(CAMERAS[0], CAMERAS, "per-camera", names)
+
+
+def test_estimate_multiplies_only_view_subtasks_not_primary_memory():
+    one = studio.estimate(selection(), ["Subtasks"], .1, 1, .0005)
+    three = studio.estimate(selection(), ["Subtasks"], .1, 1, .0005, subtask_camera_count=3)
+    assert three['inference'] == pytest.approx(3 * one['inference'])
+    assert three['calls'] == 3 * one['calls']
+    assert three['cpu_max'] == one['cpu_max']
+    primary = studio.estimate(selection(), ["Subtasks", "Memory"], .1, 1, .0005)
+    multi = studio.estimate(selection(), ["Subtasks", "Memory"], .1, 1, .0005, subtask_camera_count=3)
+    assert multi['inference'] - primary['inference'] == pytest.approx(2 * one['inference'])
+
+
+def test_quote_pins_views_and_rejects_missing_cameras_before_hardware_lookup(monkeypatch):
+    picked = {**selection(), 'cameras': CAMERAS, 'info': {'fps': 30, 'features': {}}}
+    monkeypatch.setattr(studio, 'read_selection', lambda *a, **k: picked)
+    hardware = Mock(return_value=SimpleNamespace(list_jobs_hardware=lambda: [SimpleNamespace(name='cpu-upgrade', unit_label='minute', unit_cost_usd=.0005)]))
+    monkeypatch.setattr(studio, 'HfApi', hardware)
+    q = studio.prepare('owner/data', 0, 2, ['Subtasks'], CAMERAS[0], .1, 1, studio.COPY, subtask_views='per-camera', subtask_cameras=CAMERAS)
+    assert q['subtask_cameras'] == CAMERAS and q['costs']['subtask_camera_count'] == 2
+    primary = studio.prepare('owner/data', 0, 2, ['Subtasks'], CAMERAS[0], .1, 1, studio.COPY)
+    assert q['fingerprint'] != primary['fingerprint']
+    hardware.reset_mock()
+    with pytest.raises(ValueError):
+        studio.prepare('owner/data', 0, 2, ['Subtasks'], CAMERAS[0], .1, 1, studio.COPY, subtask_views='per-camera', subtask_cameras=['missing'])
+    hardware.assert_not_called()
+
+
+def test_real_pipeline_reads_both_videos_keeps_primary_language_and_exact_clocks(tmp_path, monkeypatch):
+    row = {'episode_index': 42, 'length': 40}
+    info = {'fps': 10, 'video_path': 'videos/{video_key}/chunk-{chunk_index:03d}/file-{file_index:03d}.mp4', 'features': {c: {'dtype': 'video'} for c in CAMERAS}}
+    for camera, color in zip(CAMERAS, [(180, 0, 0), (0, 0, 180)]):
+        path = tmp_path / 'videos' / camera / 'chunk-000/file-000.mp4';path.parent.mkdir(parents=True)
+        with av.open(str(path), 'w') as container:
+            stream = container.add_stream('mpeg4', rate=10);stream.width=stream.height=64;stream.pix_fmt='yuv420p'
+            for i in range(40):
+                image=np.zeros((64,64,3),dtype=np.uint8);image[:]=color
+                for packet in stream.encode(av.VideoFrame.from_ndarray(image, format='rgb24')):container.mux(packet)
+            for packet in stream.encode():container.mux(packet)
+        for key,value in [('chunk_index',0),('file_index',0),('from_timestamp',0.)]:row[f'videos/{camera}/{key}']=value
+    record = EpisodeRecord(42,'Move the cup',tuple(i/10 for i in range(40)),tuple(range(40)),tmp_path/'unused.parquet',0,40)
+    observed=[]
+    def respond(messages):
+        blocks=[block for message in messages for block in message['content'] if isinstance(block,dict)]
+        image=next(block['image'] for block in blocks if block.get('type')=='image')
+        rgb=np.array(image).mean((0,1));view='red' if rgb[0]>rgb[2] else 'blue';observed.append(view)
+        return {'subtasks':[{'start':0.,'end':1.2,'text':f'Move the {view} cup'},{'start':1.8,'end':3.9,'text':f'Hold the {view} cup'}]}
+    manifest={'features':['Subtasks'],'camera':CAMERAS[0],'subtask_views':'per-camera','subtask_cameras':CAMERAS,'prompts':{},'vqa_hz':.1}
+    result=worker.annotate_episode(record,row,info,tmp_path,manifest,StubVlmClient(respond),tmp_path/'staging')
+    assert set(observed)=={'red','blue'}
+    assert result['subtasks_by_camera'][CAMERAS[0]][0]['text']=='Move the red cup'
+    assert result['subtasks_by_camera'][CAMERAS[1]][0]['text']=='Move the blue cup'
+    assert result['subtasks_by_camera'][CAMERAS[0]][0]['end']==1.8
+    staged=EpisodeStaging(tmp_path/'staging',42).read_all();labels=[r['content'] for rows in staged.values() for r in rows if r.get('style')=='subtask']
+    assert labels==['Move the red cup','Hold the red cup']
+    assert worker.CUSTOM_PROMPTS.get() is None
+    file=write_view_subtasks(result,0,tmp_path/'output',{'dataset':'owner/data','dataset_revision':'a'*40})
+    table=pq.read_table(file);assert table['episode_index'].to_pylist()==[0]*4;assert table['source_episode_index'].to_pylist()==[42]*4
+    assert table['end_timestamp'].to_pylist()==[1.8,3.9,1.8,3.9]
+    preview=preview_timelines({'manifest':manifest,'source_episode_map':[{'source_episode_index':42,'episode_index':0}],'episodes':[result]})
+    assert preview['episodes'][0]['cameras']==result['subtasks_by_camera']
+
+
+def test_preview_requires_owner_session_and_hides_unexported_results(monkeypatch):
+    server.SESSIONS.clear()
+    with TestClient(server.app) as client:
+        assert client.get('/api/jobs/job/subtask-views').status_code in (401,403)
+        login_client(client)
+        report={'manifest':{'subtask_views':'per-camera'},'source_episode_map':[], 'episodes':[{'episode_index':42,'status':'completed','subtask_view_duration_s':4.,'subtasks_by_camera':{CAMERAS[0]:[{'start':0.,'end':4.,'text':'Move'}]}}]}
+        monkeypatch.setattr(server,'studio_output',lambda *a:(None,'alice/output'))
+        monkeypatch.setattr(server,'read_output_file',lambda *a:json.dumps(report).encode())
+        assert client.get('/api/jobs/job/subtask-views').json()['episodes']==[]
+        report['source_episode_map']=[{'source_episode_index':42,'episode_index':0}]
+        data=client.get('/api/jobs/job/subtask-views').json();assert data['episodes'][0]['episode_index']==0
+        assert 'never-return-this-token' not in json.dumps(data)

worker.py

@@ -25,6 +25,7 @@ from provider import Budget, BudgetExceededError, DeadlineExceededError, Provide
 from studio import COPY, FEATURES, HUMAN_VIDEOS, needs_provider, read_selection, relative_path
 from studio_runtime import run_episode as run_runtime_episode
 from subset import EpisodeSubset
+from subtask_views import VIEW_INSTRUCTIONS, select_subtask_cameras, write_view_subtasks
 
 from lerobot.annotations.steerable_pipeline.config import InterjectionsConfig, PlanConfig, VqaConfig
 from lerobot.annotations.steerable_pipeline.executor import Executor
@@ -307,6 +308,32 @@ def human_video_summary(entries):
 
 
 def _annotate_episode(record, row, info, root, manifest, provider, staging_dir, media=None):
+    if manifest.get("subtask_views", "primary") != "per-camera":
+        return _annotate_single_camera(record, row, info, root, manifest, provider, staging_dir, media)
+    available = [key for key, feature in info["features"].items() if feature["dtype"] == "video"]
+    cameras = select_subtask_cameras(manifest["camera"], available, "per-camera", manifest.get("subtask_cameras"))
+    prompts = dict(CUSTOM_PROMPTS.get() or {})
+    prompts["plan_subtasks"] = prompts.get("plan_subtasks", ACTION_PROMPT) + "\n" + VIEW_INSTRUCTIONS
+    context = CUSTOM_PROMPTS.set(prompts)
+    try:
+        result = _annotate_single_camera(record, row, info, root, manifest, provider, staging_dir, media)
+        by_camera = {manifest["camera"]: result["subtask_spans"]}
+        recovery = {manifest["camera"]: result["recovery_used"]}
+        for camera in cameras[1:]:
+            view_dir = staging_dir.parent / "camera-subtasks" / str(record.episode_index) / hashlib.sha256(camera.encode()).hexdigest()[:16]
+            view_manifest = {**manifest, "camera": camera, "features": ["Subtasks"], "hand_actions": None, "human_video": None}
+            view = _annotate_single_camera(record, row, info, root, view_manifest, provider, view_dir, None)
+            by_camera[camera] = view["subtask_spans"]
+            recovery[camera] = view["recovery_used"]
+            shutil.rmtree(view_dir, ignore_errors=True)
+        result.update(subtasks_by_camera=by_camera, subtask_view_recovery=recovery,
+                      subtask_view_duration_s=int(row["length"]) / float(info["fps"]))
+        return result
+    finally:
+        CUSTOM_PROMPTS.reset(context)
+
+
+def _annotate_single_camera(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)
@@ -535,6 +562,9 @@ def run(manifest, work=Path("/workspace/run")):
             vision_frame_batching=os.environ.get("FLORA_VISION_FRAME_BATCHING", "0") == "1",
         )
         flora_parallelism = int(os.environ.get("FLORA_EPISODE_PARALLELISM", "1"))
+        if manifest.get("subtask_views") == "per-camera":
+            # A resident captioner is shared; do not key different views by the same episode ID.
+            flora_parallelism = 1
         if flora_parallelism > 1:
             if FLORA_CAPTIONER.execution != "accelerated" or manifest["features"] != ["Subtasks"]:
                 raise ValueError("Cross-episode Flora requires independent accelerated subtasks only")
@@ -582,6 +612,10 @@ def run(manifest, work=Path("/workspace/run")):
             revision=manifest["dataset_revision"],
             cache_dir=str(work / "metadata-cache"),
         )
+
+    subtask_cameras = select_subtask_cameras(manifest["camera"], selection["cameras"], manifest.get("subtask_views", "primary"), manifest.get("subtask_cameras"))
+    if manifest.get("subtask_views") == "per-camera" and "Subtasks" not in manifest["features"]:
+        raise ValueError("Per-camera subtasks must be selected for export")
     ids = [r["episode_index"] for r in selection["episodes"]]
     if hashlib.sha256(json.dumps(ids).encode()).hexdigest() != manifest["episode_ids_sha256"]:
         raise ValueError("Pinned episode selection differs from the estimate.")
@@ -736,7 +770,7 @@ def run(manifest, work=Path("/workspace/run")):
             print('HAND_RUNTIME {"event":"camera_engines_ready","native_worker":"starting"}', flush=True)
             HAND_ANNOTATOR = HandAnnotator(backend, work / "hands-native")
         for data_path, rows in shards.items():
-            files = {data_path} | {video_path(row, info, manifest["camera"]) for row in rows}
+            files = {data_path} | {video_path(row, info, camera) for row in rows for camera in subtask_cameras}
             sizes = api.get_paths_info(
                 manifest["dataset"], list(files), repo_type="dataset", revision=manifest["dataset_revision"]
             )
@@ -827,6 +861,9 @@ def run(manifest, work=Path("/workspace/run")):
                     new_id, written = subset.add(
                         record.episode_index, annotated_table, hand_features=hand_features
                     )
+                    if manifest.get("subtask_views") == "per-camera":
+                        report = next(e for e in progress["episodes"] if e["episode_index"] == record.episode_index)
+                        written = [*written, write_view_subtasks(report, new_id, subset.root, manifest)]
                     extra_operations = []
                     with commit_lock:
                         for entry in human_index: