Skip to content

Edit plans

VideoEdit is a multi-segment editing plan: a Pydantic model whose fields are the JSON wire format. Each segment selects a time range from a source and carries an ordered list of operations to run against it.

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

Usage

from videopython.editing import VideoEdit

edit = VideoEdit.from_dict({
    "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,
         "operations": [{"op": "resize", "width": 1080, "height": 1920}]},
    ],
    "post_operations": [{"op": "color_adjust", "brightness": 0.05}],
})

predicted = edit.validate()
edit.run_to_file("output.mp4", crf=20, preset="medium")

Render progress

Pass on_progress to run_to_file() to receive immutable RenderProgress events:

from videopython.editing import RenderProgress


def report(event: RenderProgress) -> None:
    print(event.stage, event.segment_index, event.completed, event.total, event.finished)


edit.run_to_file("output.mp4", on_progress=report)
Field Meaning
stage compilation, segment, assembly, post_operations, audio_mix, or complete
segment_index Zero-based segment index during segment rendering; otherwise None
completed Work observed within this stage, reset at each stage or segment
total 1 for a single stage step; None when the actual frame count is unknown
unit frames or steps
finished The current stage succeeded; it does not imply whole-job success

Only a complete event with finished=True means the render succeeded. Stages that are not needed are omitted. Compilation includes execution checks and plan compilation; assembly includes hard cuts and transitions. Assembly and audio mixing report start and completion of one step. Segment and post-operation passes report FFmpeg output frames or processed Python frames. These are different measurements when an encode filter changes frame rate. Frame totals remain unknown rather than treating a duration prediction as an exact decoded count.

Intermediate updates are limited to one per 0.25 seconds. Stage boundaries are always reported. Callbacks run synchronously on the calling thread; keep them short. Callback exceptions stop execution and propagate, with active FFmpeg processes cleaned up. A failed render does not emit final success. There is no estimated finish time or overall percentage, and callbacks are not part of saved plans. The returned path is unchanged.

RenderProgress dataclass

Work completed within one render stage; only complete means job success.

Source code in src/videopython/editing/progress.py 
 8
 9
10
11
12
13
14
15
16
17
@dataclass(frozen=True)
class RenderProgress:
    """Work completed within one render stage; only ``complete`` means job success."""

    stage: RenderStage
    completed: int
    total: int | None
    unit: Literal["frames", "steps"]
    segment_index: int | None
    finished: bool

JSON wire 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 carries an op discriminator; the remaining fields belong to that op's 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").
  • The cut is the segment's start/end. There is no cut operation — cut/cut_frames are engine-internal.

Execution order

Each segment's operations run in order, the segments are concatenated, then post_operations are applied to the assembled program. What happens under the hood, and which plan shapes are rejected as unstreamable, is described in the streaming engine.

Streamability report

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

Purely structural — it touches no media, so it works as a job-admission gate.

StreamabilityReport dataclass

Per-op streaming classification for a whole plan.

Built by :meth:VideoEdit.streamability from the plan structure alone -- no source files, metadata, or runtime context needed -- so a consumer can gate job admission on it before downloading anything. streamable is the plan-level verdict: one UNSTREAMABLE op rejects the entire plan (streaming is the only engine).

Source code in src/videopython/editing/streaming.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@dataclass(frozen=True)
class StreamabilityReport:
    """Per-op streaming classification for a whole plan.

    Built by :meth:`VideoEdit.streamability` from the plan structure alone --
    no source files, metadata, or runtime context needed -- so a consumer can
    gate job admission on it before downloading anything. ``streamable`` is
    the plan-level verdict: one ``UNSTREAMABLE`` op rejects the entire plan
    (streaming is the only engine).
    """

    entries: tuple[OpStreamability, ...]

    @property
    def streamable(self) -> bool:
        """True when the plan runs (no op is unstreamable at its position)."""
        return all(e.streams for e in self.entries)

    @property
    def unstreamable(self) -> tuple[OpStreamability, ...]:
        """The unstreamable ops, in plan order."""
        return tuple(e for e in self.entries if not e.streams)

    def errors(self) -> list[PlanError]:
        """The unstreamable ops as structured ``STREAMING_UNSUPPORTED`` plan errors.

        The same shape :meth:`VideoEdit.check` returns, so an LLM refine loop
        can treat "would not stream" exactly like any other plan violation.
        """
        return [
            PlanError(code=PlanErrorCode.STREAMING_UNSUPPORTED, location=e.location, op=e.op, detail=e.reason)
            for e in self.unstreamable
        ]

