server/artifact_ingest.py

"""Artifact ingestion and signal extraction helpers for TalkingHeadBench.

This module enables user-provided files (reference images, clips, and LoRA
weights) to be uploaded once, converted into pre-extracted signal JSON, and
reused in OpenEnv episodes through an ingestion ID.
"""

from __future__ import annotations

import copy
import json
import logging
import os
import shutil
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path
from threading import Lock
from typing import Any, Optional
from uuid import uuid4

import cv2
import numpy as np
from fastapi import UploadFile

from src.envs.subenv2.node4_clip_extractor import (
    _BLUR_CALIBRATION_CEILING,
    extract_clip_signals,
)
from src.envs.subenv3.node7_weight_extractor import extract_weight_signals
from src.schemas.subenv1 import ImageDiagnosticsObservation
from src.schemas.subenv2 import ClipSignalObservation
from src.schemas.subenv3 import WeightSignalObservation

log = logging.getLogger(__name__)

_UPLOAD_ROOT = Path(tempfile.gettempdir()) / "talkingheadbench_uploads"
_UPLOAD_ROOT.mkdir(parents=True, exist_ok=True)

_BUNDLE_STORE: dict[str, dict[str, Any]] = {}
_BUNDLE_STORE_LOCK = Lock()


def _env_int(name: str, default: int, *, minimum: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return max(minimum, int(raw))
    except ValueError:
        return default


API_VERSION = "1.0"
_UPLOAD_CHUNK_BYTES = 1024 * 1024
_MAX_UPLOAD_BYTES = _env_int("THB_MAX_UPLOAD_BYTES", 512 * 1024 * 1024, minimum=1)
_MAX_CLIPS_PER_REQUEST = _env_int("THB_MAX_CLIPS_PER_REQUEST", 12, minimum=1)
_MAX_BUNDLES_IN_MEMORY = _env_int("THB_MAX_BUNDLES_IN_MEMORY", 128, minimum=1)
_UPLOAD_TTL_SECONDS = _env_int("THB_UPLOAD_TTL_SECONDS", 24 * 60 * 60, minimum=300)

_ALLOWED_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"}
_ALLOWED_CLIP_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
_ALLOWED_WEIGHT_EXTS = {".safetensors"}
_ALLOWED_TOKENIZER_EXTS = {".json"}

_DEFAULT_PARAM_CONFIG: dict[str, float] = {
    "cfg": 7.5,
    "denoise_alt": 0.30,
    "eta": 0.10,
}

_FACE_DETECTOR_URL = (
    "https://storage.googleapis.com/mediapipe-models/face_detector/"
    "blaze_face_short_range/float16/latest/blaze_face_short_range.tflite"
)
_DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES: tuple[Path, ...] = (
    Path(__file__).resolve().parents[1] / "data" / "models" / "face_detector.tflite",
    Path.home() / ".cache" / "talkingheadbench" / "models" / "face_detector.tflite",
)


def _clamp(value: float, low: float, high: float) -> float:
    return max(low, min(high, value))


def _env_truthy(name: str, *, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _detect_conflicting_descriptors(prompt: str) -> list[str]:
    text = prompt.lower()
    pairs = [
        ("young", "old"),
        ("male", "female"),
        ("smiling", "serious"),
        ("frontal", "profile"),
        ("bright", "dark"),
    ]
    conflicts: list[str] = []
    for left, right in pairs:
        if left in text and right in text:
            conflicts.append(f"{left}|{right}")
    return conflicts


def _candidate_face_detector_model_paths() -> list[Path]:
    env_path = os.getenv("THB_FACE_DETECTOR_MODEL", "").strip()
    candidates: list[Path] = []
    if env_path:
        candidates.append(Path(env_path).expanduser())
    candidates.extend(_DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES)

    deduped: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        key = str(path)
        if key in seen:
            continue
        seen.add(key)
        deduped.append(path)
    return deduped


def _download_face_detector_model(dest: Path) -> Path:
    dest.parent.mkdir(parents=True, exist_ok=True)
    urllib.request.urlretrieve(_FACE_DETECTOR_URL, dest)
    return dest


def _resolve_face_detector_model_path() -> Path | None:
    for candidate in _candidate_face_detector_model_paths():
        if candidate.exists() and candidate.is_file():
            return candidate

    if not _env_truthy("THB_AUTO_DOWNLOAD_FACE_DETECTOR", default=True):
        return None

    cache_target = _DEFAULT_FACE_DETECTOR_MODEL_CANDIDATES[-1]
    try:
        return _download_face_detector_model(cache_target)
    except (OSError, urllib.error.URLError, ValueError) as exc:
        log.warning(
            "Unable to auto-download FaceDetector model to %s: %s",
            cache_target,
            exc,
        )
        return None


def _create_tasks_face_detector() -> Any | None:
    model_path = _resolve_face_detector_model_path()
    if model_path is None:
        return None

    try:
        from mediapipe.tasks.python import BaseOptions
        from mediapipe.tasks.python.vision import (
            FaceDetector,
            FaceDetectorOptions,
            RunningMode,
        )

        options = FaceDetectorOptions(
            base_options=BaseOptions(model_asset_path=str(model_path)),
            running_mode=RunningMode.IMAGE,
            min_detection_confidence=0.5,
        )
        return FaceDetector.create_from_options(options)
    except Exception as exc:  # noqa: BLE001
        log.warning("Failed to initialize Tasks FaceDetector from %s: %s", model_path, exc)
        return None


def _get_haar_face_cascade() -> cv2.CascadeClassifier | None:
    cascade_path = Path(cv2.data.haarcascades) / "haarcascade_frontalface_default.xml"
    if not cascade_path.exists():
        return None

    cascade = cv2.CascadeClassifier(str(cascade_path))
    if cascade.empty():
        return None
    return cascade


def _estimate_face_occupancy_ratio(image_bgr) -> tuple[float, bool]:
    """Estimate face occupancy as (ratio, is_measured).

    is_measured=False means no detector backend was available and a synthetic
    fallback ratio was returned.
    """
    height, width = image_bgr.shape[:2]
    detector = _create_tasks_face_detector()
    if detector is not None:
        try:
            import mediapipe as mp

            rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
            mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb.copy())
            result = detector.detect(mp_image)

            detections = getattr(result, "detections", [])
            areas: list[float] = []
            for detection in detections:
                bbox = getattr(detection, "bounding_box", None)
                if bbox is None:
                    continue
                box_width = float(getattr(bbox, "width", 0.0))
                box_height = float(getattr(bbox, "height", 0.0))
                if box_width <= 0.0 or box_height <= 0.0:
                    continue
                areas.append((box_width * box_height) / float(max(width * height, 1)))

            if areas:
                return float(_clamp(max(areas), 0.0, 1.0)), True

            # Detector ran but no face was detected.
            return 0.05, True
        except Exception as exc:  # noqa: BLE001
            log.warning("Tasks face occupancy detection failed; falling back to Haar: %s", exc)
        finally:
            if hasattr(detector, "close"):
                detector.close()

    cascade = _get_haar_face_cascade()
    if cascade is not None:
        gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)
        faces = cascade.detectMultiScale(
            gray,
            scaleFactor=1.1,
            minNeighbors=4,
            minSize=(max(24, width // 10), max(24, height // 10)),
        )
        if len(faces) > 0:
            areas = [
                (float(face_w) * float(face_h)) / float(max(width * height, 1))
                for (_, _, face_w, face_h) in faces
            ]
            return float(_clamp(max(areas), 0.0, 1.0)), True

        # Haar backend was available but no face was detected.
        return 0.05, True

    log.warning(
        "Face occupancy ratio fell back to synthetic default because no detector backend was available."
    )
    return 0.35, False


def extract_image_signals(image_path: Path, prompt: str) -> tuple[ImageDiagnosticsObservation, bool]:
    """Extract a Node 1-compatible observation from a reference image file."""
    image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR)
    if image_bgr is None or image_bgr.size == 0:
        raise ValueError(f"Failed to read reference image: {image_path}")

    height, width = image_bgr.shape[:2]
    gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)

    face_occupancy_ratio, face_detection_measured = _estimate_face_occupancy_ratio(image_bgr)

    lap_var = float(cv2.Laplacian(gray, cv2.CV_64F).var())
    sharpness = _clamp((lap_var / float(max(width * height, 1))) / 0.001, 0.0, 1.0)

    lab = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2LAB)
    lighting_uniformity = _clamp(1.0 - float(lab[:, :, 0].astype("float32").std() / 80.0), 0.0, 1.0)

    edges = cv2.Canny(gray, 50, 150)
    background_complexity = _clamp(float((edges.mean() / 255.0) * 3.0), 0.0, 1.0)

    tokens = [tok for tok in prompt.split() if tok.strip()]
    token_count = len(tokens)
    density = len({tok.lower() for tok in tokens}) / max(1, token_count)
    conflicts = _detect_conflicting_descriptors(prompt)

    anchoring = 0.9 if "ohwx" in prompt.lower() else _clamp(0.35 + 0.50 * density, 0.0, 1.0)

    return ImageDiagnosticsObservation(
        face_occupancy_ratio=round(face_occupancy_ratio, 4),
        estimated_yaw_degrees=0.0,
        estimated_pitch_degrees=0.0,
        background_complexity_score=round(background_complexity, 4),
        lighting_uniformity_score=round(lighting_uniformity, 4),
        skin_tone_bucket=3,
        occlusion_detected=False,
        image_resolution=(width, height),
        estimated_sharpness=round(sharpness, 4),
        prompt_token_count=token_count,
        prompt_semantic_density=round(float(density), 4),
        conflicting_descriptors=conflicts,
        identity_anchoring_strength=round(float(anchoring), 4),
    ), face_detection_measured


