#!/usr/bin/env python
"""Deterministic 60 fps, data-driven camera cut planning.

The public entry point is :func:`plan_cut_boundaries`.  Its ``inputs`` value
uses the CameraDirector prepared-input contract: 63 joint-position values,
63 joint-velocity values, 32 mel bands, RMS, and onset (160 values total).
Duration samples are expressed in *frames* so callers cannot silently mix
seconds and frames.
"""

from __future__ import annotations

import hashlib
import json
import math
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping, Protocol, Sequence

import numpy as np


CUT_PLANNER_FORMAT = "cw-ai-data-driven-cut-plan"
CUT_PLANNER_SCHEMA_VERSION = 2
CANDIDATE_RANKER_FORMAT = "cw-ai-cut-candidate-ranker"
CANDIDATE_RANKER_SCHEMA_VERSION = 2
DURATION_PRIOR_FORMAT = "cw-ai-cut-duration-prior"
DURATION_PRIOR_SCHEMA_VERSION = 1
REQUIRED_SAMPLE_RATE = 60
PREPARED_INPUT_SIZE = 160

CANDIDATE_FEATURE_NAMES = (
    "beat_source",
    "onset_source",
    "motion_source",
    "mel_novelty_source",
    "fallback_source",
    "preferred_source",
    "same_audio_source",
    "beat_proximity",
    "onset_strength",
    "onset_delta",
    "motion_strength",
    "motion_delta",
    "mel_novelty",
    "mel_delta",
    "preferred_proximity",
    "same_audio_proximity",
    "timeline_progress",
    "duration_since_previous_cut_seconds",
    "previous_cut_progress",
    "source_count",
    "history_count",
    "previous_1_duration_seconds",
    "previous_1_music_energy",
    "previous_1_motion_energy",
    "previous_1_cut_beat_proximity",
    "previous_2_duration_seconds",
    "previous_2_music_energy",
    "previous_2_motion_energy",
    "previous_2_cut_beat_proximity",
    "previous_3_duration_seconds",
    "previous_3_music_energy",
    "previous_3_motion_energy",
    "previous_3_cut_beat_proximity",
)
HISTORY_CONTEXT_DEPTH = 3

_SOURCE_NAMES = frozenset(
    {
        "endpoint",
        "beat",
        "onset",
        "motion",
        "mel_novelty",
        "fallback",
        "preferred",
        "locked",
        "same_audio",
    }
)
_DENSITY_VALUES = frozenset({"auto", "slow", "normal", "fast"})


def _finite_array(value: Any, name: str, *, ndim: int = 1) -> np.ndarray:
    result = np.asarray(value, dtype=np.float64)
    if result.ndim != ndim:
        raise ValueError(f"{name} must be {ndim}-dimensional.")
    if not np.all(np.isfinite(result)):
        raise ValueError(f"{name} must contain only finite values.")
    return result


def _strict_frames(
    value: Sequence[int] | np.ndarray,
    name: str,
    *,
    frame_count: int,
    allow_endpoints: bool = False,
) -> np.ndarray:
    raw = np.asarray(value)
    if raw.ndim != 1:
        raise ValueError(f"{name} must be one-dimensional.")
    if not (
        np.issubdtype(raw.dtype, np.integer)
        or all(
            isinstance(item, (int, np.integer))
            and not isinstance(item, (bool, np.bool_))
            for item in raw.tolist()
        )
    ):
        raise ValueError(f"{name} must contain integer frame indices.")
    result = raw.astype(np.int64, copy=True)
    if len(result) and np.any(np.diff(result) <= 0):
        raise ValueError(f"{name} must be strictly increasing and unique.")
    lower = 0 if allow_endpoints else 1
    upper = frame_count if allow_endpoints else frame_count - 1
    if len(result) and (result[0] < lower or result[-1] > upper):
        range_text = "[0, frame_count]" if allow_endpoints else "(0, frame_count)"
        raise ValueError(f"{name} must lie inside {range_text}.")
    return result


def _readonly(value: np.ndarray) -> np.ndarray:
    result = value.copy()
    result.setflags(write=False)
    return result


def _positive_integer(value: Any, name: str) -> int:
    if (
        isinstance(value, (bool, np.bool_))
        or not isinstance(value, (int, np.integer))
        or int(value) < 1
    ):
        raise ValueError(f"{name} must be a positive integer.")
    return int(value)


@dataclass(frozen=True)
class CutSignals:
    """Validated frame-aligned signals used to propose cut candidates."""

    frame_count: int
    sample_rate: int
    beat_frames: Sequence[int] | np.ndarray
    onset_peak_frames: Sequence[int] | np.ndarray
    onset_strength: Sequence[float] | np.ndarray
    motion_strength: Sequence[float] | np.ndarray
    mel_novelty: Sequence[float] | np.ndarray

    def __post_init__(self) -> None:
        frame_count = _positive_integer(self.frame_count, "frame_count")
        sample_rate = _positive_integer(self.sample_rate, "sample_rate")
        if frame_count == 1:
            raise ValueError("frame_count must be at least two.")
        if sample_rate != REQUIRED_SAMPLE_RATE:
            raise ValueError("The cut planner requires exactly 60 fps input.")
        beat_frames = _strict_frames(
            self.beat_frames, "beat_frames", frame_count=frame_count
        )
        onset_peak_frames = _strict_frames(
            self.onset_peak_frames,
            "onset_peak_frames",
            frame_count=frame_count,
        )
        arrays = {}
        for name in ("onset_strength", "motion_strength", "mel_novelty"):
            values = _finite_array(getattr(self, name), name)
            if values.shape != (frame_count,):
                raise ValueError(
                    f"{name} must have shape ({frame_count},), got {values.shape}."
                )
            arrays[name] = _readonly(values)
        object.__setattr__(self, "frame_count", frame_count)
        object.__setattr__(self, "sample_rate", sample_rate)
        object.__setattr__(self, "beat_frames", _readonly(beat_frames))
        object.__setattr__(
            self, "onset_peak_frames", _readonly(onset_peak_frames)
        )
        for name, values in arrays.items():
            object.__setattr__(self, name, values)


@dataclass(frozen=True)
class CutDirectives:
    """Hard and soft editor intent; no-cut ranges are half-open [start, end)."""

    locked_frames: tuple[int, ...] = ()
    preferred_frames: tuple[int, ...] = ()
    no_cut_ranges: tuple[tuple[int, int], ...] = ()
    preferred_tolerance_frames: int = 6

    @classmethod
    def from_mapping(cls, value: Mapping[str, Any]) -> CutDirectives:
        allowed = {
            "lockedFrames",
            "preferredFrames",
            "noCutRanges",
            "preferredToleranceFrames",
        }
        unknown = set(value) - allowed
        if unknown:
            raise ValueError(
                "Unknown cut directive keys: " + ", ".join(sorted(unknown))
            )
        ranges = value.get("noCutRanges", ())
        return cls(
            locked_frames=tuple(value.get("lockedFrames", ())),
            preferred_frames=tuple(value.get("preferredFrames", ())),
            no_cut_ranges=tuple(tuple(item) for item in ranges),
            preferred_tolerance_frames=value.get(
                "preferredToleranceFrames", 6
            ),
        )


@dataclass(frozen=True)
class CandidatePool:
    """Sorted candidate frames and their complete union-source provenance."""

    frame_count: int
    frames: np.ndarray
    sources: tuple[tuple[str, ...], ...]

    def __post_init__(self) -> None:
        frame_count = _positive_integer(self.frame_count, "frame_count")
        frames = _strict_frames(
            self.frames,
            "candidate frames",
            frame_count=frame_count,
            allow_endpoints=True,
        )
        if len(frames) < 2 or frames[0] != 0 or frames[-1] != frame_count:
            raise ValueError("Candidate pool must contain both timeline endpoints.")
        if len(self.sources) != len(frames):
            raise ValueError("Candidate source count does not match frame count.")
        normalized_sources = []
        for source_values in self.sources:
            normalized = tuple(sorted(set(source_values)))
            if not normalized or set(normalized) - _SOURCE_NAMES:
                raise ValueError("Candidate pool contains an invalid source name.")
            normalized_sources.append(normalized)
        object.__setattr__(self, "frame_count", frame_count)
        object.__setattr__(self, "frames", _readonly(frames))
        object.__setattr__(self, "sources", tuple(normalized_sources))


@dataclass(frozen=True)
class HistoryShot:
    """One completed shot used as causal context for the next cut decision."""

    start_frame: int
    end_frame: int
    duration_frames: int
    music_energy: float
    motion_energy: float
    cut_beat_proximity: float

    def to_trace(self) -> dict[str, Any]:
        return {
            "startFrame": self.start_frame,
            "endFrame": self.end_frame,
            "durationFrames": self.duration_frames,
            "durationSeconds": self.duration_frames / REQUIRED_SAMPLE_RATE,
            "musicEnergy": self.music_energy,
            "motionEnergy": self.motion_energy,
            "cutBeatProximity": self.cut_beat_proximity,
        }