streamable property 

streamable: bool

True when the plan runs (no op is unstreamable at its position).

unstreamable property 

unstreamable: tuple[OpStreamability, ...]

The unstreamable ops, in plan order.

errors 

errors() -> list[PlanError]

The unstreamable ops as structured STREAMING_UNSUPPORTED plan errors.

The same shape :meth:VideoEdit.check returns, so an LLM refine loop can treat "would not stream" exactly like any other plan violation.

Source code in src/videopython/editing/streaming.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def errors(self) -> list[PlanError]:
    """The unstreamable ops as structured ``STREAMING_UNSUPPORTED`` plan errors.

    The same shape :meth:`VideoEdit.check` returns, so an LLM refine loop
    can treat "would not stream" exactly like any other plan violation.
    """
    return [
        PlanError(code=PlanErrorCode.STREAMING_UNSUPPORTED, location=e.location, op=e.op, detail=e.reason)
        for e in self.unstreamable
    ]

OpStreamability dataclass

Streaming classification for a single op within a plan.

Source code in src/videopython/editing/streaming.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass(frozen=True)
class OpStreamability:
    """Streaming classification for a single op within a plan."""

    location: str
    """Path into the plan, e.g. ``'segments[1].operations[0]'``."""
    op: str
    """The op discriminator, e.g. ``'add_subtitles'``."""
    streaming_class: StreamingClass
    reason: str | None = None
    """Why the op cannot stream; ``None`` unless ``UNSTREAMABLE``."""

    @property
    def streams(self) -> bool:
        return self.streaming_class is not StreamingClass.UNSTREAMABLE

location instance-attribute 

location: str

Path into the plan, e.g. 'segments[1].operations[0]'.

op instance-attribute 

op: str

The op discriminator, e.g. 'add_subtitles'.

reason class-attribute instance-attribute 

reason: str | None = None

Why the op cannot stream; None unless UNSTREAMABLE.

StreamingClass

Bases: str, Enum

How an op executes on the streaming engine -- its memory class.

FILTER and FRAME_EFFECT stream in O(1) memory w.r.t. video length. UNSTREAMABLE means the op (or its plan position) has no streaming strategy; such plans are rejected with structured STREAMING_UNSUPPORTED errors.

Source code in src/videopython/editing/streaming.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class StreamingClass(str, Enum):
    """How an op executes on the streaming engine -- its memory class.

    ``FILTER`` and ``FRAME_EFFECT`` stream in O(1) memory w.r.t. video
    length. ``UNSTREAMABLE`` means the op (or its plan position) has no
    streaming strategy; such plans are rejected with structured
    ``STREAMING_UNSUPPORTED`` errors.
    """

    FILTER = "filter"
    """Compiles to an ffmpeg filter -- the decode chain, or the encode chain
    when ordered after frame effects."""
    FRAME_EFFECT = "frame_effect"
    """Shape-preserving per-frame Python (``streaming_init`` + ``process_frame``)."""
    UNSTREAMABLE = "unstreamable"
    """No streaming strategy at this plan position; the plan is rejected."""

FILTER class-attribute instance-attribute 

FILTER = 'filter'

Compiles to an ffmpeg filter -- the decode chain, or the encode chain when ordered after frame effects.

FRAME_EFFECT class-attribute instance-attribute 

FRAME_EFFECT = 'frame_effect'

Shape-preserving per-frame Python (streaming_init + process_frame).

UNSTREAMABLE class-attribute instance-attribute 

UNSTREAMABLE = 'unstreamable'

No streaming strategy at this plan position; the plan is rejected.

Context data

Operations declaring requires: ClassVar[tuple[str, ...]] (for example silence_removal and add_subtitles, which need "transcription") receive their input from the runner:

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

Time-based values are sliced and shifted onto each segment's local timeline. A bare value is shared by all sources. For multiple sources, use a map keyed by the exact str(segment.source) value:

context = {"transcription": {"a.mp4": transcript_a, "b.mp4": transcript_b}}
edit.validate(context=context)
edit.run_to_file("out.mp4", context=context)

A missing source entry is a validation error for operations that require it. Time-based context in post_operations is unsupported on multi-segment plans.