def _parse_param_config(param_config_json: Optional[str]) -> dict[str, float]:
    config = dict(_DEFAULT_PARAM_CONFIG)
    if not param_config_json:
        return config

    try:
        payload = json.loads(param_config_json)
    except json.JSONDecodeError as exc:
        raise ValueError("param_config_json must be valid JSON") from exc

    if not isinstance(payload, dict):
        raise ValueError("param_config_json must decode to an object")

    for key, value in payload.items():
        if isinstance(value, (int, float)):
            config[key] = float(value)

    return config


def _validate_upload_extension(
    upload: UploadFile,
    *,
    fallback_name: str,
    allowed_extensions: set[str],
    field_name: str,
) -> None:
    suffix = Path(upload.filename or fallback_name).suffix.lower()
    if suffix not in allowed_extensions:
        allowed = ", ".join(sorted(allowed_extensions))
        raise ValueError(f"{field_name} must use one of the following extensions: {allowed}")


def _remove_upload_dir(ingestion_id: str) -> None:
    upload_dir = _UPLOAD_ROOT / str(ingestion_id)
    if upload_dir.exists():
        shutil.rmtree(upload_dir, ignore_errors=True)


def _prune_expired_assets() -> None:
    now = time.time()
    expired_ids: list[str] = []
    active_ids: set[str] = set()

    with _BUNDLE_STORE_LOCK:
        for ingestion_id, bundle in list(_BUNDLE_STORE.items()):
            active_ids.add(ingestion_id)
            metadata = bundle.get("ingestion_metadata", {})
            created_at = metadata.get("created_at_unix")
            if isinstance(created_at, (int, float)) and now - float(created_at) > _UPLOAD_TTL_SECONDS:
                expired_ids.append(ingestion_id)

        for ingestion_id in expired_ids:
            _BUNDLE_STORE.pop(ingestion_id, None)
            active_ids.discard(ingestion_id)

    for ingestion_id in expired_ids:
        _remove_upload_dir(ingestion_id)

    for child in _UPLOAD_ROOT.iterdir():
        if not child.is_dir() or child.name in active_ids:
            continue
        try:
            age_seconds = now - child.stat().st_mtime
        except OSError:
            continue
        if age_seconds > _UPLOAD_TTL_SECONDS:
            shutil.rmtree(child, ignore_errors=True)


