#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import copy
import hashlib
import json
import math
import os
import re
import shutil
import subprocess
import sys
import time
import zipfile
from collections.abc import Iterator
from pathlib import Path
from typing import Any, Callable

import cv2
import numpy as np

try:
    import resource
except ImportError:  # pragma: no cover - Windows runtime
    resource = None

SCRIPT_DIR = Path(__file__).resolve().parent
sys.dont_write_bytecode = True
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from contract import load_schema, resolve_media_path, validate_contract
from matching_math import (
    clamp_candidate_start,
    frame_aligned_candidate_starts,
    placement_safe_range_seconds,
    source_start_bounds,
)

# Bump DESCRIPTOR_VERSION whenever descriptor math, variant generation, or
# coarse decode geometry changes. Cache shape checks cannot detect semantic drift.
DESCRIPTOR_VERSION = "classical-edge-gray-flip-v1"
ANALYSIS_VERSION = f"{DESCRIPTOR_VERSION}+analysis-diagnostics-v6"
INDEX_CACHE_FORMAT_VERSION = 2
LEGACY_INDEX_CACHE_FORMAT_VERSION = 1
COARSE_INDEX_MAX_DIM = 360
DEFAULT_INDEX_CHUNK_SECONDS = 300.0
DEFAULT_RETRIEVAL_SHOT_BATCH_SIZE = 64
MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS = 5
REVIEW_MIN_NORMALIZED_SCORE = 0.75
REVIEW_MIN_PIXEL_IDENTITY_SCORE = 0.96
AUTO_MIN_NORMALIZED_SEPARATION = 0.08
AUTO_MIN_PIXEL_IDENTITY_SEPARATION = 0.03
DecodeSeriesRequest = tuple[
    Path,
    float,
    float,
    float,
    Path,
    str,
    dict[str, object],
    str,
]
PreparedDecodeSeriesRequest = tuple[
    Path,
    float,
    float,
    float,
    Path,
    str,
    dict[str, object],
    str,
    str,
]
MergedDecodeSeriesGroup = tuple[list[int], DecodeSeriesRequest]

ANALYSIS_DIAGNOSTIC_TIMING_KEYS = (
    "referenceExportFfmpegSeconds",
    "refinementWindowFfmpegSeconds",
    "fixedFrameExportFfmpegSeconds",
    "alignmentExportFfmpegSeconds",
    "exactAnchorExportFfmpegSeconds",
    "referenceImageReadSeconds",
    "sourceImageReadSeconds",
    "referenceDescriptorSeconds",
    "windowDescriptorSeconds",
    "alignmentDescriptorSeconds",
    "exactDescriptorSeconds",
    "referencePixelIdentitySeconds",
    "sourcePixelIdentitySeconds",
    "referenceFeatureSeconds",
    "sourceFeatureSeconds",
    "featureComparisonSeconds",
    "referenceMotionSeconds",
    "sourceMotionSeconds",
    "motionComparisonSeconds",
    "featureDetectorSetupSeconds",
)

ANALYSIS_DIAGNOSTIC_COUNT_KEYS = (
    "ffmpegInvocations",
    "referenceFrameExports",
    "refinementWindowDecodes",
    "refinementWindowFrames",
    "fixedFrameExports",
    "alignmentFrameExports",
    "exactAnchorFrameExports",
    "decodedFrames",
    "referenceMediaInputs",
    "originalMediaInputs",
    "proxyMediaInputs",
    "frameCacheHits",
    "frameCacheMisses",
    "descriptorVariantCalls",
    "descriptorVariantsGenerated",
    "pixelIdentityCalls",
    "featureCalls",
    "referenceFeatureCacheHits",
    "referenceFeatureCacheMisses",
    "motionComparisons",
    "exactExportTimestampWarnings",
)

MAX_EXACT_EXPORT_EVIDENCE = 64
FIRST_EXACT_EXPORT_EVIDENCE = 32

EXPORT_ROLE_METRICS = {
    "REFERENCE_ANCHOR": ("referenceExportFfmpegSeconds", "referenceFrameExports"),
    "REFINEMENT_FIXED": ("fixedFrameExportFfmpegSeconds", "fixedFrameExports"),
    "ALIGNMENT": ("alignmentExportFfmpegSeconds", "alignmentFrameExports"),
    "EXACT_ANCHOR": ("exactAnchorExportFfmpegSeconds", "exactAnchorFrameExports"),
}


def new_analysis_diagnostics() -> dict[str, Any]:
    return {
        "timing": {key: 0.0 for key in ANALYSIS_DIAGNOSTIC_TIMING_KEYS},
        "counts": {key: 0 for key in ANALYSIS_DIAGNOSTIC_COUNT_KEYS},
        "exactExports": [],
        "exactExportRetention": {
            "totalRecords": 0,
            "retainedRecords": 0,
            "truncatedRecords": 0,
        },
        "shots": [],
    }


