Editing Plans

VideoEdit is a multi-segment editing plan modeled as a Pydantic BaseModel. Each segment selects a time range from a source video and carries an ordered list of Operation instances to run against it.

At a Glance

• One operations list per segment — transforms and effects are sequenced together.
• post_operations runs against the concatenated result.
• validate() is a dry-run via metadata; no frames are loaded.
• run_to_file() streams directly to disk — the only execution engine.

Quick Start

from videopython.editing import VideoEdit

plan = {
    "segments": [
        {
            "source": "input.mp4",
            "start": 5.0,
            "end": 12.0,
            "operations": [
                {"op": "crop", "width": 0.5, "height": 1.0, "mode": "center"},
                {"op": "resize", "width": 1080, "height": 1920},
                {
                    "op": "blur_effect",
                    "mode": "constant",
                    "iterations": 1,
                    "window": {"start": 0.0, "stop": 1.0},
                },
            ],
        },
        {"source": "input.mp4", "start": 20.0, "end": 28.0},
    ],
    "post_operations": [
        {"op": "color_adjust", "brightness": 0.05},
    ],
}

edit = VideoEdit.from_dict(plan)
predicted = edit.validate()        # dry-run via VideoMetadata
edit.run_to_file("output.mp4", crf=20, preset="medium")  # streams to disk (constant memory, any video length)

JSON Plan Format

{
  "segments": [
    {
      "source": "path/to/video.mp4",
      "start": 5.0,
      "end": 15.0,
      "operations": [
        {"op": "resize", "width": 1080, "height": 1920},
        {"op": "blur_effect", "mode": "constant", "iterations": 2,
         "window": {"start": 0.0, "stop": 3.0}}
      ]
    }
  ],
  "post_operations": [
    {"op": "color_adjust", "brightness": 0.05}
  ],
  "match_to_lowest_fps": true,
  "match_to_lowest_resolution": true
}

Rules:

• segments is required and must be non-empty.
• Each op object has an op discriminator field; remaining fields belong to that op's Pydantic schema. Unknown fields are rejected.
• Effect time windows go in the op's window field ({"start": s, "stop": e}). Either endpoint may be omitted.
• Top-level and segment-level keys are strict (extra="forbid").

Pipeline Order

VideoEdit runs each segment's operations in order, concatenates the results, then applies post_operations to the assembled output.

The Streaming Engine

run_to_file() is the only execution engine. It streams ffmpeg decode → filter/effect chain → ffmpeg encode, so memory stays constant (~a frame at a time) regardless of video length.

Each operation is one of two kinds:

• a filter — compiles to a native ffmpeg filter (op.to_ffmpeg_filter(ctx)). This is reserved for ops ffmpeg does that numpy can't (or can't do as well): all transforms (resize, crop, resample_fps, the duration-changing speed_change/freeze_frame, the transcription-consuming silence_removal, and face_crop from the ai extra), and the two text-rendering effects text_overlay (drawtext) and add_subtitles (libass).
• a per-frame effect — shape-preserving Python over each decoded frame (streaming_init + process_frame). Every pixel effect lives here: blur, sharpen, zoom, film_grain, chromatic_aberration, mirror_flip, vignette, color_adjust, kaleidoscope, shake, flash, glitch, pixelate, ken_burns, punch_in, and the image overlays. These run vectorised numpy/cv2 — benchmarks showed the native-filter equivalents were at best ~1.1–1.4x faster (the win came from skipping the rawvideo round-trip, not faster math) and in some cases slower (gblur), so the per-frame path is kept for its simplicity and exact output.

A segment whose ops are all filters renders in a single ffmpeg invocation — no rawvideo round-trip, no Python loop. A per-frame effect switches that segment to the decode → Python → encode pipeline, where filters before the effect join the decode chain and filters after it the encode chain (so [fade, add_subtitles] streams). Duration-changing transforms fold their predicted metadata through the chain, so later effect windows and the audio follow the new timeline.

post_operations run as a second pass over the assembled program, so any op — filter, effect, or transform — applies to the whole concatenated timeline.

