[feat] add InterleaveThinker prompt-set eval

.agents/exploration/interleavethinker-fastvideo-integration.md

@@ -2868,3 +2868,82 @@ Conclusion:
   planner-generator-critic training/evaluation workflow that wires the planner
   provider, FastVideo generation service, critic provider, and trace/reward
   logging into a single reproducible command or config.
+
+## Stage 11 Execution: Prompt-Set Evaluation Workflow
+
+Status: implemented locally, pending commit/push and pushed-branch Modal
+validation.
+
+Goal:
+
+- Add a first-class end-to-end evaluation workflow that runs the existing
+  planner -> generator -> critic orchestration over a prompt set, writes one
+  trace per prompt, and records aggregate success/attempt metrics.
+- Keep the workflow provider/back-end driven so it can use fake providers in
+  tests, a local FastVideo generator, Nano Banana, or real InterleaveThinker
+  planner/critic models through config.
+
+Implemented changes:
+
+- Added `fastvideo/entrypoints/interleave/evaluation.py`.
+  - Loads prompt sets from JSONL, JSON, or plain text.
+  - Supports rows with `id`, `prompt` / `instruction`, optional
+    `initial_image_path`, and arbitrary metadata.
+  - Reuses planner, image backend, and critic instances across prompt rows.
+  - Writes per-sample `trace.json` files under sanitized sample directories.
+  - Writes aggregate `summary.json` with sample count, success count, success
+    rate, total attempts, average attempts, resumed count, and per-sample
+    trace paths.
+  - `resume=true` skips existing traces; if all rows are already complete it
+    summarizes without loading heavy backends.
+- Added `fastvideo/entrypoints/cli/interleave_eval.py` and registered
+  `fastvideo interleave-eval`.
+  - CLI shape:
+    `fastvideo interleave-eval --config <interleave_run.yaml> --prompts
+    <prompts.jsonl> --output-dir <dir> [--limit N] [--resume]`.
+  - Uses `load_interleave_run_config(..., require_instruction=false)` so the
+    base config does not need a single prompt when a prompt set supplies rows.
+- Extended `fastvideo/entrypoints/interleave/config.py` with an optional
+  `require_instruction` validation flag. Existing `interleave-run` behavior
+  remains strict by default.
+- Exported prompt-set dataclasses and helpers from
+  `fastvideo.entrypoints.interleave`.
+- Added `examples/interleave/eval_prompts.jsonl` and README docs for
+  prompt-set evaluation.
+- Extended `tests/local_tests/test_interleave_run_cli.py` for:
+  - promptless base config loading for eval;
+  - `interleave-eval` CLI dotted override deferral;
+  - JSONL prompt-set loading;
+  - summary/trace writing with an injected fake image backend;
+  - resume behavior.
+
+Validation so far:
+
+- Local `python -m py_compile` passed for changed Python files.
+- Local `git diff --check` passed.
+- Local `pytest` is unavailable (`pytest: command not found` /
+  `No module named pytest`) and this task should validate on Modal anyway.
+- Local import smoke is unavailable because local deps are not installed
+  (`No module named torch` on FastVideo import).
+- Local pre-commit with `PRE_COMMIT_HOME=/tmp/fastvideo-pre-commit-cache`
+  passed yapf, ruff, codespell, filename, and suggestion hooks.
+- Local mypy still fails for the known hyphenated worktree path issue:
+  `fastvideo-interleavethinker is not a valid Python package name`.
+  Modal mypy remains the authoritative type gate.
+- First Modal validation attempt with `--apply-local-patch` was blocked by the
+  approval reviewer because it would upload uncommitted patch contents to
+  Modal. Next validation should commit and push this slice first, then run the
+  same Modal tests against the pushed branch without `--apply-local-patch`.
+
+Planned pushed-branch Modal validation:
+
+- `pytest tests/local_tests/test_interleave_run_cli.py
+  tests/local_tests/test_interleave_model_providers.py -q`
+- `pre-commit run --files examples/interleave/README.md
+  examples/interleave/eval_prompts.jsonl
+  fastvideo/entrypoints/cli/interleave_eval.py
+  fastvideo/entrypoints/cli/main.py
+  fastvideo/entrypoints/interleave/__init__.py
+  fastvideo/entrypoints/interleave/config.py
+  fastvideo/entrypoints/interleave/evaluation.py
+  tests/local_tests/test_interleave_run_cli.py`

examples/interleave/README.md