Validation, repair, normalization

Call Result
validate(context=..., clamp_windows=False) Predicted final VideoMetadata; raises on the first failure
validate_with_metadata(meta, context=..., clamp_windows=False) Same, using supplied source metadata
check(meta, context=..., clamp_windows=False) Collects independent plan errors; [] means no reported errors
repair(meta, context=..., clamp_op_params=True, clamp_segment_end=False) (repaired_edit, list[PlanRepair]); a segment end past its source raises unless clamping is enabled
normalize_dimensions(meta, target, context=...) (normalized_edit, list[PlanRepair]) with appended resize operations

meta is one VideoMetadata shared by all segments, or a map keyed by str(segment.source). An incomplete map raises ValueError, including in check, repair, and normalize_dimensions. Supplying metadata avoids source-video probes; referenced assets such as music and overlays can still be read or probed.

Validation predicts operations in order. A failure can prevent later checks on the same chain. check() also reports structural streamability errors. Repair and normalization are separate, best-effort steps; check their returned plans again.

normalize_dimensions accepts (width, height), "first", "largest" (greatest predicted area), or "match" (minimum predicted width and height when resolution matching is enabled, otherwise the first predictable size).

What each stage owns — and why numeric bounds parse cleanly and fail at validation — is the plan lifecycle.

Error types

PlanValidationError subclasses ValueError and carries structured .errors.

PlanError dataclass

A structured validation failure within an edit plan.

Consumers branch on code; detail is human-readable feedback.

Source code in src/videopython/editing/errors.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass
class PlanError:
    """A structured validation failure within an edit plan.

    Consumers branch on ``code``; ``detail`` is human-readable feedback.
    """

    code: PlanErrorCode
    location: str | None = None
    op: str | None = None
    field: str | None = None
    value: float | None = None
    limit: float | None = None
    detail: str | None = None

    def to_prompt_line(self) -> str:
        """Render this error as one deterministic feedback line."""
        line = self.code.name
        if self.location is not None:
            line += f" at {self.location}"
        if self.op is not None:
            line += f" (op '{self.op}')"

        clauses: list[str] = []
        if self.field is not None:
            if self.value is not None:
                clauses.append(f"{self.field}={_fmt_num(self.value)}")
            else:
                clauses.append(self.field)
        elif self.value is not None:
            clauses.append(f"value={_fmt_num(self.value)}")
        if self.limit is not None:
            clauses.append(f"limit {_fmt_num(self.limit)}")

        if clauses:
            line += ": " + ", ".join(clauses)
        if self.detail is not None:
            line += f" -- {self.detail}"
        return line

to_prompt_line 

to_prompt_line() -> str

Render this error as one deterministic feedback line.

Source code in src/videopython/editing/errors.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def to_prompt_line(self) -> str:
    """Render this error as one deterministic feedback line."""
    line = self.code.name
    if self.location is not None:
        line += f" at {self.location}"
    if self.op is not None:
        line += f" (op '{self.op}')"

    clauses: list[str] = []
    if self.field is not None:
        if self.value is not None:
            clauses.append(f"{self.field}={_fmt_num(self.value)}")
        else:
            clauses.append(self.field)
    elif self.value is not None:
        clauses.append(f"value={_fmt_num(self.value)}")
    if self.limit is not None:
        clauses.append(f"limit {_fmt_num(self.limit)}")

    if clauses:
        line += ": " + ", ".join(clauses)
    if self.detail is not None:
        line += f" -- {self.detail}"
    return line

PlanErrorCode

Bases: str, Enum

Machine-readable edit-plan failure classes.