def retained_exact_export_evidence(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if len(records) <= MAX_EXACT_EXPORT_EVIDENCE:
        return records
    first = records[:FIRST_EXACT_EXPORT_EVIDENCE]
    indexed_tail = list(enumerate(records[FIRST_EXACT_EXPORT_EVIDENCE:], start=FIRST_EXACT_EXPORT_EVIDENCE))
    largest_deltas = sorted(
        indexed_tail,
        key=lambda item: (
            -abs(float(item[1].get("requestedToDecodedDeltaSeconds", 0.0))),
            item[0],
        ),
    )[: MAX_EXACT_EXPORT_EVIDENCE - FIRST_EXACT_EXPORT_EVIDENCE]
    return first + [record for _index, record in sorted(largest_deltas, key=lambda item: item[0])]


def record_exact_export(
    diagnostics: dict[str, Any] | None,
    record: dict[str, Any],
) -> None:
    if diagnostics is None:
        return
    retention = diagnostics["exactExportRetention"]
    retention["totalRecords"] += 1
    diagnostics["exactExports"] = retained_exact_export_evidence(
        [*diagnostics["exactExports"], record]
    )
    retention["retainedRecords"] = len(diagnostics["exactExports"])
    retention["truncatedRecords"] = (
        int(retention["totalRecords"]) - int(retention["retainedRecords"])
    )


def merge_analysis_diagnostics(
    target: dict[str, Any],
    source: dict[str, Any],
) -> dict[str, Any]:
    for key in ANALYSIS_DIAGNOSTIC_TIMING_KEYS:
        target["timing"][key] += float(source["timing"][key])
    for key in ANALYSIS_DIAGNOSTIC_COUNT_KEYS:
        target["counts"][key] += int(source["counts"][key])
    combined_exports = [
        *target["exactExports"],
        *copy.deepcopy(source["exactExports"]),
    ]
    target["exactExports"] = retained_exact_export_evidence(combined_exports)
    target_retention = target["exactExportRetention"]
    source_retention = source["exactExportRetention"]
    target_retention["totalRecords"] += int(source_retention["totalRecords"])
    target_retention["retainedRecords"] = len(target["exactExports"])
    target_retention["truncatedRecords"] = (
        int(target_retention["totalRecords"]) - int(target_retention["retainedRecords"])
    )
    target["shots"].extend(copy.deepcopy(source.get("shots") or []))
    return target


def append_shot_analysis_diagnostics(
    target: dict[str, Any],
    shot_id: str,
    source: dict[str, Any],
) -> None:
    merge_analysis_diagnostics(target, source)
    max_pts_delta = max(
        (
            abs(float(record.get("requestedToDecodedDeltaSeconds", 0.0)))
            for record in source["exactExports"]
        ),
        default=0.0,
    )
    target["shots"].append(
        {
            "shotId": shot_id,
            "timing": copy.deepcopy(source["timing"]),
            "counts": copy.deepcopy(source["counts"]),
            "exactExportRecords": int(source["exactExportRetention"]["totalRecords"]),
            "maximumAbsRequestedToDecodedDeltaSeconds": max_pts_delta,
        }
    )


def ordered_analysis_diagnostics(
    shots: list[dict[str, Any]],
    diagnostics_by_shot_id: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    expected_ids = [str(shot["shotId"]) for shot in shots]
    if len(set(expected_ids)) != len(expected_ids):
        raise AutoConformFailure(
            "analysis-diagnostics",
            "DUPLICATE_SHOT_ID",
            "Analysis diagnostics require unique shot IDs.",
        )
    missing = [shot_id for shot_id in expected_ids if shot_id not in diagnostics_by_shot_id]
    unexpected = sorted(set(diagnostics_by_shot_id) - set(expected_ids))
    if missing or unexpected:
        raise AutoConformFailure(
            "analysis-diagnostics",
            "INCOMPLETE_SHOT_DIAGNOSTICS",
            "Analysis diagnostics do not cover exactly the selected shot range.",
            {
                "missingShotIds": missing,
                "unexpectedShotIds": unexpected,
            },
        )
    result = new_analysis_diagnostics()
    for shot_id in expected_ids:
        append_shot_analysis_diagnostics(
            result,
            shot_id,
            diagnostics_by_shot_id[shot_id],
        )
    return result


def add_diagnostic_time(
    diagnostics: dict[str, Any] | None,
    key: str,
    seconds: float,
) -> None:
    if diagnostics is not None:
        diagnostics["timing"][key] += max(0.0, float(seconds))


def add_diagnostic_count(
    diagnostics: dict[str, Any] | None,
    key: str,
    value: int = 1,
) -> None:
    if diagnostics is not None:
        diagnostics["counts"][key] += max(0, int(value))


def measured_call(
    diagnostics: dict[str, Any] | None,
    timing_key: str,
    operation: Callable[[], Any],
) -> Any:
    started = time.perf_counter()
    try:
        return operation()
    finally:
        add_diagnostic_time(diagnostics, timing_key, time.perf_counter() - started)


def measured_descriptor_variant_score(
    diagnostics: dict[str, Any] | None,
    timing_key: str,
    reference: np.ndarray,
    image: np.ndarray,
) -> float:
    started = time.perf_counter()
    try:
        variants = descriptor_variants(image)
        add_diagnostic_count(diagnostics, "descriptorVariantCalls")
        add_diagnostic_count(
            diagnostics,
            "descriptorVariantsGenerated",
            int(variants.shape[0]),
        )
        return descriptor_score(reference, variants)
    finally:
        add_diagnostic_time(diagnostics, timing_key, time.perf_counter() - started)


def measured_descriptor_selected_variant_score(
    diagnostics: dict[str, Any] | None,
    timing_key: str,
    reference: np.ndarray,
    image: np.ndarray,
    variant_index: int,
) -> float:
    started = time.perf_counter()
    try:
        variant = descriptor_variant(image, variant_index)
        selected_variants = np.stack([variant])
        add_diagnostic_count(diagnostics, "descriptorVariantCalls")
        add_diagnostic_count(
            diagnostics,
            "descriptorVariantsGenerated",
            int(selected_variants.shape[0]),
        )
        return descriptor_score(reference, selected_variants)
    finally:
        add_diagnostic_time(diagnostics, timing_key, time.perf_counter() - started)


def progress(stage: str, **values: object) -> None:
    fields = ["PROGRESS", f"stage={stage}"]
    fields.extend(f"{key}={value}" for key, value in values.items())
    print("|".join(fields), flush=True)


class AutoConformFailure(RuntimeError):
    def __init__(
        self,
        stage: str,
        code: str,
        message: str,
        details: dict[str, object] | None = None,
    ) -> None:
        super().__init__(message)
        self.stage = stage
        self.code = code
        self.details = details or {}

    def record(self) -> dict[str, object]:
        return {
            "stage": self.stage,
            "code": self.code,
            "message": str(self),
            "details": self.details,
        }


def run(
    command: list[str],
    timeout_seconds: float,
    *,
    stage: str = "external-process",
    details: dict[str, object] | None = None,
) -> subprocess.CompletedProcess[bytes]:
    try:
        result = subprocess.run(
            command,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=timeout_seconds,
        )
    except subprocess.TimeoutExpired as exc:
        raise AutoConformFailure(
            stage,
            "EXTERNAL_PROCESS_TIMEOUT",
            f"External media processing exceeded {timeout_seconds:g} seconds.",
            {
                **(details or {}),
                "timeoutSeconds": timeout_seconds,
                "program": command[0] if command else "",
            },
        ) from exc
    if result.returncode:
        stderr = result.stderr.decode("utf-8", errors="replace")
        raise AutoConformFailure(
            stage,
            "EXTERNAL_PROCESS_FAILED",
            f"External media processing failed with exit code {result.returncode}: {stderr[-2000:]}",
            {
                **(details or {}),
                "exitCode": result.returncode,
                "program": command[0] if command else "",
            },
        )
    return result


SHOWINFO_PATTERN = re.compile(
    r"\bn:\s*(?P<index>\d+)\s+pts:\s*[^\s]+\s+pts_time:(?P<seconds>[-+0-9.eE]+)"
)
NAMED_SHOWINFO_PATTERN = re.compile(
    r"\[(?:Parsed_)?showinfo@(?P<label>[A-Za-z0-9_-]+)\s+@[^\]]+\]\s+"
    r"n:\s*(?P<index>\d+)\s+pts:\s*[^\s]+\s+pts_time:(?P<seconds>[-+0-9.eE]+)"
)


def parse_showinfo_pts(
    stderr: str,
    *,
    stage: str = "decode",
    details: dict[str, object] | None = None,
) -> list[float]:
    indexed: dict[int, float] = {}
    for match in SHOWINFO_PATTERN.finditer(stderr):
        indexed[int(match.group("index"))] = float(match.group("seconds"))
    if not indexed:
        raise AutoConformFailure(
            stage,
            "FFMPEG_TIMESTAMPS_MISSING",
            "FFmpeg produced images without machine-readable output timestamps.",
            details,
        )
    expected = list(range(len(indexed)))
    if sorted(indexed) != expected:
        raise AutoConformFailure(
            stage,
            "FFMPEG_TIMESTAMPS_NON_CONTIGUOUS",
            "FFmpeg output timestamp records were not contiguous.",
            {
                **(details or {}),
                "observedIndices": sorted(indexed)[:100],
            },
        )
    return [indexed[index] for index in expected]


def parse_named_showinfo_pts(
    stderr: str,
    expected_labels: list[str],
    *,
    stage: str,
    details: dict[str, object] | None = None,
) -> dict[str, float]:
    expected = set(expected_labels)
    observed: dict[str, float] = {}
    unexpected: list[str] = []
    for match in NAMED_SHOWINFO_PATTERN.finditer(stderr):
        label = match.group("label")
        if label not in expected:
            unexpected.append(label)
            continue
        index = int(match.group("index"))
        if index != 0 or label in observed:
            raise AutoConformFailure(
                stage,
                "FFMPEG_NAMED_TIMESTAMP_COUNT_MISMATCH",
                "Each grouped exact export must produce exactly one timestamp record.",
                {
                    **(details or {}),
                    "label": label,
                    "frameIndex": index,
                },
            )
        observed[label] = float(match.group("seconds"))
    missing = [label for label in expected_labels if label not in observed]
    if missing or unexpected:
        raise AutoConformFailure(
            stage,
            "FFMPEG_NAMED_TIMESTAMPS_INCOMPLETE",
            "Grouped exact-export timestamps did not cover exactly the requested outputs.",
            {
                **(details or {}),
                "missingLabels": missing,
                "unexpectedLabels": sorted(set(unexpected)),
            },
        )
    return observed


def parse_named_showinfo_pts_series(
    stderr: str,
    expected_labels: list[str],
    *,
    stage: str,
    details: dict[str, object] | None = None,
) -> dict[str, list[float]]:
    expected = set(expected_labels)
    observed: dict[str, dict[int, float]] = {label: {} for label in expected_labels}
    unexpected: list[str] = []
    for match in NAMED_SHOWINFO_PATTERN.finditer(stderr):
        label = match.group("label")
        if label not in expected:
            unexpected.append(label)
            continue
        index = int(match.group("index"))
        if index in observed[label]:
            raise AutoConformFailure(
                stage,
                "FFMPEG_NAMED_TIMESTAMP_COUNT_MISMATCH",
                "A grouped decode produced a duplicate timestamp record.",
                {
                    **(details or {}),
                    "label": label,
                    "frameIndex": index,
                },
            )
        observed[label][index] = float(match.group("seconds"))
    missing = [label for label in expected_labels if not observed[label]]
    if missing or unexpected:
        raise AutoConformFailure(
            stage,
            "FFMPEG_NAMED_TIMESTAMPS_INCOMPLETE",
            "Grouped decode timestamps did not cover exactly the requested outputs.",
            {
                **(details or {}),
                "missingLabels": missing,
                "unexpectedLabels": sorted(set(unexpected)),
            },
        )
    result: dict[str, list[float]] = {}
    for label in expected_labels:
        indices = sorted(observed[label])
        if indices != list(range(len(indices))):
            raise AutoConformFailure(
                stage,
                "FFMPEG_TIMESTAMPS_NON_CONTIGUOUS",
                "Grouped decode timestamp records were not contiguous.",
                {
                    **(details or {}),
                    "label": label,
                    "observedIndices": indices[:100],
                },
            )
        result[label] = [observed[label][index] for index in indices]
    return result


def half_open_timestamp_prefix_length(
    observed_times: list[float] | np.ndarray,
    start: float,
    end: float,
    interval: float,
) -> int:
    tolerance = max(1e-7, interval * 1e-6)
    if len(observed_times) and float(observed_times[0]) < start - tolerance:
        raise ValueError("decoded timestamps start before the requested range")
    keep_count = len(observed_times)
    while keep_count and float(observed_times[keep_count - 1]) >= end - tolerance:
        keep_count -= 1
    return keep_count


def finalize_decoded_series(
    images: list[np.ndarray],
    relative_times: list[float],
    *,
    start: float,
    end: float,
    fps: float,
    stage: str,
    details: dict[str, object] | None,
    media: Path,
    diagnostics: dict[str, Any] | None,
) -> tuple[list[np.ndarray], np.ndarray]:
    decoded_frame_count = len(images)
    if len(relative_times) != len(images):
        raise AutoConformFailure(
            stage,
            "FFMPEG_TIMESTAMP_IMAGE_COUNT_MISMATCH",
            "FFmpeg timestamp and decoded-image counts differ.",
            {
                **(details or {}),
                "timestampCount": len(relative_times),
                "imageCount": len(images),
                "mediaPath": str(media),
            },
        )
    times = np.asarray([start + value for value in relative_times], dtype=np.float64)
    if len(times) > 1 and np.any(np.diff(times) <= 0):
        raise AutoConformFailure(
            stage,
            "FFMPEG_TIMESTAMPS_NON_MONOTONE",
            "FFmpeg output timestamps were not strictly increasing.",
            details,
        )
    duration = max(1.0 / fps, end - start)
    effective_end = start + duration
    try:
        keep_count = half_open_timestamp_prefix_length(
            times,
            start,
            effective_end,
            1.0 / fps,
        )
    except ValueError as exc:
        raise AutoConformFailure(
            stage,
            "FFMPEG_TIMESTAMP_OUT_OF_RANGE",
            str(exc),
            {
                **(details or {}),
                "firstTimestampSeconds": float(times[0]),
                "lastTimestampSeconds": float(times[-1]),
                "rangeStartSeconds": start,
                "rangeEndSeconds": effective_end,
            },
        ) from exc
    if keep_count < len(times):
        images = images[:keep_count]
        times = times[:keep_count]
    if not images:
        raise AutoConformFailure(
            stage,
            "FFMPEG_SAMPLES_OUT_OF_RANGE",
            "FFmpeg produced no sample inside the requested half-open decode range.",
            {
                **(details or {}),
                "rangeStartSeconds": start,
                "rangeEndSeconds": effective_end,
            },
        )
    add_diagnostic_count(diagnostics, "decodedFrames", decoded_frame_count)
    add_diagnostic_count(diagnostics, "refinementWindowFrames", len(images))
    return images, times


def decode_series(
    ffmpeg: str,
    media: Path,
    start: float,
    end: float,
    fps: float,
    folder: Path,
    prefix: str,
    max_dim: int,
    timeout_seconds: float,
    *,
    stage: str = "decode",
    details: dict[str, object] | None = None,
    diagnostics: dict[str, Any] | None = None,
    media_kind: str = "ORIGINAL_SOURCE",
) -> tuple[list[np.ndarray], np.ndarray]:
    folder.mkdir(parents=True, exist_ok=True)
    for old in folder.glob(f"{prefix}_*.jpg"):
        old.unlink(missing_ok=True)
    pattern = folder / f"{prefix}_%06d.jpg"
    if fps <= 0:
        raise AutoConformFailure(stage, "INVALID_FRAME_RATE", "Decode frame rate must be positive.")
    requested_duration = end - start
    duration = max(1.0 / fps, requested_duration)
    filter_duration = duration + (1.0 / fps)
    ffmpeg_started = time.perf_counter()
    result = run(
        [
            ffmpeg,
            "-y",
            "-ss",
            f"{max(0.0, start):.9f}",
            "-i",
            str(media),
            "-vf",
            (
                f"tpad=stop_mode=clone:stop_duration={2.0 / fps:.12f},"
                f"fps={fps:.12f},trim=duration={filter_duration:.12f},showinfo,"
                f"scale='if(gt(iw,ih),min({max_dim},iw),-2)':'if(gt(iw,ih),-2,min({max_dim},ih))'"
            ),
            "-fps_mode",
            "passthrough",
            "-q:v",
            "4",
            str(pattern),
        ],
        timeout_seconds,
        stage=stage,
        details={
            **(details or {}),
            "mediaPath": str(media),
            "rangeStartSeconds": start,
            "rangeEndSeconds": end,
            "framesPerSecond": fps,
        },
    )
    add_diagnostic_time(
        diagnostics,
        "refinementWindowFfmpegSeconds",
        time.perf_counter() - ffmpeg_started,
    )
    add_diagnostic_count(diagnostics, "ffmpegInvocations")
    add_diagnostic_count(diagnostics, "refinementWindowDecodes")
    if media_kind == "REFERENCE":
        add_diagnostic_count(diagnostics, "referenceMediaInputs")
    elif media_kind == "PROXY_SOURCE":
        add_diagnostic_count(diagnostics, "proxyMediaInputs")
    else:
        add_diagnostic_count(diagnostics, "originalMediaInputs")
    images: list[np.ndarray] = []
    image_read_started = time.perf_counter()
    for image_path in sorted(folder.glob(f"{prefix}_*.jpg")):
        image = cv2.imread(str(image_path), cv2.IMREAD_GRAYSCALE)
        if image is not None:
            images.append(image)
    add_diagnostic_time(
        diagnostics,
        (
            "referenceImageReadSeconds"
            if media_kind == "REFERENCE"
            else "sourceImageReadSeconds"
        ),
        time.perf_counter() - image_read_started,
    )
    relative_times = parse_showinfo_pts(
        result.stderr.decode("utf-8", errors="replace"),
        stage=stage,
        details=details,
    )
    return finalize_decoded_series(
        images,
        relative_times,
        start=start,
        end=end,
        fps=fps,
        stage=stage,
        details=details,
        media=media,
        diagnostics=diagnostics,
    )


def decode_series_grouped(
    ffmpeg: str,
    requests: list[DecodeSeriesRequest],
    max_dim: int,
    timeout_seconds: float,
    *,
    diagnostics: dict[str, Any] | None,
    stage: str,
) -> list[tuple[list[np.ndarray], np.ndarray]]:
    if (
        not requests
        or len(requests) > MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS
    ):
        raise AutoConformFailure(
            stage,
            "INVALID_GROUPED_DECODE_COUNT",
            "Grouped refinement decode requires a bounded non-empty request set.",
            {
                "requestCount": len(requests),
                "maximumRequestCount": MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS,
            },
        )

    command = [ffmpeg, "-y"]
    prepared: list[PreparedDecodeSeriesRequest] = []
    for index, (
        media,
        start,
        end,
        fps,
        folder,
        prefix,
        details,
        media_kind,
    ) in enumerate(requests):
        if not math.isfinite(fps) or fps <= 0:
            raise AutoConformFailure(
                stage,
                "INVALID_FRAME_RATE",
                "Grouped refinement decode frame rates must be positive and finite.",
                {
                    **details,
                    "frameRate": fps,
                    "mediaPath": str(media),
                },
            )
        folder.mkdir(parents=True, exist_ok=True)
        for old in folder.glob(f"{prefix}_*.jpg"):
            old.unlink(missing_ok=True)
        pattern = folder / f"{prefix}_%06d.jpg"
        label = f"refinement_window_{index}"
        prepared.append(
            (
                media,
                start,
                end,
                fps,
                pattern,
                prefix,
                details,
                media_kind,
                label,
            )
        )
        command.extend(["-ss", f"{max(0.0, start):.9f}", "-i", str(media)])

    filters: list[str] = []
    for index, (
        _media,
        start,
        end,
        fps,
        _pattern,
        _prefix,
        _details,
        _media_kind,
        label,
    ) in enumerate(prepared):
        duration = max(1.0 / fps, end - start)
        filter_duration = duration + (1.0 / fps)
        filters.append(
            (
                f"[{index}:v]tpad=stop_mode=clone:stop_duration={2.0 / fps:.12f},"
                f"fps={fps:.12f},trim=duration={filter_duration:.12f},"
                f"showinfo@{label},"
                f"scale='if(gt(iw,ih),min({max_dim},iw),-2)':"
                f"'if(gt(iw,ih),-2,min({max_dim},ih))'"
                f"[grouped_refinement_output_{index}]"
            )
        )
    command.extend(["-filter_complex", ";".join(filters)])
    for index, item in enumerate(prepared):
        command.extend(
            [
                "-map",
                f"[grouped_refinement_output_{index}]",
                "-fps_mode",
                "passthrough",
                "-q:v",
                "4",
                str(item[4]),
            ]
        )

    ffmpeg_started = time.perf_counter()
    result = run(
        command,
        timeout_seconds,
        stage=stage,
        details={
            "requestCount": len(prepared),
            "mediaPaths": [str(item[0]) for item in prepared],
            "rangeStartsSeconds": [item[1] for item in prepared],
            "rangeEndsSeconds": [item[2] for item in prepared],
        },
    )
    add_diagnostic_time(
        diagnostics,
        "refinementWindowFfmpegSeconds",
        time.perf_counter() - ffmpeg_started,
    )
    add_diagnostic_count(diagnostics, "ffmpegInvocations")
    add_diagnostic_count(diagnostics, "refinementWindowDecodes", len(prepared))
    for item in prepared:
        media_kind = item[7]
        if media_kind == "REFERENCE":
            add_diagnostic_count(diagnostics, "referenceMediaInputs")
        elif media_kind == "PROXY_SOURCE":
            add_diagnostic_count(diagnostics, "proxyMediaInputs")
        else:
            add_diagnostic_count(diagnostics, "originalMediaInputs")

    named_pts = parse_named_showinfo_pts_series(
        result.stderr.decode("utf-8", errors="replace"),
        [item[8] for item in prepared],
        stage=stage,
        details={"requestCount": len(prepared)},
    )
    decoded: list[tuple[list[np.ndarray], np.ndarray]] = []
    for (
        media,
        start,
        end,
        fps,
        pattern,
        prefix,
        details,
        media_kind,
        label,
    ) in prepared:
        image_read_started = time.perf_counter()
        images: list[np.ndarray] = []
        for image_path in sorted(pattern.parent.glob(f"{prefix}_*.jpg")):
            image = cv2.imread(str(image_path), cv2.IMREAD_GRAYSCALE)
            if image is not None:
                images.append(image)
        add_diagnostic_time(
            diagnostics,
            (
                "referenceImageReadSeconds"
                if media_kind == "REFERENCE"
                else "sourceImageReadSeconds"
            ),
            time.perf_counter() - image_read_started,
        )
        decoded.append(
            finalize_decoded_series(
                images,
                named_pts[label],
                start=start,
                end=end,
                fps=fps,
                stage=stage,
                details=details,
                media=media,
                diagnostics=diagnostics,
            )
        )
    return decoded


def merge_compatible_decode_requests(
    requests: list[DecodeSeriesRequest],
) -> list[MergedDecodeSeriesGroup]:
    groups: list[dict[str, Any]] = []
    ordered = sorted(
        enumerate(requests),
        key=lambda item: (
            str(item[1][0].resolve()),
            float(item[1][3]),
            str(item[1][7]),
            str(item[1][4].resolve()),
            float(item[1][1]),
            item[0],
        ),
    )
    for original_index, request in ordered:
        media, start, end, fps, folder, prefix, details, media_kind = request
        if (
            not math.isfinite(start)
            or not math.isfinite(end)
            or not math.isfinite(fps)
            or end <= start
            or fps <= 0
        ):
            groups.append(
                {
                    "indices": [original_index],
                    "requests": [request],
                    "media": media.resolve(),
                    "start": start,
                    "end": end,
                    "fps": fps,
                    "folder": folder,
                    "mediaKind": media_kind,
                }
            )
            continue
        selected: dict[str, Any] | None = None
        if groups:
            group = groups[-1]
            if (
                group["media"] == media.resolve()
                and abs(float(group["fps"]) - fps) <= 1e-9
                and group["folder"] == folder
                and group["mediaKind"] == media_kind
            ):
                frame_delta = (start - float(group["start"])) * fps
                frame_aligned = abs(frame_delta - round(frame_delta)) <= 1e-6
                timestamp_tolerance = max(1e-7, (1.0 / fps) * 1e-6)
                if (
                    frame_aligned
                    and start <= float(group["end"]) + timestamp_tolerance
                ):
                    selected = group
        if selected is None:
            groups.append(
                {
                    "indices": [original_index],
                    "requests": [request],
                    "media": media.resolve(),
                    "start": start,
                    "end": end,
                    "fps": fps,
                    "folder": folder,
                    "mediaKind": media_kind,
                }
            )
            continue
        selected["indices"].append(original_index)
        selected["requests"].append(request)
        selected["end"] = max(float(selected["end"]), end)

    planned: list[MergedDecodeSeriesGroup] = []
    for group in groups:
        member_requests: list[DecodeSeriesRequest] = group["requests"]
        first = member_requests[0]
        first_details = first[6]
        planned.append(
            (
                list(group["indices"]),
                (
                    first[0],
                    float(group["start"]),
                    float(group["end"]),
                    first[3],
                    first[4],
                    f"{first[5]}_merged_{len(member_requests)}",
                    {
                        **first_details,
                        "mergedRequestCount": len(member_requests),
                    },
                    first[7],
                ),
            )
        )
    return planned


def decode_series_grouped_merged(
    ffmpeg: str,
    requests: list[DecodeSeriesRequest],
    max_dim: int,
    timeout_seconds: float,
    *,
    diagnostics: dict[str, Any] | None,
    stage: str,
) -> list[tuple[list[np.ndarray], np.ndarray]]:
    planned = merge_compatible_decode_requests(requests)
    merged_results = decode_series_grouped(
        ffmpeg,
        [request for _indices, request in planned],
        max_dim,
        timeout_seconds,
        diagnostics=diagnostics,
        stage=stage,
    )
    results: list[tuple[list[np.ndarray], np.ndarray] | None] = [
        None for _request in requests
    ]
    fallback_indices: list[int] = []
    for (member_indices, _merged_request), (frames, times) in zip(
        planned,
        merged_results,
    ):
        if len(member_indices) == 1:
            results[member_indices[0]] = (frames, times)
            continue
        group_failed = False
        group_results: list[tuple[int, list[np.ndarray], np.ndarray]] = []
        for original_index in member_indices:
            request = requests[original_index]
            start = float(request[1])
            fps = float(request[3])
            duration = max(1.0 / fps, float(request[2]) - start)
            end = start + duration
            tolerance = max(1e-7, (1.0 / fps) * 1e-6)
            selected_indices = np.flatnonzero(
                (times >= start - tolerance)
                & (times < end - tolerance)
            )
            selected_times = times[selected_indices]
            selected_frames = [frames[int(index)] for index in selected_indices]
            expected_count = max(
                1,
                math.ceil((duration - tolerance) * fps),
            )
            if (
                len(selected_frames) != expected_count
                or not len(selected_times)
                or abs(float(selected_times[0]) - start) > tolerance
            ):
                group_failed = True
                break
            group_results.append(
                (original_index, selected_frames, selected_times)
            )
        if group_failed:
            fallback_indices.extend(member_indices)
            continue
        for original_index, selected_frames, selected_times in group_results:
            results[original_index] = (selected_frames, selected_times)

    if fallback_indices:
        fallback_results = decode_series_grouped(
            ffmpeg,
            [requests[index] for index in fallback_indices],
            max_dim,
            timeout_seconds,
            diagnostics=diagnostics,
            stage=stage,
        )
        for original_index, fallback in zip(fallback_indices, fallback_results):
            results[original_index] = fallback
    if any(result is None for result in results):
        raise AutoConformFailure(
            stage,
            "MERGED_DECODE_RESULT_INCOMPLETE",
            "Merged refinement decoding did not cover every requested window.",
            {
                "requestCount": len(requests),
                "plannedGroupCount": len(planned),
                "fallbackCount": len(fallback_indices),
            },
        )
    return [result for result in results if result is not None]


def load_and_record_exact_export(
    output: Path,
    requested_seconds: float,
    relative_pts_seconds: float,
    *,
    diagnostics: dict[str, Any] | None,
    role: str,
    media_kind: str,
    evidence: dict[str, object] | None,
    stage: str,
    media: Path,
    expected_frame_rate: float | None,
) -> np.ndarray:
    image_read_started = time.perf_counter()
    image = cv2.imread(str(output), cv2.IMREAD_GRAYSCALE)
    add_diagnostic_time(
        diagnostics,
        (
            "referenceImageReadSeconds"
            if media_kind == "REFERENCE"
            else "sourceImageReadSeconds"
        ),
        time.perf_counter() - image_read_started,
    )
    if image is None:
        raise AutoConformFailure(
            stage,
            "EXPORTED_FRAME_UNREADABLE",
            "The exact exported frame could not be decoded.",
            {
                **(evidence or {}),
                "mediaPath": str(media),
                "requestedSeconds": requested_seconds,
                "role": role,
            },
        )
    add_diagnostic_count(diagnostics, "decodedFrames")
    decoded_seconds = requested_seconds + relative_pts_seconds
    requested_delta_seconds = decoded_seconds - requested_seconds
    if not math.isfinite(decoded_seconds) or not math.isfinite(requested_delta_seconds):
        raise AutoConformFailure(
            stage,
            "FFMPEG_EXACT_EXPORT_TIMESTAMP_INVALID",
            "Exact frame export produced a non-finite decoded timestamp.",
            {
                **(evidence or {}),
                "mediaPath": str(media),
                "requestedSeconds": requested_seconds,
                "relativePtsSeconds": relative_pts_seconds,
                "role": role,
            },
        )
    timestamp_plausibility = "NOT_CHECKED"
    if expected_frame_rate is not None:
        maximum_delta_seconds = (1.0 / expected_frame_rate) + 1e-6
        if abs(requested_delta_seconds) > maximum_delta_seconds:
            timestamp_plausibility = "OUTSIDE_NOMINAL_FRAME"
            add_diagnostic_count(diagnostics, "exactExportTimestampWarnings")
        else:
            timestamp_plausibility = "WITHIN_NOMINAL_FRAME"
    if diagnostics is not None:
        record: dict[str, object] = {
            "role": role,
            "mediaKind": media_kind,
            "requestedSeconds": requested_seconds,
            "relativePtsSeconds": relative_pts_seconds,
            "decodedPtsSeconds": decoded_seconds,
            "requestedToDecodedDeltaSeconds": requested_delta_seconds,
            "timestampPlausibility": timestamp_plausibility,
            "imageWidth": int(image.shape[1]),
            "imageHeight": int(image.shape[0]),
        }
        record.update(evidence or {})
        record_exact_export(diagnostics, record)
    return image


def export_frame(
    ffmpeg: str,
    media: Path,
    when: float,
    output: Path,
    max_dim: int,
    timeout_seconds: float,
    *,
    diagnostics: dict[str, Any] | None = None,
    role: str = "REFERENCE_ANCHOR",
    media_kind: str = "REFERENCE",
    evidence: dict[str, object] | None = None,
    stage: str = "frame-export",
    expected_frame_rate: float | None = None,
) -> np.ndarray:
    if role not in EXPORT_ROLE_METRICS:
        raise AutoConformFailure(
            stage,
            "INVALID_EXPORT_ROLE",
            "Exact frame export received an unsupported diagnostic role.",
            {
                **(evidence or {}),
                "role": role,
                "mediaPath": str(media),
            },
        )
    if expected_frame_rate is not None and (
        not math.isfinite(expected_frame_rate) or expected_frame_rate <= 0
    ):
        raise AutoConformFailure(
            stage,
            "INVALID_FRAME_RATE",
            "Exact frame export frame rate must be positive and finite.",
            {
                **(evidence or {}),
                "frameRate": expected_frame_rate,
                "mediaPath": str(media),
            },
        )
    output.parent.mkdir(parents=True, exist_ok=True)
    output.unlink(missing_ok=True)
    requested_seconds = max(0.0, when)
    ffmpeg_started = time.perf_counter()
    result = run(
        [
            ffmpeg,
            "-y",
            "-ss",
            f"{requested_seconds:.9f}",
            "-i",
            str(media),
            "-frames:v",
            "1",
            "-vf",
            (
                "select='eq(n,0)',showinfo,"
                f"scale='if(gt(iw,ih),min({max_dim},iw),-2)':'if(gt(iw,ih),-2,min({max_dim},ih))'"
            ),
            "-update",
            "1",
            str(output),
        ],
        timeout_seconds,
        stage=stage,
        details={
            **(evidence or {}),
            "mediaPath": str(media),
            "requestedSeconds": requested_seconds,
            "role": role,
        },
    )
    ffmpeg_seconds = time.perf_counter() - ffmpeg_started
    timing_key, count_key = EXPORT_ROLE_METRICS[role]
    add_diagnostic_time(diagnostics, timing_key, ffmpeg_seconds)
    add_diagnostic_count(diagnostics, "ffmpegInvocations")
    add_diagnostic_count(diagnostics, count_key)
    if media_kind == "REFERENCE":
        add_diagnostic_count(diagnostics, "referenceMediaInputs")
    elif media_kind == "PROXY_SOURCE":
        add_diagnostic_count(diagnostics, "proxyMediaInputs")
    else:
        add_diagnostic_count(diagnostics, "originalMediaInputs")
    relative_pts = parse_showinfo_pts(
        result.stderr.decode("utf-8", errors="replace"),
        stage=stage,
        details={
            **(evidence or {}),
            "mediaPath": str(media),
            "requestedSeconds": requested_seconds,
            "role": role,
        },
    )
    if len(relative_pts) != 1:
        raise AutoConformFailure(
            stage,
            "FFMPEG_EXACT_EXPORT_TIMESTAMP_COUNT_MISMATCH",
            "Exact frame export must produce exactly one timestamp record.",
            {
                **(evidence or {}),
                "mediaPath": str(media),
                "requestedSeconds": requested_seconds,
                "role": role,
                "timestampCount": len(relative_pts),
            },
        )
    relative_pts_seconds = float(relative_pts[0])
    return load_and_record_exact_export(
        output,
        requested_seconds,
        relative_pts_seconds,
        diagnostics=diagnostics,
        role=role,
        media_kind=media_kind,
        evidence=evidence,
        stage=stage,
        media=media,
        expected_frame_rate=expected_frame_rate,
    )


def export_exact_frames_grouped(
    ffmpeg: str,
    media: Path,
    requests: list[tuple[float, Path, dict[str, object]]],
    max_dim: int,
    timeout_seconds: float,
    *,
    diagnostics: dict[str, Any] | None,
    role: str,
    media_kind: str,
    stage: str,
    expected_frame_rate: float,
) -> list[np.ndarray]:
    if role not in EXPORT_ROLE_METRICS:
        raise AutoConformFailure(
            stage,
            "INVALID_EXPORT_ROLE",
            "Grouped exact frame export received an unsupported diagnostic role.",
            {"role": role, "mediaPath": str(media)},
        )
    if not requests or len(requests) > 3:
        raise AutoConformFailure(
            stage,
            "INVALID_GROUPED_EXPORT_COUNT",
            "Grouped exact frame export requires between one and three frame requests.",
            {"requestCount": len(requests), "role": role, "mediaPath": str(media)},
        )
    if not math.isfinite(expected_frame_rate) or expected_frame_rate <= 0:
        raise AutoConformFailure(
            stage,
            "INVALID_FRAME_RATE",
            "Grouped exact frame export frame rate must be positive and finite.",
            {"frameRate": expected_frame_rate, "role": role, "mediaPath": str(media)},
        )

    command = [ffmpeg, "-y"]
    prepared: list[tuple[float, Path, dict[str, object], str]] = []
    label_prefix = role.lower()
    for index, (when, output, evidence) in enumerate(requests):
        requested_seconds = max(0.0, when)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.unlink(missing_ok=True)
        label = f"{label_prefix}_{index}"
        prepared.append((requested_seconds, output, evidence, label))
        command.extend(["-ss", f"{requested_seconds:.9f}", "-i", str(media)])

    scale = (
        f"scale='if(gt(iw,ih),min({max_dim},iw),-2)':"
        f"'if(gt(iw,ih),-2,min({max_dim},ih))'"
    )
    filters = [
        (
            f"[{index}:v]select='eq(n,0)',showinfo@{label},{scale}"
            f"[grouped_exact_output_{index}]"
        )
        for index, (_requested, _output, _evidence, label) in enumerate(prepared)
    ]
    command.extend(["-filter_complex", ";".join(filters)])
    for index, (_requested, output, _evidence, _label) in enumerate(prepared):
        command.extend(
            [
                "-map",
                f"[grouped_exact_output_{index}]",
                "-frames:v",
                "1",
                "-update",
                "1",
                str(output),
            ]
        )

    ffmpeg_started = time.perf_counter()
    result = run(
        command,
        timeout_seconds,
        stage=stage,
        details={
            "mediaPath": str(media),
            "requestCount": len(prepared),
            "requestedSeconds": [item[0] for item in prepared],
            "role": role,
        },
    )
    timing_key, count_key = EXPORT_ROLE_METRICS[role]
    add_diagnostic_time(
        diagnostics,
        timing_key,
        time.perf_counter() - ffmpeg_started,
    )
    add_diagnostic_count(diagnostics, "ffmpegInvocations")
    add_diagnostic_count(diagnostics, count_key, len(prepared))
    if media_kind == "REFERENCE":
        add_diagnostic_count(diagnostics, "referenceMediaInputs", len(prepared))
    elif media_kind == "PROXY_SOURCE":
        add_diagnostic_count(diagnostics, "proxyMediaInputs", len(prepared))
    else:
        add_diagnostic_count(diagnostics, "originalMediaInputs", len(prepared))

    named_pts = parse_named_showinfo_pts(
        result.stderr.decode("utf-8", errors="replace"),
        [item[3] for item in prepared],
        stage=stage,
        details={"mediaPath": str(media), "requestCount": len(prepared), "role": role},
    )
    return [
        load_and_record_exact_export(
            output,
            requested_seconds,
            float(named_pts[label]),
            diagnostics=diagnostics,
            role=role,
            media_kind=media_kind,
            evidence=evidence,
            stage=stage,
            media=media,
            expected_frame_rate=expected_frame_rate,
        )
        for requested_seconds, output, evidence, label in prepared
    ]


def export_alignment_frames_grouped(
    ffmpeg: str,
    media: Path,
    requests: list[tuple[float, Path, dict[str, object]]],
    max_dim: int,
    timeout_seconds: float,
    *,
    diagnostics: dict[str, Any] | None,
    media_kind: str,
    stage: str,
    expected_frame_rate: float,
) -> list[np.ndarray]:
    return export_exact_frames_grouped(
        ffmpeg,
        media,
        requests,
        max_dim,
        timeout_seconds,
        diagnostics=diagnostics,
        role="ALIGNMENT",
        media_kind=media_kind,
        stage=stage,
        expected_frame_rate=expected_frame_rate,
    )


def normalized_canvas(image: np.ndarray, size: int = 96) -> np.ndarray:
    height, width = image.shape[:2]
    scale = size / max(height, width)
    resized = cv2.resize(
        image,
        (max(8, round(width * scale)), max(8, round(height * scale))),
        interpolation=cv2.INTER_AREA,
    )
    canvas = np.zeros((size, size), dtype=np.uint8)
    y = (size - resized.shape[0]) // 2
    x = (size - resized.shape[1]) // 2
    canvas[y : y + resized.shape[0], x : x + resized.shape[1]] = resized
    return cv2.equalizeHist(canvas)


def descriptor(image: np.ndarray) -> np.ndarray:
    gray = normalized_canvas(image)
    gx = cv2.Sobel(gray, cv2.CV_32F, 1, 0, ksize=3)
    gy = cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3)
    edge = cv2.magnitude(gx, gy)
    edge = cv2.resize(edge, (48, 48), interpolation=cv2.INTER_AREA)
    tone = cv2.resize(gray, (32, 32), interpolation=cv2.INTER_AREA).astype(np.float32)
    vector = np.concatenate([edge.reshape(-1), tone.reshape(-1) * 0.35]).astype(np.float32)
    vector -= float(vector.mean())
    norm = float(np.linalg.norm(vector))
    return vector if norm < 1e-9 else vector / norm


def descriptor_variants(image: np.ndarray) -> np.ndarray:
    spatial = (image, cv2.flip(image, 1), cv2.flip(image, 0), cv2.flip(image, -1))
    return np.stack([descriptor(variant) for variant in spatial] + [descriptor(255 - variant) for variant in spatial])


def descriptor_variant(image: np.ndarray, variant_index: int) -> np.ndarray:
    if variant_index < 0 or variant_index >= 8:
        raise ValueError("descriptor variant index must be in the range 0..7")
    spatial_index = variant_index % 4
    spatial = (
        image,
        cv2.flip(image, 1),
        cv2.flip(image, 0),
        cv2.flip(image, -1),
    )[spatial_index]
    return descriptor(255 - spatial if variant_index >= 4 else spatial)


def motion_image(first: np.ndarray, second: np.ndarray) -> np.ndarray:
    return cv2.absdiff(normalized_canvas(first), normalized_canvas(second))


def prepare_pixel_identity_reference(
    reference: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> tuple[np.ndarray, float]:
    def prepare() -> tuple[np.ndarray, float]:
        ref = normalized_canvas(reference).astype(np.float32).reshape(-1)
        ref -= float(ref.mean())
        return ref, float(np.linalg.norm(ref))

    return measured_call(
        diagnostics,
        "referencePixelIdentitySeconds",
        prepare,
    )


def get_pixel_identity_reference(
    cache: dict[int, tuple[np.ndarray, tuple[np.ndarray, float]]],
    reference: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> tuple[np.ndarray, float]:
    cache_key = id(reference)
    cached = cache.get(cache_key)
    if cached is not None and cached[0] is reference:
        return cached[1]
    prepared = prepare_pixel_identity_reference(reference, diagnostics)
    cache[cache_key] = (reference, prepared)
    return prepared


def get_reference_motion_descriptor(
    cache: dict[int, np.ndarray],
    index: int,
    reference_images: list[np.ndarray],
    diagnostics: dict[str, Any] | None = None,
) -> np.ndarray:
    cached = cache.get(index)
    if cached is not None:
        return cached
    prepared = measured_call(
        diagnostics,
        "referenceMotionSeconds",
        lambda: descriptor(
            motion_image(
                reference_images[index - 1],
                reference_images[index],
            )
        ),
    )
    cache[index] = prepared
    return prepared


def pixel_identity_score_prepared(
    prepared_reference: tuple[np.ndarray, float],
    source: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> float:
    ref, ref_norm = prepared_reference
    if ref_norm < 1e-9:
        return 0.0

    def compare_source_variants() -> float:
        best = 0.0
        for variant in (source, cv2.flip(source, 1), cv2.flip(source, 0), cv2.flip(source, -1)):
            canvas = normalized_canvas(variant).astype(np.float32)
            for candidate in (canvas, 255.0 - canvas):
                vector = candidate.reshape(-1)
                vector -= float(vector.mean())
                norm = float(np.linalg.norm(vector))
                if norm >= 1e-9:
                    best = max(best, float(np.dot(ref, vector) / (ref_norm * norm)))
        return best

    best = measured_call(
        diagnostics,
        "sourcePixelIdentitySeconds",
        compare_source_variants,
    )
    return max(0.0, min(1.0, best))


def pixel_identity_score(
    reference: np.ndarray,
    source: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> float:
    return pixel_identity_score_prepared(
        prepare_pixel_identity_reference(reference, diagnostics),
        source,
        diagnostics,
    )


def descriptor_score(reference: np.ndarray, source_variants: np.ndarray) -> float:
    correlation = float(np.max(source_variants @ reference))
    return max(0.0, min(1.0, (correlation + 1.0) * 0.5))


def descriptor_score_with_variant_evidence(
    reference: np.ndarray,
    source_variants: np.ndarray,
) -> tuple[float, int, float]:
    correlations = source_variants @ reference
    variant_index = int(np.argmax(correlations))
    correlation = float(correlations[variant_index])
    ordered = np.sort(correlations)
    separation = (
        float(ordered[-1] - ordered[-2])
        if len(ordered) > 1
        else 1.0
    )
    return (
        max(0.0, min(1.0, (correlation + 1.0) * 0.5)),
        variant_index,
        max(0.0, separation),
    )


def consensus_descriptor_variant(
    variant_indices: list[int],
    variant_separations: list[float],
    minimum_separation: float,
) -> int | None:
    if (
        len(variant_indices) < 3
        or len(variant_indices) != len(variant_separations)
        or len(set(variant_indices)) != 1
        or any(index < 0 or index >= 8 for index in variant_indices)
        or any(
            not math.isfinite(separation)
            or separation < minimum_separation
            for separation in variant_separations
        )
    ):
        return None
    return variant_indices[0]


def alignment_winner_margin(scores: list[float]) -> float:
    if not scores:
        raise ValueError("Alignment winner margin requires at least one score")
    if len(scores) == 1:
        return 1.0
    ordered = sorted((float(score) for score in scores), reverse=True)
    return max(0.0, min(1.0, ordered[0] - ordered[1]))


def create_feature_detector() -> Any:
    return cv2.SIFT_create(nfeatures=2500) if hasattr(cv2, "SIFT_create") else cv2.AKAZE_create()


def prepare_reference_features(
    reference: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> tuple[Any, Any, int] | None:
    detector = measured_call(
        diagnostics,
        "featureDetectorSetupSeconds",
        create_feature_detector,
    )
    key_ref, desc_ref = measured_call(
        diagnostics,
        "referenceFeatureSeconds",
        lambda: detector.detectAndCompute(reference, None),
    )
    if desc_ref is None or len(key_ref) < 8:
        return None
    norm = cv2.NORM_L2 if desc_ref.dtype == np.float32 else cv2.NORM_HAMMING
    return key_ref, desc_ref, norm


def get_reference_features(
    cache: dict[int, tuple[np.ndarray, tuple[Any, Any, int] | None]],
    reference: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> tuple[Any, Any, int] | None:
    cache_key = id(reference)
    cached = cache.get(cache_key)
    if cached is not None and cached[0] is reference:
        add_diagnostic_count(diagnostics, "referenceFeatureCacheHits")
        return cached[1]
    add_diagnostic_count(diagnostics, "referenceFeatureCacheMisses")
    prepared = prepare_reference_features(reference, diagnostics)
    cache[cache_key] = (reference, prepared)
    return prepared


def feature_score(
    prepared_reference: tuple[Any, Any, int] | None,
    source: np.ndarray,
    diagnostics: dict[str, Any] | None = None,
) -> float:
    if prepared_reference is None:
        return 0.0
    key_ref, desc_ref, norm = prepared_reference
    # Keep source comparisons isolated as before; only the immutable reference
    # keypoints/descriptors are shared across calls.
    detector = measured_call(
        diagnostics,
        "featureDetectorSetupSeconds",
        create_feature_detector,
    )
    best = 0.0
    for variant in (source, cv2.flip(source, 1), cv2.flip(source, 0), cv2.flip(source, -1)):
        key_src, desc_src = measured_call(
            diagnostics,
            "sourceFeatureSeconds",
            lambda variant=variant: detector.detectAndCompute(variant, None),
        )
        if desc_src is None or len(key_src) < 8:
            continue

        def compare_features() -> float:
            pairs = cv2.BFMatcher(norm).knnMatch(desc_ref, desc_src, k=2)
            good = [
                pair[0]
                for pair in pairs
                if len(pair) == 2 and pair[0].distance < 0.76 * pair[1].distance
            ]
            if len(good) < 8:
                return 0.0
            points_ref = np.float32([key_ref[item.queryIdx].pt for item in good]).reshape(-1, 1, 2)
            points_src = np.float32([key_src[item.trainIdx].pt for item in good]).reshape(-1, 1, 2)
            _, mask = cv2.estimateAffinePartial2D(
                points_ref,
                points_src,
                method=cv2.RANSAC,
                ransacReprojThreshold=5.0,
            )
            if mask is None:
                return 0.0
            inliers = int(mask.sum())
            ratio = inliers / max(1, len(good))
            return min(1.0, (inliers / 35.0) * ratio)

        best = max(
            best,
            measured_call(
                diagnostics,
                "featureComparisonSeconds",
                compare_features,
            ),
        )
    return best


def shot_anchor_times(shot: dict[str, Any]) -> list[float]:
    start = float(shot["referenceStartSeconds"])
    duration = max(0.0, float(shot["durationSeconds"]))
    if duration <= 0.12:
        return [start + duration * 0.5]
    return [start + duration * fraction for fraction in (0.2, 0.5, 0.8)]


def source_cache_metadata(
    source: dict[str, Any],
    interval: float,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
) -> dict[str, Any]:
    return {
        "cacheFormatVersion": INDEX_CACHE_FORMAT_VERSION,
        "descriptorVersion": DESCRIPTOR_VERSION,
        "samplingIntervalSeconds": interval,
        "targetChunkSeconds": chunk_seconds,
        "decodeGeometry": {
            "maxDimensionPixels": COARSE_INDEX_MAX_DIM,
            "colorMode": "GRAYSCALE",
        },
        "cacheIdentity": source["cacheIdentity"],
        "usableRange": source["usableRange"],
    }


def legacy_source_cache_metadata(source: dict[str, Any], interval: float) -> dict[str, Any]:
    return {
        "cacheFormatVersion": LEGACY_INDEX_CACHE_FORMAT_VERSION,
        "descriptorVersion": DESCRIPTOR_VERSION,
        "samplingIntervalSeconds": interval,
        "cacheIdentity": source["cacheIdentity"],
        "usableRange": source["usableRange"],
    }


def cache_metadata_key(metadata: dict[str, Any]) -> str:
    canonical = json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


def source_cache_path(cache_dir: Path, metadata: dict[str, Any]) -> Path:
    return cache_dir / "source-index-v2" / cache_metadata_key(metadata)


def legacy_source_cache_path(cache_dir: Path, source: dict[str, Any], interval: float) -> Path:
    metadata = legacy_source_cache_metadata(source, interval)
    return cache_dir / f"source-index-{cache_metadata_key(metadata)}.npz"


def atomic_write_json(path: Path, document: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        temporary.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def source_chunk_plan(
    source: dict[str, Any],
    interval: float,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
) -> list[dict[str, Any]]:
    if interval <= 0:
        raise ValueError("sampling interval must be positive")
    if chunk_seconds <= 0:
        raise ValueError("index chunk duration must be positive")
    start = float(source["usableRange"]["startSeconds"])
    end = float(source["usableRange"]["endSeconds"])
    duration = end - start
    if duration <= 0:
        raise ValueError("source usable range must be positive")
    # Include every globally anchored grid point inside the half-open usable
    # range. A partial final interval still contains a valid last sample; only a
    # grid point at the exact range end is excluded.
    total_samples = max(1, int(math.ceil((duration / interval) - 1e-12)))
    samples_per_chunk = max(1, int(math.floor((chunk_seconds / interval) + 1e-12)))
    chunks: list[dict[str, Any]] = []
    for chunk_index, first_sample in enumerate(range(0, total_samples, samples_per_chunk)):
        end_sample = min(total_samples, first_sample + samples_per_chunk)
        chunk_start = start + first_sample * interval
        chunk_end = end if end_sample == total_samples else start + end_sample * interval
        chunks.append(
            {
                "chunkIndex": chunk_index,
                "fileName": f"chunk-{chunk_index:06d}.npz",
                "startSeconds": chunk_start,
                "endSeconds": chunk_end,
                "firstSampleIndex": first_sample,
                "endSampleIndexExclusive": end_sample,
                "expectedSampleCount": end_sample - first_sample,
            }
        )
    return chunks


def expected_chunk_times(
    source: dict[str, Any],
    interval: float,
    chunk: dict[str, Any],
) -> np.ndarray:
    source_start = float(source["usableRange"]["startSeconds"])
    return source_start + np.arange(
        int(chunk["firstSampleIndex"]),
        int(chunk["endSampleIndexExclusive"]),
        dtype=np.float64,
    ) * interval


def planned_chunk_prefix_length(
    source: dict[str, Any],
    interval: float,
    chunk: dict[str, Any],
    observed_times: list[float] | np.ndarray,
) -> int:
    expected_count = int(chunk["expectedSampleCount"])
    if len(observed_times) < expected_count:
        raise ValueError(
            f"decoded chunk is missing planned samples ({len(observed_times)} < {expected_count})"
        )
    if len(observed_times) > expected_count:
        raise ValueError(
            f"decoded chunk contains samples beyond the planned grid ({len(observed_times)} > {expected_count})"
        )
    source_start = float(source["usableRange"]["startSeconds"])
    first_sample = int(chunk["firstSampleIndex"])
    tolerance = max(1e-7, interval * 1e-5)
    for local_index in range(expected_count):
        expected_time = source_start + (first_sample + local_index) * interval
        if not math.isclose(
            float(observed_times[local_index]),
            expected_time,
            rel_tol=0.0,
            abs_tol=tolerance,
        ):
            raise ValueError(
                "decoded chunk timestamps do not cover the globally anchored sampling grid"
            )
    return expected_count


def validate_chunk_arrays(
    source: dict[str, Any],
    interval: float,
    chunk: dict[str, Any],
    times: np.ndarray,
    vectors: np.ndarray,
) -> None:
    expected = expected_chunk_times(source, interval, chunk)
    if times.ndim != 1 or vectors.ndim != 3:
        raise ValueError("invalid chunk array dimensions")
    if vectors.shape != (len(times), 8, 3328):
        raise ValueError("invalid chunk descriptor shape")
    if len(times) != int(chunk["expectedSampleCount"]):
        raise ValueError("unexpected chunk sample count")
    if len(times) > 1 and np.any(np.diff(times) <= 0):
        raise ValueError("chunk timestamps are not strictly increasing")
    tolerance = max(1e-7, interval * 1e-5)
    if not np.allclose(times, expected, rtol=0.0, atol=tolerance):
        raise ValueError("chunk timestamps do not match the globally anchored sampling grid")
    chunk_start = float(chunk["startSeconds"])
    chunk_end = float(chunk["endSeconds"])
    if len(times) and (
        float(times[0]) < chunk_start - tolerance
        or float(times[-1]) >= chunk_end - tolerance
    ):
        raise ValueError("chunk timestamps violate the half-open chunk range")


def source_cache_manifest(
    metadata: dict[str, Any],
    chunks: list[dict[str, Any]],
    completed_chunks: list[int],
    index_path: Path,
) -> dict[str, Any]:
    completed_set = set(completed_chunks)
    completed_files: list[dict[str, int]] = []
    for chunk in chunks:
        chunk_index = int(chunk["chunkIndex"])
        if chunk_index not in completed_set:
            continue
        path = index_path / str(chunk["fileName"])
        try:
            size_bytes = path.stat().st_size
        except OSError:
            continue
        completed_files.append(
            {"chunkIndex": chunk_index, "sizeBytes": size_bytes}
        )
    return {
        "format": "automation-agent-auto-conform-source-index/v2",
        "cacheMetadata": metadata,
        "chunks": chunks,
        "completedChunks": sorted(completed_chunks),
        "completedChunkFiles": completed_files,
    }


def chunk_metadata(metadata: dict[str, Any], chunk: dict[str, Any]) -> dict[str, Any]:
    return {"cacheMetadata": metadata, "chunk": chunk}


def load_source_chunk(
    path: Path,
    metadata: dict[str, Any],
    source: dict[str, Any],
    interval: float,
    chunk: dict[str, Any],
) -> tuple[tuple[np.ndarray, np.ndarray] | None, str]:
    if not path.is_file():
        return None, "missing"
    try:
        with np.load(path, allow_pickle=False) as cached:
            cached_metadata = json.loads(str(cached["metadataJson"].item()))
            times = np.asarray(cached["times"], dtype=np.float64)
            vectors = np.asarray(cached["vectors"], dtype=np.float32)
        if cached_metadata != chunk_metadata(metadata, chunk):
            return None, "metadata-mismatch"
        validate_chunk_arrays(source, interval, chunk, times, vectors)
        try:
            os.utime(path, None)
        except OSError:
            pass
        return (times, vectors), "hit"
    except (
        OSError,
        ValueError,
        KeyError,
        TypeError,
        EOFError,
        zipfile.BadZipFile,
        json.JSONDecodeError,
    ):
        return None, "invalid-format"


def load_source_cache(
    path: Path,
    metadata: dict[str, Any],
    chunks: list[dict[str, Any]],
) -> tuple[bool, str]:
    manifest_path = path / "index.json"
    if not manifest_path.is_file():
        return False, "missing"
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        if manifest.get("format") != "automation-agent-auto-conform-source-index/v2":
            return False, "invalid-format"
        if manifest.get("cacheMetadata") != metadata or manifest.get("chunks") != chunks:
            return False, "metadata-mismatch"
        expected_completed = [int(chunk["chunkIndex"]) for chunk in chunks]
        if manifest.get("completedChunks") != expected_completed:
            return False, "partial"
        expected_files: list[dict[str, int]] = []
        for chunk in chunks:
            chunk_path = path / str(chunk["fileName"])
            try:
                size_bytes = chunk_path.stat().st_size
            except OSError:
                return False, "partial"
            expected_files.append(
                {"chunkIndex": int(chunk["chunkIndex"]), "sizeBytes": size_bytes}
            )
        if manifest.get("completedChunkFiles") != expected_files:
            return False, "size-mismatch"
        try:
            os.utime(manifest_path, None)
        except OSError:
            pass
        return True, "hit"
    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
        return False, "invalid-format"


def save_source_chunk(
    path: Path,
    metadata: dict[str, Any],
    chunk: dict[str, Any],
    times: np.ndarray,
    vectors: np.ndarray,
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        with temporary.open("wb") as handle:
            np.savez(
                handle,
                metadataJson=np.array(
                    json.dumps(chunk_metadata(metadata, chunk), sort_keys=True, separators=(",", ":"))
                ),
                times=times,
                vectors=vectors,
            )
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def prune_source_cache(
    cache_dir: Path,
    max_bytes: int,
    protected_paths: set[Path],
) -> dict[str, int]:
    cache_dir.mkdir(parents=True, exist_ok=True)
    entries: list[tuple[Path, int, float, bool]] = []
    total_bytes = 0
    for path in cache_dir.glob("source-index-*.npz"):
        try:
            stat = path.stat()
        except OSError:
            continue
        total_bytes += stat.st_size
        entries.append((path.resolve(), stat.st_size, stat.st_atime, False))

    v2_root = cache_dir / "source-index-v2"
    if v2_root.is_dir():
        for path in v2_root.iterdir():
            if not path.is_dir():
                continue
            try:
                files = [item for item in path.rglob("*") if item.is_file()]
                size_bytes = sum(item.stat().st_size for item in files)
                manifest = path / "index.json"
                last_access = manifest.stat().st_atime if manifest.is_file() else path.stat().st_atime
            except OSError:
                continue
            total_bytes += size_bytes
            entries.append((path.resolve(), size_bytes, last_access, True))

    removed_files = 0
    removed_bytes = 0
    for path, size_bytes, _last_access, is_directory in sorted(entries, key=lambda item: item[2]):
        if total_bytes <= max_bytes:
            break
        if path in protected_paths:
            continue
        try:
            if is_directory:
                shutil.rmtree(path)
            else:
                path.unlink()
        except OSError:
            continue
        total_bytes -= size_bytes
        removed_bytes += size_bytes
        removed_files += 1

    return {
        "remainingBytes": total_bytes,
        "removedBytes": removed_bytes,
        "removedFiles": removed_files,
    }


def write_placement_table(
    path: Path,
    shots: list[dict[str, Any]],
    sources: list[dict[str, Any]],
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    sources_by_id = {source["sourceId"]: source for source in sources}
    columns = (
        "shotId",
        "timelineItemRef",
        "sequenceStartSeconds",
        "sequenceEndSeconds",
        "decision",
        "sourceId",
        "projectItemRef",
        "sourceStartSeconds",
        "sourceEndSeconds",
        "normalizedScore",
        "runnerUpSeparation",
        "placementSourceStartSeconds",
        "placementSourceEndSeconds",
    )
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, delimiter="\t", lineterminator="\n")
        writer.writeheader()
        for shot in shots:
            candidates = shot.get("candidates") or []
            best = candidates[0] if candidates else {}
            source = sources_by_id.get(best.get("sourceId", ""))
            source_start: float | str = best.get("sourceStartSeconds", "")
            source_end: float | str = best.get("sourceEndSeconds", "")
            placement_start = source_start
            placement_end = source_end
            if source is not None and placement_start != "" and placement_end != "":
                usable_range = source["usableRange"]
                bounds = (
                    float(usable_range["startSeconds"]),
                    float(usable_range["endSeconds"]),
                )
                placement_start, placement_end = placement_safe_range_seconds(
                    float(placement_start),
                    float(placement_end),
                    bounds,
                )
            writer.writerow(
                {
                    "shotId": shot["shotId"],
                    "timelineItemRef": shot.get("timelineItemRef", ""),
                    "sequenceStartSeconds": shot["sequenceStartSeconds"],
                    "sequenceEndSeconds": shot["sequenceEndSeconds"],
                    "decision": shot["decision"],
                    "sourceId": best.get("sourceId", ""),
                    "projectItemRef": source.get("projectItemRef", "") if source is not None else "",
                    "sourceStartSeconds": source_start,
                    "sourceEndSeconds": source_end,
                    "normalizedScore": best.get("normalizedScore", ""),
                    "runnerUpSeparation": shot.get("runnerUpSeparation", ""),
                    "placementSourceStartSeconds": placement_start,
                    "placementSourceEndSeconds": placement_end,
                }
            )


def alternative_candidate_rows(
    shots: list[dict[str, Any]],
    sources: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    sources_by_id = {str(source["sourceId"]): source for source in sources}
    rows: list[dict[str, Any]] = []
    for shot in shots:
        if shot.get("decision") == "AUTO_CONFIRMED":
            continue
        duration = float(shot["durationSeconds"])
        distinct_spacing = max(0.5, duration * 0.5)
        accepted: list[tuple[str, float]] = []
        for winner in shot.get("candidates") or []:
            if int(winner.get("rank", 0)) != 1:
                continue
            try:
                winner_start = float(winner.get("sourceStartSeconds", ""))
            except (TypeError, ValueError):
                break
            winner_source_id = str(winner.get("sourceId", ""))
            if winner_source_id and math.isfinite(winner_start):
                accepted.append((winner_source_id, winner_start))
            break
        for candidate in shot.get("candidates") or []:
            rank = int(candidate.get("rank", 0))
            if rank <= 1:
                continue
            source_id = str(candidate.get("sourceId", ""))
            source = sources_by_id.get(source_id)
            source_start = candidate.get("sourceStartSeconds", "")
            source_end = candidate.get("sourceEndSeconds", "")
            eligible = True
            rejection_reason = ""
            try:
                source_start_number = float(source_start)
                source_end_number = float(source_end)
            except (TypeError, ValueError):
                source_start_number = 0.0
                source_end_number = 0.0
                eligible = False
                rejection_reason = "INVALID_SOURCE_RANGE"
            if (
                eligible
                and (
                    not math.isfinite(source_start_number)
                    or not math.isfinite(source_end_number)
                    or source_end_number <= source_start_number
                )
            ):
                eligible = False
                rejection_reason = "INVALID_SOURCE_RANGE"
            if (
                eligible
                and abs((source_end_number - source_start_number) - duration) > 1e-6
            ):
                eligible = False
                rejection_reason = "SOURCE_DURATION_MISMATCH"
            project_item_ref = ""
            if source is None:
                eligible = False
                rejection_reason = "UNKNOWN_SOURCE"
            else:
                project_item_ref = str(source.get("projectItemRef", ""))
                if not project_item_ref:
                    eligible = False
                    rejection_reason = "MISSING_PROJECT_ITEM_REFERENCE"
            if eligible and source is not None:
                usable_range = source["usableRange"]
                usable_start = float(usable_range["startSeconds"])
                usable_end = float(usable_range["endSeconds"])
                tolerance = 1e-9
                if (
                    source_start_number < usable_start - tolerance
                    or source_end_number > usable_end + tolerance
                ):
                    eligible = False
                    rejection_reason = "SOURCE_RANGE_OUTSIDE_USABLE_MEDIA"
            if eligible and any(
                prior_source_id == source_id
                and abs(prior_start - source_start_number) < distinct_spacing
                for prior_source_id, prior_start in accepted
            ):
                eligible = False
                rejection_reason = "REDUNDANT_SOURCE_TIME"
            placement_start: float | str = source_start
            placement_end: float | str = source_end
            if eligible and source is not None:
                bounds = (
                    float(source["usableRange"]["startSeconds"]),
                    float(source["usableRange"]["endSeconds"]),
                )
                placement_start, placement_end = placement_safe_range_seconds(
                    source_start_number,
                    source_end_number,
                    bounds,
                )
                accepted.append((source_id, source_start_number))
            rows.append(
                {
                    "shotId": shot["shotId"],
                    "rank": rank,
                    "sequenceStartSeconds": shot["sequenceStartSeconds"],
                    "sequenceEndSeconds": shot["sequenceEndSeconds"],
                    "decision": shot["decision"],
                    "sourceId": source_id,
                    "projectItemRef": project_item_ref,
                    "sourceStartSeconds": source_start,
                    "sourceEndSeconds": source_end,
                    "normalizedScore": candidate.get("normalizedScore", ""),
                    "placementSourceStartSeconds": placement_start,
                    "placementSourceEndSeconds": placement_end,
                    "eligible": "true" if eligible else "false",
                    "rejectionReason": rejection_reason,
                }
            )
    return rows


def write_alternative_table(
    path: Path,
    shots: list[dict[str, Any]],
    sources: list[dict[str, Any]],
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    columns = (
        "shotId",
        "rank",
        "sequenceStartSeconds",
        "sequenceEndSeconds",
        "decision",
        "sourceId",
        "projectItemRef",
        "sourceStartSeconds",
        "sourceEndSeconds",
        "normalizedScore",
        "placementSourceStartSeconds",
        "placementSourceEndSeconds",
        "eligible",
        "rejectionReason",
    )
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, delimiter="\t", lineterminator="\n")
        writer.writeheader()
        writer.writerows(alternative_candidate_rows(shots, sources))


def source_index_context(
    manifest_path: Path,
    source: dict[str, Any],
    work_dir: Path,
    interval: float,
    cache_dir: Path | None,
    chunk_seconds: float,
) -> dict[str, Any]:
    media = resolve_media_path(manifest_path, source["mediaPath"])
    cache_metadata = source_cache_metadata(source, interval, chunk_seconds)
    chunks = source_chunk_plan(source, interval, chunk_seconds)
    persistent_cache_path = source_cache_path(cache_dir, cache_metadata) if cache_dir is not None else None
    index_path = (
        persistent_cache_path
        if persistent_cache_path is not None
        else work_dir / "ephemeral-source-index-v2" / cache_metadata_key(cache_metadata)
    )
    return {
        "media": media,
        "cacheMetadata": cache_metadata,
        "chunks": chunks,
        "persistentCachePath": persistent_cache_path,
        "indexPath": index_path,
    }


def plan_source_index(
    manifest_path: Path,
    source: dict[str, Any],
    work_dir: Path,
    interval: float,
    cache_dir: Path | None,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
) -> dict[str, Any]:
    context = source_index_context(
        manifest_path,
        source,
        work_dir,
        interval,
        cache_dir,
        chunk_seconds,
    )
    cache_metadata = context["cacheMetadata"]
    chunks = context["chunks"]
    persistent_cache_path = context["persistentCachePath"]
    index_path = context["indexPath"]
    cache_complete, cache_reason = (
        load_source_cache(index_path, cache_metadata, chunks)
        if persistent_cache_path is not None
        else (False, "disabled")
    )
    if persistent_cache_path is not None and not cache_complete:
        progress(
            "cache-rebuild",
            sourceId=source["sourceId"],
            reason=cache_reason,
            cachePath=persistent_cache_path,
        )

    index_path.mkdir(parents=True, exist_ok=True)
    if cache_complete:
        return {
            **context,
            "cacheComplete": True,
            "cacheReason": cache_reason,
            "completedChunks": [int(chunk["chunkIndex"]) for chunk in chunks],
            "missingChunks": [],
        }

    completed_chunks: list[int] = []
    missing_chunks: list[int] = []
    for chunk in chunks:
        chunk_index = int(chunk["chunkIndex"])
        entry, _chunk_reason = load_source_chunk(
            index_path / str(chunk["fileName"]),
            cache_metadata,
            source,
            interval,
            chunk,
        )
        if entry is None:
            missing_chunks.append(chunk_index)
        else:
            completed_chunks.append(chunk_index)
            del entry
    atomic_write_json(
        index_path / "index.json",
        source_cache_manifest(cache_metadata, chunks, completed_chunks, index_path),
    )
    return {
        **context,
        "cacheComplete": False,
        "cacheReason": cache_reason,
        "completedChunks": completed_chunks,
        "missingChunks": missing_chunks,
    }


def build_source_index_chunk(
    manifest_path: Path,
    source: dict[str, Any],
    ffmpeg: str,
    work_dir: Path,
    interval: float,
    timeout_seconds: float,
    cache_dir: Path | None,
    chunk_index: int,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
) -> dict[str, Any]:
    context = source_index_context(
        manifest_path,
        source,
        work_dir,
        interval,
        cache_dir,
        chunk_seconds,
    )
    chunks = context["chunks"]
    if chunk_index < 0 or chunk_index >= len(chunks):
        raise AutoConformFailure(
            "source-index",
            "INVALID_CHUNK_INDEX",
            f"chunk index {chunk_index} is outside the source plan",
            {"sourceId": source["sourceId"], "chunkIndex": chunk_index, "chunkCount": len(chunks)},
        )
    chunk = chunks[chunk_index]
    cache_metadata = context["cacheMetadata"]
    index_path = context["indexPath"]
    index_path.mkdir(parents=True, exist_ok=True)
    chunk_path = index_path / str(chunk["fileName"])
    entry, chunk_reason = load_source_chunk(
        chunk_path,
        cache_metadata,
        source,
        interval,
        chunk,
    )
    if entry is None:
        progress(
            "index-chunk",
            sourceId=source["sourceId"],
            chunk=chunk_index + 1,
            chunks=len(chunks),
            reason=chunk_reason,
        )
        images, times = decode_series(
            ffmpeg,
            context["media"],
            float(chunk["startSeconds"]),
            float(chunk["endSeconds"]),
            1.0 / interval,
            work_dir / "index" / str(source["sourceId"]) / f"chunk-{chunk_index + 1:06d}",
            "sample",
            COARSE_INDEX_MAX_DIM,
            timeout_seconds,
            stage="source-index",
            details={
                "sourceId": source["sourceId"],
                "chunkIndex": chunk_index,
                "chunkCount": len(chunks),
                "sourceDurationSeconds": (
                    float(source["usableRange"]["endSeconds"])
                    - float(source["usableRange"]["startSeconds"])
                ),
            },
        )
        try:
            planned_chunk_prefix_length(source, interval, chunk, times)
        except ValueError as exc:
            raise AutoConformFailure(
                "source-index",
                "INVALID_CHUNK_OUTPUT",
                str(exc),
                {
                    "sourceId": source["sourceId"],
                    "chunkIndex": chunk_index,
                    "chunkStartSeconds": chunk["startSeconds"],
                    "chunkEndSeconds": chunk["endSeconds"],
                    "observedSampleCount": len(times),
                    "expectedSampleCount": int(chunk["expectedSampleCount"]),
                },
            ) from exc
        vectors = (
            np.stack([descriptor_variants(image) for image in images])
            if images
            else np.empty((0, 8, 3328), dtype=np.float32)
        )
        try:
            validate_chunk_arrays(source, interval, chunk, times, vectors)
        except ValueError as exc:
            raise AutoConformFailure(
                "source-index",
                "INVALID_CHUNK_OUTPUT",
                str(exc),
                {
                    "sourceId": source["sourceId"],
                    "chunkIndex": chunk_index,
                    "chunkStartSeconds": chunk["startSeconds"],
                    "chunkEndSeconds": chunk["endSeconds"],
                },
            ) from exc
        entry = (times, vectors)
        save_source_chunk(chunk_path, cache_metadata, chunk, times, vectors)

    manifest_path_on_disk = index_path / "index.json"
    completed_chunks: set[int] = set()
    try:
        manifest = json.loads(manifest_path_on_disk.read_text(encoding="utf-8"))
        if manifest.get("cacheMetadata") == cache_metadata and manifest.get("chunks") == chunks:
            completed_chunks.update(int(value) for value in manifest.get("completedChunks", []))
    except (OSError, ValueError, TypeError, json.JSONDecodeError):
        pass
    completed_chunks.add(chunk_index)
    atomic_write_json(
        manifest_path_on_disk,
        source_cache_manifest(cache_metadata, chunks, sorted(completed_chunks), index_path),
    )
    progress(
        "index-chunk-complete",
        sourceId=source["sourceId"],
        chunk=chunk_index + 1,
        chunks=len(chunks),
        cacheHit=str(chunk_reason == "hit").lower(),
        samples=int(chunk["expectedSampleCount"]),
    )
    del entry
    return {**context, "chunk": chunk, "cacheHit": chunk_reason == "hit"}


def finalize_source_index(
    manifest_path: Path,
    source: dict[str, Any],
    work_dir: Path,
    interval: float,
    cache_dir: Path | None,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
    cache_hit: bool = False,
) -> dict[str, Any]:
    context = source_index_context(
        manifest_path,
        source,
        work_dir,
        interval,
        cache_dir,
        chunk_seconds,
    )
    cache_metadata = context["cacheMetadata"]
    chunks = context["chunks"]
    index_path = context["indexPath"]
    persistent_cache_path = context["persistentCachePath"]
    completed_chunks = [int(chunk["chunkIndex"]) for chunk in chunks]
    if cache_hit:
        complete, reason = load_source_cache(index_path, cache_metadata, chunks)
        if not complete:
            raise AutoConformFailure(
                "source-index",
                "INCOMPLETE_SOURCE_INDEX",
                "source index changed after planning and can no longer be finalized",
                {
                    "sourceId": source["sourceId"],
                    "reason": reason,
                },
            )
    else:
        for chunk in chunks:
            entry, reason = load_source_chunk(
                index_path / str(chunk["fileName"]),
                cache_metadata,
                source,
                interval,
                chunk,
            )
            if entry is None:
                raise AutoConformFailure(
                    "source-index",
                    "INCOMPLETE_SOURCE_INDEX",
                    "source index cannot be finalized because a planned chunk is unavailable",
                    {
                        "sourceId": source["sourceId"],
                        "chunkIndex": int(chunk["chunkIndex"]),
                        "reason": reason,
                    },
                )
            del entry
        atomic_write_json(
            index_path / "index.json",
            source_cache_manifest(cache_metadata, chunks, completed_chunks, index_path),
        )

    times = np.concatenate(
        [expected_chunk_times(source, interval, chunk) for chunk in chunks]
    )
    legacy_removed = False
    if cache_dir is not None and persistent_cache_path is not None:
        legacy_path = legacy_source_cache_path(cache_dir, source, interval)
        if legacy_path.is_file():
            try:
                legacy_path.unlink()
                legacy_removed = True
                progress("cache-legacy-removed", sourceId=source["sourceId"], path=legacy_path)
            except OSError:
                pass

    return {
        "source": source,
        "media": context["media"],
        "times": times,
        "cacheMetadata": cache_metadata,
        "chunks": chunks,
        "indexPath": index_path,
        "cacheHit": bool(cache_hit),
        "cachePath": str(persistent_cache_path) if persistent_cache_path is not None else None,
        "legacyCacheRemoved": legacy_removed,
    }


def build_source_index(
    manifest_path: Path,
    source: dict[str, Any],
    ffmpeg: str,
    work_dir: Path,
    interval: float,
    timeout_seconds: float,
    cache_dir: Path | None,
    chunk_seconds: float = DEFAULT_INDEX_CHUNK_SECONDS,
) -> dict[str, Any]:
    plan = plan_source_index(
        manifest_path,
        source,
        work_dir,
        interval,
        cache_dir,
        chunk_seconds,
    )
    if plan["cacheComplete"]:
        for chunk_number, chunk in enumerate(plan["chunks"], start=1):
            progress(
                "index-chunk-complete",
                sourceId=source["sourceId"],
                chunk=chunk_number,
                chunks=len(plan["chunks"]),
                cacheHit="true",
                samples=int(chunk["expectedSampleCount"]),
            )
    else:
        for chunk_index in plan["missingChunks"]:
            build_source_index_chunk(
                manifest_path,
                source,
                ffmpeg,
                work_dir,
                interval,
                timeout_seconds,
                cache_dir,
                int(chunk_index),
                chunk_seconds,
            )
    return finalize_source_index(
        manifest_path,
        source,
        work_dir,
        interval,
        cache_dir,
        chunk_seconds,
        cache_hit=bool(plan["cacheComplete"]),
    )


def distinct_candidates(raw: list[tuple[float, str, int]], count: int, spacing: float, indexes: dict[str, Any]) -> list[tuple[float, str, int]]:
    chosen: list[tuple[float, str, int]] = []
    for score, source_id, sample_index in sorted(raw, reverse=True):
        source_time = float(indexes[source_id]["times"][sample_index])
        if all(
            prior_source != source_id
            or abs(float(indexes[prior_source]["times"][prior_index]) - source_time) >= spacing
            for _, prior_source, prior_index in chosen
        ):
            chosen.append((score, source_id, sample_index))
        if len(chosen) >= count:
            break
    return chosen


def invalidate_source_chunk(
    index: dict[str, Any],
    chunk: dict[str, Any],
    reason: str,
) -> None:
    index_path = Path(index["indexPath"])
    chunk_path = index_path / str(chunk["fileName"])
    chunk_path.unlink(missing_ok=True)
    completed_chunks = [
        int(candidate["chunkIndex"])
        for candidate in index["chunks"]
        if candidate != chunk
        and (index_path / str(candidate["fileName"])).is_file()
    ]
    atomic_write_json(
        index_path / "index.json",
        source_cache_manifest(
            index["cacheMetadata"],
            index["chunks"],
            completed_chunks,
            index_path,
        ),
    )
    progress(
        "cache-chunk-invalidated",
        sourceId=index["source"]["sourceId"],
        chunk=int(chunk["chunkIndex"]),
        reason=reason,
    )


def require_source_chunk(
    index: dict[str, Any],
    chunk: dict[str, Any],
    stage: str,
) -> tuple[np.ndarray, np.ndarray]:
    entry, reason = load_source_chunk(
        Path(index["indexPath"]) / str(chunk["fileName"]),
        index["cacheMetadata"],
        index["source"],
        float(index["cacheMetadata"]["samplingIntervalSeconds"]),
        chunk,
    )
    if entry is None:
        invalidation_error = ""
        try:
            invalidate_source_chunk(index, chunk, reason)
        except OSError as exc:
            invalidation_error = str(exc)
        raise AutoConformFailure(
            stage,
            "SOURCE_INDEX_CHUNK_UNAVAILABLE",
            f"A validated source-index chunk became unavailable ({reason}).",
            {
                "sourceId": index["source"]["sourceId"],
                "chunkIndex": chunk["chunkIndex"],
                "reason": reason,
                "invalidationError": invalidation_error,
            },
        )
    return entry


def load_index_vectors(
    index: dict[str, Any],
    sample_indices: set[int],
) -> dict[int, np.ndarray]:
    remaining = set(sample_indices)
    selected: dict[int, np.ndarray] = {}
    for chunk in index["chunks"]:
        first = int(chunk["firstSampleIndex"])
        end = int(chunk["endSampleIndexExclusive"])
        wanted = sorted(value for value in remaining if first <= value < end)
        if not wanted:
            continue
        _times, vectors = require_source_chunk(index, chunk, "verification")
        for sample_index in wanted:
            selected[sample_index] = np.array(vectors[sample_index - first], copy=True)
            remaining.remove(sample_index)
        del vectors
        if not remaining:
            break
    if remaining:
        raise AutoConformFailure(
            "verification",
            "SOURCE_INDEX_SAMPLES_MISSING",
            "Required source-index samples were not found in the chunk plan.",
            {
                "sourceId": index["source"]["sourceId"],
                "sampleIndices": sorted(remaining),
            },
        )
    return selected


def prepare_shot_query(
    reference_media: Path,
    reference_frame_rate: float,
    shot: dict[str, Any],
    args: argparse.Namespace,
    work_dir: Path,
    diagnostics: dict[str, Any],
) -> dict[str, Any]:
    anchors = shot_anchor_times(shot)
    shot["sampleTimesSeconds"] = anchors
    reference_images = export_exact_frames_grouped(
        args.ffmpeg,
        reference_media,
        [
            (
                when,
                work_dir / "reference" / f"{shot['shotId']}_{index}.png",
                {
                    "shotId": str(shot["shotId"]),
                    "anchorIndex": index,
                },
            )
            for index, when in enumerate(anchors)
        ],
        720,
        args.ffmpeg_timeout_seconds,
        diagnostics=diagnostics,
        role="REFERENCE_ANCHOR",
        media_kind="REFERENCE",
        stage="reference-frame-export",
        expected_frame_rate=reference_frame_rate,
    )
    reference_vectors = [
        measured_call(
            diagnostics,
            "referenceDescriptorSeconds",
            lambda image=image: descriptor(image),
        )
        for image in reference_images
    ]
    center_index = len(anchors) // 2
    center_offset = anchors[center_index] - float(shot["referenceStartSeconds"])
    return {
        "shot": shot,
        "anchors": anchors,
        "referenceImages": reference_images,
        "referenceVectors": reference_vectors,
        "centerIndex": center_index,
        "centerOffset": center_offset,
    }


def retrieve_shot_batch(
    queries: list[dict[str, Any]],
    indexes: dict[str, Any],
    args: argparse.Namespace,
) -> list[list[tuple[float, str, int]]]:
    if not queries:
        return []
    scores_by_source: dict[str, np.ndarray] = {}
    for source_id, index in indexes.items():
        source_scores = np.empty(
            (len(queries), len(index["times"])),
            dtype=np.float32,
        )
        for chunk in index["chunks"]:
            chunk_times, vectors = require_source_chunk(index, chunk, "retrieval")
            first_sample = int(chunk["firstSampleIndex"])
            end_sample = int(chunk["endSampleIndexExclusive"])
            for query_index, query in enumerate(queries):
                reference_vector = query["referenceVectors"][query["centerIndex"]]
                correlations = np.max(vectors @ reference_vector, axis=1)
                source_scores[query_index, first_sample:end_sample] = np.clip(
                    (correlations + 1.0) * 0.5,
                    0.0,
                    1.0,
                )
            del chunk_times, vectors
        scores_by_source[source_id] = source_scores

    shortlists: list[list[tuple[float, str, int]]] = []
    for query_index, query in enumerate(queries):
        raw: list[tuple[float, str, int]] = []
        for source_id, index in indexes.items():
            raw.extend(
                (float(score), source_id, sample_index)
                for sample_index, score in enumerate(scores_by_source[source_id][query_index])
            )
        shortlists.append(
            distinct_candidates(
                raw,
                args.shortlist,
                max(0.5, float(query["shot"]["durationSeconds"]) * 0.5),
                indexes,
            )
        )
    return shortlists


def analyze_prepared_shot(
    shot: dict[str, Any],
    query: dict[str, Any],
    shortlist: list[tuple[float, str, int]],
    indexes: dict[str, Any],
    args: argparse.Namespace,
    work_dir: Path,
    diagnostics: dict[str, Any],
) -> tuple[dict[str, Any], float, float]:
    anchors = query["anchors"]
    reference_images = query["referenceImages"]
    reference_vectors = query["referenceVectors"]
    center_index = int(query["centerIndex"])
    center_offset = float(query["centerOffset"])

    verification_started = time.perf_counter()
    pending_verification: list[dict[str, Any]] = []
    required_samples: dict[str, set[int]] = {}
    for coarse_score, source_id, sample_index in shortlist:
        index = indexes[source_id]
        start_bounds = source_start_bounds(
            float(index["source"]["usableRange"]["startSeconds"]),
            float(index["source"]["usableRange"]["endSeconds"]),
            float(shot["durationSeconds"]),
        )
        if start_bounds is None:
            continue
        center_time = float(index["times"][sample_index])
        candidate_start = clamp_candidate_start(center_time - center_offset, start_bounds)
        sample_indices: list[int] = []
        sample_distances: list[float] = []
        for anchor_time, reference_vector in zip(anchors, reference_vectors):
            source_time = candidate_start + (anchor_time - float(shot["referenceStartSeconds"]))
            nearest = int(np.argmin(np.abs(index["times"] - source_time)))
            distance = abs(float(index["times"][nearest]) - source_time)
            sample_indices.append(nearest)
            sample_distances.append(distance)
            if distance <= args.sample_interval_seconds * 0.75:
                required_samples.setdefault(source_id, set()).add(nearest)
        pending_verification.append(
            {
                "sourceId": source_id,
                "candidateStart": candidate_start,
                "coarseScore": coarse_score,
                "sampleIndices": sample_indices,
                "sampleDistances": sample_distances,
            }
        )

    selected_vectors = {
        source_id: load_index_vectors(indexes[source_id], sample_indices)
        for source_id, sample_indices in required_samples.items()
    }
    verified: list[dict[str, Any]] = []
    window_descriptor_variant_by_candidate_identity: dict[int, int] = {}
    for candidate in pending_verification:
        sample_scores: list[float] = []
        window_descriptor_variant_indices: list[int] = []
        window_descriptor_variant_separations: list[float] = []
        source_id = str(candidate["sourceId"])
        for (
            reference_vector,
            sample_index,
            distance,
        ) in zip(
            reference_vectors,
            candidate.pop("sampleIndices"),
            candidate.pop("sampleDistances"),
        ):
            if distance > args.sample_interval_seconds * 0.75:
                sample_scores.append(0.0)
                continue
            source_variants = selected_vectors[source_id][sample_index]
            if args.window_descriptor_mode == "coarse-consensus":
                (
                    sample_score,
                    variant_index,
                    variant_separation,
                ) = descriptor_score_with_variant_evidence(
                    reference_vector,
                    source_variants,
                )
                sample_scores.append(sample_score)
                window_descriptor_variant_indices.append(variant_index)
                window_descriptor_variant_separations.append(variant_separation)
            else:
                sample_scores.append(
                    descriptor_score(reference_vector, source_variants)
                )
        consensus_variant = (
            consensus_descriptor_variant(
                window_descriptor_variant_indices,
                window_descriptor_variant_separations,
                args.window_descriptor_min_correlation_separation,
            )
            if (
                args.window_descriptor_mode == "coarse-consensus"
                and len(window_descriptor_variant_indices)
                == len(reference_vectors)
            )
            else None
        )
        if consensus_variant is not None:
            window_descriptor_variant_by_candidate_identity[id(candidate)] = consensus_variant
        temporal = float(np.median(sample_scores)) * 0.8 + min(sample_scores) * 0.2
        candidate["temporal"] = temporal
        candidate["sampleScores"] = sample_scores
        verified.append(candidate)
    verification_seconds = time.perf_counter() - verification_started

    refinement_started = time.perf_counter()
    reference_feature_cache: dict[
        int,
        tuple[np.ndarray, tuple[Any, Any, int] | None],
    ] = {}
    reference_pixel_identity_cache: dict[
        int,
        tuple[np.ndarray, tuple[np.ndarray, float]],
    ] = {}
    reference_motion_descriptor_cache: dict[int, np.ndarray] = {}
    refinement_contexts: list[dict[str, Any]] = []
    for candidate_index, candidate in enumerate(
        sorted(verified, key=lambda item: item["temporal"], reverse=True)[: args.top_k]
    ):
        window_descriptor_variant_index = (
            window_descriptor_variant_by_candidate_identity.get(id(candidate))
        )
        source_index = indexes[candidate["sourceId"]]
        start_bounds = source_start_bounds(
            float(source_index["source"]["usableRange"]["startSeconds"]),
            float(source_index["source"]["usableRange"]["endSeconds"]),
            float(shot["durationSeconds"]),
        )
        if start_bounds is None:
            continue
        center_bounds = (start_bounds[0] + center_offset, start_bounds[1] + center_offset)
        center_time = candidate["candidateStart"] + center_offset
        source_fps = float(source_index["source"]["frameRate"])
        window_start = max(center_bounds[0], center_time - args.sample_interval_seconds)
        window_end = min(center_bounds[1], center_time + args.sample_interval_seconds)
        refinement_contexts.append(
            {
                "candidateIndex": candidate_index,
                "candidate": candidate,
                "sourceIndex": source_index,
                "startBounds": start_bounds,
                "centerBounds": center_bounds,
                "centerTime": center_time,
                "sourceFps": source_fps,
                "windowStart": window_start,
                "windowEnd": window_end,
                "fixed": window_end <= window_start + 1e-9,
                "windowDescriptorVariantIndex": window_descriptor_variant_index,
            }
        )

    def decoded_refinement_contexts() -> Iterator[dict[str, Any]]:
        for group_start in range(
            0,
            len(refinement_contexts),
            args.refinement_window_group_size,
        ):
            group = refinement_contexts[
                group_start : group_start + args.refinement_window_group_size
            ]
            grouped_contexts = [
                context for context in group if not bool(context["fixed"])
            ]
            if grouped_contexts:
                decoded = decode_series_grouped_merged(
                    args.ffmpeg,
                    [
                        (
                            context["sourceIndex"]["media"],
                            float(context["windowStart"]),
                            float(context["windowEnd"]),
                            float(context["sourceFps"]),
                            work_dir / "refine",
                            f"{shot['shotId']}_{context['candidateIndex']}",
                            {
                                "shotId": shot["shotId"],
                                "sourceId": context["candidate"]["sourceId"],
                                "candidateIndex": context["candidateIndex"],
                            },
                            "ORIGINAL_SOURCE",
                        )
                        for context in grouped_contexts
                    ],
                    720,
                    args.ffmpeg_timeout_seconds,
                    diagnostics=diagnostics,
                    stage="shot-refinement",
                )
                for context, (frames, frame_times) in zip(grouped_contexts, decoded):
                    context["frames"] = frames
                    context["frameTimes"] = frame_times
                del decoded
            yield from group

    for context in decoded_refinement_contexts():
        candidate_index = int(context["candidateIndex"])
        candidate = context["candidate"]
        source_index = context["sourceIndex"]
        start_bounds = context["startBounds"]
        center_bounds = context["centerBounds"]
        center_time = float(context["centerTime"])
        source_fps = float(context["sourceFps"])
        if bool(context["fixed"]):
            frames = [
                export_frame(
                    args.ffmpeg,
                    source_index["media"],
                    center_time,
                    work_dir / "refine" / f"{shot['shotId']}_{candidate_index}_fixed.png",
                    720,
                    args.ffmpeg_timeout_seconds,
                    diagnostics=diagnostics,
                    role="REFINEMENT_FIXED",
                    media_kind="ORIGINAL_SOURCE",
                    evidence={
                        "shotId": str(shot["shotId"]),
                        "sourceId": str(candidate["sourceId"]),
                        "candidateIndex": candidate_index,
                    },
                    stage="shot-refinement-fixed-export",
                    expected_frame_rate=source_fps,
                )
            ]
            frame_times = np.asarray([center_time], dtype=np.float64)
        else:
            frames = context.pop("frames")
            frame_times = context.pop("frameTimes")
        window_descriptor_variant_index = context["windowDescriptorVariantIndex"]
        if (
            args.window_descriptor_mode == "coarse-consensus"
            and window_descriptor_variant_index is not None
        ):
            frame_scores = [
                measured_descriptor_selected_variant_score(
                    diagnostics,
                    "windowDescriptorSeconds",
                    reference_vectors[center_index],
                    frame,
                    int(window_descriptor_variant_index),
                )
                for frame in frames
            ]
        else:
            frame_scores = [
                measured_descriptor_variant_score(
                    diagnostics,
                    "windowDescriptorSeconds",
                    reference_vectors[center_index],
                    frame,
                )
                for frame in frames
            ]
        if frame_scores:
            best_frame = int(np.argmax(frame_scores))
            refined_center = min(
                max(float(frame_times[best_frame]), center_bounds[0]),
                center_bounds[1],
            )
            approximate_start = clamp_candidate_start(
                refined_center - center_offset,
                start_bounds,
            )
            aligned_options: list[tuple[float, float, np.ndarray, float]] = []
            aligned_starts = frame_aligned_candidate_starts(
                approximate_start,
                start_bounds,
                source_fps,
            )
            aligned_images = export_alignment_frames_grouped(
                args.ffmpeg,
                source_index["media"],
                [
                    (
                        aligned_start + center_offset,
                        work_dir
                        / "aligned"
                        / f"{shot['shotId']}_{candidate_index}_{local_index}.png",
                        {
                            "shotId": str(shot["shotId"]),
                            "sourceId": str(candidate["sourceId"]),
                            "candidateIndex": candidate_index,
                            "alignmentIndex": local_index,
                        },
                    )
                    for local_index, aligned_start in enumerate(aligned_starts)
                ],
                720,
                args.ffmpeg_timeout_seconds,
                diagnostics=diagnostics,
                media_kind="ORIGINAL_SOURCE",
                stage="shot-alignment-export",
                expected_frame_rate=source_fps,
            )
            for aligned_start, aligned_center_image in zip(
                aligned_starts,
                aligned_images,
            ):
                aligned_descriptor = measured_descriptor_variant_score(
                    diagnostics,
                    "alignmentDescriptorSeconds",
                    reference_vectors[center_index],
                    aligned_center_image,
                )
                add_diagnostic_count(diagnostics, "pixelIdentityCalls")
                aligned_identity = pixel_identity_score_prepared(
                    get_pixel_identity_reference(
                        reference_pixel_identity_cache,
                        reference_images[center_index],
                        diagnostics,
                    ),
                    aligned_center_image,
                    diagnostics,
                )
                add_diagnostic_count(diagnostics, "featureCalls")
                aligned_feature = feature_score(
                    get_reference_features(
                        reference_feature_cache,
                        reference_images[center_index],
                        diagnostics,
                    ),
                    aligned_center_image,
                    diagnostics,
                )
                aligned_score = (
                    aligned_descriptor * 0.45
                    + aligned_identity * 0.45
                    + aligned_feature * 0.10
                )
                aligned_options.append(
                    (aligned_score, aligned_start, aligned_center_image, aligned_feature)
                )

            candidate["alignmentWinnerMargin"] = alignment_winner_margin(
                [float(option[0]) for option in aligned_options]
            )
            candidate["alignmentOptionCount"] = len(aligned_options)
            _, aligned_start, aligned_center_image, aligned_feature = max(
                aligned_options,
                key=lambda option: option[0],
            )
            candidate["candidateStart"] = aligned_start
            non_center_anchor_indices = [
                anchor_index
                for anchor_index in range(len(anchors))
                if anchor_index != center_index
            ]
            non_center_anchor_images = (
                export_exact_frames_grouped(
                    args.ffmpeg,
                    source_index["media"],
                    [
                        (
                            candidate["candidateStart"]
                            + (
                                anchors[anchor_index]
                                - float(shot["referenceStartSeconds"])
                            ),
                            work_dir
                            / "exact"
                            / f"{shot['shotId']}_{candidate_index}_{anchor_index}.png",
                            {
                                "shotId": str(shot["shotId"]),
                                "sourceId": str(candidate["sourceId"]),
                                "candidateIndex": candidate_index,
                                "anchorIndex": anchor_index,
                            },
                        )
                        for anchor_index in non_center_anchor_indices
                    ],
                    720,
                    args.ffmpeg_timeout_seconds,
                    diagnostics=diagnostics,
                    role="EXACT_ANCHOR",
                    media_kind="ORIGINAL_SOURCE",
                    stage="shot-exact-anchor-export",
                    expected_frame_rate=source_fps,
                )
                if non_center_anchor_indices
                else []
            )
            exact_source_images_by_index = {
                center_index: aligned_center_image,
                **dict(
                    zip(
                        non_center_anchor_indices,
                        non_center_anchor_images,
                    )
                ),
            }
            exact_source_images = [
                exact_source_images_by_index[anchor_index]
                for anchor_index in range(len(anchors))
            ]
            exact_scores: list[float] = []
            identity_scores: list[float] = []
            for reference_vector, reference_image, source_image in zip(
                reference_vectors,
                reference_images,
                exact_source_images,
            ):
                exact_scores.append(
                    measured_descriptor_variant_score(
                        diagnostics,
                        "exactDescriptorSeconds",
                        reference_vector,
                        source_image,
                    )
                )
                add_diagnostic_count(diagnostics, "pixelIdentityCalls")
                identity_scores.append(
                    pixel_identity_score_prepared(
                        get_pixel_identity_reference(
                            reference_pixel_identity_cache,
                            reference_image,
                            diagnostics,
                        ),
                        source_image,
                        diagnostics,
                    )
                )
            motion_scores: list[float] = []
            for index in range(1, len(reference_images)):
                add_diagnostic_count(diagnostics, "motionComparisons")
                reference_motion_descriptor = get_reference_motion_descriptor(
                    reference_motion_descriptor_cache,
                    index,
                    reference_images,
                    diagnostics,
                )
                source_motion_variants = measured_call(
                    diagnostics,
                    "sourceMotionSeconds",
                    lambda index=index: descriptor_variants(
                        motion_image(
                            exact_source_images[index - 1],
                            exact_source_images[index],
                        )
                    ),
                )
                add_diagnostic_count(diagnostics, "descriptorVariantCalls")
                add_diagnostic_count(
                    diagnostics,
                    "descriptorVariantsGenerated",
                    int(source_motion_variants.shape[0]),
                )
                motion_scores.append(
                    measured_call(
                        diagnostics,
                        "motionComparisonSeconds",
                        lambda reference_motion_descriptor=reference_motion_descriptor, source_motion_variants=source_motion_variants: descriptor_score(
                            reference_motion_descriptor,
                            source_motion_variants,
                        ),
                    )
                )
            exact_temporal = float(np.median(exact_scores)) * 0.8 + min(exact_scores) * 0.2
            pixel_identity = float(np.median(identity_scores)) * 0.8 + min(identity_scores) * 0.2
            motion_consistency = float(np.mean(motion_scores)) if motion_scores else exact_temporal
            candidate["sampleScores"] = exact_scores
            candidate["temporal"] = exact_temporal * 0.7 + motion_consistency * 0.3
            candidate["refinement"] = exact_temporal
            candidate["pixelIdentity"] = pixel_identity
            candidate["feature"] = aligned_feature
        else:
            candidate["refinement"] = 0.0
            candidate["feature"] = 0.0
            candidate["pixelIdentity"] = 0.0
            candidate["alignmentWinnerMargin"] = 0.0
            candidate["alignmentOptionCount"] = 0
        candidate["score"] = (
            candidate["temporal"] * 0.40
            + candidate["refinement"] * 0.15
            + candidate["feature"] * 0.10
            + candidate["pixelIdentity"] * 0.25
            + candidate["coarseScore"] * 0.10
        )
    refinement_seconds = time.perf_counter() - refinement_started

    verified = [candidate for candidate in verified if "score" in candidate]
    verified.sort(key=lambda item: item["score"], reverse=True)
    duration = float(shot["durationSeconds"])
    distinct_verified: list[dict[str, Any]] = []
    distinct_spacing = max(0.5, duration * 0.5)
    for candidate in verified:
        if all(
            prior["sourceId"] != candidate["sourceId"]
            or abs(float(prior["candidateStart"]) - float(candidate["candidateStart"])) >= distinct_spacing
            for prior in distinct_verified
        ):
            distinct_verified.append(candidate)
        if len(distinct_verified) >= args.top_k:
            break
    candidates: list[dict[str, Any]] = []
    for rank, candidate in enumerate(distinct_verified, start=1):
        total_samples = len(candidate["sampleScores"])
        supporting = sum(score >= 0.55 for score in candidate["sampleScores"])
        source_start = float(candidate["candidateStart"])
        candidates.append(
            {
                "rank": rank,
                "sourceId": candidate["sourceId"],
                "sourceStartSeconds": source_start,
                "sourceEndSeconds": source_start + duration,
                "normalizedScore": max(0.0, min(1.0, float(candidate.get("score", candidate["temporal"])))),
                "evidence": {
                    "coarseScore": max(0.0, min(1.0, float(candidate["coarseScore"]))),
                    "temporalConsistency": max(0.0, min(1.0, float(candidate["temporal"]))),
                    "featureScore": max(0.0, min(1.0, float(candidate.get("feature", 0.0)))),
                    "pixelIdentityScore": max(0.0, min(1.0, float(candidate.get("pixelIdentity", 0.0)))),
                    "refinementScore": max(0.0, min(1.0, float(candidate.get("refinement", 0.0)))),
                    "alignmentWinnerMargin": max(
                        0.0,
                        min(1.0, float(candidate.get("alignmentWinnerMargin", 0.0))),
                    ),
                    "alignmentOptionCount": int(
                        candidate.get("alignmentOptionCount", 0)
                    ),
                    "supportingSamples": supporting,
                    "totalSamples": total_samples,
                    "sampleScores": [max(0.0, min(1.0, float(score))) for score in candidate["sampleScores"]],
                },
            }
        )
    best_score = candidates[0]["normalizedScore"] if candidates else 0.0
    runner_score = candidates[1]["normalizedScore"] if len(candidates) > 1 else 0.0
    separation = max(0.0, best_score - runner_score)
    support = candidates[0]["evidence"]["supportingSamples"] if candidates else 0
    identity_score = candidates[0]["evidence"].get("pixelIdentityScore", 0.0) if candidates else 0.0
    runner_identity = candidates[1]["evidence"].get("pixelIdentityScore", 0.0) if len(candidates) > 1 else 0.0
    identity_separation = max(0.0, identity_score - runner_identity)
    review_quality = candidates and (
        best_score >= REVIEW_MIN_NORMALIZED_SCORE
        or identity_score >= REVIEW_MIN_PIXEL_IDENTITY_SCORE
    )
    auto_distinct = (
        best_score >= REVIEW_MIN_NORMALIZED_SCORE
        and separation >= AUTO_MIN_NORMALIZED_SEPARATION
    ) or (
        identity_score >= REVIEW_MIN_PIXEL_IDENTITY_SCORE
        and identity_separation >= AUTO_MIN_PIXEL_IDENTITY_SEPARATION
    )
    if review_quality and support >= min(2, len(anchors)) and auto_distinct:
        decision = "AUTO_CONFIRMED"
    elif review_quality:
        decision = "REVIEW_RECOMMENDED"
    else:
        decision = "NO_RELIABLE_MATCH"
    shot.update(
        {
            "analysisStatus": "MATCHED" if candidates else "NO_MATCH",
            "candidates": candidates,
            "runnerUpSeparation": separation,
            "decision": decision,
            "outcome": {"status": "NOT_PROCESSED"},
            "diagnostics": [],
        }
    )
    return shot, verification_seconds, refinement_seconds


def benchmark_summary(shots: list[dict[str, Any]], top_k: int) -> dict[str, int]:
    summary = {
        "matchShots": 0,
        "absentShots": 0,
        "top1Correct": 0,
        "topKCorrect": 0,
        "temporalCorrect": 0,
        "falseAutoConfirms": 0,
        "reviewRecommended": 0,
        "noReliableMatch": 0,
    }
    for shot in shots:
        truth = shot.get("groundTruth") or {}
        candidates = shot.get("candidates") or []
        if shot.get("decision") == "REVIEW_RECOMMENDED":
            summary["reviewRecommended"] += 1
        if shot.get("decision") == "NO_RELIABLE_MATCH":
            summary["noReliableMatch"] += 1
        if truth.get("kind") == "ABSENT":
            summary["absentShots"] += 1
            if shot.get("decision") == "AUTO_CONFIRMED":
                summary["falseAutoConfirms"] += 1
            continue
        if truth.get("kind") != "MATCH":
            continue
        summary["matchShots"] += 1
        expected_source = truth.get("sourceId")
        expected_start = float(truth.get("sourceStartSeconds", -1))
        tolerance = float(truth.get("toleranceSeconds", 1.0 / 30.0))
        if candidates and candidates[0]["sourceId"] == expected_source:
            summary["top1Correct"] += 1
            if abs(float(candidates[0]["sourceStartSeconds"]) - expected_start) <= tolerance:
                summary["temporalCorrect"] += 1
        if any(candidate["sourceId"] == expected_source for candidate in candidates[:top_k]):
            summary["topKCorrect"] += 1
    return summary


def mark_shot_failed(
    shot: dict[str, Any],
    exc: Exception,
    stage: str,
) -> dict[str, Any]:
    shot.update(
        {
            "analysisStatus": "FAILED",
            "candidates": [],
            "decision": "NO_RELIABLE_MATCH",
            "outcome": {"status": "NOT_PROCESSED"},
            "diagnostics": [
                {
                    "stage": stage,
                    "severity": "ERROR",
                    "code": "SHOT_ANALYSIS_FAILED",
                    "message": str(exc)[:2000],
                }
            ],
        }
    )
    return shot


def directory_size(path: Path) -> int:
    return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())


def peak_memory_bytes() -> int:
    if resource is None:
        return 0
    value = int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
    return value if sys.platform == "darwin" else value * 1024


def select_shot_range(
    shots: list[dict[str, Any]],
    start_index: int,
    shot_count: int | None,
) -> list[dict[str, Any]]:
    if start_index < 0:
        raise ValueError("--shot-start-index must be greater than or equal to zero")
    if shot_count is not None and shot_count <= 0:
        raise ValueError("--shot-count must be greater than zero")
    if start_index > len(shots):
        raise ValueError("--shot-start-index is outside the manifest shot list")
    end_index = len(shots) if shot_count is None else start_index + shot_count
    if end_index > len(shots):
        raise ValueError("selected shot range extends beyond the manifest shot list")
    selected = shots[start_index:end_index]
    if not selected and shots:
        raise ValueError("selected shot range is empty")
    return selected


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", required=True)
    parser.add_argument("--schema", required=True)
    parser.add_argument("--ffmpeg", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--work-dir", required=True)
    parser.add_argument("--sample-interval-seconds", type=float, default=0.5)
    parser.add_argument("--shortlist", type=int, default=20)
    parser.add_argument("--top-k", type=int, default=5)
    parser.add_argument(
        "--window-descriptor-mode",
        choices=["all", "coarse-consensus"],
        default="coarse-consensus",
    )
    parser.add_argument(
        "--window-descriptor-min-correlation-separation",
        type=float,
        default=0.05,
    )
    parser.add_argument(
        "--refinement-window-group-size",
        type=int,
        default=MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS,
    )
    parser.add_argument("--ffmpeg-timeout-seconds", type=float, default=600.0)
    parser.add_argument("--index-ffmpeg-timeout-seconds", type=float, default=3600.0)
    parser.add_argument("--index-chunk-seconds", type=float, default=DEFAULT_INDEX_CHUNK_SECONDS)
    parser.add_argument("--keep-work-dir", choices=["true", "false"], default="false")
    parser.add_argument("--cache-dir")
    parser.add_argument("--cache-max-bytes", type=int, default=10 * 1024 * 1024 * 1024)
    parser.add_argument("--placement-table-output")
    parser.add_argument("--shot-start-index", type=int, default=0)
    parser.add_argument("--shot-count", type=int)
    args = parser.parse_args()

    if args.sample_interval_seconds <= 0:
        raise ValueError("--sample-interval-seconds must be greater than zero")
    if (
        not math.isfinite(args.window_descriptor_min_correlation_separation)
        or args.window_descriptor_min_correlation_separation < 0
    ):
        raise ValueError(
            "--window-descriptor-min-correlation-separation must be finite and non-negative"
        )
    if not (
        1
        <= args.refinement_window_group_size
        <= MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS
    ):
        raise ValueError(
            "--refinement-window-group-size must be in the range "
            f"1..{MAX_SUPPORTED_GROUPED_REFINEMENT_WINDOWS}"
        )
    if args.ffmpeg_timeout_seconds <= 0:
        raise ValueError("--ffmpeg-timeout-seconds must be greater than zero")
    if args.index_ffmpeg_timeout_seconds <= 0:
        raise ValueError("--index-ffmpeg-timeout-seconds must be greater than zero")
    if args.index_chunk_seconds <= 0:
        raise ValueError("--index-chunk-seconds must be greater than zero")

    started = time.perf_counter()
    manifest_path = Path(args.manifest).resolve()
    output_path = Path(args.output).resolve()
    work_dir = Path(args.work_dir).resolve()
    cache_dir = Path(args.cache_dir).resolve() if args.cache_dir else None
    placement_table_output = (
        Path(args.placement_table_output).resolve() if args.placement_table_output else None
    )
    schema = load_schema(Path(args.schema).resolve())
    document = json.loads(manifest_path.read_text(encoding="utf-8"))
    validate_contract(document, schema)
    result = copy.deepcopy(document)
    result["documentKind"] = "ANALYSIS_RESULT"
    result["analysisVersion"] = ANALYSIS_VERSION
    work_dir.mkdir(parents=True, exist_ok=True)

    protected_cache_paths: set[Path] = set()
    if cache_dir is not None:
        if args.cache_max_bytes <= 0:
            raise ValueError("--cache-max-bytes must be greater than zero")
        protected_cache_paths = set()
        for source in result["sources"]:
            protected_cache_paths.add(
                source_cache_path(
                    cache_dir,
                    source_cache_metadata(
                        source,
                        args.sample_interval_seconds,
                        args.index_chunk_seconds,
                    ),
                ).resolve()
            )
            protected_cache_paths.add(
                legacy_source_cache_path(cache_dir, source, args.sample_interval_seconds).resolve()
            )
        prune = prune_source_cache(cache_dir, args.cache_max_bytes, protected_cache_paths)
        progress("cache-prune", phase="before-index", **prune)

    indexing_started = time.perf_counter()
    indexes: dict[str, Any] = {}
    for source_number, source in enumerate(result["sources"], start=1):
        progress("index", source=source_number, total=len(result["sources"]), sourceId=source["sourceId"])
        index = build_source_index(
            manifest_path,
            source,
            args.ffmpeg,
            work_dir,
            args.sample_interval_seconds,
            args.index_ffmpeg_timeout_seconds,
            cache_dir,
            args.index_chunk_seconds,
        )
        indexes[source["sourceId"]] = index
        source["index"] = {
            "samplingIntervalSeconds": args.sample_interval_seconds,
            "descriptorVersion": DESCRIPTOR_VERSION,
            "sampleCount": len(index["times"]),
            "cacheHit": bool(index["cacheHit"]),
        }
        if index["cachePath"] is not None:
            source["index"]["cachePath"] = index["cachePath"]
        progress(
            "index-complete",
            source=source_number,
            total=len(result["sources"]),
            sourceId=source["sourceId"],
            cacheHit=str(bool(index["cacheHit"])).lower(),
            samples=len(index["times"]),
        )
    indexing_seconds = time.perf_counter() - indexing_started

    reference_media = resolve_media_path(manifest_path, result["reference"]["mediaPath"])
    reference_frame_rate = float(result["reference"]["frameRate"])
    diagnostics_by_shot_id: dict[str, dict[str, Any]] = {}
    retrieval_seconds = 0.0
    verification_seconds = 0.0
    refinement_seconds = 0.0
    all_shots = result["shots"]
    shots = select_shot_range(all_shots, args.shot_start_index, args.shot_count)
    completed: list[dict[str, Any] | None] = [None] * len(shots)
    batch_size = DEFAULT_RETRIEVAL_SHOT_BATCH_SIZE
    total_batches = max(1, math.ceil(len(shots) / batch_size))
    for batch_number, batch_start in enumerate(range(0, len(shots), batch_size), start=1):
        batch_end = min(len(shots), batch_start + batch_size)
        progress(
            "retrieval-batch",
            batch=batch_number,
            batches=total_batches,
            firstShot=args.shot_start_index + batch_start + 1,
            lastShot=args.shot_start_index + batch_end,
        )
        retrieval_started = time.perf_counter()
        prepared: list[tuple[int, dict[str, Any]]] = []
        for shot_index in range(batch_start, batch_end):
            shot = shots[shot_index]
            shot_diagnostics = new_analysis_diagnostics()
            progress(
                "match",
                shot=args.shot_start_index + shot_index + 1,
                total=len(all_shots),
                shotId=shot["shotId"],
            )
            try:
                prepared.append(
                    (
                        shot_index,
                        prepare_shot_query(
                            reference_media,
                            reference_frame_rate,
                            shot,
                            args,
                            work_dir,
                            shot_diagnostics,
                        ),
                    )
                )
                prepared[-1][1]["analysisDiagnostics"] = shot_diagnostics
            except Exception as exc:
                completed[shot_index] = mark_shot_failed(shot, exc, "RETRIEVAL")
                diagnostics_by_shot_id[str(shot["shotId"])] = shot_diagnostics
                progress(
                    "match-complete",
                    shot=args.shot_start_index + shot_index + 1,
                    total=len(all_shots),
                    shotId=shot["shotId"],
                    decision=completed[shot_index]["decision"],
                )

        try:
            shortlists = retrieve_shot_batch(
                [query for _shot_index, query in prepared],
                indexes,
                args,
            )
        except Exception as exc:
            retrieval_seconds += time.perf_counter() - retrieval_started
            for shot_index, query in prepared:
                shot = query["shot"]
                completed[shot_index] = mark_shot_failed(shot, exc, "RETRIEVAL")
                diagnostics_by_shot_id[str(shot["shotId"])] = query["analysisDiagnostics"]
                progress(
                    "match-complete",
                    shot=args.shot_start_index + shot_index + 1,
                    total=len(all_shots),
                    shotId=shot["shotId"],
                    decision=completed[shot_index]["decision"],
                )
            continue
        retrieval_seconds += time.perf_counter() - retrieval_started

        for (shot_index, query), shortlist in zip(prepared, shortlists):
            shot = query["shot"]
            try:
                analyzed, verification, refinement = analyze_prepared_shot(
                    shot,
                    query,
                    shortlist,
                    indexes,
                    args,
                    work_dir,
                    query["analysisDiagnostics"],
                )
                verification_seconds += verification
                refinement_seconds += refinement
                completed[shot_index] = analyzed
            except Exception as exc:
                completed[shot_index] = mark_shot_failed(shot, exc, "REFINEMENT")
            diagnostics_by_shot_id[str(shot["shotId"])] = query["analysisDiagnostics"]
            progress(
                "match-complete",
                shot=args.shot_start_index + shot_index + 1,
                total=len(all_shots),
                shotId=shot["shotId"],
                decision=completed[shot_index]["decision"],
            )
    if any(shot is None for shot in completed):
        raise AutoConformFailure(
            "retrieval",
            "INCOMPLETE_SHOT_BATCH",
            "One or more shots did not receive a terminal analysis record.",
        )
    completed_shots = [shot for shot in completed if shot is not None]
    result["shots"] = completed_shots
    analysis_diagnostics = ordered_analysis_diagnostics(
        completed_shots,
        diagnostics_by_shot_id,
    )
    if cache_dir is not None:
        prune = prune_source_cache(cache_dir, args.cache_max_bytes, protected_cache_paths)
        progress("cache-prune", phase="after-index", **prune)
    total_seconds = time.perf_counter() - started
    result["run"] = {
        "status": "COMPLETED",
        "timing": {
            "indexingSeconds": indexing_seconds,
            "retrievalSeconds": retrieval_seconds,
            "verificationSeconds": verification_seconds,
            "refinementSeconds": refinement_seconds,
            "totalSeconds": total_seconds,
        },
        "peakMemoryBytes": peak_memory_bytes(),
        "temporaryDiskBytes": directory_size(work_dir),
        "analysisDiagnostics": analysis_diagnostics,
        "completedShots": len(completed_shots),
    }
    result["benchmarkSummary"] = benchmark_summary(completed_shots, args.top_k)
    validate_contract(result, schema)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    if placement_table_output is not None:
        write_placement_table(placement_table_output, completed_shots, result["sources"])
    progress("complete", shots=len(completed_shots), output=output_path)
    print(f"SUMMARY|shots={len(completed_shots)}|output={output_path}|seconds={total_seconds:.3f}", flush=True)
    if args.keep_work_dir != "true":
        shutil.rmtree(work_dir, ignore_errors=True)
    return 0


def argument_value(name: str) -> str | None:
    try:
        index = sys.argv.index(name)
    except ValueError:
        return None
    return sys.argv[index + 1] if index + 1 < len(sys.argv) else None


def write_failure_report(failure: AutoConformFailure) -> Path | None:
    work_dir_value = argument_value("--work-dir")
    if not work_dir_value:
        return None
    path = Path(work_dir_value).resolve() / "auto-conform-failure.json"
    try:
        atomic_write_json(
            path,
            {
                "format": "automation-agent-auto-conform-failure/v1",
                "status": "FAILED",
                **failure.record(),
            },
        )
        return path
    except OSError:
        return None


def structured_error_value(value: object) -> str:
    return str(value).replace("|", "/").replace("\r", " ").replace("\n", " ")


def cli() -> int:
    try:
        return main()
    except AutoConformFailure as exc:
        failure = exc
    except Exception as exc:
        failure = AutoConformFailure(
            "analysis",
            "UNEXPECTED_FAILURE",
            str(exc) or exc.__class__.__name__,
            {"exceptionType": exc.__class__.__name__},
        )
    report_path = write_failure_report(failure)
    fields = [
        "ERROR",
        f"stage={structured_error_value(failure.stage)}",
        f"code={structured_error_value(failure.code)}",
        f"message={structured_error_value(failure)}",
    ]
    if report_path is not None:
        fields.append(f"report={structured_error_value(report_path)}")
    for key, value in failure.details.items():
        fields.append(f"{key}={structured_error_value(value)}")
    print("|".join(fields), file=sys.stderr, flush=True)
    return 2


if __name__ == "__main__":
    raise SystemExit(cli())