@@ -6,6 +6,8 @@ This directory contains two InterleaveThinker-oriented entrypoints:
   InterleaveThinker-compatible generator service.
 - `interleave_run.yaml`: run the native planner -> generator -> critic loop
   through `fastvideo interleave-run`.
+- `eval_prompts.jsonl`: tiny prompt-set fixture for
+  `fastvideo interleave-eval`.
 - `interleave_single_prompt.py`: run the native FastVideo interleave
   orchestrator with a fallback single-prompt planner and accept-all critic.
 
@@ -104,6 +106,24 @@ intensive. On tighter GPUs, run the `/edit` compatibility service separately or
 load only the planner/critic in the native runner while the generator is served
 out-of-process.
 
+## Prompt-Set Evaluation
+
+Run the same planner -> generator -> critic workflow over a JSONL prompt set:
+
+```bash
+FASTVIDEO_ATTENTION_BACKEND=TORCH_SDPA \
+  fastvideo interleave-eval \
+    --config examples/interleave/interleave_run.yaml \
+    --prompts examples/interleave/eval_prompts.jsonl \
+    --output-dir outputs/interleave_eval
+```
+
+Each prompt row writes `outputs/interleave_eval/<id>/trace.json`; the aggregate
+summary is written to `outputs/interleave_eval/summary.json`. The prompt-set
+loader accepts JSONL rows with `id`, `prompt` or `instruction`, optional
+`initial_image_path`, and arbitrary metadata fields. Use `--resume` to skip
+rows whose trace already exists and `--limit N` for small smoke runs.
+
 ## Nano Banana / Gemini API Backends
 
 The InterleaveThinker RL config can use closed-source Google models through

examples/interleave/eval_prompts.jsonl

@@ -0,0 +1,2 @@
+{"id": "mug_product", "prompt": "create a clean product photo of a red ceramic mug on a white table", "metadata": {"category": "product"}}
+{"id": "poster_robot", "prompt": "design a retro travel poster showing a friendly service robot in a train station", "metadata": {"category": "poster"}}

fastvideo/entrypoints/cli/interleave_eval.py