@dataclass(frozen=True)
class EmpiricalDurationPrior:
    """Gaussian-kernel empirical prior that preserves multiple duration modes."""

    duration_frames: np.ndarray
    sample_weights: np.ndarray
    bandwidth_frames: float
    sample_rate: int = REQUIRED_SAMPLE_RATE

    def __post_init__(self) -> None:
        durations = _finite_array(self.duration_frames, "duration_frames")
        weights = _finite_array(self.sample_weights, "sample_weights")
        bandwidth = float(self.bandwidth_frames)
        sample_rate = _positive_integer(self.sample_rate, "sample_rate")
        if sample_rate != REQUIRED_SAMPLE_RATE:
            raise ValueError("Duration prior must use exactly 60 fps.")
        if not len(durations) or weights.shape != durations.shape:
            raise ValueError("Duration samples and weights must be non-empty and aligned.")
        if np.any(durations <= 0.0) or np.any(weights <= 0.0):
            raise ValueError("Duration samples and weights must be positive.")
        if not math.isfinite(bandwidth) or bandwidth <= 0.0:
            raise ValueError("bandwidth_frames must be finite and positive.")
        weights = weights / float(np.sum(weights))
        object.__setattr__(self, "duration_frames", _readonly(durations))
        object.__setattr__(self, "sample_weights", _readonly(weights))
        object.__setattr__(self, "bandwidth_frames", bandwidth)
        object.__setattr__(self, "sample_rate", sample_rate)

    @classmethod
    def from_duration_frames(
        cls,
        duration_frames: Sequence[float] | np.ndarray,
        *,
        sample_weights: Sequence[float] | np.ndarray | None = None,
        bandwidth_frames: float | None = None,
        sample_rate: int = REQUIRED_SAMPLE_RATE,
    ) -> EmpiricalDurationPrior:
        durations = _finite_array(duration_frames, "duration_frames")
        if sample_weights is None:
            weights = np.ones(len(durations), dtype=np.float64)
        else:
            weights = _finite_array(sample_weights, "sample_weights")
        if bandwidth_frames is None:
            if len(durations) < 2:
                bandwidth_frames = sample_rate * 0.25
            else:
                spread = float(np.std(durations))
                bandwidth_frames = max(
                    sample_rate * 0.12,
                    1.06 * max(spread, 1.0) * len(durations) ** -0.2,
                )
        return cls(durations, weights, bandwidth_frames, sample_rate)

    @classmethod
    def from_duration_seconds(
        cls,
        duration_seconds: Sequence[float] | np.ndarray,
        **kwargs: Any,
    ) -> EmpiricalDurationPrior:
        seconds = _finite_array(duration_seconds, "duration_seconds")
        return cls.from_duration_frames(
            seconds * REQUIRED_SAMPLE_RATE, **kwargs
        )

    def log_scores(self, duration_frames: np.ndarray) -> np.ndarray:
        query = _finite_array(duration_frames, "duration query")

        def log_density(points: np.ndarray) -> np.ndarray:
            result = np.empty(len(points), dtype=np.float64)
            for start in range(0, len(points), 2048):
                end = min(len(points), start + 2048)
                delta = (
                    points[start:end, None]
                    - self.duration_frames[None, :]
                ) / self.bandwidth_frames
                kernels = (
                    np.exp(-0.5 * np.square(delta)) @ self.sample_weights
                )
                result[start:end] = np.log(
                    np.maximum(kernels, np.finfo(np.float64).tiny)
                )
            return result

        # This reference is independent of the query batch, so scoring one
        # duration and scoring the same duration in a batch are identical.
        reference = float(np.max(log_density(self.duration_frames)))
        return log_density(query) - reference

    def to_artifact(self) -> dict[str, Any]:
        return {
            "format": DURATION_PRIOR_FORMAT,
            "schemaVersion": DURATION_PRIOR_SCHEMA_VERSION,
            "sampleRate": self.sample_rate,
            "durationFrames": self.duration_frames.tolist(),
            "sampleWeights": self.sample_weights.tolist(),
            "bandwidthFrames": self.bandwidth_frames,
        }

    @classmethod
    def from_artifact(cls, value: Mapping[str, Any]) -> EmpiricalDurationPrior:
        expected = {
            "format",
            "schemaVersion",
            "sampleRate",
            "durationFrames",
            "sampleWeights",
            "bandwidthFrames",
        }
        if set(value) != expected:
            raise ValueError("Duration-prior artifact keys do not match schema 1.")
        if (
            value["format"] != DURATION_PRIOR_FORMAT
            or type(value["schemaVersion"]) is not int
            or value["schemaVersion"] != DURATION_PRIOR_SCHEMA_VERSION
        ):
            raise ValueError("Unsupported duration-prior artifact format or schema.")
        return cls(
            np.asarray(value["durationFrames"], dtype=np.float64),
            np.asarray(value["sampleWeights"], dtype=np.float64),
            float(value["bandwidthFrames"]),
            int(value["sampleRate"]),
        )


class CandidateScoreModel(Protocol):
    def score(self, feature_matrix: np.ndarray) -> np.ndarray:
        """Return one finite logit-like score per feature row."""