Source code in src/videopython/editing/errors.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class PlanErrorCode(str, Enum):
    """Machine-readable edit-plan failure classes."""

    SEGMENT_END_EXCEEDS_SOURCE = "segment_end_exceeds_source"
    SEGMENT_NEGATIVE = "segment_negative"
    SEGMENT_RANGE = "segment_range"
    EFFECT_WINDOW_EXCEEDS_DURATION = "effect_window_exceeds_duration"
    WINDOW_NEGATIVE = "window_negative"
    WINDOW_ORDER = "window_order"
    CUT_EXCEEDS_DURATION = "cut_exceeds_duration"
    OP_TIMESTAMP_OUT_OF_RANGE = "op_timestamp_out_of_range"
    CROP_EXCEEDS_SOURCE = "crop_exceeds_source"
    DEGENERATE_DURATION = "degenerate_duration"
    SOURCE_UNREADABLE = "source_unreadable"
    OP_PREDICTION_FAILED = "op_prediction_failed"
    UNKNOWN_OP = "unknown_op"
    CONCAT_MISMATCH = "concat_mismatch"
    POST_OP_REQUIRES_CONTEXT = "post_op_requires_context"
    CONTEXT_SOURCE_MISSING = "context_source_missing"
    TRANSITION_TOO_LONG = "transition_too_long"
    MUSIC_BED_DUCK_MULTISEGMENT = "music_bed_duck_multisegment"
    STREAMING_UNSUPPORTED = "streaming_unsupported"

PlanRepair dataclass

A field change made while repairing or normalizing an edit plan.

Source code in src/videopython/editing/errors.py 
78
79
80
81
82
83
84
85
86
@dataclass
class PlanRepair:
    """A field change made while repairing or normalizing an edit plan."""

    location: str
    field: str
    old: float | str | None
    new: float | str | None
    code: PlanErrorCode

PlanValidationError

Bases: ValueError

A ValueError whose errors attribute contains structured failures.

Source code in src/videopython/editing/errors.py 
89
90
91
92
93
94
95
96
97
98
class PlanValidationError(ValueError):
    """A ``ValueError`` whose ``errors`` attribute contains structured failures."""

    def __init__(self, message: str, errors: list[PlanError]):
        super().__init__(message)
        self.errors = errors

    def prompt_feedback(self) -> str:
        """Return every carried error as newline-separated feedback."""
        return "\n".join(error.to_prompt_line() for error in self.errors)

prompt_feedback 

prompt_feedback() -> str

Return every carried error as newline-separated feedback.

Source code in src/videopython/editing/errors.py 
96
97
98
def prompt_feedback(self) -> str:
    """Return every carried error as newline-separated feedback."""
    return "\n".join(error.to_prompt_line() for error in self.errors)

Matching sources

For multiple segments, match_to_lowest_fps=True and match_to_lowest_resolution=True normalize source metadata before each segment's operations. Resolution matching uses the minimum width and minimum height across sources. A single-segment plan needs no matching.

Operations can change those dimensions or fps again. The final segment outputs must agree before concatenation. Set a flag to False to skip that source normalization; your operations must then produce matching outputs. Use normalize_dimensions() to append resizes for a common output canvas. Exact width-and-height resizes can distort aspect ratio; crop to the target aspect first when that matters.

Transitions

Set transition_in on the incoming segment. The first segment must leave it None. The overlap must be shorter than both adjacent segments after operations.

from videopython.editing import SegmentConfig, TransitionSpec, VideoEdit

edit = VideoEdit(segments=[
    SegmentConfig(source="input.mp4", start=0, end=5),
    SegmentConfig(source="input.mp4", start=5, end=10,
                  transition_in=TransitionSpec(type="dissolve", duration=0.5)),
])
edit.validate()
edit.run_to_file("dissolve.mp4")

This produces a 9.5-second program. Each transition subtracts its overlap from the sum of segment durations. audio=True crossfades when both adjacent segments have audio; otherwise audio joins at the boundary. The generated schema lists the accepted transition types.

MusicBed

music_bed mixes music across the assembled program, after transitions and post_operations. It is a plan field, not an operation.

from videopython.editing.audio_ops import MusicBed

edit.music_bed = MusicBed(source="music.mp3", gain=0.25, fade_in=0.5, fade_out=1.0)

The bed loops by default, is trimmed to the program duration, and does not extend output length. With loop=False, a shorter bed is padded with silence. The source must have a readable audio stream and is probed during validation.

duck reduces bed gain during transcription-derived speech windows: 0 leaves it unchanged and 1 silences it. Ducking accepts only a single-segment plan. Pass its source transcription in context; without one, the bed mixes at a flat gain. Speech windows use source timing, so use ducking with operations that keep that timing.

MusicBed

Bases: BaseModel

A music bed mixed under the WHOLE assembled program in a final pass.

The bed spans the entire timeline (after concat / transitions) and is mixed in one amix over the assembled program audio plus this bed input -- not a per-segment op. Frozen and closed so it surfaces as a constrained object in :meth:VideoEdit.json_schema and never accepts stray fields.