def _update_phoneme_coverage(coverage: dict[str, int], clip_obs: ClipSignalObservation) -> None:
    for phoneme in clip_obs.phoneme_sequence:
        coverage[phoneme] = int(coverage.get(phoneme, 0)) + 1


def _extract_clip_signals_fallback(
    clip_path: Path,
    dataset_context: dict[str, Any],
) -> ClipSignalObservation:
    """Deterministic OpenCV-only fallback when full extractor deps are unavailable."""
    cap = cv2.VideoCapture(str(clip_path))
    if not cap.isOpened():
        raise ValueError(f"OpenCV could not open clip: {clip_path}")

    frames = []
    while True:
        ok, frame = cap.read()
        if not ok:
            break
        frames.append(frame)
        if len(frames) >= 180:
            break
    cap.release()

    if len(frames) < 2:
        raise ValueError(f"Clip has too few frames: {clip_path}")

    blur_scores: list[float] = []
    exposure_scores: list[float] = []
    diffs: list[float] = []
    gray_frames: list[Any] = []
    mouth_brightness_series: list[float] = []
    eye_brightness_series: list[float] = []

    height, width = frames[0].shape[:2]
    x1 = max(0, int(width * 0.30))
    x2 = min(width, int(width * 0.70))
    y1 = max(0, int(height * 0.20))
    y2 = min(height, int(height * 0.80))

    mouth_y1 = y1 + int((y2 - y1) * 0.66)
    eye_y1 = y1
    eye_y2 = y1 + max(1, int((y2 - y1) * 0.25))

    cascade = _get_haar_face_cascade()
    occlusion_frames = 0

    for idx, frame in enumerate(frames):
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        gray_frames.append(gray)

        face_roi = gray[y1:y2, x1:x2]
        if face_roi.size == 0:
            face_roi = gray

        lap_var = float(cv2.Laplacian(face_roi, cv2.CV_64F).var())
        pixel_count = max(1, face_roi.shape[0] * face_roi.shape[1])
        blur_scores.append(_clamp((lap_var / pixel_count) / _BLUR_CALIBRATION_CEILING, 0.0, 1.0))

        brightness = float(face_roi.mean() / 255.0)
        exposure_scores.append(_clamp(1.0 - abs(brightness - 0.5) * 2.0, 0.0, 1.0))

        mouth_roi = gray[mouth_y1:y2, x1:x2]
        eye_roi = gray[eye_y1:eye_y2, x1:x2]
        mouth_brightness_series.append(float(mouth_roi.mean()) if mouth_roi.size else float(gray.mean()))
        eye_brightness_series.append(float(eye_roi.mean()) if eye_roi.size else float(gray.mean()))

        if cascade is not None:
            faces = cascade.detectMultiScale(
                gray,
                scaleFactor=1.1,
                minNeighbors=4,
                minSize=(max(24, width // 8), max(24, height // 8)),
            )
            if len(faces) == 0:
                occlusion_frames += 1
        else:
            hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
            center_hsv = hsv[y1:y2, x1:x2]
            if center_hsv.size:
                skin_like = (
                    (center_hsv[..., 0] >= 0)
                    & (center_hsv[..., 0] <= 25)
                    & (center_hsv[..., 1] >= 30)
                    & (center_hsv[..., 1] <= 200)
                    & (center_hsv[..., 2] >= 60)
                )
                skin_ratio = float(skin_like.mean())
                if skin_ratio < 0.06:
                    occlusion_frames += 1

        if idx > 0:
            prev = frames[idx - 1].astype("float32")
            cur = frame.astype("float32")
            diffs.append(float(abs(cur - prev).mean()))

    frame_difference_mean = float(sum(diffs) / max(1, len(diffs)))
    identity_drift_proxy = _clamp(frame_difference_mean / 45.0, 0.0, 1.0)
    landmark_jitter_proxy = _clamp(frame_difference_mean / 500.0, 0.0, 1.0)

    if len(mouth_brightness_series) >= 2:
        mouth_delta_std = float(np.std(np.diff(np.array(mouth_brightness_series, dtype=np.float32))))
        # Keep a computed proxy for diagnostics, but return a conservative fixed
        # fallback confidence because true A/V lip-sync is not measured here.
        _ = _clamp(mouth_delta_std / 2.5, 0.0, 1.0)
    else:
        _ = 0.0

    # Fallback extractor does not perform true audio-visual sync analysis.
    lip_sync_confidence = 0.35

    blink_count = 0
    if eye_brightness_series:
        eye_arr = np.array(eye_brightness_series, dtype=np.float32)
        baseline = float(np.median(eye_arr))
        threshold = baseline * 0.88
        in_blink = False
        for value in eye_arr:
            if value < threshold:
                if not in_blink:
                    blink_count += 1
                    in_blink = True
            else:
                in_blink = False

    optical_flow_magnitude = 1.0
    if len(gray_frames) >= 2:
        face_flows: list[float] = []
        bg_flows: list[float] = []

        face_mask = np.zeros_like(gray_frames[0], dtype=bool)
        face_mask[y1:y2, x1:x2] = True
        for i in range(min(len(gray_frames) - 1, 30)):
            flow = cv2.calcOpticalFlowFarneback(
                gray_frames[i],
                gray_frames[i + 1],
                None,
                0.5,
                3,
                15,
                3,
                5,
                1.2,
                0,
            )
            mag = np.sqrt(flow[..., 0] ** 2 + flow[..., 1] ** 2)
            face_mean = float(mag[face_mask].mean()) if face_mask.any() else 0.0
            bg_mean = float(mag[~face_mask].mean() + 1e-6)
            face_flows.append(face_mean)
            bg_flows.append(bg_mean)

        if face_flows and bg_flows:
            optical_flow_magnitude = (
                float(sum(face_flows) / len(face_flows))
                / float(sum(bg_flows) / len(bg_flows))
            )

    return ClipSignalObservation(
        clip_id=clip_path.stem,
        face_embedding_variance=round(identity_drift_proxy, 4),
        landmark_stability_score=round(landmark_jitter_proxy, 4),
        identity_cosine_drift=round(identity_drift_proxy, 4),
        frame_difference_mean=round(frame_difference_mean, 4),
        optical_flow_magnitude=round(optical_flow_magnitude, 4),
        blink_count=blink_count,
        lip_sync_confidence=round(lip_sync_confidence, 4),
        phoneme_sequence=[],
        phoneme_coverage_new=0.0,
        blur_score=round(float(sum(blur_scores) / max(1, len(blur_scores))), 4),
        exposure_score=round(float(sum(exposure_scores) / max(1, len(exposure_scores))), 4),
        occlusion_frames=occlusion_frames,
        clips_audited_so_far=int(dataset_context.get("clips_audited_so_far", 0)),
        current_phoneme_coverage=dataset_context.get("current_phoneme_coverage", {}),
        current_pose_distribution=dataset_context.get("current_pose_distribution", {}),
        similar_clips_accepted=int(dataset_context.get("similar_clips_accepted", 0)),
    )


def _extract_clip_signal_observations(
    clip_paths: list[Path],
) -> tuple[list[ClipSignalObservation], int, list[str]]:
    observations: list[ClipSignalObservation] = []
    phoneme_coverage: dict[str, int] = {}
    fallback_count = 0
    fallback_clip_ids: list[str] = []

    for idx, clip_path in enumerate(clip_paths):
        dataset_context = {
            "clips_audited_so_far": idx,
            "current_phoneme_coverage": phoneme_coverage,
            "current_pose_distribution": {},
            "similar_clips_accepted": 0,
        }

        try:
            clip_obs = extract_clip_signals(
                clip_path=clip_path,
                dataset_context=dataset_context,
                aligner_output=None,
            )
            fallback_used = False
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "Full clip extractor failed for %s; using fallback extractor: %s",
                clip_path,
                exc,
            )
            fallback_count += 1
            clip_obs = _extract_clip_signals_fallback(clip_path, dataset_context)
            fallback_used = True

        if fallback_used:
            fallback_clip_ids.append(str(clip_obs.clip_id))

        observations.append(clip_obs)
        _update_phoneme_coverage(phoneme_coverage, clip_obs)

    return observations, fallback_count, fallback_clip_ids


async def _save_upload(
    upload: UploadFile,
    output_dir: Path,
    fallback_name: str,
    *,
    max_upload_bytes: int = _MAX_UPLOAD_BYTES,
) -> Path:
    original_name = Path(upload.filename or fallback_name).name
    suffix = Path(original_name).suffix.lower() or Path(fallback_name).suffix.lower() or ".bin"
    destination = output_dir / f"{Path(fallback_name).stem}_{uuid4().hex[:8]}{suffix}"

    bytes_written = 0
    try:
        with destination.open("wb") as handle:
            while True:
                chunk = await upload.read(_UPLOAD_CHUNK_BYTES)
                if not chunk:
                    break
                bytes_written += len(chunk)
                if bytes_written > max_upload_bytes:
                    raise ValueError(
                        f"Uploaded file exceeds the size limit of {max_upload_bytes} bytes"
                    )
                handle.write(chunk)
    except Exception:
        destination.unlink(missing_ok=True)
        await upload.close()
        raise

    await upload.close()
    return destination


async def ingest_artifacts_to_bundle(
    *,
    reference_image: UploadFile | None,
    clips: list[UploadFile] | None,
    lora_weights: UploadFile | None,
    tokenizer_config: UploadFile | None,
    prompt: str,
    param_config_json: str,
) -> dict[str, Any]:
    """Extract a custom signal bundle from uploaded user artifacts."""
    clips = clips or []
    _prune_expired_assets()

    if reference_image is None and not clips and lora_weights is None:
        raise ValueError("At least one artifact is required (image, clips, or lora_weights).")

    if len(clips) > _MAX_CLIPS_PER_REQUEST:
        raise ValueError(f"A maximum of {_MAX_CLIPS_PER_REQUEST} clips is allowed per request")

    if reference_image is not None:
        _validate_upload_extension(
            reference_image,
            fallback_name="reference_image.png",
            allowed_extensions=_ALLOWED_IMAGE_EXTS,
            field_name="reference_image",
        )

    for clip_upload in clips:
        _validate_upload_extension(
            clip_upload,
            fallback_name="clip.mp4",
            allowed_extensions=_ALLOWED_CLIP_EXTS,
            field_name="clips",
        )

    if lora_weights is not None:
        _validate_upload_extension(
            lora_weights,
            fallback_name="weights.safetensors",
            allowed_extensions=_ALLOWED_WEIGHT_EXTS,
            field_name="lora_weights",
        )

    if tokenizer_config is not None:
        _validate_upload_extension(
            tokenizer_config,
            fallback_name="tokenizer_config.json",
            allowed_extensions=_ALLOWED_TOKENIZER_EXTS,
            field_name="tokenizer_config",
        )

    ingestion_id = str(uuid4())
    created_at = int(time.time())
    output_dir = _UPLOAD_ROOT / ingestion_id
    output_dir.mkdir(parents=True, exist_ok=True)

    bundle: dict[str, Any] = {
        "case_id": ingestion_id,
        "prompt": prompt,
        "param_config": _parse_param_config(param_config_json),
        "source_files": {},
        "ingestion_metadata": {
            "api_version": API_VERSION,
            "created_at_unix": created_at,
            "limits": {
                "max_upload_bytes": _MAX_UPLOAD_BYTES,
                "max_clips_per_request": _MAX_CLIPS_PER_REQUEST,
                "upload_ttl_seconds": _UPLOAD_TTL_SECONDS,
                "max_bundles_in_memory": _MAX_BUNDLES_IN_MEMORY,
            },
            "extractor_metadata": {
                "opencv_version": cv2.__version__,
                "clip_extractor_fallback_count": 0,
            },
        },
    }

    if reference_image is not None:
        image_path = await _save_upload(reference_image, output_dir, "reference_image.png")
        image_obs, face_detection_measured = extract_image_signals(image_path, prompt)
        bundle["image_observation"] = image_obs.model_dump(mode="json")
        bundle["source_files"]["reference_image"] = image_path.name
        bundle["ingestion_metadata"]["extractor_metadata"][
            "face_detection_measured"
        ] = face_detection_measured

    if clips:
        clip_paths: list[Path] = []
        for idx, clip_upload in enumerate(clips):
            clip_path = await _save_upload(clip_upload, output_dir, f"clip_{idx:03d}.mp4")
            clip_paths.append(clip_path)

        clip_obs_list, fallback_count, fallback_clip_ids = _extract_clip_signal_observations(clip_paths)
        fallback_set = set(fallback_clip_ids)
        bundle["clip_signal_observations"] = []
        for clip_obs in clip_obs_list:
            clip_payload = clip_obs.model_dump(mode="json")
            fallback_used = str(clip_obs.clip_id) in fallback_set
            clip_payload["fallback_extractor_used"] = fallback_used
            clip_payload["lip_sync_reliable"] = not fallback_used
            bundle["clip_signal_observations"].append(clip_payload)

        bundle["source_files"]["clips"] = [p.name for p in clip_paths]
        bundle["ingestion_metadata"]["extractor_metadata"][
            "clip_extractor_fallback_count"
        ] = fallback_count
        bundle["ingestion_metadata"]["extractor_metadata"][
            "fallback_clip_ids"
        ] = fallback_clip_ids

    if lora_weights is not None:
        weight_path = await _save_upload(lora_weights, output_dir, "weights.safetensors")
        tokenizer_path: Path | None = None
        if tokenizer_config is not None:
            tokenizer_path = await _save_upload(tokenizer_config, output_dir, "tokenizer_config.json")

        weight_obs: WeightSignalObservation = extract_weight_signals(
            weight_path=weight_path,
            tokenizer_config_path=tokenizer_path,
        )
        bundle["weight_observation"] = weight_obs.model_dump(mode="json")
        bundle["source_files"]["lora_weights"] = weight_path.name
        if tokenizer_path is not None:
            bundle["source_files"]["tokenizer_config"] = tokenizer_path.name

    return bundle


def store_ingested_bundle(bundle: dict[str, Any]) -> str:
    _prune_expired_assets()

    ingestion_id = str(bundle.get("case_id") or uuid4())
    saved = copy.deepcopy(bundle)
    saved["case_id"] = ingestion_id

    evicted_ids: list[str] = []

    with _BUNDLE_STORE_LOCK:
        _BUNDLE_STORE[ingestion_id] = saved

        while len(_BUNDLE_STORE) > _MAX_BUNDLES_IN_MEMORY:
            oldest_id = next(iter(_BUNDLE_STORE.keys()))
            _BUNDLE_STORE.pop(oldest_id, None)
            evicted_ids.append(oldest_id)

    for evicted_id in evicted_ids:
        _remove_upload_dir(evicted_id)

    return ingestion_id


def get_ingested_bundle(ingestion_id: str) -> dict[str, Any] | None:
    _prune_expired_assets()

    with _BUNDLE_STORE_LOCK:
        bundle = _BUNDLE_STORE.get(str(ingestion_id))
    if bundle is None:
        return None
    return copy.deepcopy(bundle)


def list_ingested_bundle_ids() -> list[str]:
    _prune_expired_assets()

    with _BUNDLE_STORE_LOCK:
        return sorted(_BUNDLE_STORE.keys())


def delete_ingested_bundle(ingestion_id: str) -> bool:
    with _BUNDLE_STORE_LOCK:
        removed = _BUNDLE_STORE.pop(str(ingestion_id), None)
    if removed is not None:
        _remove_upload_dir(str(ingestion_id))
    return removed is not None