@dataclass(frozen=True)
class LinearCandidateScoreModel:
    """Portable standardized linear ranker with a strict JSON artifact."""

    weights: np.ndarray
    bias: float
    mean: np.ndarray
    scale: np.ndarray
    metadata: Mapping[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        size = len(CANDIDATE_FEATURE_NAMES)
        arrays = {}
        for name in ("weights", "mean", "scale"):
            values = _finite_array(getattr(self, name), name)
            if values.shape != (size,):
                raise ValueError(f"{name} must have shape ({size},).")
            arrays[name] = _readonly(values)
        if np.any(arrays["scale"] <= 0.0):
            raise ValueError("Model feature scales must be positive.")
        bias = float(self.bias)
        if not math.isfinite(bias):
            raise ValueError("Model bias must be finite.")
        try:
            json.dumps(dict(self.metadata), allow_nan=False)
        except (TypeError, ValueError) as error:
            raise ValueError("Model metadata must be finite JSON data.") from error
        for name, values in arrays.items():
            object.__setattr__(self, name, values)
        object.__setattr__(self, "bias", bias)
        object.__setattr__(self, "metadata", dict(self.metadata))

    def score(self, feature_matrix: np.ndarray) -> np.ndarray:
        features = _finite_array(feature_matrix, "feature_matrix", ndim=2)
        if features.shape[1] != len(CANDIDATE_FEATURE_NAMES):
            raise ValueError("Candidate feature matrix width does not match contract.")
        return ((features - self.mean) / self.scale) @ self.weights + self.bias

    def to_artifact(self) -> dict[str, Any]:
        return {
            "format": CANDIDATE_RANKER_FORMAT,
            "schemaVersion": CANDIDATE_RANKER_SCHEMA_VERSION,
            "featureNames": list(CANDIDATE_FEATURE_NAMES),
            "weights": self.weights.tolist(),
            "bias": self.bias,
            "mean": self.mean.tolist(),
            "scale": self.scale.tolist(),
            "metadata": dict(self.metadata),
        }

    @classmethod
    def from_artifact(cls, value: Mapping[str, Any]) -> LinearCandidateScoreModel:
        expected = {
            "format",
            "schemaVersion",
            "featureNames",
            "weights",
            "bias",
            "mean",
            "scale",
            "metadata",
        }
        if set(value) != expected:
            raise ValueError(
                "Candidate-ranker artifact keys do not match runtime schema 2."
            )
        if (
            value["format"] != CANDIDATE_RANKER_FORMAT
            or type(value["schemaVersion"]) is not int
            or value["schemaVersion"] != CANDIDATE_RANKER_SCHEMA_VERSION
            or tuple(value["featureNames"]) != CANDIDATE_FEATURE_NAMES
        ):
            raise ValueError(
                "Candidate-ranker artifact contract does not match runtime schema 2; "
                "retrain the ranker with history-context features."
            )
        return cls(
            np.asarray(value["weights"], dtype=np.float64),
            float(value["bias"]),
            np.asarray(value["mean"], dtype=np.float64),
            np.asarray(value["scale"], dtype=np.float64),
            value["metadata"],
        )


@dataclass(frozen=True)
class CutPlannerConfig:
    fallback_interval_frames: int = 6
    peak_minimum_distance_frames: int = 6
    onset_peak_quantile: float = 0.70
    motion_peak_quantile: float = 0.72
    mel_peak_quantile: float = 0.70
    minimum_shot_frames: int = 12
    maximum_shot_frames: int = 720
    ultrashort_frames: int = 45
    short_tail_frames: int = 60
    # The ranker refines timing inside the structural duration objective.  A
    # sub-unit weight keeps one unusual venue from turning relative rank
    # logits into an extra-cut decision during cross-domain validation.
    candidate_score_weight: float = 0.5
    duration_prior_weight: float = 1.25
    density_weight: float = 0.85
    # Calibrated on the authored Arisa exact-audio reference: 2.5 produced
    # 69 shots versus 65 authored while retaining 0.953 boundary recall at
    # +/-6 frames.  Lower values materially over-cut the same song.
    cut_count_cost: float = 2.5
    consecutive_ultrashort_cost: float = 2.5
    short_tail_cost: float = 4.0
    preferred_bonus: float = 2.4
    same_audio_bonus: float = 2.8
    history_beam_width: int = 4
    # Authored timelines commonly hold their final camera through a long
    # silent/idle export tail.  Do not manufacture a fallback cut merely to
    # satisfy the normal maximum shot duration once every causal activity
    # signal has been inactive for at least two seconds.  The threshold is
    # intentionally near zero: this is an export-tail exception, not a way to
    # lengthen quiet-but-active musical passages.
    inactive_tail_minimum_frames: int = 120
    inactive_tail_signal_threshold: float = 1e-6

    def __post_init__(self) -> None:
        integer_names = (
            "fallback_interval_frames",
            "peak_minimum_distance_frames",
            "minimum_shot_frames",
            "maximum_shot_frames",
            "ultrashort_frames",
            "short_tail_frames",
            "history_beam_width",
            "inactive_tail_minimum_frames",
        )
        for name in integer_names:
            value = getattr(self, name)
            if (
                isinstance(value, (bool, np.bool_))
                or not isinstance(value, (int, np.integer))
                or value <= 0
            ):
                raise ValueError(f"{name} must be a positive integer.")
        if self.maximum_shot_frames <= self.minimum_shot_frames:
            raise ValueError("maximum_shot_frames must exceed minimum_shot_frames.")
        for name in (
            "onset_peak_quantile",
            "motion_peak_quantile",
            "mel_peak_quantile",
        ):
            value = float(getattr(self, name))
            if not 0.0 <= value <= 1.0:
                raise ValueError(f"{name} must lie in [0, 1].")
        for name in (
            "candidate_score_weight",
            "duration_prior_weight",
            "density_weight",
            "cut_count_cost",
            "consecutive_ultrashort_cost",
            "short_tail_cost",
            "preferred_bonus",
            "same_audio_bonus",
            "inactive_tail_signal_threshold",
        ):
            value = float(getattr(self, name))
            if not math.isfinite(value) or value < 0.0:
                raise ValueError(f"{name} must be finite and non-negative.")


@dataclass(frozen=True)
class CutPlanResult:
    boundaries: tuple[int, ...]
    trace: dict[str, Any]
    metadata: dict[str, Any]


def signals_from_prepared_inputs(
    inputs: np.ndarray,
    beat_frames: Sequence[int] | np.ndarray,
    peak_frames: Sequence[int] | np.ndarray,
    *,
    sample_rate: int = REQUIRED_SAMPLE_RATE,
) -> CutSignals:
    """Convert CameraDirector's [frame, 160] prepared input into cut signals."""

    values = _finite_array(inputs, "inputs", ndim=2)
    if values.shape[1] != PREPARED_INPUT_SIZE:
        raise ValueError(
            f"inputs must have exactly {PREPARED_INPUT_SIZE} features per frame."
        )
    velocity = values[:, 63:126]
    motion = np.sqrt(np.mean(np.square(velocity), axis=1))
    mel = values[:, -34:-2]
    mel_novelty = np.zeros(len(values), dtype=np.float64)
    mel_novelty[1:] = np.sqrt(np.mean(np.square(np.diff(mel, axis=0)), axis=1))
    return CutSignals(
        frame_count=len(values),
        sample_rate=sample_rate,
        beat_frames=beat_frames,
        onset_peak_frames=peak_frames,
        onset_strength=values[:, -1],
        motion_strength=motion,
        mel_novelty=mel_novelty,
    )


def plan_cut_boundaries(
    inputs: np.ndarray | CutSignals,
    beat_frames: Sequence[int] | np.ndarray,
    peak_frames: Sequence[int] | np.ndarray,
    duration_samples_or_prior: Sequence[float] | EmpiricalDurationPrior,
    *,
    sample_rate: int = REQUIRED_SAMPLE_RATE,
    model: CandidateScoreModel | Mapping[str, Any] | Callable[[np.ndarray], Any] | None = None,
    same_audio_boundaries: Sequence[int] | np.ndarray | None = None,
    directives: CutDirectives | Mapping[str, Any] | None = None,
    density: str = "auto",
    config: CutPlannerConfig | None = None,
) -> CutPlanResult:
    """Plan globally coherent cuts; duration sequences use frame units.

    ``beat_frames`` and ``peak_frames`` remain explicit even when ``inputs`` is
    already a :class:`CutSignals`, preventing accidentally scoring a different
    event cache.  In that case they must exactly equal the signal payload.
    """

    if isinstance(inputs, CutSignals):
        signals = inputs
        if _positive_integer(sample_rate, "sample_rate") != signals.sample_rate:
            raise ValueError("sample_rate conflicts with CutSignals.")
        supplied_beats = _strict_frames(
            beat_frames, "beat_frames", frame_count=signals.frame_count
        )
        supplied_peaks = _strict_frames(
            peak_frames, "peak_frames", frame_count=signals.frame_count
        )
        if not np.array_equal(supplied_beats, signals.beat_frames) or not np.array_equal(
            supplied_peaks, signals.onset_peak_frames
        ):
            raise ValueError("Explicit beat/peak frames conflict with CutSignals.")
    else:
        signals = signals_from_prepared_inputs(
            inputs, beat_frames, peak_frames, sample_rate=sample_rate
        )
    prior = (
        duration_samples_or_prior
        if isinstance(duration_samples_or_prior, EmpiricalDurationPrior)
        else EmpiricalDurationPrior.from_duration_frames(
            duration_samples_or_prior, sample_rate=sample_rate
        )
    )
    normalized_directives = _validate_directives(
        directives, signals.frame_count
    )
    normalized_same_audio = _normalize_same_audio(
        same_audio_boundaries, signals.frame_count
    )
    if density not in _DENSITY_VALUES:
        raise ValueError("density must be auto, slow, normal, or fast.")
    config = config or CutPlannerConfig()
    pool = generate_cut_candidates(
        signals,
        normalized_directives,
        same_audio_boundaries=normalized_same_audio,
        config=config,
    )
    return _solve_global_dag(
        signals,
        pool,
        prior,
        normalized_directives,
        normalized_same_audio,
        density,
        config,
        model,
    )


# Implementation helpers are appended below to keep the public contract above
# easy to audit.


def _validate_directives(
    directives: CutDirectives | Mapping[str, Any] | None,
    frame_count: int,
) -> CutDirectives:
    if directives is None:
        directives = CutDirectives()
    elif isinstance(directives, Mapping):
        directives = CutDirectives.from_mapping(directives)
    elif not isinstance(directives, CutDirectives):
        raise TypeError("directives must be CutDirectives, a mapping, or None.")

    locked = _strict_frames(
        directives.locked_frames,
        "locked_frames",
        frame_count=frame_count,
    )
    preferred = _strict_frames(
        directives.preferred_frames,
        "preferred_frames",
        frame_count=frame_count,
    )
    tolerance = directives.preferred_tolerance_frames
    if (
        isinstance(tolerance, (bool, np.bool_))
        or not isinstance(tolerance, (int, np.integer))
        or not 0 <= int(tolerance) <= frame_count
    ):
        raise ValueError(
            "preferred_tolerance_frames must be a non-negative integer."
        )

    ranges: list[tuple[int, int]] = []
    for index, item in enumerate(directives.no_cut_ranges):
        if not isinstance(item, (tuple, list)) or len(item) != 2:
            raise ValueError(f"no_cut_ranges[{index}] must be a frame pair.")
        start, end = item
        if any(
            isinstance(value, bool)
            or not isinstance(value, (int, np.integer))
            for value in (start, end)
        ):
            raise ValueError("No-cut range endpoints must be integer frames.")
        start, end = int(start), int(end)
        if not 0 <= start < end <= frame_count:
            raise ValueError("No-cut ranges must satisfy 0 <= start < end <= frame_count.")
        if ranges and start < ranges[-1][1]:
            raise ValueError("No-cut ranges must be sorted and non-overlapping.")
        ranges.append((start, end))

    def inside_no_cut(frame: int) -> bool:
        return any(start <= frame < end for start, end in ranges)

    conflicts = [
        int(frame)
        for frame in np.concatenate((locked, preferred))
        if inside_no_cut(int(frame))
    ]
    if conflicts:
        raise ValueError(
            "Locked/preferred cuts conflict with a no-cut range: "
            + ", ".join(str(value) for value in conflicts)
        )
    return CutDirectives(
        locked_frames=tuple(int(value) for value in locked),
        preferred_frames=tuple(int(value) for value in preferred),
        no_cut_ranges=tuple(ranges),
        preferred_tolerance_frames=int(tolerance),
    )


def _normalize_same_audio(
    boundaries: Sequence[int] | np.ndarray | None,
    frame_count: int,
) -> np.ndarray:
    if boundaries is None:
        return np.empty(0, dtype=np.int64)
    normalized = _strict_frames(
        boundaries,
        "same_audio_boundaries",
        frame_count=frame_count,
        allow_endpoints=True,
    )
    return _readonly(normalized[(normalized > 0) & (normalized < frame_count)])


def _robust_unit(values: np.ndarray) -> np.ndarray:
    values = np.asarray(values, dtype=np.float64)
    low, high = np.quantile(values, (0.05, 0.95))
    if high - low <= 1e-12:
        low, high = float(np.min(values)), float(np.max(values))
        if high - low <= 1e-12:
            return np.zeros_like(values)
    return np.clip((values - low) / (high - low), 0.0, 1.0)


def _inactive_tail_start(
    signals: CutSignals,
    config: CutPlannerConfig,
) -> int | None:
    """Return the first frame of a sufficiently long causal inactive suffix.

    The suffix must contain no normalized onset, mel-novelty, performer-motion,
    beat, or onset-peak evidence above a near-zero threshold.  Requiring an
    earlier active frame prevents an entirely empty/new input from silently
    disabling the normal maximum-shot contract.
    """

    activity = np.maximum.reduce(
        (
            _robust_unit(signals.onset_strength),
            _robust_unit(signals.motion_strength),
            _robust_unit(signals.mel_novelty),
        )
    )
    if len(signals.beat_frames):
        activity[np.asarray(signals.beat_frames, dtype=np.int64)] = 1.0
    if len(signals.onset_peak_frames):
        activity[np.asarray(signals.onset_peak_frames, dtype=np.int64)] = 1.0
    active = np.flatnonzero(activity > config.inactive_tail_signal_threshold)
    if not len(active):
        return None
    start = int(active[-1]) + 1
    if (
        start <= 0
        or signals.frame_count - start < config.inactive_tail_minimum_frames
    ):
        return None
    return start


def _salient_peaks(
    values: np.ndarray,
    quantile: float,
    minimum_distance: int,
) -> np.ndarray:
    normalized = _robust_unit(values)
    if len(normalized) < 3 or float(np.max(normalized)) <= 1e-12:
        return np.empty(0, dtype=np.int64)
    threshold = float(np.quantile(normalized, quantile))
    local = np.flatnonzero(
        (normalized[1:-1] >= normalized[:-2])
        & (normalized[1:-1] >= normalized[2:])
        & (normalized[1:-1] >= threshold)
    ) + 1
    ranked = sorted(local.tolist(), key=lambda frame: (-normalized[frame], frame))
    accepted: list[int] = []
    for frame in ranked:
        if all(abs(frame - previous) >= minimum_distance for previous in accepted):
            accepted.append(int(frame))
    return np.asarray(sorted(accepted), dtype=np.int64)


def generate_cut_candidates(
    signals: CutSignals,
    directives: CutDirectives | Mapping[str, Any] | None = None,
    *,
    same_audio_boundaries: Sequence[int] | np.ndarray | None = None,
    config: CutPlannerConfig | None = None,
) -> CandidatePool:
    """Return the deterministic union of all event and fallback candidates."""

    config = config or CutPlannerConfig()
    directives = _validate_directives(directives, signals.frame_count)
    same_audio = _normalize_same_audio(
        same_audio_boundaries, signals.frame_count
    )
    source_by_frame: dict[int, set[str]] = {
        0: {"endpoint"},
        signals.frame_count: {"endpoint"},
    }

    def add(frames: Sequence[int] | np.ndarray, source: str) -> None:
        for raw_frame in frames:
            frame = int(raw_frame)
            if 0 < frame < signals.frame_count:
                source_by_frame.setdefault(frame, set()).add(source)

    add(signals.beat_frames, "beat")
    add(signals.onset_peak_frames, "onset")
    add(
        _salient_peaks(
            signals.onset_strength,
            config.onset_peak_quantile,
            config.peak_minimum_distance_frames,
        ),
        "onset",
    )
    add(
        _salient_peaks(
            signals.motion_strength,
            config.motion_peak_quantile,
            config.peak_minimum_distance_frames,
        ),
        "motion",
    )
    add(
        _salient_peaks(
            signals.mel_novelty,
            config.mel_peak_quantile,
            config.peak_minimum_distance_frames,
        ),
        "mel_novelty",
    )
    add(
        range(
            config.fallback_interval_frames,
            signals.frame_count,
            config.fallback_interval_frames,
        ),
        "fallback",
    )
    add(directives.preferred_frames, "preferred")
    add(directives.locked_frames, "locked")
    add(same_audio, "same_audio")

    def forbidden(frame: int) -> bool:
        return any(
            start <= frame < end
            for start, end in directives.no_cut_ranges
        )

    for frame in tuple(source_by_frame):
        if 0 < frame < signals.frame_count and forbidden(frame):
            del source_by_frame[frame]
    frames = np.asarray(sorted(source_by_frame), dtype=np.int64)
    return CandidatePool(
        signals.frame_count,
        frames,
        tuple(tuple(source_by_frame[int(frame)]) for frame in frames),
    )


def _proximity(
    frames: np.ndarray,
    events: Sequence[int] | np.ndarray,
    radius: int,
) -> np.ndarray:
    events = np.asarray(events, dtype=np.int64)
    if not len(events):
        return np.zeros(len(frames), dtype=np.float64)
    distance = np.min(
        np.abs(frames[:, None] - events[None, :]), axis=1
    ).astype(np.float64)
    return np.clip(1.0 - distance / max(1, radius), 0.0, 1.0)


def _local_delta(
    values: np.ndarray,
    frames: np.ndarray,
    radius: int = 6,
) -> np.ndarray:
    prefix = np.concatenate(([0.0], np.cumsum(values, dtype=np.float64)))
    starts = np.maximum(0, frames - radius)
    ends = np.minimum(len(values), frames + radius + 1)
    means = (prefix[ends] - prefix[starts]) / (ends - starts)
    return values[frames] - means


def summarize_history_shots(
    signals: CutSignals,
    history_shots: Sequence[tuple[int, int]],
) -> tuple[HistoryShot, ...]:
    """Validate and summarize completed shots in chronological order.

    The summary uses only samples inside each completed shot.  It is therefore
    safe to use at inference time and cannot expose the next authored cut.
    """

    if len(history_shots) > HISTORY_CONTEXT_DEPTH:
        raise ValueError(
            f"history_shots may contain at most {HISTORY_CONTEXT_DEPTH} shots."
        )
    onset = _robust_unit(signals.onset_strength)
    motion = _robust_unit(signals.motion_strength)
    mel = _robust_unit(signals.mel_novelty)
    music = 0.5 * onset + 0.5 * mel
    result: list[HistoryShot] = []
    previous_end: int | None = None
    for index, item in enumerate(history_shots):
        if (
            not isinstance(item, (tuple, list))
            or len(item) != 2
            or any(
                isinstance(value, (bool, np.bool_))
                or not isinstance(value, (int, np.integer))
                for value in item
            )
        ):
            raise ValueError("history_shots must contain integer (start, end) pairs.")
        start, end = (int(item[0]), int(item[1]))
        if start < 0 or end <= start or end > signals.frame_count:
            raise ValueError("Each history shot must satisfy 0 <= start < end <= frame_count.")
        if previous_end is not None and start != previous_end:
            raise ValueError("history_shots must be contiguous and chronological.")
        duration = end - start
        result.append(
            HistoryShot(
                start_frame=start,
                end_frame=end,
                duration_frames=duration,
                music_energy=float(np.mean(music[start:end])),
                motion_energy=float(np.mean(motion[start:end])),
                cut_beat_proximity=float(
                    _proximity(
                        np.asarray([end], dtype=np.int64),
                        signals.beat_frames,
                        30,
                    )[0]
                ),
            )
        )
        previous_end = end
    return tuple(result)


def _history_feature_columns(
    signals: CutSignals,
    previous: np.ndarray,
    history_shots: Sequence[tuple[int, int]],
) -> tuple[np.ndarray, ...]:
    summaries = summarize_history_shots(signals, history_shots)
    columns = [np.zeros(len(previous), dtype=np.float64) for _ in range(13)]
    if not summaries:
        return tuple(columns)
    history_end = summaries[-1].end_frame
    nonzero_previous = previous[previous > 0]
    if len(nonzero_previous) and np.any(nonzero_previous != history_end):
        raise ValueError(
            "Nonzero previous_cut_frame values must equal the end of history_shots."
        )
    active = previous == history_end
    columns[0][active] = float(len(summaries))
    for lag, summary in enumerate(reversed(summaries), start=1):
        base = 1 + (lag - 1) * 4
        columns[base][active] = summary.duration_frames / signals.sample_rate
        columns[base + 1][active] = summary.music_energy
        columns[base + 2][active] = summary.motion_energy
        columns[base + 3][active] = summary.cut_beat_proximity
    return tuple(columns)


def build_candidate_features(
    signals: CutSignals,
    pool: CandidatePool,
    directives: CutDirectives | Mapping[str, Any] | None = None,
    *,
    previous_cut_frame: int | Sequence[int] | np.ndarray = 0,
    same_audio_boundaries: Sequence[int] | np.ndarray | None = None,
    history_shots: Sequence[tuple[int, int]] = (),
) -> np.ndarray:
    """Build the stable schema-v2 candidate-ranker matrix.

    ``previous_cut_frame`` may be scalar or one value per candidate.  Every
    previous frame must be at or before its candidate, which catches training
    leakage and inference wiring mistakes early.
    """

    if pool.frame_count != signals.frame_count:
        raise ValueError("Candidate pool and signals have different frame counts.")
    directives = _validate_directives(directives, signals.frame_count)
    same_audio = _normalize_same_audio(
        same_audio_boundaries, signals.frame_count
    )
    frames = pool.frames
    previous_raw = np.asarray(previous_cut_frame)
    if previous_raw.ndim == 0:
        if isinstance(previous_cut_frame, (bool, np.bool_)) or not float(
            previous_raw
        ).is_integer():
            raise ValueError("previous_cut_frame must contain integer frames.")
        previous = np.full(len(frames), int(previous_raw), dtype=np.int64)
    else:
        if previous_raw.shape != frames.shape or not np.issubdtype(
            previous_raw.dtype, np.integer
        ):
            raise ValueError(
                "previous_cut_frame must be scalar or an aligned integer array."
            )
        previous = previous_raw.astype(np.int64, copy=False)
    if np.any(previous < 0) or np.any(previous > frames):
        raise ValueError("Each previous cut must be at or before its candidate.")

    history_columns = _history_feature_columns(signals, previous, history_shots)

    onset = _robust_unit(signals.onset_strength)
    motion = _robust_unit(signals.motion_strength)
    mel = _robust_unit(signals.mel_novelty)
    source_sets = [set(values) for values in pool.sources]
    sample_frames = np.minimum(frames, signals.frame_count - 1)

    def source(name: str) -> np.ndarray:
        return np.asarray(
            [float(name in values) for values in source_sets],
            dtype=np.float64,
        )

    feature_columns = (
        source("beat"),
        source("onset"),
        source("motion"),
        source("mel_novelty"),
        source("fallback"),
        source("preferred"),
        source("same_audio"),
        _proximity(frames, signals.beat_frames, 30),
        onset[sample_frames],
        _local_delta(onset, sample_frames),
        motion[sample_frames],
        _local_delta(motion, sample_frames),
        mel[sample_frames],
        _local_delta(mel, sample_frames),
        _proximity(
            frames,
            directives.preferred_frames,
            directives.preferred_tolerance_frames,
        ),
        _proximity(frames, same_audio, 6),
        frames.astype(np.float64) / signals.frame_count,
        (frames - previous).astype(np.float64) / signals.sample_rate,
        previous.astype(np.float64) / signals.frame_count,
        np.asarray(
            [
                float(len(values - {"endpoint", "locked"}))
                for values in source_sets
            ],
            dtype=np.float64,
        ),
        *history_columns,
    )
    result = np.column_stack(feature_columns)
    if result.shape != (len(frames), len(CANDIDATE_FEATURE_NAMES)):
        raise RuntimeError("Internal candidate feature contract mismatch.")
    return result


def _weighted_quantile(
    values: np.ndarray,
    weights: np.ndarray,
    quantile: float,
) -> float:
    order = np.argsort(values, kind="stable")
    ordered_values = values[order]
    cumulative = np.cumsum(weights[order])
    index = int(np.searchsorted(cumulative, quantile, side="left"))
    return float(ordered_values[min(index, len(ordered_values) - 1)])


def _model_scores(
    model: CandidateScoreModel
    | Mapping[str, Any]
    | Callable[[np.ndarray], Any]
    | None,
    features: np.ndarray,
) -> np.ndarray:
    if model is None:
        return np.zeros(len(features), dtype=np.float64)
    if isinstance(model, Mapping):
        model = LinearCandidateScoreModel.from_artifact(model)
    if hasattr(model, "score"):
        raw = model.score(features)  # type: ignore[union-attr]
    elif callable(model):
        raw = model(features)
    else:
        raise TypeError("model must expose score(), be callable, or be an artifact.")
    result = _finite_array(raw, "candidate model scores")
    if result.shape != (len(features),):
        raise ValueError("Candidate model must return one score per feature row.")
    # Ranker artifacts expose logit-like values whose absolute scale can vary
    # with regularization and class balance.  A bounded monotonic transform
    # preserves ordering without letting a newly trained model overwhelm the
    # independently calibrated cut-count and duration costs.
    return np.tanh(result / 2.0)


def _duration_prior_summary(prior: EmpiricalDurationPrior) -> dict[str, Any]:
    quantiles = [
        _weighted_quantile(
            prior.duration_frames, prior.sample_weights, quantile
        )
        for quantile in (0.25, 0.5, 0.75)
    ]
    margin = max(2, int(math.ceil(prior.bandwidth_frames * 3.0)))
    lower = max(
        1,
        int(math.floor(float(np.min(prior.duration_frames)))) - margin,
    )
    upper = max(
        lower + 1,
        int(math.ceil(float(np.max(prior.duration_frames)))) + margin,
    )
    grid = np.arange(lower, upper + 1, dtype=np.float64)
    scores = prior.log_scores(grid)
    local_modes = [
        index
        for index in range(1, len(grid) - 1)
        if scores[index] >= scores[index - 1]
        and scores[index] >= scores[index + 1]
    ]
    if not local_modes:
        local_modes = [int(np.argmax(scores))]
    strongest = sorted(local_modes, key=lambda index: (-scores[index], index))[:6]
    return {
        "sampleCount": len(prior.duration_frames),
        "bandwidthFrames": prior.bandwidth_frames,
        "quartileFrames": quantiles,
        "modeFrames": sorted(int(grid[index]) for index in strongest),
    }


def _legacy_solve_global_dag_v1_reference_do_not_use(
    signals: CutSignals,
    pool: CandidatePool,
    prior: EmpiricalDurationPrior,
    directives: CutDirectives,
    same_audio: np.ndarray,
    density: str,
    config: CutPlannerConfig,
    model: CandidateScoreModel
    | Mapping[str, Any]
    | Callable[[np.ndarray], Any]
    | None,
) -> CutPlanResult:
    """Frozen comparison reference; production never dispatches to this solver."""
    frames = pool.frames
    candidate_count = len(frames)
    frame_to_index = {int(frame): index for index, frame in enumerate(frames)}
    missing_locks = [
        frame for frame in directives.locked_frames if frame not in frame_to_index
    ]
    if missing_locks:
        raise RuntimeError("Locked cuts were unexpectedly removed from candidates.")

    # Features that do not depend on the previous cut are computed once.  The
    # two edge-context columns are updated in one vectorized batch per DAG node.
    base_features = build_candidate_features(
        signals,
        pool,
        directives,
        previous_cut_frame=np.zeros(candidate_count, dtype=np.int64),
        same_audio_boundaries=same_audio,
    )
    feature_index = {
        name: index for index, name in enumerate(CANDIDATE_FEATURE_NAMES)
    }
    heuristic = (
        0.70 * base_features[:, feature_index["beat_source"]]
        + 0.30 * base_features[:, feature_index["onset_source"]]
        + 0.20 * base_features[:, feature_index["motion_source"]]
        + 0.30 * base_features[:, feature_index["mel_novelty_source"]]
        + 0.25 * base_features[:, feature_index["beat_proximity"]]
        + 0.30 * base_features[:, feature_index["onset_strength"]]
        + 0.20 * base_features[:, feature_index["motion_strength"]]
        + 0.25 * base_features[:, feature_index["mel_novelty"]]
        + config.preferred_bonus
        * base_features[:, feature_index["preferred_proximity"]]
        + config.same_audio_bonus
        * base_features[:, feature_index["same_audio_proximity"]]
        + 0.06 * base_features[:, feature_index["source_count"]]
    )

    duration_grid = np.arange(1, signals.frame_count + 1, dtype=np.float64)
    duration_log_score = prior.log_scores(duration_grid)
    onset = _robust_unit(signals.onset_strength)
    motion = _robust_unit(signals.motion_strength)
    mel = _robust_unit(signals.mel_novelty)
    intensity = 0.40 * onset + 0.28 * motion + 0.32 * mel
    intensity_prefix = np.concatenate(
        ([0.0], np.cumsum(intensity, dtype=np.float64))
    )
    q25 = _weighted_quantile(
        prior.duration_frames, prior.sample_weights, 0.25
    )
    q75 = _weighted_quantile(
        prior.duration_frames, prior.sample_weights, 0.75
    )
    density_scale = {"auto": 1.0, "normal": 1.0, "slow": 1.35, "fast": 0.72}[
        density
    ]

    locked = np.asarray(directives.locked_frames, dtype=np.int64)
    best = np.full((candidate_count, 2), np.inf, dtype=np.float64)
    predecessor_node = np.full((candidate_count, 2), -1, dtype=np.int64)
    predecessor_state = np.full((candidate_count, 2), -1, dtype=np.int8)
    edge_details: list[list[dict[str, Any] | None]] = [
        [None, None] for _ in range(candidate_count)
    ]
    best[0, 0] = 0.0

    for start_index in range(candidate_count - 1):
        if not np.isfinite(best[start_index]).any():
            continue
        start_frame = int(frames[start_index])
        next_lock_values = locked[locked > start_frame]
        next_lock = int(next_lock_values[0]) if len(next_lock_values) else None
        end_limit = int(
            np.searchsorted(
                frames,
                start_frame + config.maximum_shot_frames,
                side="right",
            )
        )
        destination_indices = np.arange(
            start_index + 1, min(candidate_count, end_limit), dtype=np.int64
        )
        if not len(destination_indices):
            continue
        if next_lock is not None:
            destination_indices = destination_indices[
                frames[destination_indices] <= next_lock
            ]
        durations = frames[destination_indices] - start_frame
        valid_minimum = durations >= config.minimum_shot_frames
        valid_minimum |= frames[destination_indices] == signals.frame_count
        destination_indices = destination_indices[valid_minimum]
        durations = durations[valid_minimum]
        if not len(destination_indices):
            continue

        edge_features = base_features[destination_indices].copy()
        edge_features[:, feature_index["duration_since_previous_cut_seconds"]] = (
            durations / signals.sample_rate
        )
        edge_features[:, feature_index["previous_cut_progress"]] = (
            start_frame / signals.frame_count
        )
        learned_scores = _model_scores(model, edge_features)

        for offset, end_index_raw in enumerate(destination_indices):
            end_index = int(end_index_raw)
            end_frame = int(frames[end_index])
            duration = int(durations[offset])
            is_endpoint = end_frame == signals.frame_count
            is_short = duration <= config.ultrashort_frames
            segment_intensity = float(
                (intensity_prefix[end_frame] - intensity_prefix[start_frame])
                / duration
            )
            target_duration = (
                (q75 * (1.0 - segment_intensity) + q25 * segment_intensity)
                * density_scale
            )
            duration_cost = -config.duration_prior_weight * float(
                duration_log_score[duration - 1]
            )
            density_cost = config.density_weight * (
                (duration - target_duration) / max(target_duration, 1.0)
            ) ** 2
            tail_cost = 0.0
            if is_endpoint and duration < config.short_tail_frames:
                tail_cost = config.short_tail_cost * (
                    1.0 - duration / config.short_tail_frames
                ) ** 2
            candidate_reward = 0.0 if is_endpoint else float(
                heuristic[end_index]
                + config.candidate_score_weight * learned_scores[offset]
            )
            cut_cost = 0.0 if is_endpoint else config.cut_count_cost

            for previous_short_state in (0, 1):
                if not math.isfinite(best[start_index, previous_short_state]):
                    continue
                repeated_short_cost = (
                    config.consecutive_ultrashort_cost
                    if previous_short_state and is_short
                    else 0.0
                )
                total_edge_cost = (
                    duration_cost
                    + density_cost
                    + tail_cost
                    + cut_cost
                    + repeated_short_cost
                    - candidate_reward
                )
                next_state = int(is_short)
                proposed = best[start_index, previous_short_state] + total_edge_cost
                if proposed < best[end_index, next_state] - 1e-12:
                    best[end_index, next_state] = proposed
                    predecessor_node[end_index, next_state] = start_index
                    predecessor_state[end_index, next_state] = previous_short_state
                    edge_details[end_index][next_state] = {
                        "startFrame": start_frame,
                        "endFrame": end_frame,
                        "durationFrames": duration,
                        "candidateReward": candidate_reward,
                        "heuristicCandidateScore": (
                            0.0 if is_endpoint else float(heuristic[end_index])
                        ),
                        "learnedCandidateScore": (
                            0.0 if is_endpoint else float(learned_scores[offset])
                        ),
                        "durationPriorCost": duration_cost,
                        "densityCost": density_cost,
                        "cutCountCost": cut_cost,
                        "shortTailCost": tail_cost,
                        "consecutiveUltrashortCost": repeated_short_cost,
                        "totalEdgeCost": total_edge_cost,
                        "localIntensity": segment_intensity,
                        "targetDurationFrames": target_duration,
                    }

    end_state = min((0, 1), key=lambda state: (best[-1, state], state))
    if not math.isfinite(best[-1, end_state]):
        raise ValueError(
            "No valid cut plan satisfies locked cuts, no-cut ranges, and "
            "duration safety limits."
        )

    path_indices = [candidate_count - 1]
    selected_edges: list[dict[str, Any]] = []
    node, state = candidate_count - 1, end_state
    while node != 0:
        detail = edge_details[node][state]
        previous_node = int(predecessor_node[node, state])
        previous_state = int(predecessor_state[node, state])
        if detail is None or previous_node < 0 or previous_state < 0:
            raise RuntimeError("Global cut planner predecessor chain is corrupt.")
        selected_edges.append(detail)
        path_indices.append(previous_node)
        node, state = previous_node, previous_state
    path_indices.reverse()
    selected_edges.reverse()
    boundaries = tuple(int(frames[index]) for index in path_indices)

    if boundaries[0] != 0 or boundaries[-1] != signals.frame_count:
        raise RuntimeError("Global cut planner omitted a timeline endpoint.")
    if not set(directives.locked_frames).issubset(boundaries):
        raise RuntimeError("Global cut planner omitted a locked boundary.")
    if any(
        start <= boundary < end
        for boundary in boundaries[1:-1]
        for start, end in directives.no_cut_ranges
    ):
        raise RuntimeError("Global cut planner emitted a cut in a no-cut range.")

    by_source = {
        source: sum(source in values for values in pool.sources)
        for source in sorted(_SOURCE_NAMES)
    }
    selected_cuts = []
    for path_index in path_indices[1:-1]:
        feature_values = base_features[path_index]
        selected_cuts.append(
            {
                "frame": int(frames[path_index]),
                "sources": list(pool.sources[path_index]),
                "baseFeatures": {
                    name: float(feature_values[index])
                    for index, name in enumerate(CANDIDATE_FEATURE_NAMES)
                    if name
                    not in {
                        "duration_since_previous_cut_seconds",
                        "previous_cut_progress",
                    }
                },
            }
        )
    objective_totals = {
        key: float(sum(float(edge[key]) for edge in selected_edges))
        for key in (
            "candidateReward",
            "durationPriorCost",
            "densityCost",
            "cutCountCost",
            "shortTailCost",
            "consecutiveUltrashortCost",
            "totalEdgeCost",
        )
    }
    boundary_bytes = json.dumps(
        boundaries, separators=(",", ":")
    ).encode("utf-8")
    trace = {
        "candidateUnion": {
            "total": candidate_count,
            "bySource": by_source,
            "fallbackIntervalFrames": config.fallback_interval_frames,
        },
        "directives": {
            "lockedFrames": list(directives.locked_frames),
            "preferredFrames": list(directives.preferred_frames),
            "noCutRanges": [list(item) for item in directives.no_cut_ranges],
            "preferredToleranceFrames": directives.preferred_tolerance_frames,
        },
        "selectedCuts": selected_cuts,
        "selectedEdges": selected_edges,
        "objectiveTotals": objective_totals,
    }
    metadata = {
        "format": CUT_PLANNER_FORMAT,
        "schemaVersion": CUT_PLANNER_SCHEMA_VERSION,
        "algorithm": "deterministic-global-dag-dp-v1",
        "sampleRate": signals.sample_rate,
        "frameCount": signals.frame_count,
        "density": density,
        "candidateFeatureNames": list(CANDIDATE_FEATURE_NAMES),
        "learnedCandidateModel": model is not None,
        "learnedCandidateScoreTransform": (
            "tanh(logit/2)" if model is not None else None
        ),
        "sameAudioBoundaryCount": len(same_audio),
        "durationPrior": _duration_prior_summary(prior),
        "plannerConfig": {
            name: getattr(config, name)
            for name in config.__dataclass_fields__
        },
        "cutCount": len(boundaries) - 2,
        "shotCount": len(boundaries) - 1,
        "boundarySha256": hashlib.sha256(boundary_bytes).hexdigest(),
    }
    return CutPlanResult(boundaries, trace, metadata)


@dataclass(frozen=True)
class _HistoryBeamState:
    cost: float
    node_index: int
    previous_short: bool
    history: tuple[HistoryShot, ...]
    predecessor: _HistoryBeamState | None
    edge_detail: dict[str, Any] | None
    path_key: tuple[int, ...]


def _beam_state_key(state: _HistoryBeamState) -> tuple[Any, ...]:
    """Total ordering used for every beam prune and final selection."""

    return (state.cost, state.path_key, int(state.previous_short))


def _write_history_features(
    matrix: np.ndarray,
    feature_index: Mapping[str, int],
    history: Sequence[HistoryShot],
) -> None:
    matrix[:, feature_index["history_count"]] = float(len(history))
    for lag in range(1, HISTORY_CONTEXT_DEPTH + 1):
        prefix = f"previous_{lag}_"
        if lag > len(history):
            matrix[:, feature_index[prefix + "duration_seconds"]] = 0.0
            matrix[:, feature_index[prefix + "music_energy"]] = 0.0
            matrix[:, feature_index[prefix + "motion_energy"]] = 0.0
            matrix[:, feature_index[prefix + "cut_beat_proximity"]] = 0.0
            continue
        shot = history[-lag]
        matrix[:, feature_index[prefix + "duration_seconds"]] = (
            shot.duration_frames / REQUIRED_SAMPLE_RATE
        )
        matrix[:, feature_index[prefix + "music_energy"]] = shot.music_energy
        matrix[:, feature_index[prefix + "motion_energy"]] = shot.motion_energy
        matrix[:, feature_index[prefix + "cut_beat_proximity"]] = (
            shot.cut_beat_proximity
        )


def _solve_global_dag(
    signals: CutSignals,
    pool: CandidatePool,
    prior: EmpiricalDurationPrior,
    directives: CutDirectives,
    same_audio: np.ndarray,
    density: str,
    config: CutPlannerConfig,
    model: CandidateScoreModel
    | Mapping[str, Any]
    | Callable[[np.ndarray], Any]
    | None,
) -> CutPlanResult:
    """Plan with a deterministic bounded beam over causal three-shot history."""

    frames = pool.frames
    candidate_count = len(frames)
    frame_to_index = {int(frame): index for index, frame in enumerate(frames)}
    missing_locks = [
        frame for frame in directives.locked_frames if frame not in frame_to_index
    ]
    if missing_locks:
        raise RuntimeError("Locked cuts were unexpectedly removed from candidates.")

    base_features = build_candidate_features(
        signals,
        pool,
        directives,
        previous_cut_frame=np.zeros(candidate_count, dtype=np.int64),
        same_audio_boundaries=same_audio,
    )
    feature_index = {
        name: index for index, name in enumerate(CANDIDATE_FEATURE_NAMES)
    }
    heuristic = (
        0.70 * base_features[:, feature_index["beat_source"]]
        + 0.30 * base_features[:, feature_index["onset_source"]]
        + 0.20 * base_features[:, feature_index["motion_source"]]
        + 0.30 * base_features[:, feature_index["mel_novelty_source"]]
        + 0.25 * base_features[:, feature_index["beat_proximity"]]
        + 0.30 * base_features[:, feature_index["onset_strength"]]
        + 0.20 * base_features[:, feature_index["motion_strength"]]
        + 0.25 * base_features[:, feature_index["mel_novelty"]]
        + config.preferred_bonus
        * base_features[:, feature_index["preferred_proximity"]]
        + config.same_audio_bonus
        * base_features[:, feature_index["same_audio_proximity"]]
        + 0.06 * base_features[:, feature_index["source_count"]]
    )

    duration_grid = np.arange(1, signals.frame_count + 1, dtype=np.float64)
    duration_log_score = prior.log_scores(duration_grid)
    onset = _robust_unit(signals.onset_strength)
    motion = _robust_unit(signals.motion_strength)
    mel = _robust_unit(signals.mel_novelty)
    music = 0.5 * onset + 0.5 * mel
    intensity = 0.40 * onset + 0.28 * motion + 0.32 * mel
    intensity_prefix = np.concatenate(([0.0], np.cumsum(intensity)))
    music_prefix = np.concatenate(([0.0], np.cumsum(music)))
    motion_prefix = np.concatenate(([0.0], np.cumsum(motion)))
    beat_proximity = _proximity(frames, signals.beat_frames, 30)
    inactive_tail_start = _inactive_tail_start(signals, config)
    protected_tail_frames = frozenset(
        int(frame)
        for frame in (
            *directives.locked_frames,
            *directives.preferred_frames,
            *same_audio.tolist(),
        )
    )
    q25 = _weighted_quantile(prior.duration_frames, prior.sample_weights, 0.25)
    q75 = _weighted_quantile(prior.duration_frames, prior.sample_weights, 0.75)
    density_scale = {
        "auto": 1.0,
        "normal": 1.0,
        "slow": 1.35,
        "fast": 0.72,
    }[density]

    locked = np.asarray(directives.locked_frames, dtype=np.int64)
    states_by_node: list[list[_HistoryBeamState]] = [
        [] for _ in range(candidate_count)
    ]
    states_by_node[0].append(
        _HistoryBeamState(0.0, 0, False, (), None, None, (0,))
    )
    states_expanded = 0
    states_pruned = 0

    def retain(end_index: int, state: _HistoryBeamState) -> None:
        nonlocal states_pruned
        bucket = states_by_node[end_index]
        bucket.append(state)
        bucket.sort(key=_beam_state_key)
        if len(bucket) > config.history_beam_width:
            del bucket[config.history_beam_width :]
            states_pruned += 1

    for start_index in range(candidate_count - 1):
        start_states = states_by_node[start_index]
        if not start_states:
            continue
        start_frame = int(frames[start_index])
        next_lock_values = locked[locked > start_frame]
        next_lock = int(next_lock_values[0]) if len(next_lock_values) else None
        end_limit = int(
            np.searchsorted(
                frames,
                start_frame + config.maximum_shot_frames,
                side="right",
            )
        )
        destination_indices = np.arange(
            start_index + 1, min(candidate_count, end_limit), dtype=np.int64
        )
        endpoint_index = candidate_count - 1
        endpoint_beyond_normal_limit = endpoint_index not in destination_indices
        active_prefix_frames = (
            max(0, inactive_tail_start - start_frame)
            if inactive_tail_start is not None
            else signals.frame_count
        )
        crosses_protected_tail_boundary = any(
            start_frame < frame < signals.frame_count
            for frame in protected_tail_frames
        )
        inactive_tail_extension_allowed = bool(
            inactive_tail_start is not None
            and endpoint_beyond_normal_limit
            and (
                start_frame <= inactive_tail_start
                or start_frame in protected_tail_frames
            )
            and active_prefix_frames <= config.maximum_shot_frames
            and not crosses_protected_tail_boundary
        )
        if inactive_tail_extension_allowed:
            destination_indices = np.append(destination_indices, endpoint_index)
        if next_lock is not None:
            destination_indices = destination_indices[
                frames[destination_indices] <= next_lock
            ]
        if not len(destination_indices):
            continue
        durations = frames[destination_indices] - start_frame
        valid_minimum = durations >= config.minimum_shot_frames
        valid_minimum |= frames[destination_indices] == signals.frame_count
        destination_indices = destination_indices[valid_minimum]
        durations = durations[valid_minimum]
        if not len(destination_indices):
            continue

        end_frames = frames[destination_indices]
        inactive_tail_extension_mask = (
            (end_frames == signals.frame_count)
            & (end_frames - start_frame > config.maximum_shot_frames)
            & inactive_tail_extension_allowed
        )
        scoring_durations = durations.copy()
        if np.any(inactive_tail_extension_mask):
            scoring_durations[inactive_tail_extension_mask] = max(
                config.minimum_shot_frames,
                active_prefix_frames,
            )
        scoring_end_frames = end_frames.copy()
        if np.any(inactive_tail_extension_mask):
            scoring_end_frames[inactive_tail_extension_mask] = max(
                start_frame + config.minimum_shot_frames,
                int(inactive_tail_start),
            )
        segment_intensities = (
            intensity_prefix[scoring_end_frames] - intensity_prefix[start_frame]
        ) / scoring_durations
        segment_music = (
            music_prefix[scoring_end_frames] - music_prefix[start_frame]
        ) / scoring_durations
        segment_motion = (
            motion_prefix[scoring_end_frames] - motion_prefix[start_frame]
        ) / scoring_durations
        target_durations = (
            q75 * (1.0 - segment_intensities) + q25 * segment_intensities
        ) * density_scale
        duration_costs = -config.duration_prior_weight * duration_log_score[
            scoring_durations - 1
        ]
        density_costs = config.density_weight * np.square(
            (scoring_durations - target_durations)
            / np.maximum(target_durations, 1.0)
        )

        for state in tuple(sorted(start_states, key=_beam_state_key)):
            edge_features = base_features[destination_indices].copy()
            edge_features[
                :, feature_index["duration_since_previous_cut_seconds"]
            ] = durations / signals.sample_rate
            edge_features[:, feature_index["previous_cut_progress"]] = (
                start_frame / signals.frame_count
            )
            _write_history_features(edge_features, feature_index, state.history)
            learned_scores = _model_scores(model, edge_features)

            endpoint_mask = end_frames == signals.frame_count
            # The ranker is trained listwise, so only within-choice-set order
            # is meaningful.  Centering each causal choice set prevents a
            # positive logit baseline from becoming an accidental per-cut
            # reward (and therefore changing shot density).
            rankable_scores = learned_scores[~endpoint_mask]
            if model is not None and len(rankable_scores):
                learned_scores = np.clip(
                    learned_scores - float(np.median(rankable_scores)),
                    -1.0,
                    1.0,
                )
            short_mask = durations <= config.ultrashort_frames
            tail_costs = np.zeros(len(destination_indices), dtype=np.float64)
            tail_indices = endpoint_mask & (durations < config.short_tail_frames)
            tail_costs[tail_indices] = config.short_tail_cost * np.square(
                1.0 - durations[tail_indices] / config.short_tail_frames
            )
            candidate_rewards = (
                heuristic[destination_indices]
                + config.candidate_score_weight * learned_scores
            )
            candidate_rewards[endpoint_mask] = 0.0
            cut_costs = np.full(
                len(destination_indices), config.cut_count_cost, dtype=np.float64
            )
            cut_costs[endpoint_mask] = 0.0
            repeated_short_costs = np.zeros(
                len(destination_indices), dtype=np.float64
            )
            if state.previous_short:
                repeated_short_costs[short_mask] = (
                    config.consecutive_ultrashort_cost
                )
            total_edge_costs = (
                duration_costs
                + density_costs
                + tail_costs
                + cut_costs
                + repeated_short_costs
                - candidate_rewards
            )
            proposed_costs = state.cost + total_edge_costs
            worst_retained = np.asarray(
                [
                    (
                        states_by_node[int(index)][-1].cost
                        if len(states_by_node[int(index)])
                        >= config.history_beam_width
                        else math.inf
                    )
                    for index in destination_indices
                ],
                dtype=np.float64,
            )
            viable_offsets = np.flatnonzero(
                proposed_costs <= worst_retained + 1e-12
            )
            states_expanded += len(destination_indices)
            states_pruned += len(destination_indices) - len(viable_offsets)

            for offset_raw in viable_offsets:
                offset = int(offset_raw)
                end_index_raw = destination_indices[offset]
                end_index = int(end_index_raw)
                end_frame = int(end_frames[offset])
                duration = int(durations[offset])
                is_endpoint = bool(endpoint_mask[offset])
                is_short = bool(short_mask[offset])
                tail_cost = float(tail_costs[offset])
                candidate_reward = float(candidate_rewards[offset])
                cut_cost = float(cut_costs[offset])
                repeated_short_cost = float(repeated_short_costs[offset])
                total_edge_cost = float(total_edge_costs[offset])
                history_context = {
                    "historyCount": len(state.history),
                    "previousShots": [
                        shot.to_trace() for shot in reversed(state.history)
                    ],
                }
                detail = {
                    "startFrame": start_frame,
                    "endFrame": end_frame,
                    "durationFrames": duration,
                    "durationScoringFrames": int(scoring_durations[offset]),
                    "inactiveTailContinuation": bool(
                        inactive_tail_extension_mask[offset]
                    ),
                    "inactiveTailStartFrame": (
                        int(inactive_tail_start)
                        if inactive_tail_extension_mask[offset]
                        else None
                    ),
                    "candidateReward": candidate_reward,
                    "heuristicCandidateScore": (
                        0.0 if is_endpoint else float(heuristic[end_index])
                    ),
                    "learnedCandidateScore": (
                        0.0 if is_endpoint else float(learned_scores[offset])
                    ),
                    "durationPriorCost": float(duration_costs[offset]),
                    "densityCost": float(density_costs[offset]),
                    "cutCountCost": cut_cost,
                    "shortTailCost": tail_cost,
                    "consecutiveUltrashortCost": repeated_short_cost,
                    "totalEdgeCost": total_edge_cost,
                    "localIntensity": float(segment_intensities[offset]),
                    "targetDurationFrames": float(target_durations[offset]),
                    "historyContext": history_context,
                }
                completed = HistoryShot(
                    start_frame=start_frame,
                    end_frame=end_frame,
                    duration_frames=duration,
                    music_energy=float(segment_music[offset]),
                    motion_energy=float(segment_motion[offset]),
                    cut_beat_proximity=float(beat_proximity[end_index]),
                )
                next_history = (state.history + (completed,))[
                    -HISTORY_CONTEXT_DEPTH:
                ]
                retain(
                    end_index,
                    _HistoryBeamState(
                        float(proposed_costs[offset]),
                        end_index,
                        is_short,
                        next_history,
                        state,
                        detail,
                        state.path_key + (end_frame,),
                    ),
                )

    if not states_by_node[-1]:
        raise ValueError(
            "No valid cut plan satisfies locked cuts, no-cut ranges, and "
            "duration safety limits."
        )
    end_state = min(states_by_node[-1], key=_beam_state_key)
    selected_edges: list[dict[str, Any]] = []
    state: _HistoryBeamState | None = end_state
    while state is not None and state.predecessor is not None:
        if state.edge_detail is None:
            raise RuntimeError("History-beam predecessor chain is corrupt.")
        selected_edges.append(state.edge_detail)
        state = state.predecessor
    selected_edges.reverse()
    boundaries = tuple(edge["startFrame"] for edge in selected_edges) + (
        signals.frame_count,
    )
    path_indices = [frame_to_index[int(frame)] for frame in boundaries]

    if boundaries[0] != 0 or boundaries[-1] != signals.frame_count:
        raise RuntimeError("History-beam cut planner omitted a timeline endpoint.")
    if not set(directives.locked_frames).issubset(boundaries):
        raise RuntimeError("History-beam cut planner omitted a locked boundary.")
    if any(
        start <= boundary < end
        for boundary in boundaries[1:-1]
        for start, end in directives.no_cut_ranges
    ):
        raise RuntimeError("History-beam cut planner emitted a cut in a no-cut range.")

    by_source = {
        source: sum(source in values for values in pool.sources)
        for source in sorted(_SOURCE_NAMES)
    }
    edge_context_names = {
        "duration_since_previous_cut_seconds",
        "previous_cut_progress",
        "history_count",
        *(
            f"previous_{lag}_{suffix}"
            for lag in range(1, HISTORY_CONTEXT_DEPTH + 1)
            for suffix in (
                "duration_seconds",
                "music_energy",
                "motion_energy",
                "cut_beat_proximity",
            )
        ),
    }
    selected_cuts = []
    for path_index in path_indices[1:-1]:
        feature_values = base_features[path_index]
        selected_cuts.append(
            {
                "frame": int(frames[path_index]),
                "sources": list(pool.sources[path_index]),
                "baseFeatures": {
                    name: float(feature_values[index])
                    for index, name in enumerate(CANDIDATE_FEATURE_NAMES)
                    if name not in edge_context_names
                },
            }
        )
    objective_totals = {
        key: float(sum(float(edge[key]) for edge in selected_edges))
        for key in (
            "candidateReward",
            "durationPriorCost",
            "densityCost",
            "cutCountCost",
            "shortTailCost",
            "consecutiveUltrashortCost",
            "totalEdgeCost",
        )
    }
    boundary_bytes = json.dumps(boundaries, separators=(",", ":")).encode("utf-8")
    trace = {
        "candidateUnion": {
            "total": candidate_count,
            "bySource": by_source,
            "fallbackIntervalFrames": config.fallback_interval_frames,
        },
        "directives": {
            "lockedFrames": list(directives.locked_frames),
            "preferredFrames": list(directives.preferred_frames),
            "noCutRanges": [list(item) for item in directives.no_cut_ranges],
            "preferredToleranceFrames": directives.preferred_tolerance_frames,
        },
        "historyBeam": {
            "width": config.history_beam_width,
            "contextDepth": HISTORY_CONTEXT_DEPTH,
            "statesExpanded": states_expanded,
            "statesPruned": states_pruned,
        },
        "inactiveTail": {
            "detected": inactive_tail_start is not None,
            "startFrame": inactive_tail_start,
            "minimumFrames": config.inactive_tail_minimum_frames,
            "signalThreshold": config.inactive_tail_signal_threshold,
            "continuationSelected": any(
                bool(edge.get("inactiveTailContinuation", False))
                for edge in selected_edges
            ),
        },
        "selectedCuts": selected_cuts,
        "selectedEdges": selected_edges,
        "objectiveTotals": objective_totals,
    }
    metadata = {
        "format": CUT_PLANNER_FORMAT,
        "schemaVersion": CUT_PLANNER_SCHEMA_VERSION,
        "algorithm": "deterministic-history-beam-dp-v2",
        "sampleRate": signals.sample_rate,
        "frameCount": signals.frame_count,
        "density": density,
        "candidateFeatureNames": list(CANDIDATE_FEATURE_NAMES),
        "learnedCandidateModel": model is not None,
        "learnedCandidateScoreTransform": (
            "clip(tanh(logit/2)-choice_set_median,-1,1)"
            if model is not None
            else None
        ),
        "sameAudioBoundaryCount": len(same_audio),
        "inactiveTailStartFrame": inactive_tail_start,
        "inactiveTailContinuationSelected": any(
            bool(edge.get("inactiveTailContinuation", False))
            for edge in selected_edges
        ),
        "durationPrior": _duration_prior_summary(prior),
        "plannerConfig": {
            name: getattr(config, name) for name in config.__dataclass_fields__
        },
        "cutCount": len(boundaries) - 2,
        "shotCount": len(boundaries) - 1,
        "boundarySha256": hashlib.sha256(boundary_bytes).hexdigest(),
    }
    return CutPlanResult(boundaries, trace, metadata)
