app.py

"""
Holographic Trace Stack Reassembly Harness
Standalone Hugging Face Gradio Space proof harness.

Boundary: this is a trace-structured holographic dataset proof, not optical
holography, not a 3D hologram, and not a cinematic renderer.
"""

from __future__ import annotations

import copy
import hashlib
import json
import math
import random
import tempfile
import time
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple

import numpy as np
from PIL import Image, ImageDraw

try:
    import gradio as gr
except Exception:  # pragma: no cover - allows import/validation without UI dependency
    gr = None

APP_TITLE = "Holographic Trace Stack Reassembly Harness"
APP_SHORT_LINE = "Stack the traces. Reassemble the account."
APP_VERSION = "0.2.0"
LICENSE = "cc-by-nc-sa-4.0"
RUNTIME_ROUTE = [
    "source",
    "atomization",
    "trace capsule",
    "holographic dataset",
    "trace stack",
    "midstream reassembly",
    "reprojected dataset",
    "receipt",
]
TRACE_CLASSES = [
    "frame_atom",
    "object_atom",
    "edge_shape_atom",
    "color_atom",
    "motion_atom",
    "time_atom",
    "source_return_atom",
    "receipt_atom",
]
PRESSURE_STATES = [
    "HELD",
    "REASSEMBLED",
    "STRAINED",
    "REPAIRING",
    "CONFLICT",
    "QUARANTINED",
    "MUST_STOP",
    "CLOSED_FOR_CURRENT_SCOPE",
]
EXPORT_DIR = Path(tempfile.gettempdir()) / "holographic_trace_stack_reassembly_exports"
EXPORT_DIR.mkdir(parents=True, exist_ok=True)


# ---------------------------------------------------------------------------
# Core utilities
# ---------------------------------------------------------------------------


def canonical_json(data: Any) -> str:
    return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def sha256_text(data: str) -> str:
    return hashlib.sha256(data.encode("utf-8")).hexdigest()


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


def clamp(value: float, low: float = 0.0, high: float = 1.0) -> float:
    return max(low, min(high, value))


def round_metric(value: float) -> float:
    return round(float(clamp(value)), 4)


def image_to_png_bytes(image: Image.Image) -> bytes:
    import io

    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    return buffer.getvalue()


def write_json_file(name: str, data: Dict[str, Any]) -> str:
    safe_name = name.replace("/", "_").replace(" ", "_")
    path = EXPORT_DIR / safe_name
    path.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
    return str(path)


def make_zip_file(name: str, files: Dict[str, Dict[str, Any]]) -> str:
    safe_name = name.replace("/", "_").replace(" ", "_")
    path = EXPORT_DIR / safe_name
    with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
        for filename, data in files.items():
            zf.writestr(filename, json.dumps(data, indent=2, sort_keys=True))
    return str(path)


@dataclass(frozen=True)
class SyntheticConfig:
    source_name: str = "moving_circle"
    frames: int = 8
    width: int = 256
    height: int = 160
    shape: str = "circle"
    color_mode: str = "steady"
    motion: str = "diagonal"
    include_occlusion: bool = False
    missing_frame: Optional[int] = None
    seed: int = 610


# ---------------------------------------------------------------------------
# Synthetic source generation
# ---------------------------------------------------------------------------


def color_for_frame(index: int, mode: str, source_name: str) -> Tuple[int, int, int]:
    if mode == "color_shift":
        palette = [
            (235, 68, 68),
            (240, 145, 55),
            (242, 215, 65),
            (74, 190, 92),
            (75, 145, 235),
            (138, 92, 235),
            (220, 88, 190),
            (235, 68, 68),
        ]
        return palette[index % len(palette)]
    if source_name == "moving_square_b":
        return (80, 165, 240)
    return (230, 80, 95)


def center_for_frame(index: int, frames: int, width: int, height: int, motion: str) -> Tuple[int, int]:
    radius = 18
    left = 36
    right = width - 36
    top = 34
    bottom = height - 34
    denom = max(frames - 1, 1)
    t = index / denom
    if motion == "horizontal":
        return int(left + (right - left) * t), height // 2
    if motion == "vertical":
        return width // 2, int(top + (bottom - top) * t)
    if motion == "reverse_diagonal":
        return int(right - (right - left) * t), int(top + (bottom - top) * t)
    return int(left + (right - left) * t), int(top + (bottom - top) * t)