source is validated cheaply (an ffprobe header probe, like :class:ImageOverlay's source check) at validate/check time, surfacing SOURCE_UNREADABLE before any decode. loop loops/trims the bed to the program duration so it neither truncates early nor extends the output. duck (when set) lowers the bed under transcription-derived speech windows with duck_attack/duck_release ramps; ducking requires a single-segment assembled timeline (see module docstring).

Source code in src/videopython/editing/audio_ops.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class MusicBed(BaseModel):
    """A music bed mixed under the WHOLE assembled program in a final pass.

    The bed spans the entire timeline (after concat / transitions) and is mixed
    in one ``amix`` over the assembled program audio plus this bed input -- not a
    per-segment op. Frozen and closed so it surfaces as a constrained object in
    :meth:`VideoEdit.json_schema` and never accepts stray fields.

    ``source`` is validated cheaply (an ffprobe header probe, like
    :class:`ImageOverlay`'s source check) at validate/check time, surfacing
    ``SOURCE_UNREADABLE`` before any decode. ``loop`` loops/trims the bed to the
    program duration so it neither truncates early nor extends the output.
    ``duck`` (when set) lowers the bed under transcription-derived speech windows
    with ``duck_attack``/``duck_release`` ramps; ducking requires a
    single-segment assembled timeline (see module docstring).
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    source: Path = Field(description="Path to the music bed audio file (mixed under the whole program).")
    gain: float = Field(0.25, ge=0, description="Bed level multiplier (0 = silent, 1 = unchanged source level).")
    loop: bool = Field(True, description="Loop/trim the bed to the program duration so it spans the whole timeline.")
    fade_in: float = Field(0.0, ge=0, description="Seconds to fade the bed in at the start of the program.")
    fade_out: float = Field(0.0, ge=0, description="Seconds to fade the bed out at the end of the program.")
    duck: float | None = Field(
        None,
        ge=0,
        le=1,
        description=(
            "How far to lower the bed under speech (0 = no ducking, 1 = fully duck). "
            "None disables ducking. Requires a single-segment plan."
        ),
    )
    duck_attack: float = Field(0.2, gt=0, description="Seconds for the duck to ramp in at each speech onset.")
    duck_release: float = Field(0.5, gt=0, description="Seconds for the duck to ramp back out after speech ends.")

    def validate_source(self) -> None:
        """Reject an unreadable bed ``source`` with ``SOURCE_UNREADABLE``.

        Mirrors :meth:`ImageOverlay.predict_metadata`: a cheap ffprobe header
        probe (no decode) catches a missing / non-audio file at validate time,
        before ``run_to_file()`` would crash mid-stream after assembling the program.
        """
        try:
            info = _ffmpeg.probe(self.source)
            has_audio = any(s.get("codec_type") == "audio" for s in info.get("streams", []))
        except Exception as exc:
            message = f"music_bed source {str(self.source)!r} is not a readable audio file: {exc}"
            raise PlanValidationError(
                message,
                [PlanError(code=PlanErrorCode.SOURCE_UNREADABLE, field="source")],
            ) from exc
        if not has_audio:
            message = f"music_bed source {str(self.source)!r} has no audio stream"
            raise PlanValidationError(
                message,
                [PlanError(code=PlanErrorCode.SOURCE_UNREADABLE, field="source")],
            )

    def bed_stages(self, program_seconds: float, speech: list[tuple[float, float]] | None) -> list[str]:
        """The ordered ``filter_complex`` fragments applied to the bed stream.

        ``[gain] -> [fade in/out] -> [duck automation] -> [loop/trim pin]``: the
        bed is scaled to ``gain``, faded, ducked under ``speech`` (when given and
        ``duck`` is set), then pinned to exactly ``program_seconds`` (``atrim``
        end + ``apad`` whole_dur) so a looped bed neither truncates early nor
        extends the output past the program. Used by the file
        mix path via :func:`build_music_bed_filter_complex`.
        """
        stages: list[str] = [f"volume={self.gain:.6f}"]
        if self.fade_in > 0:
            stages.append(f"afade=t=in:st=0:d={self.fade_in:.6f}")
        if self.fade_out > 0:
            fade_start = max(0.0, program_seconds - self.fade_out)
            stages.append(f"afade=t=out:st={fade_start:.6f}:d={self.fade_out:.6f}")
        if self.duck is not None and speech:
            stages.append(
                duck_volume_expression(speech, duck=self.duck, attack=self.duck_attack, release=self.duck_release)
            )
        # Pin the bed to the program length: trim a too-long (looped) bed and pad
        # a too-short one, so amix's program input is exactly matched.
        stages.append(f"atrim=end={program_seconds:.6f}")
        stages.append(f"apad=whole_dur={program_seconds:.6f}")
        return stages

validate_source 

validate_source() -> None

Reject an unreadable bed source with SOURCE_UNREADABLE.

Mirrors :meth:ImageOverlay.predict_metadata: a cheap ffprobe header probe (no decode) catches a missing / non-audio file at validate time, before run_to_file() would crash mid-stream after assembling the program.

Source code in src/videopython/editing/audio_ops.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def validate_source(self) -> None:
    """Reject an unreadable bed ``source`` with ``SOURCE_UNREADABLE``.

    Mirrors :meth:`ImageOverlay.predict_metadata`: a cheap ffprobe header
    probe (no decode) catches a missing / non-audio file at validate time,
    before ``run_to_file()`` would crash mid-stream after assembling the program.
    """
    try:
        info = _ffmpeg.probe(self.source)
        has_audio = any(s.get("codec_type") == "audio" for s in info.get("streams", []))
    except Exception as exc:
        message = f"music_bed source {str(self.source)!r} is not a readable audio file: {exc}"
        raise PlanValidationError(
            message,
            [PlanError(code=PlanErrorCode.SOURCE_UNREADABLE, field="source")],
        ) from exc
    if not has_audio:
        message = f"music_bed source {str(self.source)!r} has no audio stream"
        raise PlanValidationError(
            message,
            [PlanError(code=PlanErrorCode.SOURCE_UNREADABLE, field="source")],
        )

bed_stages 

bed_stages(
    program_seconds: float,
    speech: list[tuple[float, float]] | None,
) -> list[str]

The ordered filter_complex fragments applied to the bed stream.

[gain] -> [fade in/out] -> [duck automation] -> [loop/trim pin]: the bed is scaled to gain, faded, ducked under speech (when given and duck is set), then pinned to exactly program_seconds (atrim end + apad whole_dur) so a looped bed neither truncates early nor extends the output past the program. Used by the file mix path via :func:build_music_bed_filter_complex.

Source code in src/videopython/editing/audio_ops.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def bed_stages(self, program_seconds: float, speech: list[tuple[float, float]] | None) -> list[str]:
    """The ordered ``filter_complex`` fragments applied to the bed stream.

    ``[gain] -> [fade in/out] -> [duck automation] -> [loop/trim pin]``: the
    bed is scaled to ``gain``, faded, ducked under ``speech`` (when given and
    ``duck`` is set), then pinned to exactly ``program_seconds`` (``atrim``
    end + ``apad`` whole_dur) so a looped bed neither truncates early nor
    extends the output past the program. Used by the file
    mix path via :func:`build_music_bed_filter_complex`.
    """
    stages: list[str] = [f"volume={self.gain:.6f}"]
    if self.fade_in > 0:
        stages.append(f"afade=t=in:st=0:d={self.fade_in:.6f}")
    if self.fade_out > 0:
        fade_start = max(0.0, program_seconds - self.fade_out)
        stages.append(f"afade=t=out:st={fade_start:.6f}:d={self.fade_out:.6f}")
    if self.duck is not None and speech:
        stages.append(
            duck_volume_expression(speech, duck=self.duck, attack=self.duck_attack, release=self.duck_release)
        )
    # Pin the bed to the program length: trim a too-long (looped) bed and pad
    # a too-short one, so amix's program input is exactly matched.
    stages.append(f"atrim=end={program_seconds:.6f}")
    stages.append(f"apad=whole_dur={program_seconds:.6f}")
    return stages

JSON Schema

schema = VideoEdit.json_schema()               # LLM-exposed ops only
strict = VideoEdit.json_schema(strict=True)    # closed provider grammar

The default excludes server-only ops such as image_overlay (why). Import the AI operation classes before schema generation to include them; see the LLM guide. strict=True closes every object, makes every property required, expresses the union as an anyOf without a discriminator, and hoists $defs to the document root. Usage: Author edit plans with your own LLM.

Classes

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 
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205
2206
2207
2208
2209
2210
2211
2212
2213
2214
2215
2216
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
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, effect-window endpoints past the running
        predicted duration are clamped to that duration. A window that starts at
        or after the duration is an empty no-op, matching ``run_to_file()``.
        """
        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 supplied video metadata; referenced assets can still be probed.

        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.

        Like :meth:`validate_with_metadata`, but collects independent plan errors.
        An incomplete source metadata map raises ``ValueError``. The walk accumulates
        errors 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`: clampable window overruns are 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}]"
                active = True
                if clamp_windows:
                    op, active = _resolve_effect_window(op, seg_meta.total_seconds)
                for message, err in _window_errors(op, seg_meta.total_seconds, location):
                    emit(message, err)
                if not active:
                    continue
                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}]"
            active = True
            if clamp_windows:
                op, active = _resolve_effect_window(op, assembled.total_seconds)
            for message, err in _window_errors(op, assembled.total_seconds, location):
                emit(message, err)
            if not active:
                continue
            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,
        *,
        on_progress: Callable[[RenderProgress], None] | None = None,
    ) -> Path:
        """Execute the plan, streaming directly to a file.

        Video frames and segment audio stream through FFmpeg. Frame buffers stay
        bounded; context and operation state can grow with duration.
        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)

        progress = _Progress(on_progress)
        on_frame = progress.advance if on_progress is not None else None
        progress.start("compilation")
        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)
            progress.finish()
            # Stage 1: assemble the segments.
            target = _stage_target(is_last=not (has_post or has_bed))
            if len(plans) == 1:
                progress.start("segment", total=None, unit="frames", segment_index=0)
                assembled = stream_segment(
                    plans[0],
                    target,
                    format=format,
                    preset=preset,
                    crf=crf,
                    on_frame=on_frame,
                )
                progress.finish()
            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 index, plan in enumerate(plans):
                        tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
                        tmp.close()
                        audible.append(source_has_audio_stream(plan.source_path))
                        temp_files.append(Path(tmp.name))
                        progress.start("segment", total=None, unit="frames", segment_index=index)
                        stream_segment(
                            plan,
                            Path(tmp.name),
                            with_audio=True,
                            format=format,
                            preset=preset,
                            crf=crf,
                            on_frame=on_frame,
                        )
                        progress.finish()

                    progress.start("assembly")
                    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
                        )
                    progress.finish()
                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:
                progress.start("post_operations", total=None, unit="frames")
                target = _stage_target(is_last=not has_bed)
                assembled = self._apply_post_operations(
                    assembled,
                    target,
                    context,
                    format=format,
                    preset=preset,
                    crf=crf,
                    on_frame=on_frame,
                )
                progress.finish()

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

                progress.finish()

            progress.start("complete")
            progress.finish()
            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,
        on_frame: Callable[[int], None] | None = None,
    ) -> 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, on_frame=on_frame)
        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):
                    op, active = _resolve_effect_window(op, running.total_seconds)
                    if not active:
                        continue
                    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, vignette, ...):
                        # 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 UNLESS the op is video_passthrough.
                        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 op.video_passthrough:
                            # Audio-only filter effect (volume_adjust): there is
                            # no video filter to place by design, but the audio
                            # twin still lands -- at the stage a video filter
                            # WOULD have landed, so audio/video stage placement
                            # stays coupled. Deliberately does not touch
                            # vf_filters/post_vf_filters/pipe_meta: the op is
                            # invisible to the video chain, so it neither opens
                            # the encode stage nor blocks a later frame effect.
                            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 
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
@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, effect-window endpoints past the running predicted duration are clamped to that duration. A window that starts at or after the duration is an empty no-op, matching run_to_file().

Source code in src/videopython/editing/video_edit.py 
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
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, effect-window endpoints past the running
    predicted duration are clamped to that duration. A window that starts at
    or after the duration is an empty no-op, matching ``run_to_file()``.
    """
    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 supplied video metadata; referenced assets can still be probed.

See :meth:validate for the clamp_windows semantics.

Source code in src/videopython/editing/video_edit.py 
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
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 supplied video metadata; referenced assets can still be probed.

    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.

Like :meth:validate_with_metadata, but collects independent plan errors. An incomplete source metadata map raises ValueError. The walk accumulates errors 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: clampable window overruns are 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 
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
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.

    Like :meth:`validate_with_metadata`, but collects independent plan errors.
    An incomplete source metadata map raises ``ValueError``. The walk accumulates
    errors 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`: clampable window overruns are 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 
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
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 
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
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 
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
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,
    *,
    on_progress: Callable[[RenderProgress], None]
    | None = None,
) -> Path