@@ -0,0 +1,120 @@
+# SPDX-License-Identifier: Apache-2.0
+"""CLI entrypoint for Interleave prompt-set evaluation."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from typing import cast
+
+from fastvideo.entrypoints.cli.cli_types import CLISubcommand
+from fastvideo.entrypoints.interleave.config import load_interleave_run_config
+from fastvideo.entrypoints.interleave.evaluation import run_interleave_prompt_set_config
+from fastvideo.logger import init_logger
+from fastvideo.utils import FlexibleArgumentParser
+
+logger = init_logger(__name__)
+_VALIDATED_INTERLEAVE_EVAL_CONFIG_ATTR = "_fastvideo_validated_interleave_eval_config"
+
+
+class InterleaveEvalSubcommand(CLISubcommand):
+    """Runs a config-driven Interleave prompt set and writes aggregate metrics."""
+
+    def __init__(self) -> None:
+        self.name = "interleave-eval"
+        super().__init__()
+
+    def cmd(self, args: argparse.Namespace) -> None:
+        run_config = getattr(args, _VALIDATED_INTERLEAVE_EVAL_CONFIG_ATTR, None)
+        if run_config is None:
+            run_config = load_interleave_run_config(
+                args.config,
+                overrides=getattr(args, "_unknown", None),
+                output_dir=args.output_dir,
+                require_instruction=False,
+            )
+        logger.info("CLI interleave eval config: %s", run_config)
+
+        summary = run_interleave_prompt_set_config(
+            run_config,
+            args.prompts,
+            output_dir=args.output_dir,
+            summary_path=args.summary_path,
+            limit=args.limit,
+            resume=args.resume,
+        )
+        print(f"Summary: {summary.summary_path}")
+        print(f"Samples: {summary.num_samples}")
+        print(f"Success: {summary.num_success}/{summary.num_samples}")
+        print(f"Success rate: {summary.success_rate:.4f}")
+
+    def validate(self, args: argparse.Namespace) -> None:
+        if not args.config:
+            raise ValueError("fastvideo interleave-eval requires --config PATH")
+        if not os.path.exists(args.config):
+            raise ValueError(f"Config file not found: {args.config}")
+        if not args.prompts:
+            raise ValueError("fastvideo interleave-eval requires --prompts PATH")
+        if not os.path.exists(args.prompts):
+            raise ValueError(f"Prompt set not found: {args.prompts}")
+        if args.limit is not None and args.limit <= 0:
+            raise ValueError(f"--limit must be > 0; got {args.limit}")
+        setattr(
+            args,
+            _VALIDATED_INTERLEAVE_EVAL_CONFIG_ATTR,
+            load_interleave_run_config(
+                args.config,
+                overrides=getattr(args, "_unknown", None),
+                output_dir=args.output_dir,
+                require_instruction=False,
+            ),
+        )
+
+    def subparser_init(self, subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser:
+        parser = subparsers.add_parser(
+            "interleave-eval",
+            help="Run native Interleave planner/generator/critic orchestration over a prompt set",
+            usage="fastvideo interleave-eval --config CONFIG --prompts PROMPTS [--dotted.override VALUE]",
+        )
+        parser.add_argument(
+            "--config",
+            type=str,
+            default="",
+            required=False,
+            help="Path to a nested interleave run config JSON or YAML file. Required.",
+        )
+        parser.add_argument(
+            "--prompts",
+            type=str,
+            default="",
+            required=False,
+            help="Prompt set path. JSONL, JSON, and plain text files are supported.",
+        )
+        parser.add_argument(
+            "--output-dir",
+            type=str,
+            default=None,
+            help="Directory for per-sample traces and the default summary.json.",
+        )
+        parser.add_argument(
+            "--summary-path",
+            type=str,
+            default=None,
+            help="Optional path for the aggregate summary JSON.",
+        )
+        parser.add_argument(
+            "--limit",
+            type=int,
+            default=None,
+            help="Optional maximum number of prompt rows to run.",
+        )
+        parser.add_argument(
+            "--resume",
+            action="store_true",
+            help="Skip prompt rows whose trace.json already exists.",
+        )
+        return cast(FlexibleArgumentParser, parser)
+
+
+def cmd_init() -> list[CLISubcommand]:
+    return [InterleaveEvalSubcommand()]

fastvideo/entrypoints/cli/main.py

@@ -2,6 +2,8 @@
 # adapted from vllm: https://github.com/vllm-project/vllm/blob/v0.7.3/vllm/entrypoints/cli/main.py
 from fastvideo.entrypoints.cli.cli_types import CLISubcommand
 from fastvideo.entrypoints.cli.generate import cmd_init as generate_cmd_init
+from fastvideo.entrypoints.cli.interleave_eval import (
+    cmd_init as interleave_eval_cmd_init, )
 from fastvideo.entrypoints.cli.interleave_run import (
     cmd_init as interleave_run_cmd_init, )
 from fastvideo.entrypoints.cli.interleave_serve import (
@@ -19,6 +21,7 @@ def cmd_init() -> list[CLISubcommand]:
     commands = []
     commands.extend(generate_cmd_init())
     commands.extend(serve_cmd_init())
+    commands.extend(interleave_eval_cmd_init())
     commands.extend(interleave_run_cmd_init())
     commands.extend(interleave_serve_cmd_init())
     commands.extend(router_serve_cmd_init())
@@ -39,7 +42,7 @@ def main() -> None:
         cmds[cmd.name] = cmd
 
     args, unknown = parser.parse_known_args()
-    if unknown and args.subparser not in {"generate", "serve", "interleave-run", "interleave-serve"}:
+    if unknown and args.subparser not in {"generate", "serve", "interleave-eval", "interleave-run", "interleave-serve"}:
         parser.error(f"unrecognized arguments: {' '.join(unknown)}")
     args._unknown = unknown
     if args.subparser in cmds:

fastvideo/entrypoints/interleave/__init__.py

@@ -10,6 +10,16 @@ from fastvideo.entrypoints.interleave.config import (
     load_interleave_run_config,
     resolve_interleave_instruction,
 )
+from fastvideo.entrypoints.interleave.evaluation import (
+    InterleavePromptItem,
+    InterleavePromptResult,
+    InterleavePromptSetSummary,
+    load_interleave_prompt_set,
+    prompt_set_summary_to_dict,
+    run_interleave_prompt_set,
+    run_interleave_prompt_set_config,
+    save_prompt_set_summary,
+)
 from fastvideo.entrypoints.interleave.generator import (
     FastVideoImageGeneratorBackend,
     ImageGeneratorBackend,
@@ -61,6 +71,9 @@ __all__ = [
     "InterleaveImageBackendConfig",
     "InterleaveOrchestrator",
     "InterleavePlannerConfig",
+    "InterleavePromptItem",
+    "InterleavePromptResult",
+    "InterleavePromptSetSummary",
     "InterleaveRunConfig",
     "InterleaveRunResult",
     "InterleaveRunStateConfig",
@@ -72,9 +85,14 @@ __all__ = [
     "PlannerProvider",
     "SinglePromptPlanner",
     "build_app",
+    "load_interleave_prompt_set",
     "load_interleave_run_config",
+    "prompt_set_summary_to_dict",
     "resolve_interleave_instruction",
+    "run_interleave_prompt_set",
+    "run_interleave_prompt_set_config",
     "run_interleave_config",
+    "save_prompt_set_summary",
     "save_trace",
     "trace_to_dict",
 ]

fastvideo/entrypoints/interleave/config.py

@@ -110,6 +110,7 @@ def load_interleave_run_config(
     input_image: str | None = None,
     output_dir: str | None = None,
     trace_path: str | None = None,
+    require_instruction: bool = True,
 ) -> InterleaveRunConfig:
     raw = load_raw_config(path)
     raw = _apply_interleave_cli_fields(
@@ -125,7 +126,10 @@ def load_interleave_run_config(
         config.request,
         raw.get("request") if isinstance(raw.get("request"), Mapping) else {},
     )
-    validate_interleave_run_config(config)
+    validate_interleave_run_config(
+        config,
+        require_instruction=require_instruction,
+    )
     return config
 
 
@@ -141,8 +145,13 @@ def resolve_interleave_instruction(config: InterleaveRunConfig) -> str:
     raise ValueError("interleave-run requires interleave.instruction or a single request.prompt")
 
 
-def validate_interleave_run_config(config: InterleaveRunConfig) -> None:
-    resolve_interleave_instruction(config)
+def validate_interleave_run_config(
+    config: InterleaveRunConfig,
+    *,
+    require_instruction: bool = True,
+) -> None:
+    if require_instruction:
+        resolve_interleave_instruction(config)
     if config.image_backend.kind == "fastvideo" and config.generator is None:
         raise ValueError("interleave-run with image_backend.kind=fastvideo requires a generator config")
     if config.planner.kind == "interleave_thinker" and config.planner.max_new_tokens is not None:

fastvideo/entrypoints/interleave/evaluation.py

@@ -0,0 +1,413 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Prompt-set runner and summary metrics for native Interleave workflows."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from fastvideo.entrypoints.interleave.generator import ImageGeneratorBackend
+from fastvideo.entrypoints.interleave.runner import (
+    build_critic,
+    build_image_backend,
+    build_planner,
+)
+from fastvideo.entrypoints.interleave.schema import InterleaveTrace
+from fastvideo.entrypoints.interleave.trace import save_trace
+
+
+@dataclass(frozen=True)
+class InterleavePromptItem:
+    """One prompt-set row for end-to-end Interleave evaluation."""
+
+    sample_id: str
+    instruction: str
+    initial_image_path: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class InterleavePromptResult:
+    sample_id: str
+    instruction: str
+    trace_path: str
+    success: bool
+    attempts: int
+    final_image_path: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+    resumed: bool = False
+
+
+@dataclass(frozen=True)
+class InterleavePromptSetSummary:
+    output_dir: str
+    summary_path: str
+    num_samples: int
+    num_success: int
+    success_rate: float
+    total_attempts: int
+    average_attempts: float
+    num_resumed: int
+    results: list[InterleavePromptResult]
+
+
+def load_interleave_prompt_set(path: str | Path) -> list[InterleavePromptItem]:
+    """Load prompt rows from JSONL, JSON, or plain text files."""
+
+    prompt_path = Path(path)
+    if not prompt_path.exists():
+        raise FileNotFoundError(f"Prompt set not found: {prompt_path}")
+
+    suffix = prompt_path.suffix.lower()
+    if suffix == ".jsonl":
+        raw_items = _load_jsonl(prompt_path)
+    elif suffix == ".json":
+        raw_items = _load_json(prompt_path)
+    elif suffix in {".txt", ".prompts"}:
+        raw_items = [line.strip() for line in prompt_path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    else:
+        raise ValueError(f"Unsupported prompt-set file format: {prompt_path}")
+
+    items = [_coerce_prompt_item(raw, index) for index, raw in enumerate(raw_items)]
+    if not items:
+        raise ValueError(f"Prompt set is empty: {prompt_path}")
+    return items
+
+
+def run_interleave_prompt_set_config(
+    config: Any,
+    prompt_set_path: str | Path,
+    *,
+    output_dir: str | None = None,
+    summary_path: str | None = None,
+    limit: int | None = None,
+    resume: bool = False,
+    image_backend: ImageGeneratorBackend | None = None,
+) -> InterleavePromptSetSummary:
+    """Run a typed Interleave config over a prompt-set file."""
+
+    prompt_items = load_interleave_prompt_set(prompt_set_path)
+    if limit is not None:
+        if limit <= 0:
+            raise ValueError(f"limit must be > 0; got {limit}")
+        prompt_items = prompt_items[:limit]
+    return run_interleave_prompt_set(
+        config,
+        prompt_items,
+        output_dir=output_dir,
+        summary_path=summary_path,
+        resume=resume,
+        image_backend=image_backend,
+    )
+
+
+def run_interleave_prompt_set(
+    config: Any,
+    prompt_items: Sequence[InterleavePromptItem],
+    *,
+    output_dir: str | None = None,
+    summary_path: str | None = None,
+    resume: bool = False,
+    image_backend: ImageGeneratorBackend | None = None,
+) -> InterleavePromptSetSummary:
+    """Run multiple Interleave traces while reusing planner/generator/critic backends."""
+
+    if not prompt_items:
+        raise ValueError("prompt_items must not be empty")
+
+    run_config = deepcopy(config)
+    root = Path(output_dir or run_config.interleave.output_dir)
+    root.mkdir(parents=True, exist_ok=True)
+    run_config.interleave.output_dir = str(root)
+
+    planned_rows = _planned_trace_rows(prompt_items, root)
+    if resume and all(trace_path.exists() for _, item, trace_path in planned_rows):
+        results = [_result_from_saved_trace(item, trace_path) for _, item, trace_path in planned_rows]
+        resolved_summary_path = Path(summary_path or root / "summary.json")
+        summary = _build_summary(
+            results,
+            output_dir=root,
+            summary_path=resolved_summary_path,
+        )
+        save_prompt_set_summary(summary, resolved_summary_path)
+        return summary
+
+    cleanup = lambda: None
+    if image_backend is None:
+        image_backend, cleanup = build_image_backend(run_config)
+
+    try:
+        orchestrator = _build_prompt_set_orchestrator(run_config, image_backend)
+        results: list[InterleavePromptResult] = []
+        for index, item, trace_path in planned_rows:
+            if resume and trace_path.exists():
+                results.append(_result_from_saved_trace(item, trace_path))
+                continue
+
+            trace = orchestrator.run(
+                item.instruction,
+                initial_image_path=item.initial_image_path or run_config.interleave.initial_image_path,
+                metadata=_trace_metadata(item, index),
+            )
+            trace.metadata.update(_trace_metadata(item, index))
+            save_trace(
+                trace,
+                trace_path,
+                include_images=run_config.interleave.include_images_in_trace,
+            )
+            results.append(_result_from_trace(item, trace, trace_path))
+
+        resolved_summary_path = Path(summary_path or root / "summary.json")
+        summary = _build_summary(
+            results,
+            output_dir=root,
+            summary_path=resolved_summary_path,
+        )
+        save_prompt_set_summary(summary, resolved_summary_path)
+        return summary
+    finally:
+        cleanup()
+
+
+def save_prompt_set_summary(
+    summary: InterleavePromptSetSummary,
+    path: str | Path | None = None,
+) -> None:
+    output_path = Path(path or summary.summary_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(
+        json.dumps(
+            prompt_set_summary_to_dict(summary),
+            indent=2,
+            sort_keys=True,
+        ) + "\n",
+        encoding="utf-8",
+    )
+
+
+def prompt_set_summary_to_dict(summary: InterleavePromptSetSummary) -> dict[str, Any]:
+    return {
+        "output_dir": summary.output_dir,
+        "summary_path": summary.summary_path,
+        "num_samples": summary.num_samples,
+        "num_success": summary.num_success,
+        "success_rate": summary.success_rate,
+        "total_attempts": summary.total_attempts,
+        "average_attempts": summary.average_attempts,
+        "num_resumed": summary.num_resumed,
+        "results": [_prompt_result_to_dict(result) for result in summary.results],
+    }
+
+
+def _build_prompt_set_orchestrator(
+    config: Any,
+    image_backend: ImageGeneratorBackend,
+) -> Any:
+    from fastvideo.entrypoints.interleave.orchestrator import InterleaveOrchestrator
+
+    return InterleaveOrchestrator(
+        planner=build_planner(config.planner),
+        generator=image_backend,
+        critic=build_critic(config.critic),
+    )
+
+
+def _load_jsonl(path: Path) -> list[Any]:
+    rows: list[Any] = []
+    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
+        if not line.strip():
+            continue
+        try:
+            rows.append(json.loads(line))
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"Invalid JSONL row in {path}:{line_number}: {exc}") from exc
+    return rows
+
+
+def _load_json(path: Path) -> list[Any]:
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(raw, list):
+        return raw
+    if isinstance(raw, Mapping):
+        for key in ("items", "prompts", "samples"):
+            value = raw.get(key)
+            if isinstance(value, list):
+                return value
+        return [raw]
+    raise ValueError(f"{path} must contain a prompt list or mapping")
+
+
+def _coerce_prompt_item(raw: Any, index: int) -> InterleavePromptItem:
+    if isinstance(raw, str):
+        return InterleavePromptItem(
+            sample_id=f"sample_{index:05d}",
+            instruction=raw,
+        )
+    if not isinstance(raw, Mapping):
+        raise ValueError(f"Prompt row {index} must be a mapping or string")
+
+    instruction = _first_text(raw, "instruction", "prompt", "text")
+    if not instruction:
+        raise ValueError(f"Prompt row {index} requires instruction, prompt, or text")
+    sample_id = _first_text(raw, "id", "sample_id", "name") or f"sample_{index:05d}"
+    initial_image_path = _first_text(raw, "initial_image_path", "input_image", "image_path", "image")
+
+    metadata: dict[str, Any] = {}
+    raw_metadata = raw.get("metadata")
+    if isinstance(raw_metadata, Mapping):
+        metadata.update(dict(raw_metadata))
+    reserved = {
+        "id",
+        "sample_id",
+        "name",
+        "instruction",
+        "prompt",
+        "text",
+        "initial_image_path",
+        "input_image",
+        "image_path",
+        "image",
+        "metadata",
+    }
+    for key, value in raw.items():
+        if key not in reserved:
+            metadata[str(key)] = value
+
+    return InterleavePromptItem(
+        sample_id=str(sample_id),
+        instruction=str(instruction),
+        initial_image_path=str(initial_image_path) if initial_image_path else None,
+        metadata=metadata,
+    )
+
+
+def _first_text(row: Mapping[str, Any], *keys: str) -> str | None:
+    for key in keys:
+        value = row.get(key)
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _trace_metadata(item: InterleavePromptItem, index: int) -> dict[str, Any]:
+    return {
+        "prompt_set_id": item.sample_id,
+        "prompt_set_index": index,
+        "prompt_set_metadata": dict(item.metadata),
+    }
+
+
+def _result_from_trace(
+    item: InterleavePromptItem,
+    trace: InterleaveTrace,
+    trace_path: Path,
+) -> InterleavePromptResult:
+    return InterleavePromptResult(
+        sample_id=item.sample_id,
+        instruction=item.instruction,
+        trace_path=str(trace_path),
+        success=trace.success,
+        attempts=len(trace.attempts),
+        final_image_path=(trace.final_image.file_path if trace.final_image is not None else None),
+        metadata=dict(item.metadata),
+    )
+
+
+def _result_from_saved_trace(
+    item: InterleavePromptItem,
+    trace_path: Path,
+) -> InterleavePromptResult:
+    payload = json.loads(trace_path.read_text(encoding="utf-8"))
+    final_image = payload.get("final_image")
+    return InterleavePromptResult(
+        sample_id=item.sample_id,
+        instruction=item.instruction,
+        trace_path=str(trace_path),
+        success=bool(payload.get("success")),
+        attempts=len(payload.get("attempts") or []),
+        final_image_path=(final_image.get("file_path") if isinstance(final_image, Mapping) else None),
+        metadata=dict(item.metadata),
+        resumed=True,
+    )
+
+
+def _build_summary(
+    results: Sequence[InterleavePromptResult],
+    *,
+    output_dir: Path,
+    summary_path: Path,
+) -> InterleavePromptSetSummary:
+    num_samples = len(results)
+    num_success = sum(1 for result in results if result.success)
+    total_attempts = sum(result.attempts for result in results)
+    return InterleavePromptSetSummary(
+        output_dir=str(output_dir),
+        summary_path=str(summary_path),
+        num_samples=num_samples,
+        num_success=num_success,
+        success_rate=(num_success / num_samples if num_samples else 0.0),
+        total_attempts=total_attempts,
+        average_attempts=(total_attempts / num_samples if num_samples else 0.0),
+        num_resumed=sum(1 for result in results if result.resumed),
+        results=list(results),
+    )
+
+
+def _prompt_result_to_dict(result: InterleavePromptResult) -> dict[str, Any]:
+    return {
+        "sample_id": result.sample_id,
+        "instruction": result.instruction,
+        "trace_path": result.trace_path,
+        "success": result.success,
+        "attempts": result.attempts,
+        "final_image_path": result.final_image_path,
+        "metadata": dict(result.metadata),
+        "resumed": result.resumed,
+    }
+
+
+def _planned_trace_rows(
+    prompt_items: Sequence[InterleavePromptItem],
+    root: Path,
+) -> list[tuple[int, InterleavePromptItem, Path]]:
+    seen_ids: dict[str, int] = {}
+    rows: list[tuple[int, InterleavePromptItem, Path]] = []
+    for index, item in enumerate(prompt_items):
+        sample_dir = root / _unique_sample_dir_name(item.sample_id, index, seen_ids)
+        rows.append((index, item, sample_dir / "trace.json"))
+    return rows
+
+
+def _unique_sample_dir_name(
+    sample_id: str,
+    index: int,
+    seen_ids: dict[str, int],
+) -> str:
+    base = _safe_sample_id(sample_id) or f"sample_{index:05d}"
+    count = seen_ids.get(base, 0)
+    seen_ids[base] = count + 1
+    if count:
+        return f"{base}_{count + 1}"
+    return base
+
+
+def _safe_sample_id(sample_id: str) -> str:
+    sanitized = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(sample_id)).strip("._-")
+    return sanitized[:96]
+
+
+__all__ = [
+    "InterleavePromptItem",
+    "InterleavePromptResult",
+    "InterleavePromptSetSummary",
+    "load_interleave_prompt_set",
+    "prompt_set_summary_to_dict",
+    "run_interleave_prompt_set",
+    "run_interleave_prompt_set_config",
+    "save_prompt_set_summary",
+]

tests/local_tests/test_interleave_run_cli.py

@@ -11,8 +11,11 @@ from fastvideo.entrypoints.cli.main import cmd_init
 from fastvideo.entrypoints.interleave import (
     GeneratedImage,
     InterleaveEditRequest,
+    InterleavePromptItem,
+    load_interleave_prompt_set,
     load_interleave_run_config,
     resolve_interleave_instruction,
+    run_interleave_prompt_set,
     run_interleave_config,
 )
 from fastvideo.entrypoints.interleave.orchestrator import SinglePromptPlanner
@@ -65,6 +68,18 @@ def test_interleave_run_config_rejects_unknown_override_prefix(tmp_path: Path) -
         )
 
 
+def test_interleave_eval_config_can_omit_single_instruction(tmp_path: Path) -> None:
+    config_path = _write_run_config(tmp_path, include_prompt=False)
+
+    config = load_interleave_run_config(
+        str(config_path),
+        require_instruction=False,
+    )
+
+    with pytest.raises(ValueError, match="requires interleave.instruction"):
+        resolve_interleave_instruction(config)
+
+
 def test_interleave_cli_defers_nested_config_loading(tmp_path: Path) -> None:
     config_path = _write_run_config(tmp_path)
     parser = FlexibleArgumentParser(description="FastVideo CLI")
@@ -85,6 +100,31 @@ def test_interleave_cli_defers_nested_config_loading(tmp_path: Path) -> None:
     assert unknown == ["--request.sampling.width", "256"]
 
 
+def test_interleave_eval_cli_defers_nested_config_loading(tmp_path: Path) -> None:
+    config_path = _write_run_config(tmp_path, include_prompt=False)
+    prompt_path = tmp_path / "prompts.jsonl"
+    prompt_path.write_text('{"id": "mug", "prompt": "draw a mug"}\n', encoding="utf-8")
+    parser = FlexibleArgumentParser(description="FastVideo CLI")
+    subparsers = parser.add_subparsers(required=False, dest="subparser")
+    for cmd in cmd_init():
+        cmd.subparser_init(subparsers).set_defaults(dispatch_function=cmd.cmd)
+
+    args, unknown = parser.parse_known_args([
+        "interleave-eval",
+        "--config",
+        str(config_path),
+        "--prompts",
+        str(prompt_path),
+        "--request.sampling.width",
+        "256",
+    ])
+
+    assert args.subparser == "interleave-eval"
+    assert args.config == str(config_path)
+    assert args.prompts == str(prompt_path)
+    assert unknown == ["--request.sampling.width", "256"]
+
+
 def test_run_interleave_config_with_injected_backend_writes_trace(tmp_path: Path) -> None:
     config_path = _write_run_config(tmp_path)
     config = load_interleave_run_config(str(config_path))
@@ -102,6 +142,85 @@ def test_run_interleave_config_with_injected_backend_writes_trace(tmp_path: Path
     assert backend.requests[0].prompt == "draw a red mug"
 
 
+def test_load_interleave_prompt_set_accepts_jsonl_rows(tmp_path: Path) -> None:
+    prompt_path = tmp_path / "prompts.jsonl"
+    prompt_path.write_text(
+        '{"id": "mug", "prompt": "draw a mug", "metadata": {"split": "smoke"}, "difficulty": "easy"}\n'
+        '"draw a kettle"\n',
+        encoding="utf-8",
+    )
+
+    items = load_interleave_prompt_set(prompt_path)
+
+    assert [item.sample_id for item in items] == ["mug", "sample_00001"]
+    assert items[0].instruction == "draw a mug"
+    assert items[0].metadata == {
+        "split": "smoke",
+        "difficulty": "easy",
+    }
+    assert items[1].instruction == "draw a kettle"
+
+
+def test_run_interleave_prompt_set_writes_traces_and_summary(tmp_path: Path) -> None:
+    config_path = _write_run_config(tmp_path, include_prompt=False)
+    config = load_interleave_run_config(
+        str(config_path),
+        require_instruction=False,
+    )
+    backend = _FakeImageBackend(tmp_path / "generated.png")
+    items = [
+        InterleavePromptItem(sample_id="red/mug", instruction="draw a red mug"),
+        InterleavePromptItem(sample_id="blue mug", instruction="draw a blue mug"),
+    ]
+
+    summary = run_interleave_prompt_set(
+        config,
+        items,
+        output_dir=str(tmp_path / "eval"),
+        image_backend=backend,
+    )
+
+    assert summary.num_samples == 2
+    assert summary.num_success == 2
+    assert summary.success_rate == 1.0
+    assert summary.total_attempts == 2
+    assert [request.prompt for request in backend.requests] == ["draw a red mug", "draw a blue mug"]
+    assert Path(summary.summary_path).exists()
+    assert Path(summary.results[0].trace_path).exists()
+    assert "red_mug" in summary.results[0].trace_path
+    summary_text = Path(summary.summary_path).read_text(encoding="utf-8")
+    assert "draw a blue mug" in summary_text
+
+
+def test_run_interleave_prompt_set_resume_uses_existing_trace(tmp_path: Path) -> None:
+    config_path = _write_run_config(tmp_path, include_prompt=False)
+    config = load_interleave_run_config(
+        str(config_path),
+        require_instruction=False,
+    )
+    item = InterleavePromptItem(sample_id="mug", instruction="draw a mug")
+    backend = _FakeImageBackend(tmp_path / "generated.png")
+    first = run_interleave_prompt_set(
+        config,
+        [item],
+        output_dir=str(tmp_path / "eval"),
+        image_backend=backend,
+    )
+
+    resumed = run_interleave_prompt_set(
+        config,
+        [item],
+        output_dir=str(tmp_path / "eval"),
+        image_backend=_FakeImageBackend(tmp_path / "unused.png"),
+        resume=True,
+    )
+
+    assert first.num_resumed == 0
+    assert resumed.num_resumed == 1
+    assert resumed.results[0].resumed is True
+    assert resumed.results[0].trace_path == first.results[0].trace_path
+
+
 def test_single_prompt_planner_uses_configured_attempt_count() -> None:
     planner = SinglePromptPlanner(max_attempts=4)
 
@@ -134,9 +253,10 @@ class _FakeImageBackend:
         )
 
 
-def _write_run_config(tmp_path: Path) -> Path:
+def _write_run_config(tmp_path: Path, *, include_prompt: bool = True) -> Path:
     output_path = tmp_path / "generated.png"
     config_path = tmp_path / "interleave_run.yaml"
+    request_prompt = "  prompt: draw a red mug\n" if include_prompt else ""
     config_path.write_text(
         f"""
 generator:
@@ -161,14 +281,13 @@ interleave:
   trace_path: {tmp_path / "trace.json"}
 
 request:
-  prompt: draw a red mug
+{request_prompt}  extensions:
+    test_output_path: {output_path}
   sampling:
     width: 512
     height: 512
     seed: 7
     num_inference_steps: 4
-  extensions:
-    test_output_path: {output_path}
 """,
         encoding="utf-8",
     )