A few plan shapes have no streaming strategy and are rejected with structured STREAMING_UNSUPPORTED errors before any decode: a per-frame effect ordered after encode-stage filters, a context-requiring op after a duration-changing transform, face_crop behind per-frame effects, and a time-based-context post-op on a multi-segment plan (its source-absolute context can't re-base onto the concat). Reorder the ops, or move the op into a segment, to fix.

add_subtitles renders via libass — the transcription is compiled to an ASS document at plan-compile time and burned in by ffmpeg's subtitles= filter (needs an ffmpeg built with libass). Pass context= to run_to_file.

Streamability report

edit.streamability() classifies every op by streaming class without touching the disk — filter (ffmpeg filter chain), frame_effect (process_frame), or unstreamable (the plan is rejected, with the reason and a reorder hint on the entry):

report = edit.streamability()
report.streamable        # will the plan run?
report.unstreamable      # the offending ops, with reasons
report.errors()          # the same as structured STREAMING_UNSUPPORTED PlanErrors

edit.check(meta) reports the same STREAMING_UNSUPPORTED errors after the regular validity errors, and run_to_file() raises them before any decode. Streamability is purely structural (op classes, order, plan shape), so a consumer can gate job admission on the report before downloading sources.

Context Data

Operations that need side-channel data (e.g. silence_removal and add_subtitles need a transcription) declare it via requires: ClassVar[tuple[str, ...]]. The runner picks matching keys out of the context dict and threads them into the streaming compile — predict_metadata and either the effect's streaming_init or the filter-compiled op's FilterCtx.context:

edit = VideoEdit.from_dict(plan)
edit.run_to_file("out.mp4", context={"transcription": my_transcription})

Time-based context values are re-based onto each cut segment's local timeline, and the resolved values are delivered to the effect's streaming_init.

Validation

VideoEdit.validate() chains Operation.predict_metadata across the plan and checks:

• segment end is within source duration
• each operation's metadata prediction succeeds
• effect window is within the predicted segment duration
• concatenation compatibility (exact fps + dimensions)

Returns the predicted final VideoMetadata. On failure it raises PlanValidationError (a ValueError subclass, so except ValueError keeps working) carrying structured .errors — each a PlanError with a code, location, and the offending field/value/limit.

For dry-run validation without disk access, pass a pre-built VideoMetadata to validate_with_metadata(meta_or_dict, context=...).

Parse vs. validate

Parsing (from_dict) owns the plan shape (field types, required fields, unknown ops). The numeric bounds of the skeleton — segment start/end and effect window ranges — are owned by validation, not parse: a negative window.start or a start >= end segment parses and is reported by validate/check. This keeps every numeric violation a structured, collectable, repairable PlanError.

Collecting every error (check)

check(source_metadata, context=..., clamp_windows=...) is the non-raising sibling of validate_with_metadata: it runs the same dry-run but accumulates every PlanError and returns the list ([] == valid), best-effort isolating failures per segment/op. Use it to re-prompt an LLM once with all problems instead of one-at-a-time.

Repairing the mechanical violations (repair)

repair(source_metadata, context=..., clamp_op_params=True, clamp_segment_end=False) returns (repaired_edit, repairs) — a corrected deep copy plus a list[PlanRepair] changelog (location, field, old, new, code), leaving the original untouched. It clamps only the unambiguous cases:

• effect window.start/window.stop into [0, duration] (segment ops and post_operations);
• op time fields past the clip end (e.g. freeze_frame.timestamp), generic via each op's declared time_fields;
• a negative segment start → 0, and with clamp_segment_end=True a segment end past the source → the source end.

It never invents intent — a concat mismatch or end <= start is left for check() / re-prompting — and never raises on an unrepairable op. A clampable window.stop overrun (a duration-shrinking op before a windowed effect) is the case run_to_file() already tolerates; validate(clamp_windows=True) and check(..., clamp_windows=True) won't report it either.

Normalizing concat geometry (normalize_dimensions)

normalize_dimensions(source_metadata, target, context=...) appends a per-segment resize to a common canvas — target is an explicit (width, height), "first", "largest", or "match" (the lowest common resolution, the same policy match_to_lowest_resolution applies in-stream) — so the "all segments share dimensions" concat invariant holds by construction. Best-effort and non-raising like repair()/check(): a segment it can't yet predict is left untouched for check() to report. Returns (normalized_edit, repairs).

Matching Sources

When multiple segments draw from sources with different fps/resolution, VideoEdit auto-matches:

• match_to_lowest_fps (default true) — resample all segments to the lowest source fps.
• match_to_lowest_resolution (default true) — resize all segments to the lowest source resolution.

Set either flag to false to require sources match natively; otherwise validate() / run_to_file() raises.

JSON Schema (json_schema)

VideoEdit.json_schema() returns a JSON Schema for the wire format, including the discriminated union over every LLM-exposed Operation (server-only ops like image_overlay are excluded — see Operations). Pass it to any LLM API as a tool/function schema or structured-output format. AI operations appear in the union only after import videopython.ai has run.

schema = VideoEdit.json_schema()
# tools=[{"input_schema": schema}]            # Anthropic
# tools=[{"type": "function", "function": {"parameters": schema}}]  # OpenAI

Pass strict=True (VideoEdit.json_schema(strict=True) / Operation.json_schema(strict=True)) for a submittable provider strict-mode grammar: every object closed (additionalProperties: false), every property required (optionality follows the Pydantic type — genuinely optional fields stay nullable, defaulted-but-required ones keep their concrete type, so a grammar-valid response always parses back), the op union expressed as an anyOf of closed variants without a discriminator, and the union's $defs hoisted to the document root so every $ref resolves. Numeric constraints are preserved, so grammar-constrained decoding makes simple bound violations impossible at decode time.

API Reference

VideoEdit

VideoEdit

Bases: BaseModel

A multi-segment editing plan.

Parse vs. validate. Parsing (from_dict/model_validate) owns the shape: field types, required fields, unknown-op/extra-field rejection, and op-local structural rules (e.g. resize needs a dimension) surface as a Pydantic ValidationError. The numeric bounds of the plan skeleton -- segment start/end and effect window ranges -- are deliberately not enforced at parse; they are owned by :meth:validate / :meth:check / :meth:repair, which report them as structured :class:PlanErrors. This keeps one code path for the LLM refine loop: from_dict (permissive) -> :meth:repair (clamp the mechanical ones) -> :meth:check (collect whatever remains) -> re-prompt with the full structured error list.

Source code in src/videopython/editing/video_edit.py

class VideoEdit(BaseModel):
    """A multi-segment editing plan.

    **Parse vs. validate.** Parsing (``from_dict``/``model_validate``) owns the
    *shape*: field types, required fields, unknown-op/extra-field rejection, and
    op-local structural rules (e.g. ``resize`` needs a dimension) surface as a
    Pydantic ``ValidationError``. The numeric *bounds* of the plan skeleton --
    segment ``start``/``end`` and effect ``window`` ranges -- are deliberately
    **not** enforced at parse; they are owned by :meth:`validate` / :meth:`check`
    / :meth:`repair`, which report them as structured :class:`PlanError`s. This
    keeps one code path for the LLM refine loop: ``from_dict`` (permissive) ->
    :meth:`repair` (clamp the mechanical ones) -> :meth:`check` (collect whatever
    remains) -> re-prompt with the full structured error list.
    """

    model_config = ConfigDict(extra="forbid")

    segments: list[SegmentConfig] = Field(
        min_length=1,
        description=(
            "Ordered list of segments. Each segment selects a time range from a "
            "source video and applies its `operations` to it; results are "
            "concatenated in order."
        ),
    )
    post_operations: list[OperationInput] = Field(
        default_factory=list,
        description="Operations applied to the concatenated output after all segments are joined.",
    )
    match_to_lowest_fps: bool = Field(
        True,
        description=(
            "When concatenating multiple segments with different fps, resample "
            "all of them to the lowest source fps. If false, mismatched fps "
            "raises during validation."
        ),
    )
    match_to_lowest_resolution: bool = Field(
        True,
        description=(
            "When concatenating multiple segments with different resolutions, "
            "resize all of them to the lowest source resolution. If false, "
            "mismatched dimensions raise during validation."
        ),
    )
    music_bed: MusicBed | None = Field(
        default=None,
        description=(
            "Optional music bed mixed under the whole assembled program in a final pass "
            "(after segment concat / transitions). Set `duck` on it to lower the bed under "
            "transcription-derived speech (single-segment plans only)."
        ),
    )

    # ------------------------------------------------------------------ I/O

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> VideoEdit:
        return cls.model_validate(data)

    @classmethod
    def from_json(cls, text: str) -> VideoEdit:
        try:
            data = json.loads(text)
        except json.JSONDecodeError as e:
            raise ValueError(f"Invalid VideoEdit JSON: {e.msg} at line {e.lineno} column {e.colno}") from e
        return cls.model_validate(data)

    def to_dict(self) -> dict[str, Any]:
        return self.model_dump(mode="json", exclude_none=False)

    @classmethod
    def json_schema(cls, *, strict: bool = False) -> dict[str, Any]:
        """LLM-facing schema: a discriminated union of operations per slot.

        A thin transform over the Pydantic models, so it cannot drift from
        them: the operations union is :meth:`Operation.json_schema` (LLM-exposed
        ops by default, per the ``llm_exposed`` ClassVar), and every other field
        shape and ``description`` is derived from ``SegmentConfig``/``VideoEdit``
        ``model_json_schema()`` rather than hand-typed. Adding a model field thus
        surfaces here automatically; in particular ``source`` carries its
        ``"format": "path"``. The Draft-07 ``$schema`` envelope and the default
        ``operations`` array shape are preserved for downstream LLM tooling.

        With ``strict=True`` the result is a submittable provider strict-mode
        grammar: a closed object root, all properties ``required`` (optionality
        kept as Pydantic emitted it -- no synthesized nulls), the op union as
        ``anyOf`` of closed variants, and the union's ``$defs`` hoisted to the
        document root so every ``$ref`` resolves. See :meth:`Operation.json_schema`
        for the strict-mode contract. Use it as a ``response_format: json_schema``
        grammar so simple bound violations (``window.start >= 0``, enums, required
        fields) become impossible at decode time. Cross-field constraints
        (``timestamp < duration``, segment-dim equality) cannot live in a grammar
        and stay with :meth:`check` / :meth:`repair` / :meth:`normalize_dimensions`.
        """
        # Build the permissive envelope, then (if strict) run one closing pass
        # over the whole thing -- including the embedded op union -- so the
        # hand-built segment/top-level objects are closed and required too.
        # `op_schema` is a self-contained union carrying its own root `$defs`;
        # the same object is inlined as the `items` of both operation slots.
        op_schema = Operation.json_schema()

        segment_schema: dict[str, Any] = {
            "type": "object",
            "description": SegmentConfig.__doc__,
            "properties": {
                "source": field_schema(SegmentConfig, "source"),
                "start": field_schema(SegmentConfig, "start"),
                "end": field_schema(SegmentConfig, "end"),
                "operations": array_field_schema(SegmentConfig, "operations", op_schema),
                "transition_in": optional_model_field_schema(TransitionSpec, SegmentConfig, "transition_in"),
            },
            "required": ["source", "start", "end"],
            "additionalProperties": False,
        }
        segments = field_schema(cls, "segments")
        segments["items"] = segment_schema
        schema = {
            "$schema": "http://json-schema.org/draft-07/schema#",
            "type": "object",
            "description": cls.__doc__,
            "properties": {
                "segments": segments,
                "post_operations": array_field_schema(cls, "post_operations", op_schema),
                "match_to_lowest_fps": field_schema(cls, "match_to_lowest_fps"),
                "match_to_lowest_resolution": field_schema(cls, "match_to_lowest_resolution"),
                "music_bed": optional_model_field_schema(MusicBed, cls, "music_bed"),
            },
            "required": ["segments"],
            "additionalProperties": False,
        }
        if not strict:
            return schema
        # Inlining the op union as `items` buried its `$defs` while its `$ref`s
        # stayed root-relative (`#/$defs/X`), so they dangled. Hoist the (shared)
        # defs to this schema's root before the strict pass, so every ref
        # resolves and the result is a submittable provider grammar.
        op_defs = op_schema.pop("$defs", None)
        if op_defs:
            schema["$defs"] = op_defs
        return _to_strict_schema(schema)

    # --------------------------------------------------------------- validate

    def validate(  # type: ignore[override]
        self,
        context: dict[str, Any] | None = None,
        *,
        clamp_windows: bool = False,
    ) -> VideoMetadata:
        """Dry-run the plan via metadata. Requires source files on disk.

        Shadows Pydantic v1's deprecated ``BaseModel.validate`` classmethod;
        use ``VideoEdit.from_dict``/``model_validate`` for plan parsing.

        When ``clamp_windows`` is True, an :class:`Effect`'s ``window.stop`` that
        overruns the running predicted duration (e.g. after a duration-shrinking
        op like ``speed_change``/``cut``) is clamped to that duration -- the same
        ``min(stop, total_seconds)`` value the streaming engine applies at run
        time -- instead of raising. Only ``window.stop`` is clamped: a ``window.start`` past the
        duration still hard-raises (a residual divergence from ``run_to_file()``, which
        degrades it to a zero-width no-op).
        """
        source_metas = [VideoMetadata.from_path(str(seg.source)) for seg in self.segments]
        return self._validate(source_metas, context, clamp_windows=clamp_windows)

    def validate_with_metadata(
        self,
        source_metadata: VideoMetadata | dict[str, VideoMetadata],
        context: dict[str, Any] | None = None,
        *,
        clamp_windows: bool = False,
    ) -> VideoMetadata:
        """Dry-run with pre-built metadata, avoiding disk access.

        See :meth:`validate` for the ``clamp_windows`` semantics.
        """
        metas = self._resolve_source_metas(source_metadata)
        return self._validate(metas, context, clamp_windows=clamp_windows)

    def check(
        self,
        source_metadata: VideoMetadata | dict[str, VideoMetadata],
        context: dict[str, Any] | None = None,
        *,
        clamp_windows: bool = False,
    ) -> list[PlanError]:
        """Collect **every** plan error in one pass; ``[]`` means valid.

        The non-raising sibling of :meth:`validate_with_metadata`: it runs the
        same dry-run but accumulates instead of aborting on the first failure,
        so an LLM refine loop can fix all problems in a single re-prompt instead
        of playing whack-a-mole across a retry budget. Best-effort: each segment
        is checked against its own source metadata, per-op and per-segment errors
        collected; a check that cannot run because an earlier one failed (a
        segment's op chain past a bad cut, the cross-segment concat check when a
        segment did not produce an output) is skipped rather than aborting.

        Returns the same :class:`PlanError` list that
        :attr:`PlanValidationError.errors` carries -- every failure is structured
        (no bare ``ValueError`` escapes the walk), so a consumer branches on
        ``code`` rather than substring-matching prose. ``clamp_windows`` matches
        :meth:`validate`: a clampable ``window.stop`` overrun is not reported.

        Streaming is the only engine, so ops that cannot stream at their
        plan position are real plan errors: one ``STREAMING_UNSUPPORTED`` per
        offending op is appended after the validity errors, in plan order,
        with the actionable cause in :attr:`PlanError.detail`. See
        :meth:`streamability` for the full per-op report including the ops
        that *do* stream.
        """
        metas = self._resolve_source_metas(source_metadata)
        _, errors = self._collect(metas, context, clamp_windows=clamp_windows, stop_first=False)
        errors.extend(self.streamability().errors())
        return errors

    def streamability(self) -> StreamabilityReport:
        """Classify every op by streaming class, without touching the disk.

        Streamability is purely structural -- it depends on op classes, their
        order, and the plan shape, never on source metadata or runtime context
        -- so this needs no source files and is safe to call before a job is
        admitted. ``report.streamable`` answers "will :meth:`run_to_file`
        stream this plan in O(1) memory, or is an op unstreamable at its
        plan position?"; each entry carries the op's memory class and, for
        unstreamable ops, the reason.
        """
        return analyze_streamability(
            [list(seg.operations) for seg in self.segments],
            list(self.post_operations),
        )

    def repair(
        self,
        source_metadata: VideoMetadata | dict[str, VideoMetadata],
        context: dict[str, Any] | None = None,
        *,
        clamp_op_params: bool = True,
        clamp_segment_end: bool = False,
    ) -> tuple[VideoEdit, list[PlanRepair]]:
        """Return a copy of this plan with the *unambiguous* violations clamped.

        Walks the chain (cut, fps/resolution matching, per-op prediction) and
        clamps only the mechanical faults whose fix is not a judgement call,
        recording each as a :class:`PlanRepair`. The returned plan is a deep copy
        (``self`` is untouched); the changelog is meant to be surfaced to the
        user ("we trimmed your effect to fit"). ``repair`` never *invents* intent
        -- genuinely semantic problems (a concat dimension mismatch, an
        ``end <= start`` range) are left for :meth:`check` / re-prompting.

        With ``clamp_op_params`` (default ``True``) it clamps each effect
        ``window.start``/``window.stop`` into ``[0, duration]`` and each declared
        :attr:`Operation.time_fields` value (e.g. ``freeze_frame.timestamp`` past
        the clip end) into range, plus a negative segment ``start`` to ``0``.
        With ``clamp_segment_end`` (default ``False``, since it changes editorial
        intent) it also clamps a segment ``end`` past the source to the source
        end; left ``False``, that case hard-raises as before. Always
        :meth:`check` / :meth:`validate` the returned plan before running it.
        """
        metas = self._resolve_source_metas(source_metadata)
        repairs: list[PlanRepair] = []

        # Pass 1: segment-level start/end repairs, and the cut metadata each
        # surviving segment's op clamps run against.
        fixed_segments: list[SegmentConfig] = []
        cut_metas: list[VideoMetadata | None] = []
        for i, (seg, meta) in enumerate(zip(self.segments, metas)):
            start, end = seg.start, seg.end
            if clamp_op_params and start < 0:
                repairs.append(PlanRepair(f"segments[{i}]", "start", start, 0.0, PlanErrorCode.SEGMENT_NEGATIVE))
                start = 0.0
            if end > meta.total_seconds + DURATION_EPS:
                if clamp_segment_end:
                    new_end = round(meta.total_seconds, 4)
                    repairs.append(
                        PlanRepair(f"segments[{i}]", "end", end, new_end, PlanErrorCode.SEGMENT_END_EXCEEDS_SOURCE)
                    )
                    end = new_end
                else:
                    message, err = _segment_end_exceeds_source(i, seg, meta)
                    raise PlanValidationError(message, [err])
            seg = seg.model_copy(update={"start": start, "end": end})
            fixed_segments.append(seg)
            repairable = start >= 0 and end > start and end <= meta.total_seconds + DURATION_EPS
            cut_metas.append(CutSeconds(start=start, end=end).predict_metadata(meta) if repairable else None)

        # Pass 2: per-segment op clamps against the matched cut meta. Capture each
        # segment's predicted output -- needed to assemble the post-op timeline.
        matched_by_index = self._match_cuttable(cut_metas)
        final_segments: list[SegmentConfig] = []
        seg_output_metas: list[VideoMetadata | None] = []
        for i, seg in enumerate(fixed_segments):
            seg_meta = matched_by_index.get(i)
            if seg_meta is None:
                final_segments.append(seg)
                seg_output_metas.append(None)
                continue
            seg_context = _segment_context(context, str(seg.source), seg.start, seg.end)
            new_ops, op_repairs, out_meta = _repair_op_chain(
                list(seg.operations),
                seg_meta,
                seg_context,
                f"segments[{i}].operations",
                clamp=clamp_op_params,
            )
            repairs.extend(op_repairs)
            final_segments.append(seg.model_copy(update={"operations": new_ops}))
            seg_output_metas.append(out_meta)

        # Pass 2b: clamp each transition's overlap down to one frame short of the
        # shorter adjacent segment -- the mechanical TRANSITION_TOO_LONG repair,
        # mirroring the window clamps. The clamp is frame-safe: it targets
        # `limit_frames - 1` overlap frames so the repaired plan satisfies the
        # strict `overlap < min(frames)` streaming constraint and actually runs
        # (a seconds clamp to the segment length lands on `overlap == frames`,
        # which passes check but crashes run_to_file). Only fires when both adjacent
        # segments predicted; a segment too short for any transition
        # (limit_frames <= 1) is left for check. A first-segment transition is a
        # structural fault, deliberately left for check.
        if clamp_op_params:
            for i in range(1, len(final_segments)):
                spec = final_segments[i].transition_in
                if spec is None:
                    continue
                left_meta = seg_output_metas[i - 1]
                right_meta = seg_output_metas[i]
                if left_meta is None or right_meta is None:
                    continue
                fps = right_meta.fps
                overlap = _transition_overlap_frames(spec, fps)
                limit_frames = min(left_meta.frame_count, right_meta.frame_count)
                if overlap >= limit_frames and limit_frames > 1 and fps:
                    new_dur = round((limit_frames - 1) / fps, 4)
                    repairs.append(
                        PlanRepair(
                            f"segments[{i}]",
                            "transition_in.duration",
                            spec.duration,
                            new_dur,
                            PlanErrorCode.TRANSITION_TOO_LONG,
                        )
                    )
                    new_spec = spec.model_copy(update={"duration": new_dur})
                    final_segments[i] = final_segments[i].model_copy(update={"transition_in": new_spec})

        # Pass 3: clamp post-op windows / time-fields against the assembled
        # timeline -- but only when every segment predicted (else its duration is
        # unknown). A negative post_operations window is the canonical attempt-3
        # failure, so this matters as much as the per-segment clamps.
        update: dict[str, Any] = {"segments": final_segments}
        outputs = [m for m in seg_output_metas if m is not None]
        if self.post_operations and len(outputs) == len(self.segments):
            transitions = [seg.transition_in for seg in final_segments]
            assembled = _assemble_timeline(outputs, transitions)
            new_post, post_repairs, _ = _repair_op_chain(
                list(self.post_operations),
                assembled,
                context,
                "post_operations",
                clamp=clamp_op_params,
            )
            repairs.extend(post_repairs)
            update["post_operations"] = new_post

        repaired = self.model_copy(update=update, deep=True)
        return repaired, repairs

    def normalize_dimensions(
        self,
        source_metadata: VideoMetadata | dict[str, VideoMetadata],
        target: tuple[int, int] | Literal["first", "largest", "match"],
        context: dict[str, Any] | None = None,
    ) -> tuple[VideoEdit, list[PlanRepair]]:
        """Make every segment concat-compatible by resizing to a common canvas.

        ``CONCAT_MISMATCH`` is the one class a consumer cannot cleanly repair in
        its own layer: detecting it needs each segment's *predicted post-op*
        dimensions, and fixing it needs a per-segment resize inserted **before**
        concat. videopython owns both, so it does it here: predict each segment's
        output dimensions, pick the ``target`` -- an explicit ``(width, height)``,
        ``"first"`` (the first predictable segment's output), or ``"largest"``
        (greatest area) -- and append a ``resize`` op to every segment whose
        output differs, recording a :class:`PlanRepair` per insertion. The
        returned plan satisfies the "all segments share dimensions" invariant for
        every segment it could predict.

        Best-effort and non-raising, matching :meth:`repair` / :meth:`check`: a
        segment that cannot be cut (bad range) or whose op chain fails prediction
        is left untouched and its fault deferred to :meth:`check`, rather than
        aborting the whole call. This keeps the documented refine flow
        (``repair -> normalize_dimensions -> check``) a single non-raising path.
        When no segment is predictable the plan is returned unchanged with an
        empty changelog.

        Expressed purely as appended ``resize`` ops, so the normal
        validate/run/stream paths need no special casing. Resizing to an exact
        canvas can distort aspect when segments genuinely differ -- intended for
        a plan whose segments already share a target aspect (resolve that
        upstream).
        """
        metas = self._resolve_source_metas(source_metadata)
        # Predict each segment's post-op output dims; None when it can't be cut
        # or an op fails prediction (best-effort: deferred to check, never raised).
        cut_metas: list[VideoMetadata | None] = []
        for i, (seg, meta) in enumerate(zip(self.segments, metas)):
            if _segment_bounds_errors(i, seg, meta):
                cut_metas.append(None)
            else:
                cut_metas.append(CutSeconds(start=seg.start, end=seg.end).predict_metadata(meta))
        matched_by_index = self._match_cuttable(cut_metas)
        outputs: list[VideoMetadata | None] = []
        for i, seg in enumerate(self.segments):
            seg_meta = matched_by_index.get(i)
            if seg_meta is None:
                outputs.append(None)
                continue
            try:
                outputs.append(self._predict_segment(i, seg, seg_meta, context))
            except (ValueError, TypeError):
                outputs.append(None)

        known = [m for m in outputs if m is not None]
        if not known:
            return self.model_copy(deep=True), []
        if target == "first":
            tw, th = known[0].width, known[0].height
        elif target == "largest":
            biggest = max(known, key=lambda m: m.width * m.height)
            tw, th = biggest.width, biggest.height
        elif target == "match":
            # The same min-resolution policy the engine's
            # match_to_lowest_resolution applies in-stream, materialized as
            # resize ops so it survives serialization. Degrades to "first" when
            # the flag is off (resolved.width is None), keeping the
            # always-produce-a-compatible-plan contract.
            resolved = self._resolve_matching_target(known)
            tw = resolved.width if resolved.width is not None else known[0].width
            th = resolved.height if resolved.height is not None else known[0].height
        else:
            tw, th = target

        repairs: list[PlanRepair] = []
        new_segments: list[SegmentConfig] = []
        for i, (seg, out) in enumerate(zip(self.segments, outputs)):
            if out is None or (out.width, out.height) == (tw, th):
                new_segments.append(seg)
                continue
            new_ops = [*seg.operations, Resize(width=tw, height=th)]
            repairs.append(
                PlanRepair(
                    location=f"segments[{i}]",
                    field="dimensions",
                    old=f"{out.width}x{out.height}",
                    new=f"{tw}x{th}",
                    code=PlanErrorCode.CONCAT_MISMATCH,
                )
            )
            new_segments.append(seg.model_copy(update={"operations": new_ops}))

        normalized = self.model_copy(update={"segments": new_segments}, deep=True)
        return normalized, repairs

    def _resolve_source_metas(self, source_metadata: VideoMetadata | dict[str, VideoMetadata]) -> list[VideoMetadata]:
        """One ``VideoMetadata`` per segment from a single meta or a by-source dict."""
        if isinstance(source_metadata, VideoMetadata):
            return [source_metadata for _ in self.segments]
        metas: list[VideoMetadata] = []
        for i, seg in enumerate(self.segments):
            key = str(seg.source)
            if key not in source_metadata:
                available = sorted(source_metadata)
                raise ValueError(f"Segment {i}: no metadata for '{key}'. Available: {available}")
            metas.append(source_metadata[key])
        return metas

    def _post_op_context_errors(self, context: dict[str, Any] | None) -> list[_LocatedError]:
        """``POST_OP_REQUIRES_CONTEXT`` errors for time-context post-ops on a multi-segment plan.

        ``post_operations`` run on the assembled, concatenated timeline. A
        source-absolute context value (e.g. a ``Transcription``) cannot be
        re-based across a multi-segment concat, and passing the raw value would
        silently mis-time the op (subtitles/silence-removal against the wrong
        timeline). Single-segment plans are unaffected -- their concatenated
        timeline is just the one segment's, handled by ``_segment_context``.
        """
        if len(self.segments) <= 1 or not self.post_operations:
            return []
        rebaseable = _rebaseable_keys(context)
        if not rebaseable:
            return []
        out: list[_LocatedError] = []
        for j, op in enumerate(self.post_operations):
            clash = sorted(set(op.requires) & rebaseable)
            if clash:
                message = (
                    f"post_operation '{op.op}' requires time-based context {clash}, but the plan "
                    f"has {len(self.segments)} segments. post_operations run on the concatenated "
                    "timeline and time-based context is not re-based across a multi-segment concat. "
                    f"Move '{op.op}' into a segment, or use a single-segment plan."
                )
                out.append(
                    (
                        message,
                        PlanError(PlanErrorCode.POST_OP_REQUIRES_CONTEXT, location=f"post_operations[{j}]", op=op.op),
                    )
                )
        return out

    def _assert_post_ops_supported(self, context: dict[str, Any] | None) -> None:
        """Raising guard used by ``run_to_file`` (see :meth:`_post_op_context_errors`)."""
        errs = self._post_op_context_errors(context)
        if errs:
            message, err = errs[0]
            raise PlanValidationError(message, [err])

    def _music_bed_errors(self) -> list[_LocatedError]:
        """Validate :attr:`music_bed`: readable source, and duck on a single segment.

        Two checks, both fail-fast (before any decode): the bed ``source`` must
        be a readable audio file (``SOURCE_UNREADABLE``, a cheap ffprobe header
        probe like :class:`ImageOverlay`'s); and transcription-derived ducking is
        only well-defined when the assembled timeline is a single segment's
        timeline -- a multi-segment plan with ``duck`` set is rejected
        (``MUSIC_BED_DUCK_MULTISEGMENT``), since the assembled-timeline
        transcription mapping across cuts is out of scope. A non-ducked bed on a
        multi-segment plan is fine.
        """
        bed = self.music_bed
        if bed is None:
            return []
        out: list[_LocatedError] = []
        if bed.duck is not None and len(self.segments) > 1:
            message = (
                f"music_bed.duck is set, but the plan has {len(self.segments)} segments. "
                "Transcription-derived ducking is only supported on a single-segment plan "
                "(the assembled-timeline transcription mapping across cuts is out of scope); "
                "drop the duck or use a single-segment plan."
            )
            out.append(
                (
                    message,
                    PlanError(PlanErrorCode.MUSIC_BED_DUCK_MULTISEGMENT, location="music_bed", field="duck"),
                )
            )
        try:
            bed.validate_source()
        except PlanValidationError as e:
            for err in e.errors:
                err.location = "music_bed"
            out.append((str(e), e.errors[0]))
        return out

    def _assert_music_bed_supported(self) -> None:
        """Raising guard used by ``run_to_file`` (see :meth:`_music_bed_errors`)."""
        errs = self._music_bed_errors()
        if errs:
            message, err = errs[0]
            raise PlanValidationError(message, [err])

    def _music_bed_speech_windows(self, context: dict[str, Any] | None) -> list[tuple[float, float]] | None:
        """Speech windows for the music-bed duck, on the single segment's timeline.

        Only reached for a single-segment plan (the duck-multisegment guard
        rejects the rest), so the assembled timeline == the segment's timeline:
        the rebased context transcription maps directly. Resolves the
        segment-local transcription exactly as the per-segment ops do
        (:func:`_segment_context`), then derives padded speech windows via the
        shared :func:`speech_windows` helper. ``None`` when the bed is not
        ducked or there is no transcription to derive windows from -- a non-ducked
        (or context-less) bed mixes flat.
        """
        bed = self.music_bed
        if bed is None or bed.duck is None:
            return None
        from videopython.base.transcription import Transcription

        segment = self.segments[0]
        seg_context = _segment_context(context, str(segment.source), segment.start, segment.end)
        transcription = (seg_context or {}).get("transcription")
        if not isinstance(transcription, Transcription):
            return None
        # The bed's padding mirrors silence_removal's default breathing room so
        # the duck eases around speech rather than clipping each word tightly.
        total = segment.end - segment.start
        return speech_windows(transcription.words, 0.15, total)

    def _validate(
        self,
        source_metas: list[VideoMetadata],
        context: dict[str, Any] | None,
        *,
        clamp_windows: bool = False,
    ) -> VideoMetadata:
        assembled, _ = self._collect(source_metas, context, clamp_windows=clamp_windows, stop_first=True)
        assert assembled is not None  # stop_first raised on any error, so a result is guaranteed
        return assembled

    def _collect(
        self,
        source_metas: list[VideoMetadata],
        context: dict[str, Any] | None,
        *,
        clamp_windows: bool,
        stop_first: bool,
    ) -> tuple[VideoMetadata | None, list[PlanError]]:
        """The single dry-run walker behind both :meth:`validate` and :meth:`check`.

        ``stop_first=True`` raises ``PlanValidationError`` on the first failure
        (byte-stable prose for existing callers); ``stop_first=False``
        accumulates every structured error and returns them. The check *order* is
        identical in both modes -- post-op context guard, per-segment cut, fps/res
        matching, per-segment op prediction, cross-segment concat, post-ops -- so
        the first collected error always matches what ``validate`` would raise.
        """
        errors: list[PlanError] = []

        def emit(message: str, err: PlanError) -> None:
            if stop_first:
                raise PlanValidationError(message, [err])
            errors.append(err)

        def raise_or_collect(exc: PlanValidationError) -> None:
            if stop_first:
                raise exc
            errors.extend(exc.errors)

        for message, err in self._post_op_context_errors(context):
            emit(message, err)

        for message, err in self._music_bed_errors():
            emit(message, err)

        for message, err in _transition_structure_errors(list(self.segments)):
            emit(message, err)

        for i, seg in enumerate(self.segments):
            for message, err in _missing_source_context_errors(context, i, seg):
                emit(message, err)

        # Cut each segment against its own source metadata. A bad range isolates
        # that segment (no cut meta -> its op chain is skipped below).
        cut_metas: list[VideoMetadata | None] = []
        for i, (seg, meta) in enumerate(zip(self.segments, source_metas)):
            bounds = _segment_bounds_errors(i, seg, meta)
            if bounds:
                for message, err in bounds:
                    emit(message, err)
                cut_metas.append(None)
                continue
            cut_metas.append(CutSeconds(start=seg.start, end=seg.end).predict_metadata(meta))

        matched_by_index = self._match_cuttable(cut_metas)

        # `_apply_matching` runs over the cuttable subset only; an uncuttable
        # segment isolates and is already reported, so the (now-invalid) plan's
        # matched dims may differ from run_to_file()'s all-segments matching -- harmless,
        # since that plan can't run until the bad segment is fixed.
        seg_outputs: dict[int, VideoMetadata] = {}
        for i, seg in enumerate(self.segments):
            seg_meta = matched_by_index.get(i)
            if seg_meta is None:
                continue
            seg_context = _segment_context(context, str(seg.source), seg.start, seg.end)
            failed = False
            for op_index, op in enumerate(seg.operations):
                location = f"segments[{i}].operations[{op_index}]"
                if clamp_windows:
                    op = _clamp_effect_window(op, seg_meta.total_seconds)
                for message, err in _window_errors(op, seg_meta.total_seconds, location):
                    emit(message, err)
                try:
                    seg_meta = _predict_with_context(op, seg_meta, seg_context)
                except PlanValidationError as e:
                    _relocate(e.errors, location)
                    raise_or_collect(e)
                    failed = True
                    break
                except (ValueError, TypeError) as e:
                    message = f"Segment {i}: metadata prediction failed for '{op.op}': {e}"
                    emit(message, PlanError(PlanErrorCode.OP_PREDICTION_FAILED, location=location, op=op.op))
                    failed = True
                    break
            if not failed:
                seg_outputs[i] = seg_meta

        # Concat + post-ops need every segment's predicted output. When a segment
        # failed, those dependent checks are skipped (best-effort): the consumer
        # fixes the isolated segment and they surface on the next pass.
        if len(seg_outputs) != len(self.segments):
            return None, errors
        outputs = [seg_outputs[i] for i in range(len(self.segments))]
        for message, err in _concat_errors(outputs):
            emit(message, err)
        for message, err in _transition_duration_errors(list(self.segments), outputs):
            emit(message, err)

        transitions = [seg.transition_in for seg in self.segments]
        assembled = _assemble_timeline(outputs, transitions)
        for j, op in enumerate(self.post_operations):
            location = f"post_operations[{j}]"
            for message, err in _window_errors(op, assembled.total_seconds, location):
                emit(message, err)
            try:
                assembled = _predict_with_context(op, assembled, context)
            except PlanValidationError as e:
                _relocate(e.errors, location)
                raise_or_collect(e)
                return None, errors
            except (ValueError, TypeError) as e:
                message = f"post_operations[{j}]: metadata prediction failed for '{op.op}': {e}"
                emit(message, PlanError(PlanErrorCode.OP_PREDICTION_FAILED, location=location, op=op.op))
                return None, errors
        return assembled, errors

    def _predict_segment(
        self,
        index: int,
        segment: SegmentConfig,
        meta: VideoMetadata,
        context: dict[str, Any] | None,
    ) -> VideoMetadata:
        """Predict one segment's post-op metadata, raising located plan errors.

        The always-raising single-segment predictor used by
        :meth:`normalize_dimensions` (which needs each segment's predicted output
        dimensions). :meth:`_collect` has its own per-op loop because it must
        also accumulate, clamp windows, and isolate failures per segment.
        """
        seg_context = _segment_context(context, str(segment.source), segment.start, segment.end)
        for op_index, op in enumerate(segment.operations):
            location = f"segments[{index}].operations[{op_index}]"
            for message, err in _window_errors(op, meta.total_seconds, location):
                raise PlanValidationError(message, [err])
            try:
                meta = _predict_with_context(op, meta, seg_context)
            except PlanValidationError as e:
                _relocate(e.errors, location)
                raise
            except (ValueError, TypeError) as e:
                raise ValueError(f"Segment {index}: metadata prediction failed for '{op.op}': {e}") from e
        return meta

    def _match_cuttable(self, cut_metas: list[VideoMetadata | None]) -> dict[int, VideoMetadata]:
        """Run fps/resolution matching over the cuttable segments only, by original index.

        An uncuttable segment (a ``None`` cut meta -- a bad range, isolated and
        reported elsewhere) is excluded so it cannot perturb the ``min`` fps/dims
        that :meth:`_apply_matching` derives. Returns ``{original_index: matched
        meta}`` so callers stay index-aligned with ``self.segments``. Shared by
        :meth:`_collect`, :meth:`repair`, and :meth:`normalize_dimensions`.
        """
        present = [(i, m) for i, m in enumerate(cut_metas) if m is not None]
        return dict(zip((i for i, _ in present), self._apply_matching([m for _, m in present])))

    def _resolve_matching_target(self, metas: list[VideoMetadata]) -> _MatchTarget:
        """The one place the min-fps/min-resolution concat policy is decided.

        Reduces per-segment metas to a single :class:`_MatchTarget` honoring
        ``match_to_lowest_fps`` / ``match_to_lowest_resolution``. An axis is
        ``None`` (left untouched) when its flag is off or there is nothing to
        match (``<= 1`` meta). Shared verbatim by the prediction pass
        (:meth:`_apply_matching`) and the execution pass
        (:meth:`_matching_targets_from_disk`), and reused by
        :meth:`normalize_dimensions` for its ``"match"`` target, so all three
        agree by construction. Pure: no disk, no mutation of the inputs.
        """
        if len(metas) <= 1:
            return _MatchTarget(fps=None, width=None, height=None)
        fps = min(m.fps for m in metas) if self.match_to_lowest_fps else None
        width = min(m.width for m in metas) if self.match_to_lowest_resolution else None
        height = min(m.height for m in metas) if self.match_to_lowest_resolution else None
        return _MatchTarget(fps=fps, width=width, height=height)

    def _apply_matching(self, metas: list[VideoMetadata]) -> list[VideoMetadata]:
        if len(metas) <= 1:
            return metas
        target = self._resolve_matching_target(metas)
        return [target.apply(m) for m in metas]

    def run_to_file(
        self,
        output_path: str | Path,
        format: ALLOWED_VIDEO_FORMATS = "mp4",
        preset: ALLOWED_VIDEO_PRESETS = "medium",
        crf: int = 23,
        context: dict[str, Any] | None = None,
    ) -> Path:
        """Execute the plan, streaming directly to a file.

        Memory usage is O(1) w.r.t. video length (video; segment audio is
        in-memory). Streaming is the only engine: a plan with an unstreamable
        shape raises :class:`PlanValidationError` carrying one
        ``STREAMING_UNSUPPORTED`` :class:`PlanError` per offending op -- before
        any decode. Gate plans early with :meth:`check` or
        :meth:`streamability`, which report the same errors without running
        anything.
        """
        output_path = Path(output_path).with_suffix(f".{format}")
        output_path.parent.mkdir(parents=True, exist_ok=True)

        plans = self._compile_streaming_plans(context)
        self._assert_music_bed_supported()

        # The program flows through up to three stages, each writing to its own
        # temp file: assemble segments -> apply post_operations -> mix music bed.
        # Only the final active stage writes straight to `output_path`.
        has_post = bool(self.post_operations)
        has_bed = self.music_bed is not None
        intermediates: list[Path] = []

        def _stage_target(*, is_last: bool) -> Path:
            if is_last:
                return output_path
            tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
            tmp.close()
            p = Path(tmp.name)
            intermediates.append(p)
            return p

        try:
            self._assert_transitions_runnable(plans)
            # Stage 1: assemble the segments.
            target = _stage_target(is_last=not (has_post or has_bed))
            if len(plans) == 1:
                assembled = stream_segment(plans[0], target, format=format, preset=preset, crf=crf)
            else:
                # Realize each segment to its own temp file (effects + audio
                # already baked by stream_segment's filter_complex), capturing
                # per-segment audibility for the crossfade-vs-butt-join decision
                # at any transition seam. Audibility is the source's audio-stream
                # presence: a source with no audio gets a native silent track
                # (anullsrc), so it never crossfades -- the same decision the old
                # materialized is_silent made for the no-audio case, without
                # decoding audio.
                temp_files: list[Path] = []
                audible: list[bool] = []
                try:
                    for plan in plans:
                        tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
                        tmp.close()
                        audible.append(source_has_audio_stream(plan.source_path))
                        stream_segment(plan, Path(tmp.name), with_audio=True, format=format, preset=preset, crf=crf)
                        temp_files.append(Path(tmp.name))

                    if all(seg.transition_in is None for seg in self.segments):
                        assembled = concat_files(temp_files, target)
                    else:
                        assembled = self._assemble_with_transitions(
                            temp_files, audible, target, format=format, preset=preset, crf=crf
                        )
                finally:
                    for f in temp_files:
                        f.unlink(missing_ok=True)

            # Stage 2: apply post_operations as ONE pass over the assembled
            # program (a synthetic single segment), so filter-class effects,
            # frame effects, and transforms all apply on the assembled timeline.
            if has_post:
                target = _stage_target(is_last=not has_bed)
                assembled = self._apply_post_operations(
                    assembled, target, context, format=format, preset=preset, crf=crf
                )

            # Stage 3: mix the music bed under the assembled program.
            if has_bed:
                speech = self._music_bed_speech_windows(context)
                assembled = self._mix_music_bed_to_file(
                    assembled, output_path, speech, format=format, preset=preset, crf=crf
                )

            return assembled
        finally:
            for p in intermediates:
                p.unlink(missing_ok=True)
            for plan in plans:
                for f in plan.owned_temp_files:
                    f.unlink(missing_ok=True)

    def _apply_post_operations(
        self,
        assembled: Path,
        output_path: Path,
        context: dict[str, Any] | None,
        *,
        format: str,
        preset: str,
        crf: int,
    ) -> Path:
        """Apply ``post_operations`` as one pass over the assembled program.

        The assembled file is treated as a single synthetic segment whose
        ``operations`` are the ``post_operations``, then run through the same
        streaming engine as any segment -- so filter-class effects (vignette,
        color_adjust, ...), frame effects, and transforms all apply uniformly to
        the whole program on the assembled timeline, with each effect ``window``
        resolving against the assembled file's own frame count. No per-segment
        fold and no cross-boundary re-basing.

        Runtime context: a single-segment plan's assembled timeline IS that
        segment's, so a rebaseable value (e.g. a ``Transcription``) is re-based
        onto it exactly as a per-segment op would see it. Multi-segment context
        post-ops are rejected up front by :meth:`_assert_post_ops_supported`
        (source-absolute context cannot be re-based across a concat), so only
        broadcast (non-time) context reaches a multi-segment post-op pass.
        """
        meta = VideoMetadata.from_path(str(assembled))
        seg = SegmentConfig(
            source=assembled,
            start=0.0,
            end=round(meta.total_seconds, 6),
            operations=list(self.post_operations),
        )
        if len(self.segments) == 1:
            src = self.segments[0]
            post_context = _segment_context(context, str(src.source), src.start, src.end)
        else:
            post_context = context
        plan = self._build_streaming_plan(seg, None, None, None, post_context)
        if plan is None:
            raise PlanValidationError(
                "post_operations did not compile to a streaming plan",
                [PlanError(code=PlanErrorCode.STREAMING_UNSUPPORTED, location="post_operations")],
            )
        try:
            return stream_segment(plan, output_path, format=format, preset=preset, crf=crf)
        finally:
            for f in plan.owned_temp_files:
                f.unlink(missing_ok=True)

    def _mix_music_bed_to_file(
        self,
        assembled: Path,
        output_path: Path,
        speech: list[tuple[float, float]] | None,
        *,
        format: str,
        preset: str,
        crf: int,
    ) -> Path:
        """Mix the music bed under an assembled program file in one ffmpeg pass.

        The bed mix: the assembled program is input 0, the bed
        a second ``-i`` input, and :func:`build_music_bed_filter_complex`
        compiles the ``amix`` graph. The program duration is the
        assembled file's PROBED length so the bed's loop/trim pin matches the
        realized timeline exactly. The video stream is copied (``-c:v copy`` --
        the bed touches only audio); the mixed audio is re-encoded to AAC.
        """
        bed = self.music_bed
        assert bed is not None
        meta = VideoMetadata.from_path(str(assembled))
        program_seconds = meta.total_seconds
        bed_inputs, graph, out_label = build_music_bed_filter_complex(
            bed, program_seconds, speech=speech, bed_input_index=1, prog_label="0:a"
        )
        cmd = [
            "ffmpeg",
            "-y",
            "-hide_banner",
            "-loglevel",
            "error",
            "-i",
            str(assembled),
            *bed_inputs,
            "-filter_complex",
            ";".join(graph),
            "-map",
            "0:v",
            "-map",
            out_label,
            "-c:v",
            "copy",
            "-c:a",
            "aac",
            "-b:a",
            "192k",
            "-movflags",
            "+faststart",
            str(output_path),
        ]
        try:
            _ffmpeg.run(cmd)
        except Exception:
            if output_path.exists():
                output_path.unlink()
            raise
        return output_path

    def _assemble_with_transitions(
        self,
        seg_files: list[Path],
        audible: list[bool],
        output_path: Path,
        *,
        format: str,
        preset: str,
        crf: int,
    ) -> Path:
        """Boundary-aware assembly: concat-copy hard-cut runs, xfade transitions.

        Walks segments left to right. Maximal runs of hard-cut (``transition_in
        is None``) boundaries are grouped and joined with ``concat_files``
        (``-c copy``, no re-encode), so only transition seams pay an encode.
        Each transition then xfade-joins the running tail with the next group:
        ``offset`` is derived from the realized TAIL's PROBED duration (not the
        prediction) so the seam is frame-aligned, the video blend goes through
        the shared :func:`xfade_filter` builder (matching ``run_to_file()``), and the
        audio is ``acrossfade``-d when ``spec.audio`` and both boundary
        segments are audible, else hard butt-joined.

        Note xfade re-encodes the whole left tail at each seam, so a
        transition-heavy reel re-encodes its cumulative tail repeatedly; a
        mixed reel only re-encodes at the seams (the hard-cut runs stay copies).
        """
        # Partition into hard-cut groups; a new group starts at each segment
        # whose transition_in is set. Each group concat-copies to one file.
        groups: list[list[int]] = [[0]]
        for i in range(1, len(self.segments)):
            if self.segments[i].transition_in is None:
                groups[-1].append(i)
            else:
                groups.append([i])

        owned: list[Path] = []

        def _group_file(indices: list[int]) -> Path:
            if len(indices) == 1:
                return seg_files[indices[0]]
            tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
            tmp.close()
            owned.append(Path(tmp.name))
            return concat_files([seg_files[i] for i in indices], Path(tmp.name))

        try:
            tail = _group_file(groups[0])
            tail_left_audible = audible[groups[0][-1]]
            for g in range(1, len(groups)):
                right_indices = groups[g]
                right_file = _group_file(right_indices)
                first_idx = right_indices[0]
                spec = self.segments[first_idx].transition_in
                assert spec is not None  # a group (after the first) always opens on a transition
                meta = VideoMetadata.from_path(str(tail))
                right_meta = VideoMetadata.from_path(str(right_file))
                # Offset from the realized tail's PROBED frame count (matching
                # the xfade input trim), not the prediction, so the seam is
                # frame-aligned -- ``(n_left - overlap) / fps``.
                overlap = _transition_overlap_frames(spec, meta.fps)
                offset = round((meta.frame_count - overlap) / meta.fps, 6)
                crossfade_audio = bool(spec.audio and tail_left_audible and audible[first_idx])

                is_last = g == len(groups) - 1
                if is_last:
                    target = output_path
                else:
                    tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
                    tmp.close()
                    target = Path(tmp.name)
                    owned.append(target)
                new_tail = stream_transition_pair(
                    tail,
                    right_file,
                    spec.type,
                    spec.duration,
                    offset,
                    target,
                    width=meta.width,
                    height=meta.height,
                    fps=meta.fps,
                    left_frame_count=meta.frame_count,
                    right_frame_count=right_meta.frame_count,
                    left_has_audio=True,
                    right_has_audio=True,
                    crossfade_audio=crossfade_audio,
                    format=format,
                    preset=preset,
                    crf=crf,
                )
                # Once we xfade past a seam the new tail's trailing content is
                # the right group's, so the next seam's audibility is that
                # group's last segment.
                tail = new_tail
                tail_left_audible = audible[right_indices[-1]]
            return output_path
        finally:
            for f in owned:
                if f != output_path:
                    f.unlink(missing_ok=True)

    def _compile_streaming_plans(self, context: dict[str, Any] | None) -> list[StreamingSegmentPlan]:
        """Compile every per-segment streaming plan, or raise.

        Post-operations are NOT compiled here -- they run as a separate pass over
        the assembled program (:meth:`_apply_post_operations`).

        The single admission point for :meth:`run_to_file`. Unstreamable shapes raise
        :class:`PlanValidationError` with the streamability report's
        structured errors; a builder/report drift (e.g. an op whose
        ``to_ffmpeg_filter`` returns ``None`` at its plan position despite
        overriding it) raises with a generic ``STREAMING_UNSUPPORTED``.
        On raise, any compile-time temp files already created are deleted;
        on success the caller owns ``plan.owned_temp_files``.
        """
        self._assert_post_ops_supported(context)
        report = self.streamability()
        if not report.streamable:
            causes = "; ".join(f"{e.location} '{e.op}': {e.reason}" for e in report.unstreamable)
            message = (
                f"Plan cannot stream: {len(report.unstreamable)} op(s) have no streaming "
                f"strategy at their plan position -- {causes}"
            )
            raise PlanValidationError(message, report.errors())

        def drift(detail: str) -> PlanValidationError:
            return PlanValidationError(
                f"plan stopped streaming despite a clean streamability report ({detail})",
                [PlanError(code=PlanErrorCode.STREAMING_UNSUPPORTED, detail=detail)],
            )

        target_fps, target_w, target_h = self._matching_targets_from_disk()
        plans: list[StreamingSegmentPlan] = []
        try:
            for i, segment in enumerate(self.segments):
                plan = self._build_streaming_plan(segment, target_fps, target_w, target_h, context)
                if plan is None:
                    raise drift(f"segments[{i}] did not compile to a streaming plan")
                plans.append(plan)
        except BaseException:
            for plan in plans:
                for f in plan.owned_temp_files:
                    f.unlink(missing_ok=True)
            raise
        return plans

    # ----------------------------------------------------------------- helpers

    def _matching_targets_from_disk(self) -> tuple[float | None, int | None, int | None]:
        if len(self.segments) <= 1 or (not self.match_to_lowest_fps and not self.match_to_lowest_resolution):
            return None, None, None
        metas = [VideoMetadata.from_path(str(seg.source)) for seg in self.segments]
        target = self._resolve_matching_target(metas)
        return target.fps, target.width, target.height

    def _build_streaming_plan(
        self,
        segment: SegmentConfig,
        target_fps: float | None,
        target_w: int | None,
        target_h: int | None,
        context: dict[str, Any] | None = None,
    ) -> StreamingSegmentPlan | None:
        source_meta = VideoMetadata.from_path(str(segment.source))
        # Fold real metadata through the chain -- the same walk validation
        # does -- so duration-changing transforms (speed, freeze) keep frame
        # counts, effect ranges, and audio in sync with the actual output.
        running = CutSeconds(start=segment.start, end=segment.end).predict_metadata(source_meta)
        if running.frame_count <= 0:
            message = (
                f"Segment [{segment.start}, {segment.end}) is shorter than one frame "
                f"at {running.fps} fps and cannot be rendered"
            )
            raise PlanValidationError(
                message,
                [PlanError(code=PlanErrorCode.DEGENERATE_DURATION, op=None, field="end", value=segment.end)],
            )

        vf_filters: list[str] = []
        if target_w and target_h and (target_w != source_meta.width or target_h != source_meta.height):
            vf_filters.append(f"scale={target_w}:{target_h}")
            running = running.with_dimensions(target_w, target_h)
        if target_fps and target_fps != source_meta.fps:
            vf_filters.append(f"fps={target_fps}")
            running = running.with_fps(target_fps)

        # Resolve requires-context onto the segment's local timeline once; each
        # context-requiring effect gets its keys on the schedule entry, which
        # stream_segment forwards to streaming_init. A missing/empty key is
        # passed as absent so the effect raises its own clear "requires ..."
        # error -- before that segment's decode.
        seg_context = _segment_context(context, str(segment.source), segment.start, segment.end)

        owned_files: list[Path] = []

        def abandon() -> None:
            """Delete compile-time temp files of a plan that won't run."""
            for f in owned_files:
                f.unlink(missing_ok=True)

        def make_ctx(decode_filters: tuple[str, ...] | None | object = _DECODE_FILTERS_DEFAULT) -> FilterCtx:
            """A :class:`FilterCtx` snapshot of the current ``running`` state.

            Captures the per-op pipeline geometry (dims/fps/frame-count) and the
            segment's location/context every op compiles against. ``decode_filters``
            is the decode-stage filter prefix ahead of this op; left unset it keeps
            :class:`FilterCtx`'s own default (the op consumes no decode pass).
            """
            kwargs: dict[str, Any] = {
                "width": running.width,
                "height": running.height,
                "fps": running.fps,
                "frame_count": running.frame_count,
                "context": seg_context or {},
                "source_path": segment.source,
                "start_second": segment.start,
                "end_second": segment.end,
            }
            if decode_filters is not _DECODE_FILTERS_DEFAULT:
                kwargs["decode_filters"] = decode_filters
            return FilterCtx(**kwargs)

        effect_schedule: list[EffectScheduleEntry] = []
        post_vf_filters: list[str] = []
        af_filters: list[str] = []
        post_af_filters: list[str] = []
        audio_idx = 0
        duration_changed = False

        def compile_audio_twin(op: Operation, ctx: FilterCtx, encode_stage: bool) -> None:
            """Append the op's audio-domain filter at the same stage the video
            filter landed, keeping audio and video stage placement coupled.

            ``ctx`` is the SAME FilterCtx the video side compiled this op with
            (pre-op ``running``), plus a per-op ``audio_label`` so a
            multi-statement fragment (freeze_frame's splice) cannot collide with
            another op's internal labels."""
            nonlocal audio_idx
            audio_ctx = dataclasses.replace(ctx, audio_label=f"f{audio_idx}")
            af = op.to_ffmpeg_audio_filter(audio_ctx)
            audio_idx += 1
            if af is None:
                return
            (post_af_filters if encode_stage else af_filters).append(af)

        # The pipe stage: dims/fps/frame-count of the frames flowing through
        # process_frame and into the encoder's rawvideo stdin. Frozen the
        # moment encode-stage content appears (a scheduled effect or any
        # post_vf entry); transforms after that fold `running` further (for
        # audio durations and the final output), but the pipe stays put.
        pipe_meta: VideoMetadata | None = None
        try:
            for op in segment.operations:
                if isinstance(op, Effect):
                    if not op.streams():
                        abandon()
                        return None
                    if op.requires and duration_changed:
                        # A duration-changing transform earlier in the chain
                        # moved the timeline; segment-local context (e.g. the
                        # transcription) is not re-mapped through the warp
                        # yet. Not streamable here -- rejected as
                        # UNSTREAMABLE by the streamability report.
                        abandon()
                        return None
                    if op.compiles_to_filter:
                        # Filter-class effect (add_subtitles): consumes its
                        # context at compile time and joins the filter chain at
                        # this op's plan position -- the decode chain when no
                        # frame effect precedes it, else the encode chain
                        # (FrameEncoder -vf), which runs after every
                        # process_frame. Either way plan order is preserved. A
                        # None compile falls through to the frame-effect path.
                        encode_stage_effect = bool(effect_schedule or post_vf_filters)
                        ctx = make_ctx(decode_filters=None if encode_stage_effect else tuple(vf_filters))
                        filter_expr = op.to_ffmpeg_filter(ctx)
                        owned_files.extend(ctx.owned_files)
                        if filter_expr is not None:
                            if encode_stage_effect:
                                pipe_meta = pipe_meta or running
                                post_vf_filters.append(filter_expr)
                            else:
                                vf_filters.append(filter_expr)
                            # Audio twin at the same stage (add_subtitles has
                            # none today; kept coupled for extensibility).
                            compile_audio_twin(op, ctx, encode_stage_effect)
                            continue
                    if post_vf_filters:
                        # A frame effect after an encode-stage filter would run
                        # before it (process_frame precedes the encoder), so
                        # plan order cannot be preserved -- not streamable
                        # (rejected as UNSTREAMABLE by the streamability report).
                        abandon()
                        return None
                    op_context: dict[str, Any] = {}
                    if op.requires and seg_context:
                        op_context = {k: seg_context[k] for k in op.requires if k in seg_context}
                    pipe_meta = pipe_meta or running
                    start_f, end_f = _effect_frame_range(op, running.fps, running.frame_count)
                    effect_schedule.append(EffectScheduleEntry(op, start_f, end_f, op_context))
                    # Audio-coupled frame effects (fade/volume_adjust) express
                    # their gain on the audio graph. A frame effect always sits
                    # at the decode stage (one after an encode-stage filter is
                    # rejected above), so its audio twin joins af_filters; its
                    # window resolves against this position's running metadata.
                    effect_ctx = make_ctx()
                    compile_audio_twin(op, effect_ctx, encode_stage=False)
                    continue
                if op.requires and (duration_changed or not op.streams()):
                    # A context-requiring transform can stream only when its
                    # filter compile can consume the context (silence_removal
                    # does, via FilterCtx.context) -- but not after a
                    # duration-changing transform (no time-warp re-mapping
                    # yet), and not without a streaming strategy. Not
                    # streamable here -- rejected as UNSTREAMABLE by the
                    # streamability report.
                    abandon()
                    return None
                # Non-effect transform: streams iff it compiles a filter. Checked
                # structurally (op.streams() == overrides to_ffmpeg_filter) so the
                # streamability report and the builder cannot disagree; the
                # remaining drift class (overrides to_ffmpeg_filter but it compiles
                # to None here) is caught by the STREAMING_UNSUPPORTED raise in
                # _compile_streaming_plans.
                if not op.streams():
                    abandon()
                    return None
                encode_stage = bool(effect_schedule or post_vf_filters)
                if encode_stage and op.compiles_from_source:
                    # The op's compile decodes the source to see its input
                    # frames (face_crop's detection pass); frames behind
                    # per-frame Python effects are not reproducible at
                    # compile time -- not streamable (rejected as UNSTREAMABLE).
                    abandon()
                    return None
                ctx = make_ctx(decode_filters=None if encode_stage else tuple(vf_filters))
                filter_expr = op.to_ffmpeg_filter(ctx)
                owned_files.extend(ctx.owned_files)
                if filter_expr is None:
                    abandon()
                    return None
                if encode_stage:
                    # A transform following frame effects joins the encode
                    # chain (FrameEncoder -vf), which runs after every
                    # process_frame -- plan order is preserved because this
                    # shape places the filter without the whole-plan fallback.
                    pipe_meta = pipe_meta or running
                    post_vf_filters.append(filter_expr)
                else:
                    vf_filters.append(filter_expr)
                # Audio twin compiled from the SAME pre-op ctx, appended to the
                # same stage the video filter landed -- so audio and video
                # stage placement cannot drift (speed->atempo, freeze->silence
                # splice, silence_removal->aselect keep windows).
                compile_audio_twin(op, ctx, encode_stage)
                running = _predict_with_context(op, running, seg_context)
                if op.changes_duration:
                    duration_changed = True
        except BaseException:
            # A raising compile or prediction (e.g. missing subtitle context,
            # crop exceeding source) must not leak earlier ops' temp files.
            abandon()
            raise

        pipe = pipe_meta or running
        # Final output duration after every transform (decode + encode stage),
        # used to pin the audio graph length to the video timeline. `running`
        # is folded through the whole chain, so this is the encoded output's
        # duration even when an encode-stage transform (post_vf speed) shortens
        # it below the pipe frame count.
        output_total_seconds = running.frame_count / running.fps if running.fps else 0.0
        return StreamingSegmentPlan(
            source_path=segment.source,
            start_second=segment.start,
            end_second=segment.end,
            output_fps=pipe.fps,
            output_width=pipe.width,
            output_height=pipe.height,
            vf_filters=vf_filters,
            effect_schedule=effect_schedule,
            post_vf_filters=post_vf_filters,
            owned_temp_files=owned_files,
            output_total_frames=pipe.frame_count,
            af_filters=af_filters,
            post_af_filters=post_af_filters,
            output_total_seconds=output_total_seconds,
            final_fps=running.fps,
            final_width=running.width,
            final_height=running.height,
        )

    def _assert_transitions_runnable(self, plans: list[StreamingSegmentPlan]) -> None:
        """Raise before decode if any transition is structurally invalid or too long.

        Mirrors :func:`_transition_structure_errors` /
        :func:`_transition_duration_errors` but measures each segment against
        its compiled plan's predicted post-op duration -- the exact length the
        per-segment realize pass produces -- so a plan that passes here cannot
        fail the xfade pass at decode. Called by ``run_to_file`` after plan
        compilation (no frames decoded yet).
        ``repair`` clamps these mechanically; here they hard-raise.
        """
        for message, err in _transition_structure_errors(list(self.segments)):
            raise PlanValidationError(message, [err])
        for i in range(1, len(self.segments)):
            spec = self.segments[i].transition_in
            if spec is None:
                continue
            left_frames, left_fps = _plan_output_frames(plans[i - 1])
            right_frames, right_fps = _plan_output_frames(plans[i])
            fps = right_fps or left_fps
            overlap = _transition_overlap_frames(spec, fps)
            limit_frames = min(left_frames, right_frames)
            if overlap >= limit_frames:
                limit_seconds = round(limit_frames / fps, 4) if fps else 0.0
                message, err = _transition_too_long_error(
                    i,
                    spec,
                    overlap,
                    limit_frames,
                    limit_seconds,
                    "repair the plan or shorten the transition.",
                )
                raise PlanValidationError(message, [err])

json_schema classmethod

json_schema(*, strict: bool = False) -> dict[str, Any]

LLM-facing schema: a discriminated union of operations per slot.

A thin transform over the Pydantic models, so it cannot drift from them: the operations union is :meth:Operation.json_schema (LLM-exposed ops by default, per the llm_exposed ClassVar), and every other field shape and description is derived from SegmentConfig/VideoEdit model_json_schema() rather than hand-typed. Adding a model field thus surfaces here automatically; in particular source carries its "format": "path". The Draft-07 $schema envelope and the default operations array shape are preserved for downstream LLM tooling.

With strict=True the result is a submittable provider strict-mode grammar: a closed object root, all properties required (optionality kept as Pydantic emitted it -- no synthesized nulls), the op union as anyOf of closed variants, and the union's $defs hoisted to the document root so every $ref resolves. See :meth:Operation.json_schema for the strict-mode contract. Use it as a response_format: json_schema grammar so simple bound violations (window.start >= 0, enums, required fields) become impossible at decode time. Cross-field constraints (timestamp < duration, segment-dim equality) cannot live in a grammar and stay with :meth:check / :meth:repair / :meth:normalize_dimensions.

Source code in src/videopython/editing/video_edit.py

@classmethod
def json_schema(cls, *, strict: bool = False) -> dict[str, Any]:
    """LLM-facing schema: a discriminated union of operations per slot.

    A thin transform over the Pydantic models, so it cannot drift from
    them: the operations union is :meth:`Operation.json_schema` (LLM-exposed
    ops by default, per the ``llm_exposed`` ClassVar), and every other field
    shape and ``description`` is derived from ``SegmentConfig``/``VideoEdit``
    ``model_json_schema()`` rather than hand-typed. Adding a model field thus
    surfaces here automatically; in particular ``source`` carries its
    ``"format": "path"``. The Draft-07 ``$schema`` envelope and the default
    ``operations`` array shape are preserved for downstream LLM tooling.

    With ``strict=True`` the result is a submittable provider strict-mode
    grammar: a closed object root, all properties ``required`` (optionality
    kept as Pydantic emitted it -- no synthesized nulls), the op union as
    ``anyOf`` of closed variants, and the union's ``$defs`` hoisted to the
    document root so every ``$ref`` resolves. See :meth:`Operation.json_schema`
    for the strict-mode contract. Use it as a ``response_format: json_schema``
    grammar so simple bound violations (``window.start >= 0``, enums, required
    fields) become impossible at decode time. Cross-field constraints
    (``timestamp < duration``, segment-dim equality) cannot live in a grammar
    and stay with :meth:`check` / :meth:`repair` / :meth:`normalize_dimensions`.
    """
    # Build the permissive envelope, then (if strict) run one closing pass
    # over the whole thing -- including the embedded op union -- so the
    # hand-built segment/top-level objects are closed and required too.
    # `op_schema` is a self-contained union carrying its own root `$defs`;
    # the same object is inlined as the `items` of both operation slots.
    op_schema = Operation.json_schema()

    segment_schema: dict[str, Any] = {
        "type": "object",
        "description": SegmentConfig.__doc__,
        "properties": {
            "source": field_schema(SegmentConfig, "source"),
            "start": field_schema(SegmentConfig, "start"),
            "end": field_schema(SegmentConfig, "end"),
            "operations": array_field_schema(SegmentConfig, "operations", op_schema),
            "transition_in": optional_model_field_schema(TransitionSpec, SegmentConfig, "transition_in"),
        },
        "required": ["source", "start", "end"],
        "additionalProperties": False,
    }
    segments = field_schema(cls, "segments")
    segments["items"] = segment_schema
    schema = {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "description": cls.__doc__,
        "properties": {
            "segments": segments,
            "post_operations": array_field_schema(cls, "post_operations", op_schema),
            "match_to_lowest_fps": field_schema(cls, "match_to_lowest_fps"),
            "match_to_lowest_resolution": field_schema(cls, "match_to_lowest_resolution"),
            "music_bed": optional_model_field_schema(MusicBed, cls, "music_bed"),
        },
        "required": ["segments"],
        "additionalProperties": False,
    }
    if not strict:
        return schema
    # Inlining the op union as `items` buried its `$defs` while its `$ref`s
    # stayed root-relative (`#/$defs/X`), so they dangled. Hoist the (shared)
    # defs to this schema's root before the strict pass, so every ref
    # resolves and the result is a submittable provider grammar.
    op_defs = op_schema.pop("$defs", None)
    if op_defs:
        schema["$defs"] = op_defs
    return _to_strict_schema(schema)

validate

validate(
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> VideoMetadata

Dry-run the plan via metadata. Requires source files on disk.

Shadows Pydantic v1's deprecated BaseModel.validate classmethod; use VideoEdit.from_dict/model_validate for plan parsing.

When clamp_windows is True, an :class:Effect's window.stop that overruns the running predicted duration (e.g. after a duration-shrinking op like speed_change/cut) is clamped to that duration -- the same min(stop, total_seconds) value the streaming engine applies at run time -- instead of raising. Only window.stop is clamped: a window.start past the duration still hard-raises (a residual divergence from run_to_file(), which degrades it to a zero-width no-op).

Source code in src/videopython/editing/video_edit.py

def validate(  # type: ignore[override]
    self,
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> VideoMetadata:
    """Dry-run the plan via metadata. Requires source files on disk.

    Shadows Pydantic v1's deprecated ``BaseModel.validate`` classmethod;
    use ``VideoEdit.from_dict``/``model_validate`` for plan parsing.

    When ``clamp_windows`` is True, an :class:`Effect`'s ``window.stop`` that
    overruns the running predicted duration (e.g. after a duration-shrinking
    op like ``speed_change``/``cut``) is clamped to that duration -- the same
    ``min(stop, total_seconds)`` value the streaming engine applies at run
    time -- instead of raising. Only ``window.stop`` is clamped: a ``window.start`` past the
    duration still hard-raises (a residual divergence from ``run_to_file()``, which
    degrades it to a zero-width no-op).
    """
    source_metas = [VideoMetadata.from_path(str(seg.source)) for seg in self.segments]
    return self._validate(source_metas, context, clamp_windows=clamp_windows)

validate_with_metadata

validate_with_metadata(
    source_metadata: VideoMetadata
    | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> VideoMetadata

Dry-run with pre-built metadata, avoiding disk access.

See :meth:validate for the clamp_windows semantics.

Source code in src/videopython/editing/video_edit.py

def validate_with_metadata(
    self,
    source_metadata: VideoMetadata | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> VideoMetadata:
    """Dry-run with pre-built metadata, avoiding disk access.

    See :meth:`validate` for the ``clamp_windows`` semantics.
    """
    metas = self._resolve_source_metas(source_metadata)
    return self._validate(metas, context, clamp_windows=clamp_windows)

check

check(
    source_metadata: VideoMetadata
    | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> list[PlanError]

Collect every plan error in one pass; [] means valid.

The non-raising sibling of :meth:validate_with_metadata: it runs the same dry-run but accumulates instead of aborting on the first failure, so an LLM refine loop can fix all problems in a single re-prompt instead of playing whack-a-mole across a retry budget. Best-effort: each segment is checked against its own source metadata, per-op and per-segment errors collected; a check that cannot run because an earlier one failed (a segment's op chain past a bad cut, the cross-segment concat check when a segment did not produce an output) is skipped rather than aborting.

Returns the same :class:PlanError list that :attr:PlanValidationError.errors carries -- every failure is structured (no bare ValueError escapes the walk), so a consumer branches on code rather than substring-matching prose. clamp_windows matches :meth:validate: a clampable window.stop overrun is not reported.

Streaming is the only engine, so ops that cannot stream at their plan position are real plan errors: one STREAMING_UNSUPPORTED per offending op is appended after the validity errors, in plan order, with the actionable cause in :attr:PlanError.detail. See :meth:streamability for the full per-op report including the ops that do stream.

Source code in src/videopython/editing/video_edit.py

def check(
    self,
    source_metadata: VideoMetadata | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_windows: bool = False,
) -> list[PlanError]:
    """Collect **every** plan error in one pass; ``[]`` means valid.

    The non-raising sibling of :meth:`validate_with_metadata`: it runs the
    same dry-run but accumulates instead of aborting on the first failure,
    so an LLM refine loop can fix all problems in a single re-prompt instead
    of playing whack-a-mole across a retry budget. Best-effort: each segment
    is checked against its own source metadata, per-op and per-segment errors
    collected; a check that cannot run because an earlier one failed (a
    segment's op chain past a bad cut, the cross-segment concat check when a
    segment did not produce an output) is skipped rather than aborting.

    Returns the same :class:`PlanError` list that
    :attr:`PlanValidationError.errors` carries -- every failure is structured
    (no bare ``ValueError`` escapes the walk), so a consumer branches on
    ``code`` rather than substring-matching prose. ``clamp_windows`` matches
    :meth:`validate`: a clampable ``window.stop`` overrun is not reported.

    Streaming is the only engine, so ops that cannot stream at their
    plan position are real plan errors: one ``STREAMING_UNSUPPORTED`` per
    offending op is appended after the validity errors, in plan order,
    with the actionable cause in :attr:`PlanError.detail`. See
    :meth:`streamability` for the full per-op report including the ops
    that *do* stream.
    """
    metas = self._resolve_source_metas(source_metadata)
    _, errors = self._collect(metas, context, clamp_windows=clamp_windows, stop_first=False)
    errors.extend(self.streamability().errors())
    return errors

streamability

streamability() -> StreamabilityReport

Classify every op by streaming class, without touching the disk.

Streamability is purely structural -- it depends on op classes, their order, and the plan shape, never on source metadata or runtime context -- so this needs no source files and is safe to call before a job is admitted. report.streamable answers "will :meth:run_to_file stream this plan in O(1) memory, or is an op unstreamable at its plan position?"; each entry carries the op's memory class and, for unstreamable ops, the reason.

Source code in src/videopython/editing/video_edit.py

def streamability(self) -> StreamabilityReport:
    """Classify every op by streaming class, without touching the disk.

    Streamability is purely structural -- it depends on op classes, their
    order, and the plan shape, never on source metadata or runtime context
    -- so this needs no source files and is safe to call before a job is
    admitted. ``report.streamable`` answers "will :meth:`run_to_file`
    stream this plan in O(1) memory, or is an op unstreamable at its
    plan position?"; each entry carries the op's memory class and, for
    unstreamable ops, the reason.
    """
    return analyze_streamability(
        [list(seg.operations) for seg in self.segments],
        list(self.post_operations),
    )

repair

repair(
    source_metadata: VideoMetadata
    | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_op_params: bool = True,
    clamp_segment_end: bool = False,
) -> tuple[VideoEdit, list[PlanRepair]]

Return a copy of this plan with the unambiguous violations clamped.

Walks the chain (cut, fps/resolution matching, per-op prediction) and clamps only the mechanical faults whose fix is not a judgement call, recording each as a :class:PlanRepair. The returned plan is a deep copy (self is untouched); the changelog is meant to be surfaced to the user ("we trimmed your effect to fit"). repair never invents intent -- genuinely semantic problems (a concat dimension mismatch, an end <= start range) are left for :meth:check / re-prompting.

With clamp_op_params (default True) it clamps each effect window.start/window.stop into [0, duration] and each declared :attr:Operation.time_fields value (e.g. freeze_frame.timestamp past the clip end) into range, plus a negative segment start to 0. With clamp_segment_end (default False, since it changes editorial intent) it also clamps a segment end past the source to the source end; left False, that case hard-raises as before. Always :meth:check / :meth:validate the returned plan before running it.

Source code in src/videopython/editing/video_edit.py

def repair(
    self,
    source_metadata: VideoMetadata | dict[str, VideoMetadata],
    context: dict[str, Any] | None = None,
    *,
    clamp_op_params: bool = True,
    clamp_segment_end: bool = False,
) -> tuple[VideoEdit, list[PlanRepair]]:
    """Return a copy of this plan with the *unambiguous* violations clamped.

    Walks the chain (cut, fps/resolution matching, per-op prediction) and
    clamps only the mechanical faults whose fix is not a judgement call,
    recording each as a :class:`PlanRepair`. The returned plan is a deep copy
    (``self`` is untouched); the changelog is meant to be surfaced to the
    user ("we trimmed your effect to fit"). ``repair`` never *invents* intent
    -- genuinely semantic problems (a concat dimension mismatch, an
    ``end <= start`` range) are left for :meth:`check` / re-prompting.

    With ``clamp_op_params`` (default ``True``) it clamps each effect
    ``window.start``/``window.stop`` into ``[0, duration]`` and each declared
    :attr:`Operation.time_fields` value (e.g. ``freeze_frame.timestamp`` past
    the clip end) into range, plus a negative segment ``start`` to ``0``.
    With ``clamp_segment_end`` (default ``False``, since it changes editorial
    intent) it also clamps a segment ``end`` past the source to the source
    end; left ``False``, that case hard-raises as before. Always
    :meth:`check` / :meth:`validate` the returned plan before running it.
    """
    metas = self._resolve_source_metas(source_metadata)
    repairs: list[PlanRepair] = []

    # Pass 1: segment-level start/end repairs, and the cut metadata each
    # surviving segment's op clamps run against.
    fixed_segments: list[SegmentConfig] = []
    cut_metas: list[VideoMetadata | None] = []
    for i, (seg, meta) in enumerate(zip(self.segments, metas)):
        start, end = seg.start, seg.end
        if clamp_op_params and start < 0:
            repairs.append(PlanRepair(f"segments[{i}]", "start", start, 0.0, PlanErrorCode.SEGMENT_NEGATIVE))
            start = 0.0
        if end > meta.total_seconds + DURATION_EPS:
            if clamp_segment_end:
                new_end = round(meta.total_seconds, 4)
                repairs.append(
                    PlanRepair(f"segments[{i}]", "end", end, new_end, PlanErrorCode.SEGMENT_END_EXCEEDS_SOURCE)
                )
                end = new_end
            else:
                message, err = _segment_end_exceeds_source(i, seg, meta)
                raise PlanValidationError(message, [err])
        seg = seg.model_copy(update={"start": start, "end": end})
        fixed_segments.append(seg)
        repairable = start >= 0 and end > start and end <= meta.total_seconds + DURATION_EPS
        cut_metas.append(CutSeconds(start=start, end=end).predict_metadata(meta) if repairable else None)

    # Pass 2: per-segment op clamps against the matched cut meta. Capture each
    # segment's predicted output -- needed to assemble the post-op timeline.
    matched_by_index = self._match_cuttable(cut_metas)
    final_segments: list[SegmentConfig] = []
    seg_output_metas: list[VideoMetadata | None] = []
    for i, seg in enumerate(fixed_segments):
        seg_meta = matched_by_index.get(i)
        if seg_meta is None:
            final_segments.append(seg)
            seg_output_metas.append(None)
            continue
        seg_context = _segment_context(context, str(seg.source), seg.start, seg.end)
        new_ops, op_repairs, out_meta = _repair_op_chain(
            list(seg.operations),
            seg_meta,
            seg_context,
            f"segments[{i}].operations",
            clamp=clamp_op_params,
        )
        repairs.extend(op_repairs)
        final_segments.append(seg.model_copy(update={"operations": new_ops}))
        seg_output_metas.append(out_meta)

    # Pass 2b: clamp each transition's overlap down to one frame short of the
    # shorter adjacent segment -- the mechanical TRANSITION_TOO_LONG repair,
    # mirroring the window clamps. The clamp is frame-safe: it targets
    # `limit_frames - 1` overlap frames so the repaired plan satisfies the
    # strict `overlap < min(frames)` streaming constraint and actually runs
    # (a seconds clamp to the segment length lands on `overlap == frames`,
    # which passes check but crashes run_to_file). Only fires when both adjacent
    # segments predicted; a segment too short for any transition
    # (limit_frames <= 1) is left for check. A first-segment transition is a
    # structural fault, deliberately left for check.
    if clamp_op_params:
        for i in range(1, len(final_segments)):
            spec = final_segments[i].transition_in
            if spec is None:
                continue
            left_meta = seg_output_metas[i - 1]
            right_meta = seg_output_metas[i]
            if left_meta is None or right_meta is None:
                continue
            fps = right_meta.fps
            overlap = _transition_overlap_frames(spec, fps)
            limit_frames = min(left_meta.frame_count, right_meta.frame_count)
            if overlap >= limit_frames and limit_frames > 1 and fps:
                new_dur = round((limit_frames - 1) / fps, 4)
                repairs.append(
                    PlanRepair(
                        f"segments[{i}]",
                        "transition_in.duration",
                        spec.duration,
                        new_dur,
                        PlanErrorCode.TRANSITION_TOO_LONG,
                    )
                )
                new_spec = spec.model_copy(update={"duration": new_dur})
                final_segments[i] = final_segments[i].model_copy(update={"transition_in": new_spec})

    # Pass 3: clamp post-op windows / time-fields against the assembled
    # timeline -- but only when every segment predicted (else its duration is
    # unknown). A negative post_operations window is the canonical attempt-3
    # failure, so this matters as much as the per-segment clamps.
    update: dict[str, Any] = {"segments": final_segments}
    outputs = [m for m in seg_output_metas if m is not None]
    if self.post_operations and len(outputs) == len(self.segments):
        transitions = [seg.transition_in for seg in final_segments]
        assembled = _assemble_timeline(outputs, transitions)
        new_post, post_repairs, _ = _repair_op_chain(
            list(self.post_operations),
            assembled,
            context,
            "post_operations",
            clamp=clamp_op_params,
        )
        repairs.extend(post_repairs)
        update["post_operations"] = new_post

    repaired = self.model_copy(update=update, deep=True)
    return repaired, repairs

normalize_dimensions

normalize_dimensions(
    source_metadata: VideoMetadata
    | dict[str, VideoMetadata],
    target: tuple[int, int]
    | Literal["first", "largest", "match"],
    context: dict[str, Any] | None = None,
) -> tuple[VideoEdit, list[PlanRepair]]

Make every segment concat-compatible by resizing to a common canvas.

CONCAT_MISMATCH is the one class a consumer cannot cleanly repair in its own layer: detecting it needs each segment's predicted post-op dimensions, and fixing it needs a per-segment resize inserted before concat. videopython owns both, so it does it here: predict each segment's output dimensions, pick the target -- an explicit (width, height), "first" (the first predictable segment's output), or "largest" (greatest area) -- and append a resize op to every segment whose output differs, recording a :class:PlanRepair per insertion. The returned plan satisfies the "all segments share dimensions" invariant for every segment it could predict.

Best-effort and non-raising, matching :meth:repair / :meth:check: a segment that cannot be cut (bad range) or whose op chain fails prediction is left untouched and its fault deferred to :meth:check, rather than aborting the whole call. This keeps the documented refine flow (repair -> normalize_dimensions -> check) a single non-raising path. When no segment is predictable the plan is returned unchanged with an empty changelog.

Expressed purely as appended resize ops, so the normal validate/run/stream paths need no special casing. Resizing to an exact canvas can distort aspect when segments genuinely differ -- intended for a plan whose segments already share a target aspect (resolve that upstream).

Source code in src/videopython/editing/video_edit.py

def normalize_dimensions(
    self,
    source_metadata: VideoMetadata | dict[str, VideoMetadata],
    target: tuple[int, int] | Literal["first", "largest", "match"],
    context: dict[str, Any] | None = None,
) -> tuple[VideoEdit, list[PlanRepair]]:
    """Make every segment concat-compatible by resizing to a common canvas.

    ``CONCAT_MISMATCH`` is the one class a consumer cannot cleanly repair in
    its own layer: detecting it needs each segment's *predicted post-op*
    dimensions, and fixing it needs a per-segment resize inserted **before**
    concat. videopython owns both, so it does it here: predict each segment's
    output dimensions, pick the ``target`` -- an explicit ``(width, height)``,
    ``"first"`` (the first predictable segment's output), or ``"largest"``
    (greatest area) -- and append a ``resize`` op to every segment whose
    output differs, recording a :class:`PlanRepair` per insertion. The
    returned plan satisfies the "all segments share dimensions" invariant for
    every segment it could predict.

    Best-effort and non-raising, matching :meth:`repair` / :meth:`check`: a
    segment that cannot be cut (bad range) or whose op chain fails prediction
    is left untouched and its fault deferred to :meth:`check`, rather than
    aborting the whole call. This keeps the documented refine flow
    (``repair -> normalize_dimensions -> check``) a single non-raising path.
    When no segment is predictable the plan is returned unchanged with an
    empty changelog.

    Expressed purely as appended ``resize`` ops, so the normal
    validate/run/stream paths need no special casing. Resizing to an exact
    canvas can distort aspect when segments genuinely differ -- intended for
    a plan whose segments already share a target aspect (resolve that
    upstream).
    """
    metas = self._resolve_source_metas(source_metadata)
    # Predict each segment's post-op output dims; None when it can't be cut
    # or an op fails prediction (best-effort: deferred to check, never raised).
    cut_metas: list[VideoMetadata | None] = []
    for i, (seg, meta) in enumerate(zip(self.segments, metas)):
        if _segment_bounds_errors(i, seg, meta):
            cut_metas.append(None)
        else:
            cut_metas.append(CutSeconds(start=seg.start, end=seg.end).predict_metadata(meta))
    matched_by_index = self._match_cuttable(cut_metas)
    outputs: list[VideoMetadata | None] = []
    for i, seg in enumerate(self.segments):
        seg_meta = matched_by_index.get(i)
        if seg_meta is None:
            outputs.append(None)
            continue
        try:
            outputs.append(self._predict_segment(i, seg, seg_meta, context))
        except (ValueError, TypeError):
            outputs.append(None)

    known = [m for m in outputs if m is not None]
    if not known:
        return self.model_copy(deep=True), []
    if target == "first":
        tw, th = known[0].width, known[0].height
    elif target == "largest":
        biggest = max(known, key=lambda m: m.width * m.height)
        tw, th = biggest.width, biggest.height
    elif target == "match":
        # The same min-resolution policy the engine's
        # match_to_lowest_resolution applies in-stream, materialized as
        # resize ops so it survives serialization. Degrades to "first" when
        # the flag is off (resolved.width is None), keeping the
        # always-produce-a-compatible-plan contract.
        resolved = self._resolve_matching_target(known)
        tw = resolved.width if resolved.width is not None else known[0].width
        th = resolved.height if resolved.height is not None else known[0].height
    else:
        tw, th = target

    repairs: list[PlanRepair] = []
    new_segments: list[SegmentConfig] = []
    for i, (seg, out) in enumerate(zip(self.segments, outputs)):
        if out is None or (out.width, out.height) == (tw, th):
            new_segments.append(seg)
            continue
        new_ops = [*seg.operations, Resize(width=tw, height=th)]
        repairs.append(
            PlanRepair(
                location=f"segments[{i}]",
                field="dimensions",
                old=f"{out.width}x{out.height}",
                new=f"{tw}x{th}",
                code=PlanErrorCode.CONCAT_MISMATCH,
            )
        )
        new_segments.append(seg.model_copy(update={"operations": new_ops}))

    normalized = self.model_copy(update={"segments": new_segments}, deep=True)
    return normalized, repairs

run_to_file

run_to_file(
    output_path: str | Path,
    format: ALLOWED_VIDEO_FORMATS = "mp4",
    preset: ALLOWED_VIDEO_PRESETS = "medium",
    crf: int = 23,
    context: dict[str, Any] | None = None,
) -> Path

Execute the plan, streaming directly to a file.

Memory usage is O(1) w.r.t. video length (video; segment audio is in-memory). Streaming is the only engine: a plan with an unstreamable shape raises :class:PlanValidationError carrying one STREAMING_UNSUPPORTED :class:PlanError per offending op -- before any decode. Gate plans early with :meth:check or :meth:streamability, which report the same errors without running anything.

Source code in src/videopython/editing/video_edit.py

def run_to_file(
    self,
    output_path: str | Path,
    format: ALLOWED_VIDEO_FORMATS = "mp4",
    preset: ALLOWED_VIDEO_PRESETS = "medium",
    crf: int = 23,
    context: dict[str, Any] | None = None,
) -> Path:
    """Execute the plan, streaming directly to a file.

    Memory usage is O(1) w.r.t. video length (video; segment audio is
    in-memory). Streaming is the only engine: a plan with an unstreamable
    shape raises :class:`PlanValidationError` carrying one
    ``STREAMING_UNSUPPORTED`` :class:`PlanError` per offending op -- before
    any decode. Gate plans early with :meth:`check` or
    :meth:`streamability`, which report the same errors without running
    anything.
    """
    output_path = Path(output_path).with_suffix(f".{format}")
    output_path.parent.mkdir(parents=True, exist_ok=True)

    plans = self._compile_streaming_plans(context)
    self._assert_music_bed_supported()

    # The program flows through up to three stages, each writing to its own
    # temp file: assemble segments -> apply post_operations -> mix music bed.
    # Only the final active stage writes straight to `output_path`.
    has_post = bool(self.post_operations)
    has_bed = self.music_bed is not None
    intermediates: list[Path] = []

    def _stage_target(*, is_last: bool) -> Path:
        if is_last:
            return output_path
        tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
        tmp.close()
        p = Path(tmp.name)
        intermediates.append(p)
        return p

    try:
        self._assert_transitions_runnable(plans)
        # Stage 1: assemble the segments.
        target = _stage_target(is_last=not (has_post or has_bed))
        if len(plans) == 1:
            assembled = stream_segment(plans[0], target, format=format, preset=preset, crf=crf)
        else:
            # Realize each segment to its own temp file (effects + audio
            # already baked by stream_segment's filter_complex), capturing
            # per-segment audibility for the crossfade-vs-butt-join decision
            # at any transition seam. Audibility is the source's audio-stream
            # presence: a source with no audio gets a native silent track
            # (anullsrc), so it never crossfades -- the same decision the old
            # materialized is_silent made for the no-audio case, without
            # decoding audio.
            temp_files: list[Path] = []
            audible: list[bool] = []
            try:
                for plan in plans:
                    tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
                    tmp.close()
                    audible.append(source_has_audio_stream(plan.source_path))
                    stream_segment(plan, Path(tmp.name), with_audio=True, format=format, preset=preset, crf=crf)
                    temp_files.append(Path(tmp.name))

                if all(seg.transition_in is None for seg in self.segments):
                    assembled = concat_files(temp_files, target)
                else:
                    assembled = self._assemble_with_transitions(
                        temp_files, audible, target, format=format, preset=preset, crf=crf
                    )
            finally:
                for f in temp_files:
                    f.unlink(missing_ok=True)

        # Stage 2: apply post_operations as ONE pass over the assembled
        # program (a synthetic single segment), so filter-class effects,
        # frame effects, and transforms all apply on the assembled timeline.
        if has_post:
            target = _stage_target(is_last=not has_bed)
            assembled = self._apply_post_operations(
                assembled, target, context, format=format, preset=preset, crf=crf
            )

        # Stage 3: mix the music bed under the assembled program.
        if has_bed:
            speech = self._music_bed_speech_windows(context)
            assembled = self._mix_music_bed_to_file(
                assembled, output_path, speech, format=format, preset=preset, crf=crf
            )

        return assembled
    finally:
        for p in intermediates:
            p.unlink(missing_ok=True)
        for plan in plans:
            for f in plan.owned_temp_files:
                f.unlink(missing_ok=True)

SegmentConfig

SegmentConfig is exported, but most users should construct plans via VideoEdit.from_dict(...) / VideoEdit.from_json(...).

SegmentConfig

Bases: BaseModel

A single source segment with its operation chain.

Source code in src/videopython/editing/video_edit.py

class SegmentConfig(BaseModel):
    """A single source segment with its operation chain."""

    model_config = ConfigDict(extra="forbid")

    source: Path = Field(description="Path to the source video file.")
    # Parsing is permissive (plain floats, no ge=0 / end>start): the numeric
    # bounds are owned by validate/check/repair as structured PlanErrors. See
    # `_segment_bounds_errors` and the VideoEdit class docstring.
    start: float = Field(description="Segment start time in seconds.")
    end: float = Field(description="Segment end time in seconds.")
    operations: list[OperationInput] = Field(
        default_factory=list,
        description=(
            "Ordered list of operations to run against this segment. "
            "Each item is an Operation discriminated by its `op` field."
        ),
    )
    transition_in: TransitionSpec | None = Field(
        default=None,
        description=(
            "Optional crossfade from the previous segment into this one. Must be null on the "
            "first segment (there is no previous segment to transition from)."
        ),
    )

    @property
    def duration(self) -> float:
        return self.end - self.start

    def load(
        self,
        fps: float | None = None,
        width: int | None = None,
        height: int | None = None,
    ) -> Video:
        """Load the raw segment from disk with optional decode-time matching."""
        return Video.from_path(
            str(self.source),
            start_second=self.start,
            end_second=self.end,
            fps=fps,
            width=width,
            height=height,
        )

load

load(
    fps: float | None = None,
    width: int | None = None,
    height: int | None = None,
) -> Video

Load the raw segment from disk with optional decode-time matching.

Source code in src/videopython/editing/video_edit.py

def load(
    self,
    fps: float | None = None,
    width: int | None = None,
    height: int | None = None,
) -> Video:
    """Load the raw segment from disk with optional decode-time matching."""
    return Video.from_path(
        str(self.source),
        start_second=self.start,
        end_second=self.end,
        fps=fps,
        width=width,
        height=height,
    )