#!/usr/bin/env python
"""Seed-independent candidate-pool caches for hybrid camera generation.

The expensive part of hybrid generation is producing, retargeting and checking
the safe camera candidates for every shot.  Candidate selection itself is
cheap and depends on the user-visible generation seed.  This module persists
the expensive, seed-independent pool so another seed can select from the same
validated candidates without repeating the geometry work.

The cache key deliberately contains only fingerprints for the generation
input, preparation artifact, resolved directives, generation controls and the
versions of the stages that can change a candidate.  Selection-only seed
fields are recursively removed from directives and controls before they are
fingerprinted.

SECURITY: ``candidate_metadata.joblib`` is pickle-backed.  Cache bundles are
trusted local build artifacts, not an interchange or upload format.  SHA-256
validation detects corruption, partial publication and mixed bundles, but it
does not authenticate an artifact whose manifest and payloads were all
replaced by an attacker.  Loading therefore requires an explicit
``trust_pickle=True`` at the application-owned cache trust boundary.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import tempfile
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

import numpy as np


CACHE_FORMAT = "cw-ai-hybrid-candidate-pool"
CACHE_SCHEMA_VERSION = 1
CACHE_BUILDER_VERSION = "1"
CANDIDATE_ARRAY_SCHEMA_VERSION = 1
CANDIDATE_METADATA_SCHEMA_VERSION = 1

CANDIDATE_ARRAYS_FILE = "candidate_arrays.npz"
CANDIDATE_METADATA_FILE = "candidate_metadata.joblib"
MANIFEST_FILE = "manifest.json"
MANIFEST_HASH_FILE = "manifest.sha256"

_MEMBER_NAMES = (CANDIDATE_ARRAYS_FILE, CANDIDATE_METADATA_FILE)
_ARRAY_NAME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$")
_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
_SELECTION_SEED_FIELD_NAMES = frozenset(
    {
        "seed",
        "generationseed",
        "randomseed",
        "selectionseed",
        "variationseed",
    }
)

# These versions are intentionally narrower than a hash of generate_hybrid.py.
# A caller bumps only the stage whose behavior changes, avoiding invalidation
# when CLI reporting, serialization or seed selection code changes.
REQUIRED_STAGE_VERSION_NAMES = (
    "shotSegmentation",
    "shotPlanning",
    "candidateRanking",
    "trajectoryRetargeting",
    "compositionSafety",
    "variationPool",
    "sequenceContinuation",
)

REQUIRED_CAMERA_ARRAY_SHAPES: Mapping[str, tuple[int, ...]] = {
    "position": (3,),
    "rotation": (4,),
    "fov": (),
    "dutch": (),
}
OPTIONAL_CAMERA_ARRAY_SHAPES: Mapping[str, tuple[int, ...]] = {
    "target": (3,),
    "stageAnchor": (3,),
}

PICKLE_TRUST_WARNING = (
    "candidate_metadata.joblib is pickle-backed; load only from an "
    "application-owned cache root. SHA-256 integrity is not authenticity."
)


class CandidateCacheError(RuntimeError):
    """Base exception for candidate-pool cache failures."""


class CandidateCacheIntegrityError(CandidateCacheError):
    """Raised when a bundle is incomplete, corrupt or mismatched."""


class UntrustedCandidateCacheError(CandidateCacheError):
    """Raised when pickle loading was not explicitly trusted."""


@dataclass(frozen=True)
class CandidatePoolCandidate:
    """One retargeted candidate and its non-array generation metadata.

    ``metrics`` may contain local Python objects such as a template reference,
    so it is stored in the trusted joblib member.  Large numeric trajectories
    belong in ``arrays`` and are stored in an object-free NPZ member.
    """

    arrays: Mapping[str, np.ndarray]
    metrics: Any


@dataclass(frozen=True)
class CandidatePoolShot:
    """Cached candidates and deterministic continuation state for one shot."""

    shot_index: int
    start_frame: int
    end_frame_exclusive: int
    candidates: Sequence[CandidatePoolCandidate]
    sequence_continuation_state: Any


@dataclass(frozen=True)
class CandidateCacheBundleInfo:
    """Result of atomically publishing or reusing an immutable bundle."""

    path: Path
    cache_key: str
    reused: bool
    manifest: dict[str, Any]


@dataclass(frozen=True)
class LoadedCandidatePoolCache:
    """Fully validated and explicitly trusted candidate-pool payload."""

    path: Path
    manifest: dict[str, Any]
    shots: tuple[CandidatePoolShot, ...]
    bundle_metadata: Any


@dataclass(frozen=True)
class CandidateCacheLookup:
    """Validation-only lookup result; no NPZ or pickle payload was loaded."""

    status: str
    cache_key: str
    path: Path
    manifest: dict[str, Any] | None = None
    error: str | None = None

    @property
    def hit(self) -> bool:
        return self.status == "hit"


@dataclass(frozen=True)
class CandidateCacheLoadResult:
    """Fail-safe load result used by generation code."""

    status: str
    cache_key: str
    path: Path
    loaded: LoadedCandidatePoolCache | None = None
    error: str | None = None

    @property
    def hit(self) -> bool:
        return self.status == "hit" and self.loaded is not None


def canonical_json_bytes(value: Any) -> bytes:
    """Encode cache identity data deterministically and without NaN values."""

    return json.dumps(
        value,
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with Path(path).open("rb") as stream:
        while True:
            chunk = stream.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def fingerprint_value(value: Any) -> str:
    """Return a canonical SHA-256 fingerprint for JSON-compatible content."""

    return sha256_bytes(canonical_json_bytes(value))


def fingerprint_files(
    files: Mapping[str, Path],
) -> dict[str, str]:
    """Hash a stable, caller-named set of generation input files."""

    return {
        str(name): sha256_file(Path(path))
        for name, path in sorted(files.items(), key=lambda item: str(item[0]))
    }


def build_candidate_cache_identity(
    *,
    input_fingerprint: Any,
    preparation_fingerprint: Any,
    directives: Any,
    controls: Any,
    stage_versions: Mapping[str, Any],
) -> dict[str, Any]:
    """Build the seed-independent, stage-specific cache identity.

    ``input_fingerprint`` and ``preparation_fingerprint`` may be existing
    64-character SHA-256 strings or JSON-compatible identity structures.
    ``directives`` should be the resolved per-shot directives, and ``controls``
    should contain only values that affect ranking, retargeting or safety.

    Selection-only seed fields are removed recursively.  A model/training seed
    that affects preparation belongs in ``preparation_fingerprint`` and is not
    removed from that component.
    """

    normalized_stage_versions = _normalize_stage_versions(stage_versions)
    seed_independent_directives = _without_selection_seed_fields(directives)
    seed_independent_controls = _without_selection_seed_fields(controls)
    return {
        "stage": "seed-independent-candidate-pool",
        "fingerprints": {
            "generationInput": _normalize_component_fingerprint(
                input_fingerprint
            ),
            "preparation": _normalize_component_fingerprint(
                preparation_fingerprint
            ),
            "directives": fingerprint_value(seed_independent_directives),
            "controls": fingerprint_value(seed_independent_controls),
        },
        "stageVersions": normalized_stage_versions,
    }


def candidate_cache_key_for(identity: Mapping[str, Any]) -> str:
    """Return the content address for one candidate-pool identity."""

    identity_copy = _canonical_copy(dict(identity))
    _validate_identity(identity_copy)
    envelope = {
        "format": CACHE_FORMAT,
        "schemaVersion": CACHE_SCHEMA_VERSION,
        "builderVersion": CACHE_BUILDER_VERSION,
        "candidateArraySchemaVersion": CANDIDATE_ARRAY_SCHEMA_VERSION,
        "candidateMetadataSchemaVersion": CANDIDATE_METADATA_SCHEMA_VERSION,
        "identity": identity_copy,
    }
    return sha256_bytes(canonical_json_bytes(envelope))


def atomic_write_bytes(path: Path, value: bytes) -> None:
    """Replace one file atomically using a temporary sibling on Windows."""

    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.",
        suffix=".tmp",
        dir=path.parent,
    )
    temporary_path = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(value)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary_path, path)
    except BaseException:
        try:
            os.close(descriptor)
        except OSError:
            pass
        temporary_path.unlink(missing_ok=True)
        raise


def atomic_write_json(path: Path, value: Any) -> None:
    atomic_write_bytes(
        path,
        json.dumps(
            value,
            ensure_ascii=False,
            allow_nan=False,
            sort_keys=True,
            indent=2,
        ).encode("utf-8")
        + b"\n",
    )


def publish_cache_bundle(
    cache_root: Path,
    identity: Mapping[str, Any],
    shots: Sequence[CandidatePoolShot],
    *,
    bundle_metadata: Any = None,
    summary: Mapping[str, Any] | None = None,
) -> CandidateCacheBundleInfo:
    """Publish a complete immutable bundle through one directory rename."""

    identity_copy = _canonical_copy(dict(identity))
    cache_key = candidate_cache_key_for(identity_copy)
    arrays, metadata_payload, layout = _encode_payloads(
        shots,
        bundle_metadata,
    )
    summary_copy = _canonical_copy(dict(summary or {}))

    cache_root = Path(cache_root).resolve()
    cache_root.mkdir(parents=True, exist_ok=True)
    final_path = cache_root / cache_key
    if final_path.exists():
        try:
            manifest = validate_cache_bundle(final_path, cache_key)
        except (CandidateCacheError, OSError, ValueError):
            quarantined = quarantine_cache_bundle(final_path, cache_key)
            if quarantined is not None:
                try:
                    # Close the validate-to-rename race: another writer may
                    # have replaced the invalid path with a valid bundle just
                    # before our rename. Restore/reuse that bundle instead of
                    # treating it as corrupt.
                    manifest = validate_cache_bundle(quarantined, cache_key)
                except (CandidateCacheError, OSError, ValueError):
                    pass
                else:
                    try:
                        os.replace(quarantined, final_path)
                    except OSError:
                        manifest = validate_cache_bundle(final_path, cache_key)
                    return CandidateCacheBundleInfo(
                        path=final_path,
                        cache_key=cache_key,
                        reused=True,
                        manifest=manifest,
                    )
        else:
            return CandidateCacheBundleInfo(
                path=final_path,
                cache_key=cache_key,
                reused=True,
                manifest=manifest,
            )

    temporary_path = Path(
        tempfile.mkdtemp(
            prefix=f".{cache_key[:16]}.",
            suffix=".tmp",
            dir=cache_root,
        )
    )
    published = False
    try:
        _write_candidate_arrays(
            temporary_path / CANDIDATE_ARRAYS_FILE,
            arrays,
        )
        _dump_joblib(
            temporary_path / CANDIDATE_METADATA_FILE,
            metadata_payload,
        )
        members = {
            name: {
                "bytes": (temporary_path / name).stat().st_size,
                "sha256": sha256_file(temporary_path / name),
            }
            for name in _MEMBER_NAMES
        }
        manifest = {
            "format": CACHE_FORMAT,
            "schemaVersion": CACHE_SCHEMA_VERSION,
            "builderVersion": CACHE_BUILDER_VERSION,
            "candidateArraySchemaVersion": CANDIDATE_ARRAY_SCHEMA_VERSION,
            "candidateMetadataSchemaVersion": (
                CANDIDATE_METADATA_SCHEMA_VERSION
            ),
            "cacheKey": cache_key,
            "createdUtc": datetime.now(timezone.utc).isoformat(),
            "identity": identity_copy,
            "summary": summary_copy,
            "layout": layout,
            "trustBoundary": {
                "metadataEncoding": "joblib-pickle",
                "authenticated": False,
                "warning": PICKLE_TRUST_WARNING,
            },
            "members": members,
        }
        manifest_path = temporary_path / MANIFEST_FILE
        atomic_write_json(manifest_path, manifest)
        manifest_hash = sha256_file(manifest_path)
        atomic_write_bytes(
            temporary_path / MANIFEST_HASH_FILE,
            f"{manifest_hash}  {MANIFEST_FILE}\n".encode("ascii"),
        )

        # This validates all hashes without opening the NPZ or joblib payload.
        validate_cache_bundle(temporary_path, cache_key)
        try:
            os.replace(temporary_path, final_path)
            published = True
        except OSError:
            # Another process may have atomically published the same key.
            if not final_path.is_dir():
                raise
            existing = validate_cache_bundle(final_path, cache_key)
            return CandidateCacheBundleInfo(
                path=final_path,
                cache_key=cache_key,
                reused=True,
                manifest=existing,
            )
        return CandidateCacheBundleInfo(
            path=final_path,
            cache_key=cache_key,
            reused=False,
            manifest=manifest,
        )
    finally:
        if not published and temporary_path.exists():
            shutil.rmtree(temporary_path)


def validate_cache_bundle(
    bundle_path: Path,
    expected_cache_key: str | None = None,
) -> dict[str, Any]:
    """Validate identity and every payload hash without deserializing either."""

    bundle_path = Path(bundle_path).resolve()
    if not bundle_path.is_dir():
        raise CandidateCacheIntegrityError(
            f"Candidate cache directory was not found: {bundle_path}"
        )
    manifest_path = bundle_path / MANIFEST_FILE
    manifest_hash_path = bundle_path / MANIFEST_HASH_FILE
    if not manifest_path.is_file() or not manifest_hash_path.is_file():
        raise CandidateCacheIntegrityError(
            f"Candidate cache manifest is incomplete: {bundle_path}"
        )

    expected_manifest_hash = _read_manifest_hash(manifest_hash_path)
    actual_manifest_hash = sha256_file(manifest_path)
    if actual_manifest_hash != expected_manifest_hash:
        raise CandidateCacheIntegrityError(
            "Candidate cache manifest SHA-256 mismatch."
        )
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exception:
        raise CandidateCacheIntegrityError(
            "Candidate cache manifest is not valid JSON."
        ) from exception

    if manifest.get("format") != CACHE_FORMAT:
        raise CandidateCacheIntegrityError("Unexpected candidate cache format.")
    if manifest.get("schemaVersion") != CACHE_SCHEMA_VERSION:
        raise CandidateCacheIntegrityError(
            "Unsupported candidate cache schema version."
        )
    if manifest.get("builderVersion") != CACHE_BUILDER_VERSION:
        raise CandidateCacheIntegrityError(
            "Unsupported candidate cache builder version."
        )
    if (
        manifest.get("candidateArraySchemaVersion")
        != CANDIDATE_ARRAY_SCHEMA_VERSION
    ):
        raise CandidateCacheIntegrityError(
            "Unsupported candidate array schema version."
        )
    if (
        manifest.get("candidateMetadataSchemaVersion")
        != CANDIDATE_METADATA_SCHEMA_VERSION
    ):
        raise CandidateCacheIntegrityError(
            "Unsupported candidate metadata schema version."
        )

    identity = manifest.get("identity")
    if not isinstance(identity, dict):
        raise CandidateCacheIntegrityError(
            "Candidate cache identity must be an object."
        )
    try:
        calculated_key = candidate_cache_key_for(identity)
    except (TypeError, ValueError) as exception:
        raise CandidateCacheIntegrityError(
            "Candidate cache identity is invalid."
        ) from exception
    cache_key = manifest.get("cacheKey")
    if cache_key != calculated_key:
        raise CandidateCacheIntegrityError(
            "Candidate cache key does not match its identity."
        )
    if expected_cache_key is not None and cache_key != expected_cache_key:
        raise CandidateCacheIntegrityError(
            "Candidate cache key does not match the requested cache key."
        )

    _validate_manifest_layout(manifest.get("layout"))
    trust_boundary = manifest.get("trustBoundary")
    if not isinstance(trust_boundary, dict) or (
        trust_boundary.get("metadataEncoding") != "joblib-pickle"
        or trust_boundary.get("authenticated") is not False
    ):
        raise CandidateCacheIntegrityError(
            "Candidate cache trust-boundary metadata is invalid."
        )

    members = manifest.get("members")
    if not isinstance(members, dict) or set(members) != set(_MEMBER_NAMES):
        raise CandidateCacheIntegrityError(
            "Candidate cache member list is incomplete."
        )
    for name in _MEMBER_NAMES:
        member = members.get(name)
        if not isinstance(member, dict):
            raise CandidateCacheIntegrityError(
                f"Candidate cache member metadata is invalid: {name}"
            )
        path = bundle_path / name
        if not path.is_file():
            raise CandidateCacheIntegrityError(
                f"Candidate cache member is missing: {name}"
            )
        expected_bytes = member.get("bytes")
        expected_hash = member.get("sha256")
        if (
            not isinstance(expected_bytes, int)
            or expected_bytes < 0
            or not isinstance(expected_hash, str)
            or _SHA256_PATTERN.fullmatch(expected_hash) is None
        ):
            raise CandidateCacheIntegrityError(
                f"Candidate cache member metadata is invalid: {name}"
            )
        if path.stat().st_size != expected_bytes:
            raise CandidateCacheIntegrityError(
                f"Candidate cache member size mismatch: {name}"
            )
        if sha256_file(path) != expected_hash:
            raise CandidateCacheIntegrityError(
                f"Candidate cache member SHA-256 mismatch: {name}"
            )
    return manifest


def lookup_cache_bundle(
    cache_root: Path,
    identity: Mapping[str, Any],
) -> CandidateCacheLookup:
    """Return a fail-safe hit/miss after hash validation, without loading."""

    cache_key = candidate_cache_key_for(identity)
    path = Path(cache_root).resolve() / cache_key
    if not path.exists():
        return CandidateCacheLookup("miss", cache_key, path)
    try:
        manifest = validate_cache_bundle(path, cache_key)
    except (CandidateCacheError, OSError, ValueError) as exception:
        return CandidateCacheLookup(
            "miss",
            cache_key,
            path,
            error=str(exception),
        )
    return CandidateCacheLookup(
        "hit",
        cache_key,
        path,
        manifest=manifest,
    )


def load_cache_bundle(
    bundle_path: Path,
    *,
    expected_cache_key: str | None = None,
    trust_pickle: bool,
) -> LoadedCandidatePoolCache:
    """Validate all hashes, then load an explicitly trusted local bundle.

    ``trust_pickle=True`` is mandatory because joblib uses pickle.  Pass it
    only after establishing that ``bundle_path`` is inside an
    application-owned cache root; hashes alone do not establish provenance.
    """

    bundle_path = Path(bundle_path).resolve()
    manifest = validate_cache_bundle(bundle_path, expected_cache_key)
    if not trust_pickle:
        raise UntrustedCandidateCacheError(PICKLE_TRUST_WARNING)

    try:
        # Both members were fully hash-validated above.  Loading joblib before
        # or after NPZ does not change the explicit local-artifact trust rule.
        metadata_payload = _load_joblib(
            bundle_path / CANDIDATE_METADATA_FILE
        )
        arrays = _read_candidate_arrays(
            bundle_path / CANDIDATE_ARRAYS_FILE,
            manifest["layout"]["arrayKeys"],
        )
        shots, bundle_metadata = _decode_payloads(
            metadata_payload,
            arrays,
            manifest["layout"],
        )
    except CandidateCacheError:
        raise
    except Exception as exception:
        raise CandidateCacheIntegrityError(
            "Candidate cache payload could not be deserialized safely."
        ) from exception
    return LoadedCandidatePoolCache(
        path=bundle_path,
        manifest=manifest,
        shots=shots,
        bundle_metadata=bundle_metadata,
    )


def try_load_cache_bundle(
    cache_root: Path,
    identity: Mapping[str, Any],
    *,
    trust_pickle: bool,
) -> CandidateCacheLoadResult:
    """Load a trusted bundle, converting every artifact failure into a miss."""

    if not trust_pickle:
        raise UntrustedCandidateCacheError(PICKLE_TRUST_WARNING)
    cache_key = candidate_cache_key_for(identity)
    path = Path(cache_root).resolve() / cache_key
    if not path.exists():
        return CandidateCacheLoadResult("miss", cache_key, path)
    try:
        loaded = load_cache_bundle(
            path,
            expected_cache_key=cache_key,
            trust_pickle=True,
        )
    except (CandidateCacheError, OSError, ValueError) as exception:
        return CandidateCacheLoadResult(
            "miss",
            cache_key,
            path,
            error=str(exception),
        )
    return CandidateCacheLoadResult(
        "hit",
        cache_key,
        path,
        loaded=loaded,
    )


def _canonical_copy(value: Any) -> Any:
    return json.loads(canonical_json_bytes(value).decode("utf-8"))


def _normalize_component_fingerprint(value: Any) -> str:
    if isinstance(value, str):
        normalized = value.strip().lower()
        if _SHA256_PATTERN.fullmatch(normalized) is not None:
            return normalized
    return fingerprint_value(value)


def _normalize_stage_versions(
    stage_versions: Mapping[str, Any],
) -> dict[str, str]:
    if not isinstance(stage_versions, Mapping):
        raise TypeError("stage_versions must be a mapping.")
    missing = sorted(set(REQUIRED_STAGE_VERSION_NAMES) - set(stage_versions))
    if missing:
        raise ValueError(
            "Missing candidate-cache stage versions: " + ", ".join(missing)
        )
    normalized: dict[str, str] = {}
    for name, value in stage_versions.items():
        normalized_name = str(name).strip()
        normalized_value = str(value).strip()
        if not normalized_name or not normalized_value:
            raise ValueError("Stage version names and values must not be empty.")
        normalized[normalized_name] = normalized_value
    return dict(sorted(normalized.items()))


def _without_selection_seed_fields(value: Any) -> Any:
    if isinstance(value, Mapping):
        result = {}
        for key, child in value.items():
            normalized_key = re.sub(r"[^a-z0-9]", "", str(key).lower())
            if normalized_key in _SELECTION_SEED_FIELD_NAMES:
                continue
            result[str(key)] = _without_selection_seed_fields(child)
        return result
    if isinstance(value, (list, tuple)):
        return [_without_selection_seed_fields(child) for child in value]
    return value


def _validate_identity(identity: Mapping[str, Any]) -> None:
    if identity.get("stage") != "seed-independent-candidate-pool":
        raise ValueError("Unexpected candidate-cache identity stage.")
    fingerprints = identity.get("fingerprints")
    required_fingerprints = {
        "generationInput",
        "preparation",
        "directives",
        "controls",
    }
    if (
        not isinstance(fingerprints, dict)
        or set(fingerprints) != required_fingerprints
    ):
        raise ValueError("Candidate-cache fingerprints are incomplete.")
    for name, value in fingerprints.items():
        if (
            not isinstance(value, str)
            or _SHA256_PATTERN.fullmatch(value) is None
        ):
            raise ValueError(
                f"Candidate-cache fingerprint is not SHA-256: {name}"
            )
    _normalize_stage_versions(identity.get("stageVersions", {}))


def _encode_payloads(
    shots: Sequence[CandidatePoolShot],
    bundle_metadata: Any,
) -> tuple[dict[str, np.ndarray], dict[str, Any], dict[str, Any]]:
    if not shots:
        raise ValueError("At least one candidate-pool shot is required.")

    arrays: dict[str, np.ndarray] = {}
    metadata_shots: list[dict[str, Any]] = []
    layout_shots: list[dict[str, Any]] = []
    seen_shot_indices: set[int] = set()
    total_candidates = 0

    for shot_position, shot in enumerate(shots):
        shot_index = int(shot.shot_index)
        start_frame = int(shot.start_frame)
        end_frame = int(shot.end_frame_exclusive)
        if shot_index < 0 or shot_index in seen_shot_indices:
            raise ValueError("Shot indices must be unique and non-negative.")
        if start_frame < 0 or end_frame <= start_frame:
            raise ValueError("Shot frame boundaries are invalid.")
        if not shot.candidates:
            raise ValueError(f"Shot {shot_index} has no candidates.")
        seen_shot_indices.add(shot_index)
        frame_count = end_frame - start_frame

        metadata_candidates: list[dict[str, Any]] = []
        layout_candidate_arrays: list[list[str]] = []
        for candidate_index, candidate in enumerate(shot.candidates):
            normalized_arrays = _normalize_candidate_arrays(
                candidate.arrays,
                frame_count,
            )
            array_keys: dict[str, str] = {}
            for name, value in sorted(normalized_arrays.items()):
                storage_key = (
                    f"s{shot_position:04d}_c{candidate_index:04d}_{name}"
                )
                arrays[storage_key] = value
                array_keys[name] = storage_key
            metadata_candidates.append(
                {
                    "arrayKeys": array_keys,
                    "metrics": candidate.metrics,
                }
            )
            layout_candidate_arrays.append(sorted(array_keys))
            total_candidates += 1

        metadata_shots.append(
            {
                "shotIndex": shot_index,
                "startFrame": start_frame,
                "endFrameExclusive": end_frame,
                "candidates": metadata_candidates,
                "sequenceContinuationState": (
                    shot.sequence_continuation_state
                ),
            }
        )
        layout_shots.append(
            {
                "shotIndex": shot_index,
                "startFrame": start_frame,
                "endFrameExclusive": end_frame,
                "candidateCount": len(metadata_candidates),
                "candidateArrayNames": layout_candidate_arrays,
            }
        )

    metadata_payload = {
        "format": f"{CACHE_FORMAT}-metadata",
        "schemaVersion": CANDIDATE_METADATA_SCHEMA_VERSION,
        "bundleMetadata": bundle_metadata,
        "shots": metadata_shots,
    }
    layout = {
        "shotCount": len(metadata_shots),
        "candidateCount": total_candidates,
        "arrayCount": len(arrays),
        "arrayKeys": sorted(arrays),
        "shots": layout_shots,
    }
    _validate_manifest_layout(layout)
    return arrays, metadata_payload, layout


def _normalize_candidate_arrays(
    arrays: Mapping[str, Any],
    frame_count: int,
) -> dict[str, np.ndarray]:
    if not isinstance(arrays, Mapping):
        raise TypeError("Candidate arrays must be a mapping.")
    missing = sorted(set(REQUIRED_CAMERA_ARRAY_SHAPES) - set(arrays))
    if missing:
        raise ValueError(
            "Candidate camera arrays are incomplete: " + ", ".join(missing)
        )
    normalized: dict[str, np.ndarray] = {}
    for raw_name, raw_value in arrays.items():
        name = str(raw_name)
        if _ARRAY_NAME_PATTERN.fullmatch(name) is None:
            raise ValueError(f"Invalid candidate array name: {name!r}")
        value = np.asarray(raw_value)
        if value.ndim < 1 or value.shape[0] != frame_count:
            raise ValueError(
                f"Candidate array {name!r} must start with frame dimension "
                f"{frame_count}, got {value.shape}."
            )
        if value.dtype.kind not in "biuf":
            raise ValueError(
                f"Candidate array {name!r} must have a non-object, real "
                "numeric dtype."
            )
        if value.dtype.kind == "f" and not np.all(np.isfinite(value)):
            raise ValueError(
                f"Candidate array {name!r} contains non-finite values."
            )
        expected_tail = REQUIRED_CAMERA_ARRAY_SHAPES.get(
            name,
            OPTIONAL_CAMERA_ARRAY_SHAPES.get(name),
        )
        if expected_tail is not None and value.shape[1:] != expected_tail:
            raise ValueError(
                f"Candidate array {name!r} must have shape "
                f"({frame_count}, {', '.join(map(str, expected_tail))})"
                if expected_tail
                else (
                    f"Candidate array {name!r} must have shape "
                    f"({frame_count},), got {value.shape}."
                )
            )
        normalized[name] = np.ascontiguousarray(value).copy()
    return normalized


def _validate_manifest_layout(layout: Any) -> None:
    if not isinstance(layout, dict):
        raise CandidateCacheIntegrityError(
            "Candidate cache layout must be an object."
        )
    shot_count = layout.get("shotCount")
    candidate_count = layout.get("candidateCount")
    array_count = layout.get("arrayCount")
    array_keys = layout.get("arrayKeys")
    shots = layout.get("shots")
    if (
        not isinstance(shot_count, int)
        or shot_count < 1
        or not isinstance(candidate_count, int)
        or candidate_count < shot_count
        or not isinstance(array_count, int)
        or array_count < candidate_count
        or not isinstance(array_keys, list)
        or len(array_keys) != array_count
        or len(set(array_keys)) != len(array_keys)
        or any(not isinstance(key, str) or not key for key in array_keys)
        or not isinstance(shots, list)
        or len(shots) != shot_count
    ):
        raise CandidateCacheIntegrityError(
            "Candidate cache layout counts are invalid."
        )
    calculated_candidates = 0
    seen_indices: set[int] = set()
    for shot in shots:
        if not isinstance(shot, dict):
            raise CandidateCacheIntegrityError(
                "Candidate cache shot layout is invalid."
            )
        shot_index = shot.get("shotIndex")
        start_frame = shot.get("startFrame")
        end_frame = shot.get("endFrameExclusive")
        count = shot.get("candidateCount")
        names = shot.get("candidateArrayNames")
        if (
            not isinstance(shot_index, int)
            or shot_index < 0
            or shot_index in seen_indices
            or not isinstance(start_frame, int)
            or start_frame < 0
            or not isinstance(end_frame, int)
            or end_frame <= start_frame
            or not isinstance(count, int)
            or count < 1
            or not isinstance(names, list)
            or len(names) != count
            or any(
                not isinstance(candidate_names, list)
                or not set(REQUIRED_CAMERA_ARRAY_SHAPES).issubset(
                    candidate_names
                )
                for candidate_names in names
            )
        ):
            raise CandidateCacheIntegrityError(
                "Candidate cache shot layout is invalid."
            )
        seen_indices.add(shot_index)
        calculated_candidates += count
    if calculated_candidates != candidate_count:
        raise CandidateCacheIntegrityError(
            "Candidate cache candidate count does not match its layout."
        )


def _write_candidate_arrays(
    path: Path,
    arrays: Mapping[str, np.ndarray],
) -> None:
    with Path(path).open("wb") as stream:
        np.savez(stream, **arrays)
        stream.flush()
        os.fsync(stream.fileno())


def _read_candidate_arrays(
    path: Path,
    expected_keys: Sequence[str],
) -> dict[str, np.ndarray]:
    try:
        with np.load(path, allow_pickle=False) as payload:
            if set(payload.files) != set(expected_keys):
                raise CandidateCacheIntegrityError(
                    "Candidate NPZ members do not match the manifest layout."
                )
            return {
                key: np.asarray(payload[key]).copy()
                for key in expected_keys
            }
    except CandidateCacheError:
        raise
    except (OSError, KeyError, ValueError) as exception:
        raise CandidateCacheIntegrityError(
            "Candidate array payload is invalid."
        ) from exception


def _dump_joblib(path: Path, value: Any) -> None:
    import joblib

    joblib.dump(value, path, compress=0, protocol=5)
    _flush_file(path)


def _load_joblib(path: Path) -> Any:
    import joblib

    return joblib.load(path)


def _decode_payloads(
    metadata_payload: Any,
    arrays: Mapping[str, np.ndarray],
    layout: Mapping[str, Any],
) -> tuple[tuple[CandidatePoolShot, ...], Any]:
    if (
        not isinstance(metadata_payload, dict)
        or metadata_payload.get("format") != f"{CACHE_FORMAT}-metadata"
        or metadata_payload.get("schemaVersion")
        != CANDIDATE_METADATA_SCHEMA_VERSION
        or not isinstance(metadata_payload.get("shots"), list)
    ):
        raise CandidateCacheIntegrityError(
            "Candidate metadata payload header is invalid."
        )
    metadata_shots = metadata_payload["shots"]
    layout_shots = layout["shots"]
    if len(metadata_shots) != len(layout_shots):
        raise CandidateCacheIntegrityError(
            "Candidate metadata shot count does not match the manifest."
        )

    decoded_shots: list[CandidatePoolShot] = []
    consumed_array_keys: set[str] = set()
    for metadata_shot, layout_shot in zip(metadata_shots, layout_shots):
        if not isinstance(metadata_shot, dict):
            raise CandidateCacheIntegrityError(
                "Candidate metadata shot entry is invalid."
            )
        for name in ("shotIndex", "startFrame", "endFrameExclusive"):
            if metadata_shot.get(name) != layout_shot.get(name):
                raise CandidateCacheIntegrityError(
                    "Candidate metadata shot boundaries do not match the "
                    "manifest."
                )
        metadata_candidates = metadata_shot.get("candidates")
        if (
            not isinstance(metadata_candidates, list)
            or len(metadata_candidates) != layout_shot["candidateCount"]
        ):
            raise CandidateCacheIntegrityError(
                "Candidate metadata count does not match the manifest."
            )
        frame_count = (
            layout_shot["endFrameExclusive"] - layout_shot["startFrame"]
        )
        decoded_candidates: list[CandidatePoolCandidate] = []
        for candidate_index, metadata_candidate in enumerate(
            metadata_candidates
        ):
            if not isinstance(metadata_candidate, dict):
                raise CandidateCacheIntegrityError(
                    "Candidate metadata entry is invalid."
                )
            array_key_map = metadata_candidate.get("arrayKeys")
            if not isinstance(array_key_map, dict):
                raise CandidateCacheIntegrityError(
                    "Candidate array-key mapping is invalid."
                )
            expected_names = layout_shot["candidateArrayNames"][
                candidate_index
            ]
            if set(array_key_map) != set(expected_names):
                raise CandidateCacheIntegrityError(
                    "Candidate array names do not match the manifest."
                )
            candidate_arrays: dict[str, np.ndarray] = {}
            for name, storage_key in array_key_map.items():
                if (
                    not isinstance(storage_key, str)
                    or storage_key not in arrays
                    or storage_key in consumed_array_keys
                ):
                    raise CandidateCacheIntegrityError(
                        "Candidate array-key reference is invalid."
                    )
                candidate_arrays[name] = arrays[storage_key]
                consumed_array_keys.add(storage_key)
            normalized_arrays = _normalize_candidate_arrays(
                candidate_arrays,
                frame_count,
            )
            if "metrics" not in metadata_candidate:
                raise CandidateCacheIntegrityError(
                    "Candidate metadata metrics are missing."
                )
            decoded_candidates.append(
                CandidatePoolCandidate(
                    arrays=normalized_arrays,
                    metrics=metadata_candidate["metrics"],
                )
            )
        if "sequenceContinuationState" not in metadata_shot:
            raise CandidateCacheIntegrityError(
                "Sequence-continuation state is missing."
            )
        decoded_shots.append(
            CandidatePoolShot(
                shot_index=layout_shot["shotIndex"],
                start_frame=layout_shot["startFrame"],
                end_frame_exclusive=layout_shot["endFrameExclusive"],
                candidates=tuple(decoded_candidates),
                sequence_continuation_state=metadata_shot[
                    "sequenceContinuationState"
                ],
            )
        )

    if consumed_array_keys != set(layout["arrayKeys"]):
        raise CandidateCacheIntegrityError(
            "Candidate metadata does not reference every manifest array."
        )
    return (
        tuple(decoded_shots),
        metadata_payload.get("bundleMetadata"),
    )


def _flush_file(path: Path) -> None:
    # Windows requires a writable descriptor for fsync after joblib closes its
    # own stream.
    with Path(path).open("rb+") as stream:
        os.fsync(stream.fileno())


def _read_manifest_hash(path: Path) -> str:
    try:
        value = path.read_text(encoding="ascii").strip().split()[0]
    except (OSError, IndexError, UnicodeDecodeError) as exception:
        raise CandidateCacheIntegrityError(
            "Candidate cache manifest hash is invalid."
        ) from exception
    if _SHA256_PATTERN.fullmatch(value) is None:
        raise CandidateCacheIntegrityError(
            "Candidate cache manifest hash is invalid."
        )
    return value


def quarantine_cache_bundle(
    bundle_path: Path,
    expected_cache_key: str,
    *,
    expected_manifest_fingerprint: str | None = None,
) -> Path | None:
    """Atomically move one invalid immutable bundle out of the lookup path.

    A content-addressed path may be left behind by a crash or later corruption.
    Keeping it at the canonical key would make every automatic rebuild fail at
    publication forever.  Moving it to a unique sibling preserves the evidence
    for diagnosis while allowing a freshly verified bundle to take its place.
    """

    bundle_path = Path(bundle_path).absolute()
    cache_key = str(expected_cache_key)
    if bundle_path.name != cache_key:
        raise ValueError(
            "Candidate cache quarantine path does not match its cache key."
        )
    if not bundle_path.exists():
        return None
    if expected_manifest_fingerprint is not None:
        current_fingerprint = _manifest_fingerprint(bundle_path)
        if current_fingerprint != expected_manifest_fingerprint:
            # The canonical path no longer contains the semantic-invalid
            # bundle the caller inspected. Leave a concurrent replacement in
            # place and let normal publication validation decide its fate.
            return None
    quarantine_path = bundle_path.parent / (
        f".invalid-{cache_key[:16]}-{uuid.uuid4().hex}"
    )
    try:
        os.replace(bundle_path, quarantine_path)
    except FileNotFoundError:
        # A concurrent publisher may already have removed or replaced it.  The
        # caller can continue and the final atomic publication decides the race.
        return None
    if expected_manifest_fingerprint is not None and (
        _manifest_fingerprint(quarantine_path)
        != expected_manifest_fingerprint
    ):
        try:
            validate_cache_bundle(quarantine_path, cache_key)
        except (CandidateCacheError, OSError, ValueError):
            # An unexpected invalid replacement is still safer in quarantine.
            return quarantine_path
        try:
            os.replace(quarantine_path, bundle_path)
        except OSError:
            validate_cache_bundle(bundle_path, cache_key)
        return None
    return quarantine_path


def _manifest_fingerprint(bundle_path: Path) -> str | None:
    try:
        manifest = json.loads(
            (Path(bundle_path) / MANIFEST_FILE).read_text(encoding="utf-8")
        )
        return fingerprint_value(manifest)
    except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError):
        return None