Execute the plan, streaming directly to a file.

Video frames and segment audio stream through FFmpeg. Frame buffers stay bounded; context and operation state can grow with duration. 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 
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
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,
    *,
    on_progress: Callable[[RenderProgress], None] | None = None,
) -> Path:
    """Execute the plan, streaming directly to a file.

    Video frames and segment audio stream through FFmpeg. Frame buffers stay
    bounded; context and operation state can grow with duration.
    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)

    progress = _Progress(on_progress)
    on_frame = progress.advance if on_progress is not None else None
    progress.start("compilation")
    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)
        progress.finish()
        # Stage 1: assemble the segments.
        target = _stage_target(is_last=not (has_post or has_bed))
        if len(plans) == 1:
            progress.start("segment", total=None, unit="frames", segment_index=0)
            assembled = stream_segment(
                plans[0],
                target,
                format=format,
                preset=preset,
                crf=crf,
                on_frame=on_frame,
            )
            progress.finish()
        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 index, plan in enumerate(plans):
                    tmp = tempfile.NamedTemporaryFile(suffix=f".{format}", delete=False)
                    tmp.close()
                    audible.append(source_has_audio_stream(plan.source_path))
                    temp_files.append(Path(tmp.name))
                    progress.start("segment", total=None, unit="frames", segment_index=index)
                    stream_segment(
                        plan,
                        Path(tmp.name),
                        with_audio=True,
                        format=format,
                        preset=preset,
                        crf=crf,
                        on_frame=on_frame,
                    )
                    progress.finish()

                progress.start("assembly")
                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
                    )
                progress.finish()
            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:
            progress.start("post_operations", total=None, unit="frames")
            target = _stage_target(is_last=not has_bed)
            assembled = self._apply_post_operations(
                assembled,
                target,
                context,
                format=format,
                preset=preset,
                crf=crf,
                on_frame=on_frame,
            )
            progress.finish()

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

            progress.finish()

        progress.start("complete")
        progress.finish()
        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

Bases: BaseModel

A single source segment with its operation chain.

Source code in src/videopython/editing/video_edit.py 
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
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 
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
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,
    )

TransitionSpec

Bases: BaseModel

How one segment enters from the previous one: a native ffmpeg crossfade.

A transition describes the boundary on its INCOMING side -- it lives on SegmentConfig.transition_in of segment i and overlaps the last duration seconds of segment i-1 with the first duration of segment i. type is a literal ffmpeg xfade transition= mode from a curated catalog; duration is the overlap in seconds (the assembled timeline shortens by it); audio acrossfade-s the audio across the same overlap when both adjacent segments carry audio, else the audio hard butt-joins. Frozen and closed so it surfaces as a constrained object in :meth:VideoEdit.json_schema.

Source code in src/videopython/editing/video_edit.py 
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
class TransitionSpec(BaseModel):
    """How one segment enters from the previous one: a native ffmpeg crossfade.

    A transition describes the boundary on its INCOMING side -- it lives on
    ``SegmentConfig.transition_in`` of segment ``i`` and overlaps the last
    ``duration`` seconds of segment ``i-1`` with the first ``duration`` of
    segment ``i``. ``type`` is a literal ffmpeg ``xfade`` ``transition=`` mode
    from a curated catalog; ``duration`` is the overlap in seconds (the
    assembled timeline shortens by it); ``audio`` ``acrossfade``-s the audio
    across the same overlap when both adjacent segments carry audio, else the
    audio hard butt-joins. Frozen and closed so it surfaces as a constrained
    object in :meth:`VideoEdit.json_schema`.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    type: TransitionType = Field(
        description="Crossfade style: an ffmpeg xfade transition mode from the curated catalog."
    )
    duration: float = Field(gt=0, description="Overlap duration in seconds; the assembled timeline shortens by it.")
    audio: bool = Field(
        True,
        description=(
            "Crossfade (acrossfade) the audio across the same overlap when both adjacent "
            "segments have audio; otherwise the audio hard butt-joins at the seam."
        ),
    )