def draw_frame(
    width: int,
    height: int,
    shape: str,
    center: Tuple[int, int],
    color: Tuple[int, int, int],
    visible: bool,
    occlusion: bool,
    frame_index: int,
) -> Image.Image:
    img = Image.new("RGB", (width, height), (16, 18, 24))
    draw = ImageDraw.Draw(img)

    # faint grid, deterministic and intentionally simple
    for x in range(0, width, 32):
        draw.line([(x, 0), (x, height)], fill=(29, 32, 42))
    for y in range(0, height, 32):
        draw.line([(0, y), (width, y)], fill=(29, 32, 42))

    if visible:
        cx, cy = center
        r = 18
        bbox = [cx - r, cy - r, cx + r, cy + r]
        if shape == "square":
            draw.rectangle(bbox, fill=color, outline=(245, 245, 245), width=2)
        else:
            draw.ellipse(bbox, fill=color, outline=(245, 245, 245), width=2)

    if occlusion:
        # occluder deliberately covers the center channel so trace support must account for it.
        draw.rectangle([width // 2 - 20, 10, width // 2 + 20, height - 10], fill=(45, 48, 56))

    draw.text((8, 8), f"f{frame_index:02d}", fill=(230, 230, 230))
    return img


def generate_synthetic_source(config: SyntheticConfig) -> Dict[str, Any]:
    random.seed(config.seed)
    frames: List[Dict[str, Any]] = []
    object_id = f"{config.source_name}_obj_001"

    for i in range(config.frames):
        if config.missing_frame is not None and i == config.missing_frame:
            continue
        center = center_for_frame(i, config.frames, config.width, config.height, config.motion)
        color = color_for_frame(i, config.color_mode, config.source_name)
        occlusion = bool(config.include_occlusion and i in {config.frames // 2, config.frames // 2 + 1})
        visible = True
        image = draw_frame(config.width, config.height, config.shape, center, color, visible, occlusion, i)
        png = image_to_png_bytes(image)
        radius = 18
        bbox = [center[0] - radius, center[1] - radius, center[0] + radius, center[1] + radius]
        frames.append(
            {
                "frame_index": i,
                "timestamp_ms": i * 100,
                "image": image,
                "frame_hash": sha256_bytes(png),
                "object": {
                    "object_id": object_id,
                    "shape": config.shape,
                    "center": list(center),
                    "bbox": bbox,
                    "color_rgb": list(color),
                    "visible": visible,
                    "occluded": occlusion,
                },
            }
        )

    source_manifest = {
        "app": APP_TITLE,
        "version": APP_VERSION,
        "source_name": config.source_name,
        "config": {
            "frames": config.frames,
            "width": config.width,
            "height": config.height,
            "shape": config.shape,
            "color_mode": config.color_mode,
            "motion": config.motion,
            "include_occlusion": config.include_occlusion,
            "missing_frame": config.missing_frame,
            "seed": config.seed,
        },
        "frame_hashes": [f["frame_hash"] for f in frames],
        "object_id": object_id,
    }
    source_hash = sha256_text(canonical_json(source_manifest))
    source_manifest["source_hash"] = source_hash
    source_manifest["source_id"] = f"src_{source_hash[:12]}"
    source_manifest["generated_at_unix"] = int(time.time())

    return {"manifest": source_manifest, "frames": frames}


# ---------------------------------------------------------------------------
# Atomization and holographic dataset projections
# ---------------------------------------------------------------------------


def build_atom(
    atom_class: str,
    source_id: str,
    source_hash: str,
    frame_index: Optional[int],
    payload: Dict[str, Any],
    relationships: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    base = {
        "atom_class": atom_class,
        "source_id": source_id,
        "source_hash": source_hash,
        "frame_index": frame_index,
        "payload": payload,
        "relationships": relationships or {},
        "support_claims": [
            "honest_account",
            "source_return",
            "route_integrity",
            "boundary_respect",
        ],
    }
    atom_hash = sha256_text(canonical_json(base))
    base["atom_hash"] = atom_hash
    base["atom_id"] = f"{atom_class}_{source_id}_{frame_index if frame_index is not None else 'global'}_{atom_hash[:10]}"
    return base


def atomize_source(source: Dict[str, Any]) -> Dict[str, Any]:
    manifest = source["manifest"]
    source_id = manifest["source_id"]
    source_hash = manifest["source_hash"]
    frames = source["frames"]
    atoms: List[Dict[str, Any]] = []

    previous_center: Optional[List[int]] = None
    previous_index: Optional[int] = None
    for frame in frames:
        idx = frame["frame_index"]
        obj = frame["object"]
        center = obj["center"]
        bbox = obj["bbox"]

        atoms.append(
            build_atom(
                "frame_atom",
                source_id,
                source_hash,
                idx,
                {
                    "frame_hash": frame["frame_hash"],
                    "width": manifest["config"]["width"],
                    "height": manifest["config"]["height"],
                    "present": True,
                },
                {"time_index": idx, "object_id": obj["object_id"]},
            )
        )
        atoms.append(
            build_atom(
                "object_atom",
                source_id,
                source_hash,
                idx,
                {
                    "object_id": obj["object_id"],
                    "shape": obj["shape"],
                    "center": center,
                    "bbox": bbox,
                    "visible": obj["visible"],
                    "occluded": obj["occluded"],
                },
                {"frame_hash": frame["frame_hash"]},
            )
        )
        atoms.append(
            build_atom(
                "edge_shape_atom",
                source_id,
                source_hash,
                idx,
                {
                    "shape": obj["shape"],
                    "bbox": bbox,
                    "edge_signature": f"{obj['shape']}:{bbox[0]}:{bbox[1]}:{bbox[2]}:{bbox[3]}",
                    "occlusion_accounted": obj["occluded"],
                },
                {"object_id": obj["object_id"]},
            )
        )
        atoms.append(
            build_atom(
                "color_atom",
                source_id,
                source_hash,
                idx,
                {"object_id": obj["object_id"], "color_rgb": obj["color_rgb"]},
                {"frame_hash": frame["frame_hash"]},
            )
        )
        atoms.append(
            build_atom(
                "time_atom",
                source_id,
                source_hash,
                idx,
                {
                    "frame_index": idx,
                    "timestamp_ms": frame["timestamp_ms"],
                    "previous_frame_index": previous_index,
                    "expected_next_frame_index": idx + 1,
                },
                {"frame_hash": frame["frame_hash"]},
            )
        )
        if previous_center is None:
            delta = [0, 0]
            from_frame = None
        else:
            delta = [center[0] - previous_center[0], center[1] - previous_center[1]]
            from_frame = previous_index
        atoms.append(
            build_atom(
                "motion_atom",
                source_id,
                source_hash,
                idx,
                {
                    "object_id": obj["object_id"],
                    "from_frame": from_frame,
                    "to_frame": idx,
                    "delta_xy": delta,
                    "center": center,
                    "route_segment": f"{from_frame}->{idx}:{delta[0]},{delta[1]}",
                },
                {"time_index": idx, "frame_hash": frame["frame_hash"]},
            )
        )
        atoms.append(
            build_atom(
                "source_return_atom",
                source_id,
                source_hash,
                idx,
                {
                    "source_id": source_id,
                    "source_hash": source_hash,
                    "frame_hash": frame["frame_hash"],
                    "source_name": manifest["source_name"],
                },
                {"frame_index": idx},
            )
        )
        previous_center = center
        previous_index = idx

    # Global receipt atom binds the atomization route.
    trace_root = sha256_text(canonical_json([a["atom_hash"] for a in atoms]))
    receipt_payload = {
        "route": RUNTIME_ROUTE,
        "trace_root": trace_root,
        "source_id": source_id,
        "source_hash": source_hash,
        "atom_count_before_receipt": len(atoms),
        "classes": TRACE_CLASSES,
        "boundary": "holographic dataset proof; not optical holography",
    }
    atoms.append(build_atom("receipt_atom", source_id, source_hash, None, receipt_payload, {"trace_root": trace_root}))

    atomization_json = {
        "schema": "holographic_trace_atomization.v0.2",
        "source_manifest": {k: v for k, v in manifest.items() if k != "generated_at_unix"},
        "atom_count": len(atoms),
        "trace_classes": TRACE_CLASSES,
        "atoms": atoms,
        "trace_root": trace_root,
    }
    holographic_dataset = build_holographic_dataset(atomization_json)
    return {"atomization": atomization_json, "holographic_dataset": holographic_dataset}


def group_atoms_by_class(atoms: Iterable[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
    grouped: Dict[str, List[Dict[str, Any]]] = {klass: [] for klass in TRACE_CLASSES}
    for atom in atoms:
        grouped.setdefault(atom.get("atom_class", "unknown"), []).append(atom)
    return grouped


def build_holographic_dataset(atomization: Dict[str, Any]) -> Dict[str, Any]:
    atoms = atomization["atoms"]
    grouped = group_atoms_by_class(atoms)
    frame_indices = sorted({a["frame_index"] for a in atoms if a.get("frame_index") is not None})
    object_atoms = grouped.get("object_atom", [])
    motion_atoms = grouped.get("motion_atom", [])
    color_atoms = grouped.get("color_atom", [])
    time_atoms = grouped.get("time_atom", [])
    receipt_atoms = grouped.get("receipt_atom", [])
    source_return_atoms = grouped.get("source_return_atom", [])

    timeline_view = [
        {
            "frame_index": a["frame_index"],
            "timestamp_ms": a["payload"].get("timestamp_ms"),
            "previous_frame_index": a["payload"].get("previous_frame_index"),
            "expected_next_frame_index": a["payload"].get("expected_next_frame_index"),
        }
        for a in sorted(time_atoms, key=lambda x: x.get("frame_index") if x.get("frame_index") is not None else 10**9)
    ]
    object_route_view = [
        {
            "frame_index": a["frame_index"],
            "object_id": a["payload"].get("object_id"),
            "shape": a["payload"].get("shape"),
            "center": a["payload"].get("center"),
            "bbox": a["payload"].get("bbox"),
            "occluded": a["payload"].get("occluded"),
        }
        for a in sorted(object_atoms, key=lambda x: x.get("frame_index") if x.get("frame_index") is not None else 10**9)
    ]
    motion_map = [
        {
            "frame_index": a["frame_index"],
            "from_frame": a["payload"].get("from_frame"),
            "to_frame": a["payload"].get("to_frame"),
            "delta_xy": a["payload"].get("delta_xy"),
            "center": a["payload"].get("center"),
        }
        for a in sorted(motion_atoms, key=lambda x: x.get("frame_index") if x.get("frame_index") is not None else 10**9)
    ]
    source_receipt_view = {
        "source_id": atomization["source_manifest"]["source_id"],
        "source_hash": atomization["source_manifest"]["source_hash"],
        "trace_root": atomization["trace_root"],
        "source_return_atoms": len(source_return_atoms),
        "receipt_atoms": len(receipt_atoms),
        "boundary": "holographic dataset = one trace object projected through multiple accountable views; not optical holography",
    }
    atom_table = [
        {
            "atom_id": a["atom_id"],
            "atom_class": a["atom_class"],
            "frame_index": a.get("frame_index"),
            "source_id": a.get("source_id"),
            "atom_hash": a["atom_hash"],
        }
        for a in atoms
    ]
    color_view = [
        {
            "frame_index": a["frame_index"],
            "object_id": a["payload"].get("object_id"),
            "color_rgb": a["payload"].get("color_rgb"),
        }
        for a in sorted(color_atoms, key=lambda x: x.get("frame_index") if x.get("frame_index") is not None else 10**9)
    ]
    dataset = {
        "schema": "holographic_trace_dataset.v0.2",
        "dataset_id": f"holo_{sha256_text(canonical_json(atom_table))[:16]}",
        "boundary": "trace-structured holographic dataset proof only; not optical holography",
        "source_id": atomization["source_manifest"]["source_id"],
        "source_hash": atomization["source_manifest"]["source_hash"],
        "trace_root": atomization["trace_root"],
        "projections": {
            "source_view": atomization["source_manifest"],
            "atom_view": atom_table,
            "timeline_view": timeline_view,
            "object_continuity_view": object_route_view,
            "motion_route_view": motion_map,
            "pressure_state_view": {"initial_pressure_state": "HELD", "reason": "synthetic ground truth available"},
            "receipt_view": source_receipt_view,
            "reassembly_view": {"state": "NOT_RUN", "support": "pending trace stack evaluation"},
            "color_view": color_view,
        },
    }
    dataset["dataset_hash"] = sha256_text(canonical_json(dataset))
    return dataset


# ---------------------------------------------------------------------------
# Reassembly scoring
# ---------------------------------------------------------------------------


def make_trace_stack(atomization: Dict[str, Any], mode: str = "full", seed: int = 610) -> Dict[str, Any]:
    atoms = copy.deepcopy(atomization["atoms"])
    source_manifest = atomization["source_manifest"]
    expected_frames = source_manifest["config"]["frames"]

    if mode == "partial":
        # Remove selected trace support while preserving enough for repairable reassembly.
        atoms = [
            a
            for a in atoms
            if not (
                (a.get("atom_class") in {"edge_shape_atom", "color_atom"} and a.get("frame_index") in {2, 5})
                or (a.get("atom_class") == "motion_atom" and a.get("frame_index") == 4)
            )
        ]
    elif mode == "shuffled":
        rng = random.Random(seed)
        rng.shuffle(atoms)
    elif mode == "drop_frame":
        atoms = [a for a in atoms if a.get("frame_index") != expected_frames // 2]

    stack_hash = sha256_text(canonical_json([a["atom_hash"] for a in atoms]))
    return {
        "schema": "holographic_trace_stack.v0.2",
        "stack_id": f"stack_{stack_hash[:16]}",
        "mode": mode,
        "source_id": source_manifest["source_id"],
        "source_hash": source_manifest["source_hash"],
        "expected_frames": expected_frames,
        "expected_trace_classes": TRACE_CLASSES,
        "stack_hash": stack_hash,
        "atoms": atoms,
    }


def make_mixed_source_stack(atomization_a: Dict[str, Any], atomization_b: Dict[str, Any], seed: int = 610) -> Dict[str, Any]:
    atoms_a = copy.deepcopy(atomization_a["atoms"])
    atoms_b = copy.deepcopy(atomization_b["atoms"])
    mixed: List[Dict[str, Any]] = []
    for atom in atoms_a:
        idx = atom.get("frame_index")
        if idx is None or idx < 4:
            mixed.append(atom)
    for atom in atoms_b:
        idx = atom.get("frame_index")
        if idx is not None and idx >= 4:
            mixed.append(atom)
    # Include both receipt atoms to make source conflict explicit.
    mixed.extend([a for a in atoms_b if a.get("atom_class") == "receipt_atom"])
    rng = random.Random(seed)
    rng.shuffle(mixed)
    stack_hash = sha256_text(canonical_json([a["atom_hash"] for a in mixed]))
    return {
        "schema": "holographic_trace_stack.v0.2",
        "stack_id": f"mixed_stack_{stack_hash[:16]}",
        "mode": "mixed_source_false_stack",
        "source_id": atomization_a["source_manifest"]["source_id"],
        "source_hash": atomization_a["source_manifest"]["source_hash"],
        "expected_frames": atomization_a["source_manifest"]["config"]["frames"],
        "expected_trace_classes": TRACE_CLASSES,
        "stack_hash": stack_hash,
        "atoms": mixed,
    }


def count_required_atoms(grouped: Dict[str, List[Dict[str, Any]]], frame_indices: List[int]) -> Tuple[int, int, Dict[str, int]]:
    per_frame_classes = [
        "frame_atom",
        "object_atom",
        "edge_shape_atom",
        "color_atom",
        "motion_atom",
        "time_atom",
        "source_return_atom",
    ]
    expected = len(frame_indices) * len(per_frame_classes) + 1  # global receipt atom
    present = 0
    class_counts = {}
    for klass in TRACE_CLASSES:
        class_counts[klass] = len(grouped.get(klass, []))
    for idx in frame_indices:
        for klass in per_frame_classes:
            if any(a.get("frame_index") == idx for a in grouped.get(klass, [])):
                present += 1
    if grouped.get("receipt_atom"):
        present += 1
    return present, expected, class_counts


def assess_temporal_continuity(frame_indices_expected: List[int], trace_atoms: List[Dict[str, Any]]) -> Tuple[float, List[str]]:
    warnings: List[str] = []
    time_atoms = [a for a in trace_atoms if a.get("atom_class") == "time_atom"]
    present_indices = sorted({a.get("frame_index") for a in time_atoms if a.get("frame_index") is not None})
    if not frame_indices_expected:
        return 0.0, ["No expected timeline available."]
    missing = [idx for idx in frame_indices_expected if idx not in present_indices]
    if missing:
        warnings.append(f"Missing time atoms for frames: {missing}")
    adjacency_ok = 0
    adjacency_total = max(len(frame_indices_expected) - 1, 1)
    time_by_index = {a.get("frame_index"): a for a in time_atoms}
    for idx in frame_indices_expected[1:]:
        atom = time_by_index.get(idx)
        if atom and atom.get("payload", {}).get("previous_frame_index") == idx - 1:
            adjacency_ok += 1
    support_score = adjacency_ok / adjacency_total

    # Detect shuffled route by observed stack order of frame atoms. Midstream stacking can repair only
    # when the route order is supported by time atoms; it must still mark observed strain.
    observed_frame_order = [a.get("frame_index") for a in trace_atoms if a.get("atom_class") == "frame_atom"]
    observed_clean = [idx for idx in observed_frame_order if idx is not None]
    monotonic_observed = observed_clean == sorted(observed_clean)
    if not monotonic_observed:
        warnings.append("Observed trace stack order is not temporally monotonic; timeline continuity strain detected.")
    observed_penalty = 0.25 if not monotonic_observed else 0.0
    missing_penalty = len(missing) / max(len(frame_indices_expected), 1)
    return round_metric(support_score - observed_penalty - 0.5 * missing_penalty), warnings


def assess_object_continuity(grouped: Dict[str, List[Dict[str, Any]]], expected_indices: List[int]) -> Tuple[float, List[str]]:
    warnings: List[str] = []
    object_atoms = grouped.get("object_atom", [])
    if not object_atoms:
        return 0.0, ["No object atoms available."]
    object_ids = {a.get("payload", {}).get("object_id") for a in object_atoms}
    shapes = {a.get("payload", {}).get("shape") for a in object_atoms}
    present_indices = {a.get("frame_index") for a in object_atoms}
    missing = [idx for idx in expected_indices if idx not in present_indices]
    if len(object_ids) > 1:
        warnings.append(f"Multiple object identities in stack: {sorted(str(x) for x in object_ids)}")
    if len(shapes) > 1:
        warnings.append(f"Multiple shape accounts in stack: {sorted(str(x) for x in shapes)}")
    if missing:
        warnings.append(f"Missing object atoms for frames: {missing}")
    id_score = 1.0 if len(object_ids) == 1 else 0.35
    shape_score = 1.0 if len(shapes) == 1 else 0.55
    coverage_score = 1.0 - (len(missing) / max(len(expected_indices), 1))
    return round_metric(0.45 * id_score + 0.25 * shape_score + 0.30 * coverage_score), warnings


def assess_motion_route_detailed(
    grouped: Dict[str, List[Dict[str, Any]]],
    expected_indices: List[int],
    trace_atoms: Optional[List[Dict[str, Any]]] = None,
    expected_source_hash: Optional[str] = None,
) -> Tuple[float, List[str], Dict[str, Any]]:
    """Return motion score plus a v0.2 forensic segment ledger.

    v0.1 exposed motion_route_score as a number. v0.2 keeps the scoring
    intentionally simple while adding inspectable segment diagnostics:
    returned / missing / shuffled / reversed / conflicting, coverage delta,
    return contribution, affected atoms, and a receipt-facing reason line.
    """
    warnings: List[str] = []
    trace_atoms = trace_atoms or [a for atoms in grouped.values() for a in atoms]
    motion_atoms = grouped.get("motion_atom", [])
    frame_atoms = [a for a in trace_atoms if a.get("atom_class") == "frame_atom"]
    observed_frame_order = [a.get("frame_index") for a in frame_atoms if a.get("frame_index") is not None]
    observed_positions = {idx: pos for pos, idx in enumerate(observed_frame_order)}

    if not expected_indices:
        return 0.0, ["No expected motion route available."], {
            "schema": "motion_route_diagnostics.v0.2",
            "motion_route_score_formula": "0.45 * coverage_rate + 0.55 * return_rate",
            "coverage_rate": 0.0,
            "return_rate": 0.0,
            "segments": [],
            "boundary": "motion route diagnostics expose trace support; they do not claim optical holography",
        }

    if len(expected_indices) == 1:
        expected_pairs = []
    else:
        expected_pairs = list(zip(expected_indices[:-1], expected_indices[1:]))

    motion_by_index = {a.get("frame_index"): a for a in motion_atoms}
    expected_segment_count = max(len(expected_pairs), 1)
    returned_count = 0
    covered_count = 0
    segments: List[Dict[str, Any]] = []

    if not motion_atoms:
        warnings.append("No motion atoms available.")

    def affected_atoms_for_segment(frame_from: int, frame_to: int, motion_atom: Optional[Dict[str, Any]]) -> List[str]:
        relevant_classes = {"frame_atom", "object_atom", "motion_atom", "time_atom", "source_return_atom"}
        atom_ids: List[str] = []
        for atom in trace_atoms:
            if atom.get("atom_class") not in relevant_classes:
                continue
            if atom.get("frame_index") in {frame_from, frame_to}:
                atom_ids.append(atom.get("atom_id", "unknown"))
        if motion_atom and motion_atom.get("atom_id") not in atom_ids:
            atom_ids.append(motion_atom.get("atom_id", "unknown"))
        return atom_ids[:18]

    def conflict_present(frame_from: int, frame_to: int) -> bool:
        relevant = [a for a in trace_atoms if a.get("frame_index") in {frame_from, frame_to}]
        hashes = {a.get("source_hash") for a in relevant if a.get("source_hash") is not None}
        ids = {a.get("source_id") for a in relevant if a.get("source_id") is not None}
        if expected_source_hash and any(h != expected_source_hash for h in hashes):
            return True
        return len(hashes) > 1 or len(ids) > 1

    for frame_from, frame_to in expected_pairs:
        motion_atom = motion_by_index.get(frame_to)
        payload = motion_atom.get("payload", {}) if motion_atom else {}
        from_frame = payload.get("from_frame")
        observed_from_pos = observed_positions.get(frame_from)
        observed_to_pos = observed_positions.get(frame_to)
        observed_pair_strained = (
            observed_from_pos is None
            or observed_to_pos is None
            or observed_from_pos >= observed_to_pos
        )

        if motion_atom is None:
            status = "missing"
            receipt_line = f"Missing motion atom for segment {frame_from}->{frame_to}; coverage cannot support clean route settlement."
        elif conflict_present(frame_from, frame_to):
            status = "conflicting"
            covered_count += 1
            receipt_line = f"Source conflict touches segment {frame_from}->{frame_to}; route must not blend into clean reassembly."
        elif isinstance(from_frame, int) and from_frame > frame_to:
            status = "reversed"
            covered_count += 1
            receipt_line = f"Motion segment {frame_from}->{frame_to} points forward/backward inconsistently; return support is reversed."
        elif from_frame != frame_from:
            status = "shuffled"
            covered_count += 1
            receipt_line = f"Motion segment {frame_from}->{frame_to} does not return to expected prior frame; temporal route strain is active."
        elif observed_pair_strained:
            status = "shuffled"
            covered_count += 1
            receipt_line = f"Motion payload for {frame_from}->{frame_to} returns, but observed stack order is shuffled; timeline strain remains visible."
        else:
            status = "returned"
            covered_count += 1
            returned_count += 1
            receipt_line = f"Motion segment {frame_from}->{frame_to} returned to expected prior frame/object state."

        coverage_delta = round(1.0 / expected_segment_count, 4) if motion_atom else 0.0
        return_delta = round(1.0 / expected_segment_count, 4) if status == "returned" else 0.0
        segments.append(
            {
                "segment_id": f"seg_{frame_from}_{frame_to}",
                "label": f"frame {frame_from} → {frame_to}",
                "from_frame": frame_from,
                "to_frame": frame_to,
                "status": status,
                "coverage_delta": coverage_delta,
                "return_delta": return_delta,
                "weighted_coverage_contribution": round(0.45 * coverage_delta, 4),
                "weighted_return_contribution": round(0.55 * return_delta, 4),
                "motion_atom_id": motion_atom.get("atom_id") if motion_atom else None,
                "motion_atom_hash": motion_atom.get("atom_hash") if motion_atom else None,
                "from_frame_reported": from_frame,
                "observed_stack_positions": {"from": observed_from_pos, "to": observed_to_pos},
                "affected_atom_ids": affected_atoms_for_segment(frame_from, frame_to, motion_atom),
                "receipt_line": receipt_line,
            }
        )

    missing_segments = [s for s in segments if s["status"] == "missing"]
    strained_segments = [s for s in segments if s["status"] in {"shuffled", "reversed"}]
    conflict_segments = [s for s in segments if s["status"] == "conflicting"]
    if missing_segments:
        warnings.append(
            "Missing motion route segments: "
            + ", ".join(f"{s['from_frame']}->{s['to_frame']}" for s in missing_segments)
        )
    if strained_segments:
        warnings.append(
            "Motion route strain in segments: "
            + ", ".join(f"{s['from_frame']}->{s['to_frame']}:{s['status']}" for s in strained_segments)
        )
    if conflict_segments:
        warnings.append(
            "Motion route source conflict in segments: "
            + ", ".join(f"{s['from_frame']}->{s['to_frame']}" for s in conflict_segments)
        )

    coverage_rate = covered_count / expected_segment_count
    return_rate = returned_count / expected_segment_count
    score = round_metric(0.45 * coverage_rate + 0.55 * return_rate)
    diagnostics = {
        "schema": "motion_route_diagnostics.v0.2",
        "motion_route_score_formula": "0.45 * coverage_rate + 0.55 * return_rate",
        "coverage_rate": round_metric(coverage_rate),
        "return_rate": round_metric(return_rate),
        "covered_segments": covered_count,
        "returned_segments": returned_count,
        "expected_segments": expected_segment_count,
        "status_counts": {
            "returned": sum(1 for s in segments if s["status"] == "returned"),
            "missing": sum(1 for s in segments if s["status"] == "missing"),
            "shuffled": sum(1 for s in segments if s["status"] == "shuffled"),
            "reversed": sum(1 for s in segments if s["status"] == "reversed"),
            "conflicting": sum(1 for s in segments if s["status"] == "conflicting"),
        },
        "segments": segments,
        "boundary": "motion route diagnostics expose trace support; they do not claim optical holography",
    }
    return score, warnings, diagnostics


def assess_motion_route(grouped: Dict[str, List[Dict[str, Any]]], expected_indices: List[int]) -> Tuple[float, List[str]]:
    score, warnings, _diagnostics = assess_motion_route_detailed(grouped, expected_indices)
    return score, warnings


def source_return_score(grouped: Dict[str, List[Dict[str, Any]]], expected_source_hash: str) -> Tuple[float, List[str]]:
    warnings: List[str] = []
    all_atoms = [a for atoms in grouped.values() for a in atoms]
    if not all_atoms:
        return 0.0, ["Empty trace stack."]
    matching = [a for a in all_atoms if a.get("source_hash") == expected_source_hash]
    source_hashes = sorted({str(a.get("source_hash")) for a in all_atoms})
    if len(source_hashes) > 1:
        warnings.append(f"Source-return conflict: {len(source_hashes)} source hashes present.")
    return round_metric(len(matching) / len(all_atoms)), warnings


def detect_corruption_or_conflict(grouped: Dict[str, List[Dict[str, Any]]], expected_source_hash: str) -> Tuple[float, List[str], bool]:
    warnings: List[str] = []
    all_atoms = [a for atoms in grouped.values() for a in atoms]
    source_hashes = {a.get("source_hash") for a in all_atoms}
    source_ids = {a.get("source_id") for a in all_atoms}
    severe = False
    risk = 0.0
    if len(source_hashes) > 1:
        warnings.append("Mixed source hashes detected; false stack pressure is active.")
        risk += 0.45
        severe = True
    if len(source_ids) > 1:
        warnings.append("Mixed source identifiers detected; quarantine is required unless repair separates the routes.")
        risk += 0.25
        severe = True
    for atom in all_atoms:
        basis = {k: v for k, v in atom.items() if k not in {"atom_hash", "atom_id"}}
        if sha256_text(canonical_json(basis)) != atom.get("atom_hash"):
            warnings.append(f"Atom hash mismatch: {atom.get('atom_id', 'unknown')}")
            risk += 0.30
            severe = True
            break
    return round_metric(risk), warnings, severe


def build_contribution_breakdown(
    src_score: float,
    completeness: float,
    temporal_score: float,
    object_score: float,
    motion_score: float,
    conflict_risk: float,
    base_confidence: float,
    reassembly_confidence: float,
    false_settlement_risk: float,
) -> Dict[str, Any]:
    weights = {
        "source_return_score": 0.24,
        "trace_completeness_score": 0.20,
        "temporal_continuity_score": 0.20,
        "object_continuity_score": 0.18,
        "motion_route_score": 0.18,
    }
    scores = {
        "source_return_score": src_score,
        "trace_completeness_score": completeness,
        "temporal_continuity_score": temporal_score,
        "object_continuity_score": object_score,
        "motion_route_score": motion_score,
    }
    rows = []
    for key, weight in weights.items():
        rows.append(
            {
                "metric": key,
                "score": round_metric(scores[key]),
                "weight": weight,
                "weighted_contribution": round_metric(weight * scores[key]),
            }
        )
    return {
        "schema": "reassembly_contribution_breakdown.v0.2",
        "formula": "confidence = source_return*.24 + completeness*.20 + temporal*.20 + object*.18 + motion*.18; conflict pressure subtracts from reassembly confidence",
        "rows": rows,
        "base_confidence_before_conflict": base_confidence,
        "conflict_risk": conflict_risk,
        "conflict_pressure_subtract": round_metric(0.35 * conflict_risk),
        "reassembly_confidence": reassembly_confidence,
        "false_settlement_risk": false_settlement_risk,
        "boundary": "contribution panel explains trace support and state movement without claiming optical holography",
    }


def reassemble_trace_stack(trace_stack: Dict[str, Any]) -> Dict[str, Any]:
    atoms = trace_stack.get("atoms", [])
    grouped = group_atoms_by_class(atoms)
    expected_frames = int(trace_stack.get("expected_frames", 0))
    expected_indices = list(range(expected_frames))
    expected_source_hash = trace_stack.get("source_hash")

    present_count, expected_count, class_counts = count_required_atoms(grouped, expected_indices)
    completeness = round_metric(present_count / max(expected_count, 1))
    src_score, src_warnings = source_return_score(grouped, expected_source_hash)
    temporal_score, temporal_warnings = assess_temporal_continuity(expected_indices, atoms)
    object_score, object_warnings = assess_object_continuity(grouped, expected_indices)
    motion_score, motion_warnings, motion_route_diagnostics = assess_motion_route_detailed(
        grouped, expected_indices, atoms, expected_source_hash
    )
    conflict_risk, conflict_warnings, severe_conflict = detect_corruption_or_conflict(grouped, expected_source_hash)

    warnings = src_warnings + temporal_warnings + object_warnings + motion_warnings + conflict_warnings
    base_confidence = round_metric(
        0.24 * src_score
        + 0.20 * completeness
        + 0.20 * temporal_score
        + 0.18 * object_score
        + 0.18 * motion_score
    )
    false_settlement_risk = round_metric((1.0 - base_confidence) * 0.65 + conflict_risk * 0.35)
    reassembly_confidence = round_metric(base_confidence - 0.35 * conflict_risk)
    contribution_breakdown = build_contribution_breakdown(
        src_score,
        completeness,
        temporal_score,
        object_score,
        motion_score,
        conflict_risk,
        base_confidence,
        reassembly_confidence,
        false_settlement_risk,
    )

    missing_regions = []
    for idx in expected_indices:
        frame_gaps = [klass for klass in TRACE_CLASSES[:-1] if not any(a.get("frame_index") == idx for a in grouped.get(klass, []))]
        if frame_gaps:
            missing_regions.append({"frame_index": idx, "missing_atom_classes": frame_gaps})

    temporal_route_strain = any("temporally monotonic" in w for w in warnings)
    incomplete_trace_support = bool(missing_regions)

    if not atoms:
        pressure_state = "MUST_STOP"
        final_state = "MUST_STOP"
    elif severe_conflict and conflict_risk >= 0.50:
        pressure_state = "QUARANTINED"
        final_state = "QUARANTINED"
    elif conflict_risk > 0.0:
        pressure_state = "CONFLICT"
        final_state = "CONFLICT"
    elif incomplete_trace_support:
        pressure_state = "STRAINED" if reassembly_confidence >= 0.72 else "REPAIRING"
        final_state = "REPAIRING"
        warnings.append("Incomplete trace support prevents clean closure; reassembly remains repair-marked.")
    elif temporal_route_strain:
        pressure_state = "STRAINED"
        final_state = "REPAIRING"
        warnings.append("Temporal route strain prevents clean closure until order support is repaired.")
    elif reassembly_confidence >= 0.92 and false_settlement_risk <= 0.12:
        pressure_state = "HELD"
        final_state = "REASSEMBLED"
    elif reassembly_confidence >= 0.72:
        pressure_state = "STRAINED"
        final_state = "REPAIRING"
    elif reassembly_confidence >= 0.45:
        pressure_state = "REPAIRING"
        final_state = "REPAIRING"
    else:
        pressure_state = "MUST_STOP"
        final_state = "MUST_STOP"

    reconstructed_account = reconstruct_account(grouped, expected_indices, final_state, pressure_state, missing_regions)
    reatomized = reatomize_reconstructed_account(reconstructed_account, expected_source_hash)
    result = {
        "schema": "holographic_trace_reassembly_result.v0.2",
        "stack_id": trace_stack.get("stack_id"),
        "mode": trace_stack.get("mode"),
        "route": RUNTIME_ROUTE,
        "metrics": {
            "source_return_score": src_score,
            "trace_completeness_score": completeness,
            "temporal_continuity_score": temporal_score,
            "object_continuity_score": object_score,
            "motion_route_score": motion_score,
            "reassembly_confidence": reassembly_confidence,
            "false_settlement_risk": false_settlement_risk,
            "pressure_state": pressure_state,
        },
        "contribution_breakdown": contribution_breakdown,
        "motion_route_diagnostics": motion_route_diagnostics,
        "final_state": final_state,
        "missing_regions": missing_regions,
        "warnings": warnings,
        "reconstructed_account": reconstructed_account,
        "reprojected_dataset": {
            "source_account": reconstructed_account,
            "re_atomized_downstream_packet": reatomized,
            "projection_views": build_projection_views_from_result(grouped, reconstructed_account, pressure_state),
        },
    }
    result["receipt"] = make_receipt(trace_stack, result)
    return result


def reconstruct_account(
    grouped: Dict[str, List[Dict[str, Any]]],
    expected_indices: List[int],
    final_state: str,
    pressure_state: str,
    missing_regions: List[Dict[str, Any]],
) -> Dict[str, Any]:
    objects_by_frame = {a.get("frame_index"): a for a in grouped.get("object_atom", [])}
    color_by_frame = {a.get("frame_index"): a for a in grouped.get("color_atom", [])}
    motion_by_frame = {a.get("frame_index"): a for a in grouped.get("motion_atom", [])}
    frames = []
    for idx in expected_indices:
        obj = objects_by_frame.get(idx)
        color = color_by_frame.get(idx)
        motion = motion_by_frame.get(idx)
        if not obj:
            frames.append({"frame_index": idx, "state": "MISSING", "support": "no object atom"})
            continue
        frames.append(
            {
                "frame_index": idx,
                "state": "SUPPORTED" if not any(m["frame_index"] == idx for m in missing_regions) else "PARTIAL_SUPPORT",
                "object_id": obj.get("payload", {}).get("object_id"),
                "shape": obj.get("payload", {}).get("shape"),
                "center": obj.get("payload", {}).get("center"),
                "bbox": obj.get("payload", {}).get("bbox"),
                "color_rgb": color.get("payload", {}).get("color_rgb") if color else None,
                "motion_delta_xy": motion.get("payload", {}).get("delta_xy") if motion else None,
                "occluded": obj.get("payload", {}).get("occluded"),
            }
        )
    return {
        "claim": "The source account is reconstructed from trace support, not from original surface form alone.",
        "boundary": "No optical holography claim; dataset projections only.",
        "final_state": final_state,
        "pressure_state": pressure_state,
        "frames": frames,
    }


def reatomize_reconstructed_account(account: Dict[str, Any], source_hash: str) -> Dict[str, Any]:
    support_frames = [f for f in account["frames"] if f.get("state") in {"SUPPORTED", "PARTIAL_SUPPORT"}]
    packet = {
        "schema": "downstream_reatomized_packet.v0.2",
        "source_hash_returned": source_hash,
        "frame_count": len(account["frames"]),
        "supported_frame_count": len(support_frames),
        "frame_signatures": [
            {
                "frame_index": f.get("frame_index"),
                "object_id": f.get("object_id"),
                "shape": f.get("shape"),
                "center": f.get("center"),
                "color_rgb": f.get("color_rgb"),
                "motion_delta_xy": f.get("motion_delta_xy"),
                "state": f.get("state"),
            }
            for f in account["frames"]
        ],
    }
    packet["packet_hash"] = sha256_text(canonical_json(packet))
    return packet


def build_projection_views_from_result(
    grouped: Dict[str, List[Dict[str, Any]]], account: Dict[str, Any], pressure_state: str
) -> Dict[str, Any]:
    return {
        "timeline_view": [
            {"frame_index": f.get("frame_index"), "state": f.get("state")} for f in account.get("frames", [])
        ],
        "object_route_view": [
            {"frame_index": f.get("frame_index"), "center": f.get("center"), "shape": f.get("shape")}
            for f in account.get("frames", [])
        ],
        "source_receipt_view": {
            "source_hashes_present": sorted({str(a.get("source_hash")) for atoms in grouped.values() for a in atoms}),
            "receipt_atoms": len(grouped.get("receipt_atom", [])),
        },
        "motion_map": [
            {"frame_index": f.get("frame_index"), "motion_delta_xy": f.get("motion_delta_xy")}
            for f in account.get("frames", [])
        ],
        "atom_table": [
            {"atom_class": klass, "count": len(atoms)} for klass, atoms in grouped.items() if klass in TRACE_CLASSES
        ],
        "reassembly_state": {"pressure_state": pressure_state, "frame_count": len(account.get("frames", []))},
    }


def make_receipt(trace_stack: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]:
    receipt = {
        "schema": "holographic_trace_receipt.v0.2",
        "app": APP_TITLE,
        "version": APP_VERSION,
        "license": LICENSE,
        "route": RUNTIME_ROUTE,
        "stack_id": trace_stack.get("stack_id"),
        "stack_hash": trace_stack.get("stack_hash"),
        "source_id": trace_stack.get("source_id"),
        "source_hash": trace_stack.get("source_hash"),
        "final_state": result.get("final_state"),
        "pressure_state": result.get("metrics", {}).get("pressure_state"),
        "metrics": result.get("metrics", {}),
        "contribution_breakdown": result.get("contribution_breakdown", {}),
        "motion_route_status_counts": result.get("motion_route_diagnostics", {}).get("status_counts", {}),
        "warnings": result.get("warnings", []),
        "receipt_lines": [
            f"final_state={result.get('final_state')} pressure_state={result.get('metrics', {}).get('pressure_state')}",
            f"reassembly_confidence={result.get('metrics', {}).get('reassembly_confidence')} false_settlement_risk={result.get('metrics', {}).get('false_settlement_risk')}",
            "clean closure requires sufficient trace support and low false-settlement risk",
        ],
        "boundary": "This receipt accounts for trace support and source-return; it does not claim optical holography or full source truth.",
    }
    receipt["receipt_hash"] = sha256_text(canonical_json(receipt))
    return receipt


# ---------------------------------------------------------------------------
# Required validation tests
# ---------------------------------------------------------------------------


def default_source_a() -> Dict[str, Any]:
    return generate_synthetic_source(
        SyntheticConfig(
            source_name="moving_circle_a",
            frames=8,
            shape="circle",
            color_mode="color_shift",
            motion="diagonal",
            include_occlusion=False,
            seed=610,
        )
    )


def default_source_b() -> Dict[str, Any]:
    return generate_synthetic_source(
        SyntheticConfig(
            source_name="moving_square_b",
            frames=8,
            shape="square",
            color_mode="steady",
            motion="reverse_diagonal",
            include_occlusion=True,
            seed=611,
        )
    )


def run_required_tests() -> Dict[str, Any]:
    source_a = default_source_a()
    source_b = default_source_b()
    atomized_a = atomize_source(source_a)["atomization"]
    atomized_b = atomize_source(source_b)["atomization"]

    tests: Dict[str, Dict[str, Any]] = {}

    full_stack = make_trace_stack(atomized_a, mode="full")
    tests["FULL_STACK_REASSEMBLY"] = {
        "expected": "high continuity, high source-return, low pressure, HELD or REASSEMBLED",
        "result": reassemble_trace_stack(full_stack),
    }

    partial_stack = make_trace_stack(atomized_a, mode="partial")
    tests["PARTIAL_STACK_REASSEMBLY"] = {
        "expected": "degraded score, missing regions marked, no false closure",
        "result": reassemble_trace_stack(partial_stack),
    }

    shuffled_stack = make_trace_stack(atomized_a, mode="shuffled", seed=777)
    tests["SHUFFLED_TRACE_TEST"] = {
        "expected": "timeline continuity strain detected",
        "result": reassemble_trace_stack(shuffled_stack),
    }

    mixed_stack = make_mixed_source_stack(atomized_a, atomized_b, seed=778)
    tests["MIXED_SOURCE_FALSE_STACK"] = {
        "expected": "conflict, quarantine, or refusal depending severity",
        "result": reassemble_trace_stack(mixed_stack),
    }

    round_trip_initial = reassemble_trace_stack(full_stack)
    downstream_packet = round_trip_initial["reprojected_dataset"]["re_atomized_downstream_packet"]
    original_signature = [
        {
            "frame_index": a.get("frame_index"),
            "object_id": a.get("payload", {}).get("object_id"),
            "shape": a.get("payload", {}).get("shape"),
            "center": a.get("payload", {}).get("center"),
        }
        for a in atomized_a["atoms"]
        if a.get("atom_class") == "object_atom"
    ]
    reatomized_signature = [
        {
            "frame_index": f.get("frame_index"),
            "object_id": f.get("object_id"),
            "shape": f.get("shape"),
            "center": f.get("center"),
        }
        for f in downstream_packet["frame_signatures"]
        if f.get("object_id") is not None
    ]
    round_trip_score = 1.0 if original_signature == reatomized_signature else 0.0
    round_trip_result = copy.deepcopy(round_trip_initial)
    round_trip_result["round_trip_comparison"] = {
        "original_object_signature_count": len(original_signature),
        "reatomized_object_signature_count": len(reatomized_signature),
        "round_trip_signature_score": round_metric(round_trip_score),
        "comparison_hash": sha256_text(canonical_json({"original": original_signature, "reatomized": reatomized_signature})),
    }
    tests["ROUND_TRIP_TEST"] = {
        "expected": "continuity preserved above threshold or clearly marked strain",
        "result": round_trip_result,
    }

    dataset = build_holographic_dataset(atomized_a)
    projection_keys = [
        "timeline_view",
        "object_continuity_view",
        "source_receipt_view",
        "motion_route_view",
        "atom_view",
        "reassembly_view",
    ]
    projection_hashes = {key: sha256_text(canonical_json(dataset["projections"].get(key))) for key in projection_keys}
    cross_projection_result = {
        "schema": "cross_projection_result.v0.1",
        "final_state": "HELD",
        "metrics": {
            "source_return_score": 1.0,
            "trace_completeness_score": 1.0,
            "temporal_continuity_score": 1.0,
            "object_continuity_score": 1.0,
            "motion_route_score": 1.0,
            "reassembly_confidence": 1.0,
            "false_settlement_risk": 0.0,
            "pressure_state": "HELD",
        },
        "projection_keys": projection_keys,
        "projection_hashes": projection_hashes,
        "shared_source_hash": dataset["source_hash"],
        "shared_trace_root": dataset["trace_root"],
        "boundary": dataset["boundary"],
        "receipt": {
            "final_state": "HELD",
            "pressure_state": "HELD",
            "boundary": "Different projections preserve the same source account through shared source_hash and trace_root.",
        },
    }
    cross_projection_result["receipt"]["receipt_hash"] = sha256_text(canonical_json(cross_projection_result))
    tests["CROSS_PROJECTION_TEST"] = {
        "expected": "different projections preserve the same source account",
        "result": cross_projection_result,
    }

    summary = {}
    for name, payload in tests.items():
        result = payload["result"]
        metrics = result.get("metrics", {})
        final_state = result.get("final_state", result.get("receipt", {}).get("final_state", "UNKNOWN"))
        summary[name] = {
            "final_state": final_state,
            "pressure_state": metrics.get("pressure_state"),
            "reassembly_confidence": metrics.get("reassembly_confidence"),
            "false_settlement_risk": metrics.get("false_settlement_risk"),
            "passed_boundary_expectation": test_passed(name, result),
        }
    validation = {
        "schema": "holographic_trace_validation_report.v0.1",
        "app": APP_TITLE,
        "version": APP_VERSION,
        "boundary": "Controlled synthetic proof harness; no optical holography claim; no arbitrary user video in v0.1.",
        "required_tests_run": list(tests.keys()),
        "summary": summary,
        "tests": tests,
    }
    validation["validation_hash"] = sha256_text(canonical_json(validation))
    return validation


def test_passed(name: str, result: Dict[str, Any]) -> bool:
    metrics = result.get("metrics", {})
    final_state = result.get("final_state", result.get("receipt", {}).get("final_state"))
    pressure = metrics.get("pressure_state")
    if name == "FULL_STACK_REASSEMBLY":
        return final_state in {"REASSEMBLED", "HELD"} and metrics.get("source_return_score", 0) >= 0.95
    if name == "PARTIAL_STACK_REASSEMBLY":
        return final_state == "REPAIRING" and pressure in {"STRAINED", "REPAIRING"} and bool(result.get("missing_regions")) and metrics.get("false_settlement_risk", 0) > 0
    if name == "SHUFFLED_TRACE_TEST":
        return pressure in {"STRAINED", "REPAIRING"} and final_state == "REPAIRING" and any("temporally monotonic" in w for w in result.get("warnings", []))
    if name == "MIXED_SOURCE_FALSE_STACK":
        return final_state in {"CONFLICT", "QUARANTINED", "MUST_STOP"} or pressure in {"CONFLICT", "QUARANTINED", "MUST_STOP"}
    if name == "ROUND_TRIP_TEST":
        return result.get("round_trip_comparison", {}).get("round_trip_signature_score", 0) >= 0.95
    if name == "CROSS_PROJECTION_TEST":
        return result.get("shared_source_hash") is not None and result.get("shared_trace_root") is not None
    return False


# ---------------------------------------------------------------------------
# Visualization helpers for cockpit
# ---------------------------------------------------------------------------


def source_preview_gallery(source: Dict[str, Any]) -> List[Image.Image]:
    return [f["image"] for f in source["frames"]]


def draw_motion_map(account: Dict[str, Any], width: int = 420, height: int = 260) -> Image.Image:
    img = Image.new("RGB", (width, height), (18, 20, 27))
    draw = ImageDraw.Draw(img)
    for x in range(0, width, 42):
        draw.line([(x, 0), (x, height)], fill=(32, 36, 48))
    for y in range(0, height, 42):
        draw.line([(0, y), (width, y)], fill=(32, 36, 48))
    supported = [f for f in account.get("frames", []) if f.get("center")]
    if not supported:
        draw.text((20, 20), "No supported route", fill=(230, 230, 230))
        return img
    centers = [f["center"] for f in supported]
    xs = [c[0] for c in centers]
    ys = [c[1] for c in centers]
    min_x, max_x = min(xs), max(xs)
    min_y, max_y = min(ys), max(ys)

    def map_point(c: List[int]) -> Tuple[int, int]:
        x = 40 + int((c[0] - min_x) / max(max_x - min_x, 1) * (width - 80))
        y = 40 + int((c[1] - min_y) / max(max_y - min_y, 1) * (height - 80))
        return x, y

    pts = [map_point(c) for c in centers]
    for p1, p2 in zip(pts, pts[1:]):
        draw.line([p1, p2], fill=(210, 210, 220), width=3)
    for f, p in zip(supported, pts):
        r = 7
        state = f.get("state")
        fill = (90, 190, 115) if state == "SUPPORTED" else (230, 170, 65)
        draw.ellipse([p[0] - r, p[1] - r, p[0] + r, p[1] + r], fill=fill)
        draw.text((p[0] + 9, p[1] - 8), f"f{f.get('frame_index')}", fill=(238, 238, 238))
    draw.text((14, height - 24), "Motion route view from trace support", fill=(230, 230, 230))
    return img


def metrics_markdown(metrics: Dict[str, Any]) -> str:
    rows = []
    for key in [
        "source_return_score",
        "trace_completeness_score",
        "temporal_continuity_score",
        "object_continuity_score",
        "motion_route_score",
        "reassembly_confidence",
        "false_settlement_risk",
        "pressure_state",
    ]:
        rows.append(f"| `{key}` | `{metrics.get(key, 'n/a')}` |")
    return "| Metric | Value |\n|---|---|\n" + "\n".join(rows)


def atom_summary_markdown(atomization: Dict[str, Any]) -> str:
    grouped = group_atoms_by_class(atomization["atoms"])
    rows = [f"| `{klass}` | {len(grouped.get(klass, []))} |" for klass in TRACE_CLASSES]
    manifest = atomization["source_manifest"]
    return (
        f"**Source hash / receipt anchor**: `{manifest['source_hash']}`\n\n"
        f"**Atom count**: `{atomization['atom_count']}`\n\n"
        "| Atom class | Count |\n|---|---:|\n" + "\n".join(rows)
    )


def stack_layers_markdown(trace_stack: Dict[str, Any], result: Dict[str, Any]) -> str:
    grouped = group_atoms_by_class(trace_stack["atoms"])
    rows = [f"| `{klass}` | {len(grouped.get(klass, []))} |" for klass in TRACE_CLASSES]
    warnings = result.get("warnings", [])
    warn_text = "\n".join([f"- {w}" for w in warnings]) if warnings else "- No warnings."
    return (
        f"**Stack mode**: `{trace_stack['mode']}`\n\n"
        f"**Stack hash**: `{trace_stack['stack_hash']}`\n\n"
        "| Trace layer | Count |\n|---|---:|\n" + "\n".join(rows) + "\n\n"
        "**Missing / corrupt trace warnings**\n" + warn_text
    )


def reassembly_markdown(result: Dict[str, Any]) -> str:
    account = result["reconstructed_account"]
    frames = account.get("frames", [])
    rows = []
    for f in frames:
        rows.append(
            f"| {f.get('frame_index')} | `{f.get('state')}` | `{f.get('shape')}` | `{f.get('center')}` | `{f.get('motion_delta_xy')}` |"
        )
    return (
        f"**Final state**: `{result['final_state']}`\n\n"
        f"**Pressure state**: `{result['metrics']['pressure_state']}`\n\n"
        "**Reconstructed account**\n\n"
        "| Frame | Support state | Shape | Center | Motion Δ |\n|---:|---|---|---|---|\n"
        + "\n".join(rows)
    )


def receipt_markdown(receipt: Dict[str, Any]) -> str:
    return (
        f"**Receipt hash**: `{receipt.get('receipt_hash')}`\n\n"
        f"**Final state**: `{receipt.get('final_state')}`\n\n"
        f"**Pressure state**: `{receipt.get('pressure_state')}`\n\n"
        f"**Boundary**: {receipt.get('boundary')}"
    )




def contribution_breakdown_markdown(result: Dict[str, Any]) -> str:
    breakdown = result.get("contribution_breakdown", {})
    rows = []
    for row in breakdown.get("rows", []):
        rows.append(
            f"| `{row.get('metric')}` | `{row.get('score')}` | `{row.get('weight')}` | `{row.get('weighted_contribution')}` |"
        )
    return (
        "**Per-contribution panel — v0.2 forensic layer**\n\n"
        "| Support metric | Score | Weight | Weighted contribution |\n|---|---:|---:|---:|\n"
        + "\n".join(rows)
        + "\n\n"
        f"**Base confidence before conflict pressure**: `{breakdown.get('base_confidence_before_conflict')}`\n\n"
        f"**Conflict risk**: `{breakdown.get('conflict_risk')}`\n\n"
        f"**Conflict pressure subtract**: `{breakdown.get('conflict_pressure_subtract')}`\n\n"
        f"**Final reassembly confidence**: `{breakdown.get('reassembly_confidence')}`\n\n"
        f"**False-settlement risk**: `{breakdown.get('false_settlement_risk')}`"
    )


def motion_segment_choices(result: Dict[str, Any]) -> List[str]:
    segments = result.get("motion_route_diagnostics", {}).get("segments", [])
    return [f"{s.get('label')} — {s.get('status').upper()}" for s in segments]


def _segment_from_choice(result: Dict[str, Any], segment_choice: Optional[str]) -> Optional[Dict[str, Any]]:
    segments = result.get("motion_route_diagnostics", {}).get("segments", [])
    if not segments:
        return None
    if not segment_choice:
        return segments[0]
    for segment in segments:
        if segment_choice.startswith(str(segment.get("label"))):
            return segment
    return segments[0]


def motion_segment_detail_markdown(result: Dict[str, Any], segment_choice: Optional[str]) -> str:
    segment = _segment_from_choice(result, segment_choice)
    if not segment:
        return "No motion segment diagnostics available."
    affected = segment.get("affected_atom_ids", [])
    affected_md = "\n".join([f"- `{atom_id}`" for atom_id in affected]) if affected else "- none"
    return (
        f"### {segment.get('label')}\n\n"
        f"**Status**: `{segment.get('status')}`\n\n"
        f"**Coverage delta**: `{segment.get('coverage_delta')}`\n\n"
        f"**Return delta**: `{segment.get('return_delta')}`\n\n"
        f"**Weighted coverage contribution**: `{segment.get('weighted_coverage_contribution')}`\n\n"
        f"**Weighted return contribution**: `{segment.get('weighted_return_contribution')}`\n\n"
        f"**Reported from-frame**: `{segment.get('from_frame_reported')}`\n\n"
        f"**Observed stack positions**: `{segment.get('observed_stack_positions')}`\n\n"
        f"**Receipt line**: {segment.get('receipt_line')}\n\n"
        "**Affected atom IDs**\n" + affected_md
    )


def motion_ledger_html(result: Dict[str, Any]) -> str:
    diagnostics = result.get("motion_route_diagnostics", {})
    segments = diagnostics.get("segments", [])
    colors = {
        "returned": ("#1f8f4d", "green returned"),
        "missing": ("#c43d3d", "red missing"),
        "conflicting": ("#c43d3d", "red conflicting"),
        "shuffled": ("#c98924", "amber shuffled"),
        "reversed": ("#c98924", "amber reversed"),
    }
    if not segments:
        return "<div>No motion route segments available.</div>"
    chips = []
    for s in segments:
        color, label = colors.get(s.get("status"), ("#777", "unknown"))
        chips.append(
            f"<span title='{s.get('receipt_line')}' style='display:inline-block;margin:3px;padding:8px 10px;border-radius:10px;background:{color};color:white;font-family:monospace;font-size:12px;'>"
            f"{s.get('from_frame')}→{s.get('to_frame')} · {s.get('status')}"
            "</span>"
        )
    counts = diagnostics.get("status_counts", {})
    return (
        "<div style='border:1px solid #303642;border-radius:12px;padding:12px;background:#11151f;'>"
        "<div style='font-weight:700;margin-bottom:6px;'>Motion route segment ledger</div>"
        "<div style='font-size:12px;opacity:.88;margin-bottom:10px;'>Green = returned · Amber = shuffled/reversed · Red = missing/conflicting</div>"
        + "".join(chips)
        + f"<div style='font-size:12px;opacity:.9;margin-top:10px;'>coverage_rate={diagnostics.get('coverage_rate')} · return_rate={diagnostics.get('return_rate')} · counts={counts}</div>"
        "</div>"
    )


def show_motion_segment_detail(result: Optional[Dict[str, Any]], segment_choice: Optional[str]) -> str:
    if not result:
        return "Run a proof first, then select a motion segment."
    return motion_segment_detail_markdown(result, segment_choice)


# ---------------------------------------------------------------------------
# Gradio actions
# ---------------------------------------------------------------------------


def build_custom_config(
    source_kind: str,
    frame_count: int,
    include_occlusion: bool,
    missing_frame_enabled: bool,
    missing_frame_index: int,
) -> SyntheticConfig:
    if source_kind == "Moving square / color steady":
        return SyntheticConfig(
            source_name="moving_square_custom",
            frames=int(frame_count),
            shape="square",
            color_mode="steady",
            motion="horizontal",
            include_occlusion=include_occlusion,
            missing_frame=missing_frame_index if missing_frame_enabled else None,
            seed=612,
        )
    return SyntheticConfig(
        source_name="moving_circle_custom",
        frames=int(frame_count),
        shape="circle",
        color_mode="color_shift",
        motion="diagonal",
        include_occlusion=include_occlusion,
        missing_frame=missing_frame_index if missing_frame_enabled else None,
        seed=610,
    )


def run_single_proof(
    source_kind: str,
    frame_count: int,
    include_occlusion: bool,
    missing_frame_enabled: bool,
    missing_frame_index: int,
    stack_mode: str,
) -> Tuple[List[Image.Image], str, str, Image.Image, str, str, str, Any, str, str, str, str, str, str, Dict[str, Any]]:
    config = build_custom_config(source_kind, frame_count, include_occlusion, missing_frame_enabled, missing_frame_index)
    source = generate_synthetic_source(config)
    built = atomize_source(source)
    atomization = built["atomization"]
    dataset = built["holographic_dataset"]

    mode_lookup = {
        "Full stack": "full",
        "Partial stack": "partial",
        "Shuffled temporal order": "shuffled",
        "Drop middle frame": "drop_frame",
    }
    mode = mode_lookup.get(stack_mode, "full")
    trace_stack = make_trace_stack(atomization, mode=mode)
    result = reassemble_trace_stack(trace_stack)
    motion_map = draw_motion_map(result["reconstructed_account"])

    atom_path = write_json_file("atomization.json", atomization)
    stack_path = write_json_file("trace_stack.json", trace_stack)
    result_path = write_json_file("reassembly_result.json", result)
    receipt_path = write_json_file("receipt.json", result["receipt"])
    export_zip = make_zip_file(
        "holographic_trace_stack_exports.zip",
        {
            "atomization.json": atomization,
            "holographic_dataset.json": dataset,
            "trace_stack.json": trace_stack,
            "reassembly_result.json": result,
            "receipt.json": result["receipt"],
        },
    )
    choices = motion_segment_choices(result)
    selected = choices[0] if choices else None
    segment_update = gr.update(choices=choices, value=selected) if gr is not None else selected
    return (
        source_preview_gallery(source),
        atom_summary_markdown(atomization),
        stack_layers_markdown(trace_stack, result),
        motion_map,
        reassembly_markdown(result),
        metrics_markdown(result["metrics"]),
        contribution_breakdown_markdown(result),
        motion_ledger_html(result),
        segment_update,
        motion_segment_detail_markdown(result, selected),
        receipt_markdown(result["receipt"]),
        json.dumps(result, indent=2, sort_keys=True),
        atom_path,
        stack_path,
        export_zip,
        result,
    )


def run_validation_ui() -> Tuple[str, str, str]:
    validation = run_required_tests()
    validation_path = write_json_file("validation_report.json", validation)
    summary_rows = []
    for name, row in validation["summary"].items():
        summary_rows.append(
            f"| `{name}` | `{row['final_state']}` | `{row['pressure_state']}` | `{row['reassembly_confidence']}` | `{row['false_settlement_risk']}` | `{row['passed_boundary_expectation']}` |"
        )
    md = (
        "| Test | Final state | Pressure | Confidence | False settlement risk | Boundary pass |\n"
        "|---|---|---|---:|---:|---|\n"
        + "\n".join(summary_rows)
    )
    return md, json.dumps(validation, indent=2, sort_keys=True), validation_path


def build_app():
    if gr is None:
        raise RuntimeError("Gradio is not installed. Install requirements.txt to launch the Space UI.")

    description = f"""
# {APP_TITLE}
**{APP_SHORT_LINE}**

This is a falsifiable synthetic proof harness for trace-stack reassembly. It uses **holographic dataset** to mean one trace object projected through multiple accountable views. It is **not optical holography**, not a 3D hologram, not a cinematic renderer, and not a live mycelium/geometry renderer.

v0.2 keeps the stable v0.1 six-test harness and adds a scoped forensic layer: per-contribution breakdown plus a compact motion-route segment ledger for live route inspection.

Core route: `source → atomization → trace capsule → holographic dataset → trace stack → midstream reassembly → reprojected dataset → receipt`
"""

    with gr.Blocks(title=APP_TITLE) as demo:
        gr.Markdown(description)
        result_state = gr.State(value=None)
        with gr.Row():
            source_kind = gr.Radio(
                ["Moving circle / color shift", "Moving square / color steady"],
                value="Moving circle / color shift",
                label="Synthetic source",
            )
            frame_count = gr.Slider(5, 12, value=8, step=1, label="Frame count")
            include_occlusion = gr.Checkbox(value=False, label="Optional occlusion")
            missing_frame_enabled = gr.Checkbox(value=False, label="Generate source with missing frame")
            missing_frame_index = gr.Slider(1, 10, value=4, step=1, label="Missing frame index")
            stack_mode = gr.Radio(
                ["Full stack", "Partial stack", "Shuffled temporal order", "Drop middle frame"],
                value="Full stack",
                label="Trace stack test mode",
            )
        run_button = gr.Button("Run trace-stack reassembly proof", variant="primary")

        with gr.Row(equal_height=True):
            with gr.Column(scale=1):
                gr.Markdown("## Panel 1 — Atomized Intake")
                source_gallery = gr.Gallery(label="Source preview", columns=4, height=320)
                atom_summary = gr.Markdown(label="Atom summary")
                atom_file = gr.File(label="Atomization JSON")
            with gr.Column(scale=1):
                gr.Markdown("## Panel 2 — Trace Stack Reassembly")
                stack_summary = gr.Markdown(label="Trace stack layers")
                motion_map = gr.Image(label="Motion route / continuity map", type="pil")
                stack_file = gr.File(label="Trace stack JSON")
            with gr.Column(scale=1):
                gr.Markdown("## Panel 3 — Reprojected Holographic Dataset")
                reassembly_summary = gr.Markdown(label="Reconstructed account")
                metrics = gr.Markdown(label="Metrics")
                receipt = gr.Markdown(label="Receipt")
                export_zip = gr.File(label="JSON export bundle")

        gr.Markdown("## v0.2 Per-Contribution Forensics")
        with gr.Row(equal_height=True):
            with gr.Column(scale=1):
                contribution_breakdown = gr.Markdown(label="Weighted contribution breakdown")
            with gr.Column(scale=1):
                motion_timeline = gr.HTML(label="Motion route segment ledger")
                segment_selector = gr.Dropdown(label="Inspect motion segment", choices=[], interactive=True)
                segment_detail = gr.Markdown(label="Selected segment detail")

        gr.Markdown("## Reassembly Result JSON")
        result_json = gr.Code(label="reassembly_result.json", language="json", lines=18)

        gr.Markdown("## Required Validation Tests")
        validate_button = gr.Button("Run all six required tests")
        validation_summary = gr.Markdown(label="Validation summary")
        validation_json = gr.Code(label="validation_report.json", language="json", lines=18)
        validation_file = gr.File(label="Validation report JSON")

        run_button.click(
            fn=run_single_proof,
            inputs=[source_kind, frame_count, include_occlusion, missing_frame_enabled, missing_frame_index, stack_mode],
            outputs=[
                source_gallery,
                atom_summary,
                stack_summary,
                motion_map,
                reassembly_summary,
                metrics,
                contribution_breakdown,
                motion_timeline,
                segment_selector,
                segment_detail,
                receipt,
                result_json,
                atom_file,
                stack_file,
                export_zip,
                result_state,
            ],
        )
        segment_selector.change(
            fn=show_motion_segment_detail,
            inputs=[result_state, segment_selector],
            outputs=[segment_detail],
        )
        validate_button.click(fn=run_validation_ui, inputs=[], outputs=[validation_summary, validation_json, validation_file])

        demo.load(
            fn=run_single_proof,
            inputs=[source_kind, frame_count, include_occlusion, missing_frame_enabled, missing_frame_index, stack_mode],
            outputs=[
                source_gallery,
                atom_summary,
                stack_summary,
                motion_map,
                reassembly_summary,
                metrics,
                contribution_breakdown,
                motion_timeline,
                segment_selector,
                segment_detail,
                receipt,
                result_json,
                atom_file,
                stack_file,
                export_zip,
                result_state,
            ],
        )
    return demo


if gr is not None:
    demo = build_app()
else:
    demo = None


if __name__ == "__main__":
    if demo is None:
        raise RuntimeError("Gradio UI is unavailable.")
    demo.launch()