"""Throughput scheduling and receipt-owned compact NoNE page growth.

The training helpers preserve every corpus row and every validation row.  Page
growth creates transfer-initialized candidates only; it never advances an
accepted generation and never calls an initialized page trained.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import shutil
import struct
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, cast

import torch

from resynthesis.none_paging import (
    DEFAULT_GPU_PAGE_CACHE_ENTRIES,
    NoNEAcceptedTrainingSaturationPacket,
    NoNECompactTransferTemplate,
    NoNEImmutablePageStore,
    NoNEPageObjectBinding,
    NoNEScaleCohortPacket,
    NoNEScaleEvidencePacket,
    NoNEPageWeights,
    PAGE_STORE_LOCATOR_SCHEMA,
    SCALED_FLOAT8_TRANSFER_STORAGE,
    _default_page_store_registry_roots_boundary,
    _page_store_registry_root_for_boundary,
    _session_key,
    digest_tensor,
    file_sha256_authority_batch_boundary,
    plan_none_scale_cohort,
    validate_none_accepted_training_saturation_boundary,
    validate_page_weights,
)

ACTIVE_FORWARD_PARAM_BUDGET_APPROX = 14_000_000
COMPACT_TRANSFER_JOURNAL_SCHEMA = (
    "nnf.resynthesis.compact_transfer_page_journal.v1"
)
COMPACT_TRANSFER_SUMMARY_SCHEMA = (
    "nnf.resynthesis.compact_transfer_page_summary.v1"
)
COMPACT_TRANSFER_LOCATOR_SCHEMA = (
    "nnf.resynthesis.compact_transfer_page_locator.v1"
)
COMPACT_TRANSFER_REGISTRY_DIRECTORY = "compact-transfer-banks"
TEMPLATE_REFERENCE_TRANSFER_STORAGE = (
    "template_reference_scaled_float8_e4m3fn_implicit_zero_optimizer_v1"
)
TRAINING_THROUGHPUT_LANE_SCHEMA = (
    "nnf.resynthesis.training_throughput_lane.v1"
)

FAILURE_CATEGORY_ORDER: tuple[str, ...] = (
    "stop_overrun_after_exact",
    "correction_transfer_failure",
    "uncertainty_disposition",
    "source_join_chemistry",
    "present_no_provider_checksum",
    "general_knowledge_miss",
    "unknown",
)


@dataclass(frozen=True)
class ThroughputLaneConfig:
    """Successor-lane training geometry, never a grading-surface cap."""

    # Wide retained windows + large residency: physical pages only become live
    # model weight when they stay GPU-resident long enough to be routed. The
    # defaults below favor traversal of the physical bank; promotion still
    # requires held-out / lineage proof and does not relabel storage as trained.
    training_updates_per_validation: int = 512
    training_microbatch_rows: int = 8
    # Validation automatically bisects a batch on OOM, so start wide enough to
    # amortize parent/RBO traversal and ledger boundaries on 96 GiB lanes.
    validation_microbatch_rows: int = 64
    training_emission_backward_chunk_requested: int = 2
    training_emission_backward_chunk: int = 2
    # These are external diagnostic bounds, not production defaults.  ``None``
    # proves that the complete prompt and answer sequence participate.
    training_target_tokens_per_row: int | None = None
    training_prompt_window_tokens: int | None = None
    training_task_wall_clock_seconds: float = 45.0
    # Default off: compile is a post-stability accelerator. Dynamo currently
    # fights dynamic page/sequence geometry; eager traversal of the physical
    # page bank is the faster path to first optimizer commits.
    compile_science_stack: bool = False
    # Explicit canary only.  The bulk completion objective performs its
    # full-vocabulary low-rank projections in float32 before the stable
    # logsumexp reduction.  Hopper can execute those same float32 matmuls
    # through TF32 tensor cores while retaining float32 accumulation and the
    # complete target/vocabulary surface.  Keep the ordinary lane at
    # ``highest`` until a durable rate comparison proves this useful.
    tf32_training_matmul: bool = False
    gpu_resident_page_cache_entries: int = DEFAULT_GPU_PAGE_CACHE_ENTRIES


def complete_sequence_throughput_lane_boundary(value: object) -> bool:
    """Validate uncapped prompt/target participation at a receipt boundary."""

    if not isinstance(value, Mapping):
        return False
    return (
        value.get("schema") == TRAINING_THROUGHPUT_LANE_SCHEMA
        and value.get("completeTargetSequenceUsed") is True
        and value.get("completePromptSequenceUsed") is True
        and value.get("trainingTargetTokensPerRow") is None
        and value.get("trainingPromptWindowTokens") is None
    )


@dataclass(frozen=True)
class CompactPageStreamPacket:
    """Verified append-only capacity observed at an external I/O boundary.

    ``trained_t`` and ``accepted_generation_committed_t`` remain explicit so
    discovery cannot promote transfer initialization into a capability claim.
    """

    store_root: Path
    summary_path: Path
    journal_path: Path
    storage_format: str
    session_id_t: torch.Tensor
    page_ids_t: torch.Tensor
    object_sha256_t: torch.Tensor
    object_bytes_t: torch.Tensor
    trained_t: torch.Tensor
    accepted_generation_committed_t: torch.Tensor


@dataclass(frozen=True)
class CompactPageBankBinding:
    """Verified boundary identity for an untrained compact-page bank.

    Object sizes and mappings are checked for the complete bank during
    discovery. Object payload hashes are checked when a model-selected cohort
    is admitted so discovery does not reread the complete physical bank.
    Writer-source digests are observation-only diagnostics and never form
    part of the bank's authority.
    """

    summary_path: Path
    journal_path: Path
    store_root: Path
    summary_sha256_t: torch.Tensor
    journal_sha256_t: torch.Tensor
    session_id_t: torch.Tensor
    source_checkpoint_sha256_t: torch.Tensor
    writer_source_sha256s_t: torch.Tensor
    page_ids_t: torch.Tensor
    object_sha256s_t: torch.Tensor
    object_bytes_t: torch.Tensor
    page_parameter_elements_t: torch.Tensor
    physical_parameter_elements_t: torch.Tensor
    total_object_bytes_t: torch.Tensor
    storage_format: str
    template_object_path: Path | None
    template_object_sha256_t: torch.Tensor
    template_page_id_t: torch.Tensor
    template_page_id_offset_t: torch.Tensor
    physical_archive_object_bytes_t: torch.Tensor
    ready_t: torch.Tensor


@dataclass(frozen=True)
class CompactPageAdmissionPacket:
    """Tensor-owned mapping from model-selected families to physical pages.

    Parent-family page identities may repeat so one retained family can own
    multiple distinct child pages in the same evidence-bound transaction.
    """

    bank_summary_path: Path
    bank_journal_path: Path
    bank_store_root: Path
    bank_storage_format: str
    bank_template_object_path: Path | None
    bank_summary_sha256_t: torch.Tensor
    bank_journal_sha256_t: torch.Tensor
    bank_session_id_t: torch.Tensor
    bank_source_checkpoint_sha256_t: torch.Tensor
    bank_writer_source_sha256s_t: torch.Tensor
    selected_family_page_ids_t: torch.Tensor
    selected_page_ids_t: torch.Tensor
    selected_object_sha256s_t: torch.Tensor
    selected_object_bytes_t: torch.Tensor
    bank_page_count_t: torch.Tensor
    page_parameter_elements_t: torch.Tensor
    bank_physical_parameter_elements_t: torch.Tensor
    bank_total_object_bytes_t: torch.Tensor
    bank_template_object_sha256_t: torch.Tensor
    bank_template_page_id_t: torch.Tensor
    bank_template_page_id_offset_t: torch.Tensor
    bank_physical_archive_object_bytes_t: torch.Tensor
    selected_page_count_t: torch.Tensor
    ready_t: torch.Tensor


@dataclass(frozen=True)
class DiscoveredCompactAdmissionPlan:
    """Auto-discovered physical bank bound to one model-owned scale cohort."""

    bank: CompactPageBankBinding
    scale_cohort: NoNEScaleCohortPacket
    admission: CompactPageAdmissionPacket


def throughput_lane_config_from_env() -> ThroughputLaneConfig:
    """Read performance geometry without changing model-owned routing."""

    def _pos_int(name: str, default: int) -> int:
        raw = os.environ.get(name, "").strip()
        if not raw:
            return default
        value = int(raw)
        if value < 1:
            raise ValueError(f"{name} must be >= 1")
        return value

    def _nonneg_int(name: str, default: int) -> int:
        raw = os.environ.get(name, "").strip()
        if not raw:
            return default
        value = int(raw)
        if value < 0:
            raise ValueError(f"{name} must be >= 0")
        return value

    def _optional_pos_int(name: str) -> int | None:
        raw = os.environ.get(name, "").strip()
        if not raw:
            return None
        value = int(raw)
        if value < 1:
            raise ValueError(f"{name} must be >= 1")
        return value

    def _pos_float(name: str, default: float) -> float:
        raw = os.environ.get(name, "").strip()
        if not raw:
            return default
        value = float(raw)
        if value <= 0.0:
            raise ValueError(f"{name} must be > 0")
        return value

    def _bool(name: str) -> bool:
        raw = os.environ.get(name, "").strip()
        if not raw:
            return False
        if raw in {"1", "true", "True", "yes"}:
            return True
        if raw in {"0", "false", "False", "no"}:
            return False
        raise ValueError(f"{name} must be a boolean")

    compile_flag = _bool("NNF_RESYNTHESIS_COMPILE_SCIENCE")
    tf32_training_matmul = _bool("NNF_RESYNTHESIS_TF32_TRAINING_MATMUL")
    requested_emission_backward_chunk = _pos_int(
        "NNF_RESYNTHESIS_EMISSION_BACKWARD_CHUNK",
        2,
    )
    return ThroughputLaneConfig(
        training_updates_per_validation=_pos_int(
            "NNF_RESYNTHESIS_TRAINING_UPDATES_PER_VALIDATION",
            512,
        ),
        training_microbatch_rows=_pos_int(
            "NNF_RESYNTHESIS_TRAINING_MICROBATCH_ROWS",
            8,
        ),
        validation_microbatch_rows=_pos_int(
            "NNF_RESYNTHESIS_VALIDATION_MICROBATCH_ROWS",
            64,
        ),
        training_emission_backward_chunk_requested=(
            requested_emission_backward_chunk
        ),
        # The signed caller request owns the grouping geometry.  Silently
        # shrinking it creates a misleading throughput receipt and prevents a
        # clean-lane proof from ever exercising the requested traversal.  An
        # unsafe request therefore fails at the real training boundary rather
        # than being rewritten by a host-side fixed cap.
        training_emission_backward_chunk=requested_emission_backward_chunk,
        training_target_tokens_per_row=_optional_pos_int(
            "NNF_RESYNTHESIS_TRAIN_TARGET_TOKENS_PER_ROW"
        ),
        training_prompt_window_tokens=_optional_pos_int(
            "NNF_RESYNTHESIS_TRAIN_PROMPT_WINDOW_TOKENS"
        ),
        training_task_wall_clock_seconds=_pos_float(
            "NNF_RESYNTHESIS_TRAINING_TASK_WALL_CLOCK_SEC",
            45.0,
        ),
        compile_science_stack=compile_flag,
        tf32_training_matmul=tf32_training_matmul,
        gpu_resident_page_cache_entries=_pos_int(
            "NNF_RESYNTHESIS_GPU_PAGE_CACHE_ENTRIES",
            DEFAULT_GPU_PAGE_CACHE_ENTRIES,
        ),
    )


def configure_training_matmul_precision_boundary(
    lane: ThroughputLaneConfig,
) -> tuple[str, bool]:
    """Install one launch-owned float32 matmul policy before model loading.

    This is a process-global PyTorch execution setting, so it is established
    exactly once at the external CLI boundary before any model or optimizer
    tensors are constructed.  It changes only CUDA float32 matmul execution;
    target participation, loss geometry, routing, and durable cursor authority
    remain unchanged.
    """

    requested_precision = (
        "high" if lane.tf32_training_matmul else "highest"
    )
    torch.set_float32_matmul_precision(requested_precision)
    effective_precision = torch.get_float32_matmul_precision()
    tf32_activated = bool(torch.backends.cuda.matmul.allow_tf32)
    if effective_precision != requested_precision:
        raise RuntimeError("training float32 matmul precision differs")
    if tf32_activated != lane.tf32_training_matmul:
        raise RuntimeError("training TF32 matmul activation differs")
    return effective_precision, tf32_activated


def failure_category_for_row(row: dict[str, Any]) -> str:
    """Map a training-boundary row to a curriculum priority bucket."""

    explicit = row.get("failureCategory")
    if isinstance(explicit, str) and explicit.strip():
        return explicit.strip()
    axis = row.get("generalizationAxis") or row.get("generalization_axis")
    group = str(
        row.get("generalizationGroup") or row.get("generalization_group") or ""
    )
    if axis == "adaptive_source_join":
        return "source_join_chemistry"
    if "uncertainty" in group.lower() or "unsupported" in group.lower():
        return "uncertainty_disposition"
    if row.get("stopOverrunAfterExact") is True:
        return "stop_overrun_after_exact"
    if row.get("correctionTransferFailure") is True:
        return "correction_transfer_failure"
    return "unknown"


def prioritize_train_rows_by_failure_category(
    rows: list[dict[str, Any]],
    *,
    baseline_grade_rows: Iterable[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    """Order all rows by observed weakness without dropping any row."""

    overrun_ids: set[str] = set()
    correction_ids: set[str] = set()
    if baseline_grade_rows is not None:
        for grade in baseline_grade_rows:
            qid = grade.get("questionId") or grade.get("record_id")
            if qid is None:
                continue
            key = str(qid)
            audit = grade.get("generationAuditTrace")
            if isinstance(audit, dict):
                first_exact = audit.get("firstExactAnswerEmissionIndex")
                after = audit.get("tokensAfterFirstExactAnswerBoundary")
                if (
                    isinstance(first_exact, int)
                    and first_exact >= 0
                    and isinstance(after, int)
                    and after > 0
                    and not grade.get("passed")
                ):
                    overrun_ids.add(key)
            if grade.get("passed") is False and grade.get("attempt") == 2:
                correction_ids.add(key)

    def sort_key(row: dict[str, Any]) -> tuple[int, str]:
        qid = str(row.get("question_id") or row.get("record_id") or "")
        category = failure_category_for_row(row)
        if qid in overrun_ids:
            category = "stop_overrun_after_exact"
        elif qid in correction_ids:
            category = "correction_transfer_failure"
        try:
            rank = FAILURE_CATEGORY_ORDER.index(category)
        except ValueError:
            rank = len(FAILURE_CATEGORY_ORDER)
        return rank, qid

    ordered = sorted(rows, key=sort_key)
    if len(ordered) != len(rows):
        raise RuntimeError("curriculum ordering dropped training rows")
    return ordered


def group_train_rows_into_microbatches(
    rows: list[dict[str, Any]],
    *,
    microbatch_rows: int,
) -> list[list[dict[str, Any]]]:
    """Pack every row into geometry-owned microbatches exactly once."""

    if microbatch_rows < 1:
        raise ValueError("microbatch_rows must be >= 1")
    batches = [
        rows[cursor : cursor + microbatch_rows]
        for cursor in range(0, len(rows), microbatch_rows)
    ]
    if sum(len(batch) for batch in batches) != len(rows):
        raise RuntimeError("microbatch construction dropped training rows")
    return batches


def load_baseline_grade_rows(path: str) -> list[dict[str, Any]]:
    """Load pre-update grade rows from an execution-ledger boundary."""

    grades: list[dict[str, Any]] = []
    with open(path, encoding="utf-8") as handle:
        for line in handle:
            line = line.strip()
            if not line:
                continue
            try:
                row = json.loads(line)
            except json.JSONDecodeError:
                continue
            if row.get("schema") != "nnf.resynthesis.grade.v1":
                continue
            if row.get("evaluationPhase") != "pre_update_baseline":
                continue
            grades.append(row)
    return grades


def _science_compile_recurrent_types() -> tuple[type[torch.nn.Module], ...]:
    """Return recurrent module types that never enter the compiled region."""

    from resynthesis.kda_expert import CRConditionedKDAExpert

    return (
        torch.nn.RNN,
        torch.nn.LSTM,
        torch.nn.GRU,
        torch.nn.RNNCell,
        torch.nn.LSTMCell,
        torch.nn.GRUCell,
        CRConditionedKDAExpert,
    )


def _science_compile_targets(
    stack: torch.nn.Module,
) -> tuple[
    tuple[tuple[str, torch.nn.Module], ...],
    tuple[tuple[str, torch.nn.Module], ...],
]:
    """Select graph-heavy, non-recurrent kernels from the science stack."""

    from resynthesis.science_layers import IntentContextPivotAttention

    recurrent_types = _science_compile_recurrent_types()
    excluded_recurrent = tuple(
        (name, module)
        for name, module in stack.named_modules()
        if name and isinstance(module, recurrent_types)
    )
    selected: list[tuple[str, torch.nn.Module]] = []
    for name, module in stack.named_modules():
        # DeltaBlockAttnRes selects a model-owned depth between one and
        # max_blocks. Compiling that dynamic-depth module produced shape
        # recompiles and eventually a detached output on the live r152 B2
        # canary. Keep it eager; only the fixed-geometry attention kernel is a
        # supported compile target.
        if not isinstance(module, IntentContextPivotAttention):
            continue
        if any(
            nested is not module and isinstance(nested, recurrent_types)
            for nested in module.modules()
        ):
            continue
        selected.append((name, module))
    return tuple(selected), excluded_recurrent


def _science_compile_canary_forward_backward(
    module: torch.nn.Module,
    *,
    compiled: bool,
) -> tuple[
    torch.Tensor,
    torch.Tensor,
    tuple[torch.Tensor | None, ...],
]:
    """Run one real forward/backward without touching parameter gradients."""

    from resynthesis.delta_attn_res import DeltaBlockAttnRes
    from resynthesis.science_layers import IntentContextPivotAttention

    # Scalar control parameters intentionally remain fp32 while the expensive
    # projection weights may be bf16 on the live training lane.  The canary
    # input must follow the first projection that consumes it, not whichever
    # direct scalar happens to be first in Module.parameters(); otherwise a
    # valid mixed-precision model fails before torch.compile is even exercised.
    if isinstance(module, IntentContextPivotAttention):
        reference_t = module.q_proj.weight
    elif isinstance(module, DeltaBlockAttnRes):
        reference_t = module.query_proj.weight
    else:
        raise RuntimeError("science compile canary target is unsupported")
    if reference_t.device.type == "meta":
        raise RuntimeError("science compile canary has no materialized parameter")
    trainable_parameters = tuple(
        parameter
        for _name, parameter in module.named_parameters()
        if parameter.requires_grad
    )
    # Paged NoNE intentionally freezes the dense science kernels while the
    # external page/RBO/Fabric pathway remains trainable through their input.
    # Such a kernel still participates in backward and is a valid compile
    # target: the contract to prove is its exact input gradient. Parameter
    # gradients are additionally compared only when this particular runtime
    # leaves local parameters trainable.
    dtype = (
        reference_t.dtype
        if reference_t.dtype.is_floating_point
        else torch.float32
    )
    hidden_size = int(getattr(module, "hidden_size", 0))
    glyph_dim = int(getattr(module, "glyph_dim", 0))
    if hidden_size < 1 or glyph_dim < 1:
        raise RuntimeError("science compile canary geometry is invalid")
    hidden_t = torch.ones(
        1,
        2,
        hidden_size,
        device=reference_t.device,
        dtype=dtype,
        requires_grad=True,
    )
    glyph_t = torch.ones(
        1,
        2,
        glyph_dim,
        device=reference_t.device,
        dtype=dtype,
    )
    forward = module if compiled else module.forward
    if isinstance(module, IntentContextPivotAttention):
        output_t = forward(
            hidden_t,
            intent_glyph_context=glyph_t,
            action_glyph_context=glyph_t,
            relation_glyph_context=glyph_t,
        )
    elif isinstance(module, DeltaBlockAttnRes):
        output_t = forward(
            hidden_t,
            delta_bank_t=hidden_t.detach().unsqueeze(0),
            intent_glyph_t=glyph_t,
            relation_glyph_t=glyph_t,
            relation_bank_t=glyph_t.detach().unsqueeze(0),
        )
    if (
        not isinstance(output_t, torch.Tensor)
        or output_t.shape != hidden_t.shape
        or not output_t.isfinite().all()
        or not output_t.requires_grad
    ):
        raise RuntimeError("science compile forward canary failed")
    gradients = torch.autograd.grad(
        output_t.float().square().mean(),
        (hidden_t, *trainable_parameters),
        retain_graph=False,
        create_graph=False,
        allow_unused=True,
    )
    gradient_t = gradients[0]
    parameter_gradients = gradients[1:]
    if (
        gradient_t is None
        or gradient_t.shape != hidden_t.shape
        or not gradient_t.isfinite().all()
        or (
            bool(trainable_parameters)
            and not any(
                parameter_gradient_t is not None
                for parameter_gradient_t in parameter_gradients
            )
        )
        or any(
            parameter_gradient_t is not None
            and not parameter_gradient_t.isfinite().all()
            for parameter_gradient_t in parameter_gradients
        )
    ):
        raise RuntimeError("science compile backward canary failed")
    return (
        output_t.detach(),
        gradient_t.detach(),
        tuple(
            parameter_gradient_t.detach()
            if parameter_gradient_t is not None
            else None
            for parameter_gradient_t in parameter_gradients
        ),
    )


def _named_identity_stable(
    expected: tuple[tuple[str, object], ...],
    active: tuple[tuple[str, object], ...],
) -> bool:
    return len(expected) == len(active) and all(
        expected_name == active_name and expected_value is active_value
        for (expected_name, expected_value), (
            active_name,
            active_value,
        ) in zip(expected, active, strict=True)
    )


def _optimizer_parameter_names_snapshot(group: dict[str, Any]) -> object:
    names = group.get("param_names")
    return tuple(names) if isinstance(names, (list, tuple)) else names


def maybe_compile_science_stack(
    model: Any,
    *,
    optimizer: torch.optim.Optimizer | None = None,
) -> bool:
    """Compile proven non-recurrent science kernels after optimizer restore.

    ``nn.Module.compile`` is lazy, so registration alone is not activation.
    Every selected kernel must execute an eager and compiled forward/backward
    canary with matching outputs/gradients before the receipt can call the
    compiler active.  GRU, RNN, and recurrent KDA modules stay eager.
    """

    stack = getattr(model, "science_stack", None)
    if not isinstance(stack, torch.nn.Module):
        return False
    if getattr(model, "_science_stack_compiled", False):
        return True
    model._science_stack_compiled = False
    model._science_stack_compile_canary_passed = False
    model._science_stack_compile_failure = None
    model._science_stack_optimizer_parameter_names_stable = False
    if not torch.cuda.is_available():
        model._science_stack_compile_failure = "CUDA is unavailable"
        return False
    selected, excluded_recurrent = _science_compile_targets(stack)
    model._science_stack_compile_target_count = len(selected)
    model._science_stack_compile_excluded_recurrent_count = len(
        excluded_recurrent
    )
    model._science_stack_compile_target_names = tuple(
        name for name, _module in selected
    )
    model._science_stack_compile_excluded_recurrent_names = tuple(
        name for name, _module in excluded_recurrent
    )
    if not selected:
        model._science_stack_compile_failure = (
            "no supported non-recurrent science kernels were found"
        )
        return False

    module_identity = tuple(model.named_modules())
    parameter_identity = tuple(model.named_parameters())
    buffer_identity = tuple(model.named_buffers())
    optimizer_group_identity = (
        tuple(
            (
                group,
                tuple(group["params"]),
                _optimizer_parameter_names_snapshot(group),
            )
            for group in optimizer.param_groups
        )
        if optimizer is not None
        else ()
    )
    optimizer_state_identity = (
        tuple((parameter, state) for parameter, state in optimizer.state.items())
        if optimizer is not None
        else ()
    )
    recurrent_compile_identity = tuple(
        (module, getattr(module, "_compiled_call_impl", None))
        for _name, module in excluded_recurrent
    )
    target_compile_identity = tuple(
        (module, getattr(module, "_compiled_call_impl", None))
        for _name, module in selected
    )
    parameter_gradient_identity = tuple(
        (
            parameter,
            parameter.grad,
            parameter.grad.detach().clone()
            if parameter.grad is not None
            else None,
        )
        for _name, parameter in parameter_identity
    )
    delta_depth_weight_identity = tuple(
        (
            module,
            getattr(module, "last_depth_weights_t", None),
        )
        for _name, module in selected
        if hasattr(module, "last_depth_weights_t")
    )

    compile_error: Exception | None = None
    try:
        import torch._functorch.config as _functorch_config

        _functorch_config.donated_buffer = False
        for _name, target in selected:
            (
                eager_output_t,
                eager_gradient_t,
                eager_parameter_gradients,
            ) = (
                _science_compile_canary_forward_backward(
                    target,
                    compiled=False,
                )
            )
            target.compile(
                mode="default",
                fullgraph=False,
            )
            (
                compiled_output_t,
                compiled_gradient_t,
                compiled_parameter_gradients,
            ) = (
                _science_compile_canary_forward_backward(
                    target,
                    compiled=True,
                )
            )
            torch.testing.assert_close(
                compiled_output_t,
                eager_output_t,
                rtol=5.0e-2,
                atol=5.0e-3,
            )
            torch.testing.assert_close(
                compiled_gradient_t,
                eager_gradient_t,
                rtol=5.0e-2,
                atol=5.0e-3,
            )
            if (
                len(compiled_parameter_gradients)
                != len(eager_parameter_gradients)
                or any(
                    (compiled_parameter_gradient_t is None)
                    != (eager_parameter_gradient_t is None)
                    for (
                        compiled_parameter_gradient_t,
                        eager_parameter_gradient_t,
                    ) in zip(
                        compiled_parameter_gradients,
                        eager_parameter_gradients,
                        strict=True,
                    )
                )
            ):
                raise RuntimeError(
                    "science compile parameter-gradient topology differs"
                )
            for (
                compiled_parameter_gradient_t,
                eager_parameter_gradient_t,
            ) in zip(
                compiled_parameter_gradients,
                eager_parameter_gradients,
                strict=True,
            ):
                if (
                    compiled_parameter_gradient_t is not None
                    and eager_parameter_gradient_t is not None
                ):
                    torch.testing.assert_close(
                        compiled_parameter_gradient_t,
                        eager_parameter_gradient_t,
                        rtol=5.0e-2,
                        atol=5.0e-3,
                    )
    except Exception as error:
        compile_error = error
    finally:
        for delta_module, prior_depth_weights_t in (
            delta_depth_weight_identity
        ):
            setattr(
                delta_module,
                "last_depth_weights_t",
                prior_depth_weights_t,
            )

    active_module_identity = tuple(model.named_modules())
    active_parameter_identity = tuple(model.named_parameters())
    active_buffer_identity = tuple(model.named_buffers())
    optimizer_identity_stable = bool(
        optimizer is None
        or (
            len(optimizer_group_identity) == len(optimizer.param_groups)
            and all(
                expected_group is active_group
                and len(expected_parameters) == len(active_group["params"])
                and expected_parameter_names
                == _optimizer_parameter_names_snapshot(active_group)
                and all(
                    expected_parameter is active_parameter
                    for expected_parameter, active_parameter in zip(
                        expected_parameters,
                        active_group["params"],
                        strict=True,
                    )
                )
                for (
                    expected_group,
                    expected_parameters,
                    expected_parameter_names,
                ), active_group in zip(
                    optimizer_group_identity,
                    optimizer.param_groups,
                    strict=True,
                )
            )
            and len(optimizer_state_identity) == len(optimizer.state)
            and all(
                optimizer.state.get(parameter) is state
                for parameter, state in optimizer_state_identity
            )
        )
    )
    recurrent_identity_stable = all(
        getattr(module, "_compiled_call_impl", None) is compiled_call
        for module, compiled_call in recurrent_compile_identity
    )
    gradient_identity_stable = all(
        parameter.grad is prior_gradient
        and (
            prior_gradient_value is None
            or (
                parameter.grad is not None
                and torch.equal(parameter.grad, prior_gradient_value)
            )
        )
        for parameter, prior_gradient, prior_gradient_value in (
            parameter_gradient_identity
        )
    )
    identity_stable = bool(
        getattr(model, "science_stack", None) is stack
        and _named_identity_stable(module_identity, active_module_identity)
        and _named_identity_stable(parameter_identity, active_parameter_identity)
        and _named_identity_stable(buffer_identity, active_buffer_identity)
        and optimizer_identity_stable
        and recurrent_identity_stable
        and gradient_identity_stable
    )
    if not identity_stable:
        for target, compiled_call in target_compile_identity:
            target._compiled_call_impl = compiled_call
        model._science_stack_compile_failure = (
            "science-stack compilation changed model or optimizer identity"
        )
        raise RuntimeError(
            "science-stack compilation changed model, parameter, buffer, "
            "gradient, or optimizer identity"
        ) from compile_error
    if compile_error is not None:
        for target, compiled_call in target_compile_identity:
            target._compiled_call_impl = compiled_call
        model._science_stack_compile_failure = (
            f"{type(compile_error).__name__}: {compile_error}"
        )
        return False
    model._science_stack_optimizer_parameter_names_stable = bool(
        optimizer is not None and optimizer_identity_stable
    )
    model._science_stack_compile_canary_passed = True
    model._science_stack_compiled = True
    return True


def estimate_page_storage_bytes(
    *,
    page_parameter_elements: int,
    page_count: int,
    bytes_per_element: int = 2,
) -> int:
    """Return the physical model-weight storage envelope for a page cohort."""

    if page_parameter_elements < 1 or page_count < 0 or bytes_per_element < 1:
        raise ValueError("page storage geometry is malformed")
    return page_parameter_elements * page_count * bytes_per_element


def bulk_transfer_init_pages(
    *,
    source_gate_t: torch.Tensor,
    source_up_t: torch.Tensor,
    source_down_t: torch.Tensor,
    source_glyph_down_t: torch.Tensor,
    source_glyph_up_t: torch.Tensor,
    objects_root: Path,
    start_page_id: int,
    count: int,
    router_size: int = 8,
    batch_size: int = 100,
) -> list[tuple[int, str]]:
    """Persist transfer-initialized pages without making a trained claim.

    This compatibility boundary restores the original bulk-catalog API. New
    growth uses :func:`write_compact_transfer_pages`; both surfaces retain the
    page ID in the immutable payload and store objects below ``sha256/``.
    """

    from safetensors.torch import save_file

    if start_page_id < 0 or count < 0 or router_size < 1 or batch_size < 1:
        raise ValueError("bulk transfer page geometry is malformed")
    if source_gate_t.shape != source_up_t.shape:
        raise ValueError("bulk transfer gate/up geometry differs")
    shard_root = Path(objects_root).expanduser().resolve() / "sha256"
    shard_root.mkdir(parents=True, exist_ok=True)
    gate_t = source_gate_t.detach().cpu().contiguous()
    up_t = source_up_t.detach().cpu().contiguous()
    down_t = source_down_t.detach().cpu().contiguous()
    glyph_down_t = source_glyph_down_t.detach().cpu().contiguous()
    glyph_up_t = source_glyph_up_t.detach().cpu().contiguous()
    flat_parameter_count = sum(
        tensor.numel()
        for tensor in (gate_t, up_t, down_t, glyph_down_t, glyph_up_t)
    ) + 1 + 3 * router_size
    results: list[tuple[int, str]] = []
    for page_id in range(start_page_id, start_page_id + count):
        payload = {
            "format_revision_t": torch.full((1,), 2, dtype=torch.long),
            "page_ids_t": torch.full((1,), page_id, dtype=torch.long),
            "ffn_mode_t": torch.zeros(1, 1, dtype=gate_t.dtype),
            "gate_t": gate_t.unsqueeze(0),
            "up_t": up_t.unsqueeze(0),
            "down_t": down_t.unsqueeze(0),
            "glyph_down_t": glyph_down_t.unsqueeze(0),
            "glyph_up_t": glyph_up_t.unsqueeze(0),
            "translation_gate_t": torch.zeros(1, 1, dtype=gate_t.dtype),
            "outcome_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
            "repair_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
            "transfer_memory_t": torch.zeros(1, router_size, dtype=gate_t.dtype),
            "optimizer_mean_t": torch.zeros(1, flat_parameter_count),
            "optimizer_square_t": torch.zeros(1, flat_parameter_count),
            "step_t": torch.zeros(1, dtype=torch.long),
        }
        temporary = shard_root / f".bulk_page.{os.getpid()}.{page_id}.tmp"
        temporary.unlink(missing_ok=True)
        save_file(payload, str(temporary))
        with temporary.open("rb") as handle:
            os.fsync(handle.fileno())
        object_sha256 = _file_sha256(temporary)
        object_path = shard_root / f"{object_sha256}.safetensors"
        if object_path.is_file():
            if _file_sha256(object_path) != object_sha256:
                raise RuntimeError("existing bulk page object differs")
            temporary.unlink()
        else:
            os.replace(temporary, object_path)
        results.append((page_id, object_sha256))
    return results


def page_count_for_physical_target(
    *,
    target_parameter_elements: int,
    current_parameter_elements: int,
    page_parameter_elements: int,
) -> int:
    """Derive needed physical pages from observed model and page geometry."""

    if target_parameter_elements < 1:
        raise ValueError("physical parameter target must be positive")
    if current_parameter_elements < 0:
        raise ValueError("current physical parameters cannot be negative")
    if page_parameter_elements < 1:
        raise ValueError("page parameter geometry must be positive")
    remaining = max(0, target_parameter_elements - current_parameter_elements)
    return math.ceil(remaining / page_parameter_elements)


def _atomic_json(path: Path, payload: Mapping[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + f".{os.getpid()}.tmp")
    temporary.write_text(
        json.dumps(payload, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )
    with temporary.open("rb") as handle:
        os.fsync(handle.fileno())
    os.replace(temporary, path)


def _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _fsync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _digest_hex(digest_t: torch.Tensor) -> str:
    values = digest_t.detach().cpu().to(dtype=torch.uint8).reshape(-1).tolist()
    return bytes(values).hex()


def compact_transfer_weights_from_source(
    *,
    page_id: int,
    source_gate_t: torch.Tensor,
    source_up_t: torch.Tensor,
    source_down_t: torch.Tensor,
    source_glyph_down_t: torch.Tensor,
    source_glyph_up_t: torch.Tensor,
    router_size: int,
) -> NoNEPageWeights:
    """Build one zero-residual, untrained page without optimizer moments."""

    if page_id < 0 or router_size < 1:
        raise ValueError("page identity and router geometry must be nonnegative")
    dtype = source_gate_t.dtype
    weights = NoNEPageWeights(
        page_ids_t=torch.tensor([page_id], dtype=torch.long),
        ffn_mode_t=torch.zeros(1, 1, dtype=dtype),
        gate_t=source_gate_t.detach().cpu().unsqueeze(0).contiguous(),
        up_t=source_up_t.detach().cpu().unsqueeze(0).contiguous(),
        down_t=source_down_t.detach().cpu().unsqueeze(0).contiguous(),
        glyph_down_t=(
            source_glyph_down_t.detach().cpu().unsqueeze(0).contiguous()
        ),
        glyph_up_t=source_glyph_up_t.detach().cpu().unsqueeze(0).contiguous(),
        translation_gate_t=torch.zeros(1, 1, dtype=dtype),
        outcome_memory_t=torch.zeros(1, router_size, dtype=dtype),
        repair_memory_t=torch.zeros(1, router_size, dtype=dtype),
        transfer_memory_t=torch.zeros(1, router_size, dtype=dtype),
    )
    validate_page_weights(weights)
    return weights


def _read_compact_journal(
    path: Path,
    *,
    source_sha256: str,
    session_id: list[int],
) -> dict[int, dict[str, Any]]:
    if not path.is_file():
        return {}
    records: dict[int, dict[str, Any]] = {}
    with path.open(encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, start=1):
            line = line.strip()
            if not line:
                continue
            row = json.loads(line)
            page_id = row.get("pageId")
            if (
                row.get("schema") != COMPACT_TRANSFER_JOURNAL_SCHEMA
                or row.get("sourceCheckpointSha256") != source_sha256
                or row.get("sessionId") != session_id
                or row.get("trained") is not False
                or row.get("acceptedGenerationCommitted") is not False
                or not isinstance(page_id, int)
                or isinstance(page_id, bool)
                or page_id in records
            ):
                raise RuntimeError(
                    f"compact page journal differs at line {line_number}"
                )
            records[page_id] = row
    return records


def _binding_from_record(record: Mapping[str, Any]) -> NoNEPageObjectBinding:
    page_id = record.get("pageId")
    sha256 = record.get("objectSha256")
    object_bytes = record.get("objectBytes")
    if (
        not isinstance(page_id, int)
        or isinstance(page_id, bool)
        or not isinstance(sha256, str)
        or len(sha256) != 64
        or not isinstance(object_bytes, int)
        or isinstance(object_bytes, bool)
        or object_bytes < 1
    ):
        raise RuntimeError("compact page journal object identity is malformed")
    return NoNEPageObjectBinding(
        page_id_t=torch.tensor(page_id, dtype=torch.long),
        object_sha256_t=digest_tensor(sha256),
        object_bytes_t=torch.tensor(object_bytes, dtype=torch.long),
    )


def _validated_sha256_bytes(value: object, *, field: str) -> bytes:
    if not isinstance(value, str) or len(value) != 64:
        raise RuntimeError(f"{field} is not a SHA-256 digest")
    try:
        raw = bytes.fromhex(value)
    except ValueError as exc:
        raise RuntimeError(f"{field} is not a SHA-256 digest") from exc
    if len(raw) != 32 or value != value.lower():
        raise RuntimeError(f"{field} is not a canonical SHA-256 digest")
    return raw


def _diagnostic_sha256_bytes(value: object) -> bytes | None:
    """Decode an optional diagnostic digest without granting it authority."""

    if (
        not isinstance(value, str)
        or len(value) != 64
        or value != value.lower()
        or any(character not in "0123456789abcdef" for character in value)
    ):
        return None
    return bytes.fromhex(value)


def _positive_int(value: object, *, field: str) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value < 1:
        raise RuntimeError(f"{field} must be a positive integer")
    return value


def _nonnegative_int(value: object, *, field: str) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value < 0:
        raise RuntimeError(f"{field} must be a nonnegative integer")
    return value


def _template_page_id_offset_from_payload_boundary(payload: bytearray) -> int:
    """Return the sole signed page-identity slot from one safetensors object."""

    if len(payload) < 8:
        raise RuntimeError("compact template object is truncated")
    header_bytes = int(struct.unpack_from("<Q", payload, 0)[0])
    header_end = 8 + header_bytes
    if header_end > len(payload):
        raise RuntimeError("compact template header escapes payload")
    try:
        header = json.loads(payload[8:header_end])
    except json.JSONDecodeError as exc:
        raise RuntimeError("compact template header is malformed") from exc
    page_record = header.get("page_ids_t") if isinstance(header, dict) else None
    offsets = page_record.get("data_offsets") if isinstance(page_record, dict) else None
    if (
        not isinstance(offsets, list)
        or len(offsets) != 2
        or not all(isinstance(value, int) for value in offsets)
        or offsets[1] - offsets[0] != 8
    ):
        raise RuntimeError("compact template page identity offset differs")
    offset_values = cast(list[int], offsets)
    page_id_offset = header_end + int(offset_values[0])
    if page_id_offset < header_end or page_id_offset + 8 > len(payload):
        raise RuntimeError("compact template page identity escaped payload")
    return page_id_offset


def _template_reference_payload_boundary(
    *,
    template_path: Path,
    template_sha256: str,
    template_bytes: int,
    template_page_id: int,
    page_id_offset: int,
) -> NoNECompactTransferTemplate:
    """Load one immutable transfer template and verify its identity slot.

    Template-reference banks retain the original byte-for-byte transfer object
    once.  Each virtual page is reconstructed only by changing this isolated
    signed 64-bit identity slot, then checked against its journaled digest.
    """

    if (
        not template_path.is_file()
        or template_path.stat().st_size != template_bytes
        or _file_sha256(template_path) != template_sha256
    ):
        raise RuntimeError("compact template object identity differs")
    payload = bytearray(template_path.read_bytes())
    if len(payload) != template_bytes:
        raise RuntimeError("compact template object is truncated")
    observed_offset = _template_page_id_offset_from_payload_boundary(payload)
    if (
        observed_offset != page_id_offset
        or struct.unpack_from("<q", payload, page_id_offset)[0]
        != template_page_id
    ):
        raise RuntimeError("compact template page identity differs")
    return NoNECompactTransferTemplate(
        payload=payload,
        page_id_offset=page_id_offset,
    )


def _template_reference_fields_boundary(
    *,
    payload: Mapping[str, Any],
    records: Mapping[int, Mapping[str, Any]],
    objects_root: Path,
) -> tuple[Path, bytes, int, int, int]:
    """Validate the single retained object that backs virtual page identities."""

    template_digest = _validated_sha256_bytes(
        payload.get("templateObjectSha256"),
        field="templateObjectSha256",
    )
    template_bytes = _positive_int(
        payload.get("templateObjectBytes"),
        field="templateObjectBytes",
    )
    template_page_id = _nonnegative_int(
        payload.get("templatePageId"),
        field="templatePageId",
    )
    page_id_offset = _nonnegative_int(
        payload.get("templatePageIdOffset"),
        field="templatePageIdOffset",
    )
    archive_bytes = _positive_int(
        payload.get("physicalArchiveObjectBytes"),
        field="physicalArchiveObjectBytes",
    )
    row = records.get(template_page_id)
    if (
        row is None
        or row.get("objectSha256") != template_digest.hex()
        or row.get("objectBytes") != template_bytes
        or archive_bytes != template_bytes
    ):
        raise RuntimeError("compact template reference journal identity differs")
    template_path = objects_root / f"{template_digest.hex()}.safetensors"
    _template_reference_payload_boundary(
        template_path=template_path,
        template_sha256=template_digest.hex(),
        template_bytes=template_bytes,
        template_page_id=template_page_id,
        page_id_offset=page_id_offset,
    )
    return (
        template_path,
        template_digest,
        template_page_id,
        page_id_offset,
        archive_bytes,
    )


def discover_compact_transfer_page_bank(
    summary_path: Path,
) -> CompactPageBankBinding:
    """Discover and validate a complete untrained compact-page bank.

    The summary and journal are content-addressed.  A conventional bank keeps
    one object per page; a template-reference bank keeps one immutable object
    plus every expected page digest.  Both defer full page hashing to the
    selected-cohort materialization boundary.
    """

    resolved_summary = summary_path.expanduser().resolve()
    if not resolved_summary.is_file():
        raise FileNotFoundError(resolved_summary)
    payload = json.loads(resolved_summary.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise RuntimeError("compact page summary is not an object")
    requested_count = _positive_int(
        payload.get("requestedPageCount"),
        field="requestedPageCount",
    )
    start_page_id = _nonnegative_int(
        payload.get("startPageId"),
        field="startPageId",
    )
    page_parameter_elements = _positive_int(
        payload.get("pageParameterElements"),
        field="pageParameterElements",
    )
    total_object_bytes = _positive_int(
        payload.get("objectBytes"),
        field="objectBytes",
    )
    session_values = payload.get("sessionId")
    writer_values = payload.get("writerSourceSha256s", [])
    journal_value = payload.get("journalPath")
    store_value = payload.get("storeRoot")
    storage_format = payload.get("storageFormat")
    if (
        payload.get("schema") != COMPACT_TRANSFER_SUMMARY_SCHEMA
        or payload.get("passed") is not True
        or storage_format
        not in {
            SCALED_FLOAT8_TRANSFER_STORAGE,
            TEMPLATE_REFERENCE_TRANSFER_STORAGE,
        }
        or payload.get("completePageMappingRetained") is not True
        or payload.get("acceptedGenerationCommitted") is not False
        or payload.get("promotionRequiresTrainingAndHeldoutProof") is not True
        or payload.get("trainedPageCount") != 0
        or payload.get("journaledPageCount") != requested_count
        or payload.get("transferInitializedPageCount") != requested_count
        or payload.get("physicalParameterElementsInitialized")
        != requested_count * page_parameter_elements
        or not isinstance(journal_value, str)
        or not journal_value
        or not isinstance(store_value, str)
        or not store_value
        or not isinstance(session_values, list)
        or len(session_values) != 4
        or any(
            not isinstance(value, int) or isinstance(value, bool)
            for value in session_values
        )
    ):
        raise RuntimeError("compact page summary authority differs")
    source_digest = _validated_sha256_bytes(
        payload.get("sourceCheckpointSha256"),
        field="sourceCheckpointSha256",
    )
    writer_digests = tuple(
        dict.fromkeys(
            digest
            for value in writer_values
            for digest in (_diagnostic_sha256_bytes(value),)
            if digest is not None
        )
    ) if isinstance(writer_values, list) else ()
    recorded_journal = Path(journal_value).expanduser()
    recorded_store = Path(store_value).expanduser()
    resolved_store = (
        recorded_store.resolve()
        if recorded_store.is_dir()
        else resolved_summary.parent.resolve()
    )
    resolved_journal = (
        recorded_journal.resolve()
        if recorded_journal.is_file()
        else (resolved_store / recorded_journal.name).resolve()
    )
    if not resolved_journal.is_file() or not resolved_store.is_dir():
        raise RuntimeError("compact page bank storage is incomplete")
    journal_sha256 = _file_sha256(resolved_journal)
    expected_journal_digest = _validated_sha256_bytes(
        payload.get("journalSha256"),
        field="journalSha256",
    )
    if journal_sha256 != expected_journal_digest.hex():
        raise RuntimeError("compact page journal SHA-256 differs")
    records = _read_compact_journal(
        resolved_journal,
        source_sha256=source_digest.hex(),
        session_id=session_values,
    )
    requested_ids = tuple(range(start_page_id, start_page_id + requested_count))
    if tuple(sorted(records)) != requested_ids:
        raise RuntimeError("compact page journal mapping is incomplete")

    object_digests = bytearray()
    object_sizes: list[int] = []
    observed_object_digests: set[bytes] = set()
    objects_root = resolved_store / "objects" / "sha256"
    if not objects_root.is_dir():
        raise RuntimeError("compact page bank object root is absent")
    template_path: Path | None = None
    template_digest_t = torch.zeros(0, dtype=torch.uint8)
    template_page_id_t = torch.zeros(0, dtype=torch.long)
    template_page_id_offset_t = torch.zeros(0, dtype=torch.long)
    physical_archive_object_bytes = total_object_bytes
    if storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE:
        (
            template_path,
            template_digest,
            template_page_id,
            template_page_id_offset,
            physical_archive_object_bytes,
        ) = _template_reference_fields_boundary(
            payload=payload,
            records=records,
            objects_root=objects_root,
        )
        template_digest_t = torch.tensor(list(template_digest), dtype=torch.uint8)
        template_page_id_t = torch.tensor(template_page_id, dtype=torch.long)
        template_page_id_offset_t = torch.tensor(
            template_page_id_offset,
            dtype=torch.long,
        )
    for page_id in requested_ids:
        row = records[page_id]
        object_digest = _validated_sha256_bytes(
            row.get("objectSha256"),
            field=f"page[{page_id}].objectSha256",
        )
        object_bytes = _positive_int(
            row.get("objectBytes"),
            field=f"page[{page_id}].objectBytes",
        )
        object_path = objects_root / f"{object_digest.hex()}.safetensors"
        if (
            row.get("storageFormat") != SCALED_FLOAT8_TRANSFER_STORAGE
            or object_digest in observed_object_digests
            or (
                storage_format == SCALED_FLOAT8_TRANSFER_STORAGE
                and (
                    not object_path.is_file()
                    or object_path.stat().st_size != object_bytes
                )
            )
            or (
                storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE
                and template_path is not None
                and object_bytes != template_path.stat().st_size
            )
        ):
            raise RuntimeError(
                f"compact page object mapping differs for page {page_id}"
            )
        observed_object_digests.add(object_digest)
        object_digests.extend(object_digest)
        object_sizes.append(object_bytes)
    if sum(object_sizes) != total_object_bytes:
        raise RuntimeError("compact page bank object-byte total differs")

    object_sha256s_t = torch.frombuffer(
        object_digests,
        dtype=torch.uint8,
    ).clone().reshape(requested_count, 32)
    writer_sha256s_t = (
        torch.tensor(
            [list(value) for value in writer_digests],
            dtype=torch.uint8,
        )
        if writer_digests
        else torch.zeros((0, 32), dtype=torch.uint8)
    )
    return CompactPageBankBinding(
        summary_path=resolved_summary,
        journal_path=resolved_journal,
        store_root=resolved_store,
        summary_sha256_t=digest_tensor(_file_sha256(resolved_summary)),
        journal_sha256_t=digest_tensor(journal_sha256),
        session_id_t=torch.tensor(session_values, dtype=torch.long),
        source_checkpoint_sha256_t=torch.tensor(
            list(source_digest),
            dtype=torch.uint8,
        ),
        writer_source_sha256s_t=writer_sha256s_t,
        page_ids_t=torch.arange(
            start_page_id,
            start_page_id + requested_count,
            dtype=torch.long,
        ),
        object_sha256s_t=object_sha256s_t,
        object_bytes_t=torch.tensor(object_sizes, dtype=torch.long),
        page_parameter_elements_t=torch.tensor(
            page_parameter_elements,
            dtype=torch.long,
        ),
        physical_parameter_elements_t=torch.tensor(
            requested_count * page_parameter_elements,
            dtype=torch.long,
        ),
        total_object_bytes_t=torch.tensor(total_object_bytes, dtype=torch.long),
        storage_format=str(storage_format),
        template_object_path=template_path,
        template_object_sha256_t=template_digest_t,
        template_page_id_t=template_page_id_t,
        template_page_id_offset_t=template_page_id_offset_t,
        physical_archive_object_bytes_t=torch.tensor(
            physical_archive_object_bytes,
            dtype=torch.long,
        ),
        ready_t=torch.ones((), dtype=torch.bool),
    )


def compact_transfer_page_bank_to_template_reference(
    *,
    source_summary_path: Path,
    output_root: Path,
) -> dict[str, Any]:
    """Archive a fully verified untrained bank as one template plus identities.

    Every source payload is rehashed before publication.  The resulting bank
    preserves the complete page-to-digest journal and can later hydrate only
    a model-selected cohort into ordinary immutable page objects.  It neither
    moves an accepted generation nor makes a trained-capability claim.
    """

    source_bank = discover_compact_transfer_page_bank(source_summary_path)
    if source_bank.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE:
        raise RuntimeError("template-reference compaction requires full objects")
    resolved_output = output_root.expanduser().resolve()
    resolved_output.parent.mkdir(parents=True, exist_ok=True)
    if resolved_output.exists():
        raise FileExistsError(resolved_output)
    temporary_root = resolved_output.parent / (
        f".{resolved_output.name}.template_reference_pending"
    )
    if temporary_root.exists():
        raise FileExistsError(temporary_root)

    source_summary = json.loads(
        source_bank.summary_path.read_text(encoding="utf-8")
    )
    if not isinstance(source_summary, dict):
        raise RuntimeError("compact source summary is not an object")
    session_values = source_bank.session_id_t.detach().cpu().long().tolist()
    source_checkpoint_sha256 = bytes(
        source_bank.source_checkpoint_sha256_t.detach().cpu().tolist()
    ).hex()
    records = _read_compact_journal(
        source_bank.journal_path,
        source_sha256=source_checkpoint_sha256,
        session_id=session_values,
    )
    page_ids = source_bank.page_ids_t.detach().cpu().long().tolist()
    if not page_ids or tuple(page_ids) != tuple(sorted(records)):
        raise RuntimeError("compact source page mapping differs")
    template_page_id = page_ids[0]
    template_row = records[template_page_id]
    template_sha256 = str(template_row["objectSha256"])
    template_bytes = int(template_row["objectBytes"])
    template_source = (
        source_bank.store_root
        / "objects"
        / "sha256"
        / f"{template_sha256}.safetensors"
    )
    if _file_sha256(template_source) != template_sha256:
        raise RuntimeError("compact source template hash differs")
    template_payload = bytearray(template_source.read_bytes())
    page_id_offset = _template_page_id_offset_from_payload_boundary(template_payload)
    if struct.unpack_from("<q", template_payload, page_id_offset)[0] != template_page_id:
        raise RuntimeError("compact source template page identity differs")

    # Verify all old payloads and all virtual reconstructions before retaining
    # only the template.  This is intentionally exhaustive: storage release
    # has a higher bar than candidate-capacity discovery.
    for page_id in page_ids:
        row = records[page_id]
        expected_sha256 = str(row["objectSha256"])
        expected_bytes = int(row["objectBytes"])
        source_object = (
            source_bank.store_root
            / "objects"
            / "sha256"
            / f"{expected_sha256}.safetensors"
        )
        if (
            source_object.stat().st_size != expected_bytes
            or _file_sha256(source_object) != expected_sha256
        ):
            raise RuntimeError("compact source page hash differs")
        struct.pack_into("<q", template_payload, page_id_offset, page_id)
        if hashlib.sha256(template_payload).hexdigest() != expected_sha256:
            raise RuntimeError("compact virtual page reconstruction differs")
    struct.pack_into(
        "<q",
        template_payload,
        page_id_offset,
        template_page_id,
    )
    if hashlib.sha256(template_payload).hexdigest() != template_sha256:
        raise RuntimeError("compact template restoration differs")

    temporary_objects = temporary_root / "objects" / "sha256"
    temporary_objects.mkdir(parents=True, exist_ok=False)
    template_destination = temporary_objects / template_source.name
    if template_source.stat().st_dev == temporary_objects.stat().st_dev:
        os.link(template_source, template_destination)
    else:
        shutil.copyfile(template_source, template_destination)
        with template_destination.open("rb") as handle:
            os.fsync(handle.fileno())
    temporary_journal = temporary_root / "pages.jsonl"
    shutil.copyfile(source_bank.journal_path, temporary_journal)
    with temporary_journal.open("rb") as handle:
        os.fsync(handle.fileno())
    _fsync_directory(temporary_objects)
    _fsync_directory(temporary_journal.parent)

    summary_payload: dict[str, Any] = {
        "schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
        "passed": True,
        "storageFormat": TEMPLATE_REFERENCE_TRANSFER_STORAGE,
        "storeRoot": str(resolved_output),
        "sessionId": session_values,
        "sourceCheckpointSha256": source_checkpoint_sha256,
        "startPageId": int(source_summary["startPageId"]),
        "requestedPageCount": len(page_ids),
        "journaledPageCount": len(page_ids),
        "journalPath": str(resolved_output / "pages.jsonl"),
        "journalSha256": _file_sha256(temporary_journal),
        "pageParameterElements": int(source_bank.page_parameter_elements_t),
        "physicalParameterElementsInitialized": int(
            source_bank.physical_parameter_elements_t
        ),
        "objectBytes": int(source_bank.total_object_bytes_t),
        "physicalArchiveObjectBytes": template_bytes,
        "templateObjectSha256": template_sha256,
        "templateObjectBytes": template_bytes,
        "templatePageId": template_page_id,
        "templatePageIdOffset": page_id_offset,
        "sourceSummaryPath": str(source_bank.summary_path),
        "sourceSummarySha256": _file_sha256(source_bank.summary_path),
        "completePageMappingRetained": True,
        "trainedPageCount": 0,
        "transferInitializedPageCount": len(page_ids),
        "acceptedGenerationCommitted": False,
        "promotionRequiresTrainingAndHeldoutProof": True,
    }
    _atomic_json(temporary_root / "summary.json", summary_payload)
    os.replace(temporary_root, resolved_output)
    _fsync_directory(resolved_output.parent)

    cold_bank = discover_compact_transfer_page_bank(resolved_output / "summary.json")
    if (
        cold_bank.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE
        or not torch.equal(cold_bank.page_ids_t, source_bank.page_ids_t)
        or not torch.equal(cold_bank.object_sha256s_t, source_bank.object_sha256s_t)
        or int(cold_bank.physical_archive_object_bytes_t) != template_bytes
    ):
        raise RuntimeError("compact template-reference cold proof differs")
    publish_compact_transfer_page_bank_locator_boundary(
        resolved_output / "summary.json"
    )
    receipt: dict[str, Any] = {
        "schema": "nnf.resynthesis.compact_transfer_template_reference.v1",
        "passed": True,
        "sourceSummaryPath": str(source_bank.summary_path),
        "sourceSummarySha256": _file_sha256(source_bank.summary_path),
        "outputSummaryPath": str(resolved_output / "summary.json"),
        "outputSummarySha256": _file_sha256(resolved_output / "summary.json"),
        "sourcePageCount": len(page_ids),
        "trainedPageCount": 0,
        "acceptedGenerationCommitted": False,
        "logicalObjectBytes": int(source_bank.total_object_bytes_t),
        "physicalArchiveObjectBytes": template_bytes,
        "sourcePayloadHashesExhaustivelyVerified": True,
        "virtualPayloadHashesExhaustivelyVerified": True,
        "templateObjectSha256": template_sha256,
        "templatePageId": template_page_id,
        "promotionRequiresTrainingAndHeldoutProof": True,
    }
    _atomic_json(resolved_output / "template_reference_receipt.json", receipt)
    return receipt


def hydrate_compact_page_admission_objects_boundary(
    packet: CompactPageAdmissionPacket,
) -> NoNEImmutablePageStore:
    """Expose selected virtual objects as ordinary immutable page files.

    The model owns cohort selection before this boundary.  Hydration only
    recreates those receipt-bound page identities, validates their exact hashes,
    and gives the existing paged runtime its normal object-store interface.
    """

    bank = discover_compact_transfer_page_bank(packet.bank_summary_path)
    if (
        bank.summary_path != packet.bank_summary_path.expanduser().resolve()
        or bank.journal_path != packet.bank_journal_path.expanduser().resolve()
        or bank.store_root != packet.bank_store_root.expanduser().resolve()
        or bank.storage_format != packet.bank_storage_format
        or not torch.equal(bank.summary_sha256_t, packet.bank_summary_sha256_t)
        or not torch.equal(bank.journal_sha256_t, packet.bank_journal_sha256_t)
        or not torch.equal(bank.session_id_t, packet.bank_session_id_t)
        or not torch.equal(
            bank.source_checkpoint_sha256_t,
            packet.bank_source_checkpoint_sha256_t,
        )
    ):
        raise RuntimeError("compact page admission bank identity differs")
    selected_page_ids_t = packet.selected_page_ids_t.detach().cpu().long().reshape(-1)
    selected_hashes_t = packet.selected_object_sha256s_t.detach().cpu().to(
        dtype=torch.uint8
    )
    selected_bytes_t = packet.selected_object_bytes_t.detach().cpu().long().reshape(-1)
    if (
        selected_hashes_t.shape != (selected_page_ids_t.numel(), 32)
        or selected_bytes_t.shape != selected_page_ids_t.shape
    ):
        raise RuntimeError("compact page admission object geometry differs")
    if bank.storage_format == SCALED_FLOAT8_TRANSFER_STORAGE:
        source_store = NoNEImmutablePageStore(bank.store_root)
        source_store.begin_session(bank.session_id_t)
        return source_store
    if bank.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE:
        raise RuntimeError("compact page admission storage format is unsupported")
    if (
        bank.template_object_path is None
        or not torch.equal(
            bank.template_object_sha256_t,
            packet.bank_template_object_sha256_t.detach().cpu().to(
                dtype=torch.uint8
            ),
        )
        or not torch.equal(
            bank.template_page_id_t,
            packet.bank_template_page_id_t.detach().cpu().long(),
        )
        or not torch.equal(
            bank.template_page_id_offset_t,
            packet.bank_template_page_id_offset_t.detach().cpu().long(),
        )
        or not torch.equal(
            bank.physical_archive_object_bytes_t,
            packet.bank_physical_archive_object_bytes_t.detach().cpu().long(),
        )
    ):
        raise RuntimeError("compact template admission identity differs")
    hydration_root = bank.store_root / ".template_reference_hydrated"
    hydration_store = NoNEImmutablePageStore(
        hydration_root,
        advertise_locator=False,
    )
    hydration_store.begin_session(bank.session_id_t)
    template = _template_reference_payload_boundary(
        template_path=bank.template_object_path,
        template_sha256=bytes(bank.template_object_sha256_t.tolist()).hex(),
        template_bytes=int(bank.physical_archive_object_bytes_t),
        template_page_id=int(bank.template_page_id_t),
        page_id_offset=int(bank.template_page_id_offset_t),
    )
    bindings: list[NoNEPageObjectBinding] = []
    for index, page_id in enumerate(selected_page_ids_t.tolist()):
        binding = hydration_store.write_compact_transfer_template_page_boundary(
            template,
            page_id=int(page_id),
        )
        if (
            not torch.equal(
                binding.object_sha256_t,
                selected_hashes_t[index],
            )
            or not torch.equal(binding.object_bytes_t, selected_bytes_t[index])
        ):
            raise RuntimeError("compact template hydrated object differs")
        bindings.append(binding)
    if not bindings:
        raise RuntimeError("compact template admission selected no pages")
    hydration_store.commit_compact_transfer_chunk_boundary(tuple(bindings))
    return hydration_store


def compose_compact_transfer_page_banks(
    *,
    summary_paths: tuple[Path, ...],
    output_root: Path,
) -> dict[str, Any]:
    """Compose contiguous immutable banks without copying page payloads.

    This is an external storage-authority operation. It hardlinks already
    verified content-addressed objects on one filesystem, preserves every
    source journal row, and publishes one movable compact-bank locator only
    after the aggregate summary cold-validates. Composition never marks a
    page trained and never advances an accepted graph generation.
    """

    if len(summary_paths) < 2:
        raise ValueError("compact page composition requires at least two banks")
    resolved_output = output_root.expanduser().resolve()
    resolved_output.parent.mkdir(parents=True, exist_ok=True)
    if resolved_output.exists():
        raise FileExistsError(resolved_output)

    banks = tuple(
        sorted(
            (
                discover_compact_transfer_page_bank(path)
                for path in summary_paths
            ),
            key=lambda bank: int(bank.page_ids_t[0]),
        )
    )
    reference = banks[0]
    if reference.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE or any(
        bank.storage_format != SCALED_FLOAT8_TRANSFER_STORAGE for bank in banks
    ):
        raise RuntimeError(
            "compact template-reference banks are independently discoverable "
            "and cannot be hardlink-composed"
        )
    for bank in banks[1:]:
        if (
            not torch.equal(bank.session_id_t, reference.session_id_t)
            or not torch.equal(
                bank.source_checkpoint_sha256_t,
                reference.source_checkpoint_sha256_t,
            )
            or not torch.equal(
                bank.page_parameter_elements_t,
                reference.page_parameter_elements_t,
            )
        ):
            raise RuntimeError("compact page banks do not share one authority")

    page_ids_t = torch.cat(tuple(bank.page_ids_t for bank in banks), dim=0)
    first_page_id = int(page_ids_t[0])
    expected_page_ids_t = torch.arange(
        first_page_id,
        first_page_id + page_ids_t.numel(),
        dtype=torch.long,
    )
    if not torch.equal(page_ids_t, expected_page_ids_t):
        raise RuntimeError("compact page banks are not one contiguous interval")

    temporary_root = resolved_output.with_name(
        f".{resolved_output.name}.compose.{os.getpid()}.tmp"
    )
    if temporary_root.exists():
        raise FileExistsError(temporary_root)
    temporary_root.mkdir()
    objects_root = temporary_root / "objects" / "sha256"
    objects_root.mkdir(parents=True)
    journal_path = temporary_root / "pages.jsonl"
    linked_digests: set[str] = set()
    renamed = False
    try:
        destination_device = temporary_root.stat().st_dev
        with journal_path.open("wb") as output_journal:
            for bank in banks:
                with bank.journal_path.open("rb") as source_journal:
                    shutil.copyfileobj(
                        source_journal,
                        output_journal,
                        length=8 * 1024 * 1024,
                    )
                for object_sha256_t in bank.object_sha256s_t:
                    object_sha256 = bytes(
                        object_sha256_t.detach()
                        .cpu()
                        .to(dtype=torch.uint8)
                        .tolist()
                    ).hex()
                    if object_sha256 in linked_digests:
                        raise RuntimeError(
                            "compact page banks repeat one page object"
                        )
                    source_object = (
                        bank.store_root
                        / "objects"
                        / "sha256"
                        / f"{object_sha256}.safetensors"
                    )
                    if source_object.stat().st_dev != destination_device:
                        raise RuntimeError(
                            "compact page banks require one hardlink filesystem"
                        )
                    os.link(
                        source_object,
                        objects_root / source_object.name,
                    )
                    linked_digests.add(object_sha256)
            output_journal.flush()
            os.fsync(output_journal.fileno())

        requested_page_count = page_ids_t.numel()
        page_parameter_elements = int(reference.page_parameter_elements_t)
        source_checkpoint_sha256 = bytes(
            reference.source_checkpoint_sha256_t.detach()
            .cpu()
            .to(dtype=torch.uint8)
            .tolist()
        ).hex()
        session_id = reference.session_id_t.detach().cpu().long().tolist()
        records = _read_compact_journal(
            journal_path,
            source_sha256=source_checkpoint_sha256,
            session_id=session_id,
        )
        if tuple(sorted(records)) != tuple(expected_page_ids_t.tolist()):
            raise RuntimeError("composed compact page journal is incomplete")

        final_journal_path = resolved_output / journal_path.name
        final_summary_path = resolved_output / "summary.json"
        journal_sha256 = _file_sha256(journal_path)
        summary: dict[str, Any] = {
            "schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
            "startPageId": first_page_id,
            "requestedPageCount": requested_page_count,
            "pageParameterElements": page_parameter_elements,
            "physicalParameterElementsInitialized": (
                requested_page_count * page_parameter_elements
            ),
            "objectBytes": sum(
                int(bank.total_object_bytes_t) for bank in banks
            ),
            "journalPath": str(final_journal_path),
            "journalSha256": journal_sha256,
            "journaledPageCount": requested_page_count,
            "storeRoot": str(resolved_output),
            "sessionId": session_id,
            "sourceCheckpointSha256": source_checkpoint_sha256,
            "storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
            "passed": True,
            "trainedPageCount": 0,
            "transferInitializedPageCount": requested_page_count,
            "completePageMappingRetained": True,
            "acceptedGenerationCommitted": False,
            "promotionRequiresTrainingAndHeldoutProof": True,
        }
        _atomic_json(temporary_root / final_summary_path.name, summary)
        _fsync_directory(objects_root)
        os.replace(temporary_root, resolved_output)
        renamed = True
        _fsync_directory(resolved_output.parent)

        cold_bank = discover_compact_transfer_page_bank(final_summary_path)
        if (
            not torch.equal(cold_bank.page_ids_t, expected_page_ids_t)
            or not torch.equal(cold_bank.session_id_t, reference.session_id_t)
            or int(cold_bank.physical_parameter_elements_t)
            != requested_page_count * page_parameter_elements
        ):
            raise RuntimeError("composed compact page bank cold proof differs")
        publish_compact_transfer_page_bank_locator_boundary(
            final_summary_path
        )

        receipt: dict[str, Any] = {
            "schema": "nnf.resynthesis.compact_transfer_page_composition.v1",
            "passed": True,
            "sourceBanks": [
                {
                    "summaryPath": str(bank.summary_path),
                    "summarySha256": _digest_hex(bank.summary_sha256_t),
                    "startPageId": int(bank.page_ids_t[0]),
                    "pageCount": bank.page_ids_t.numel(),
                }
                for bank in banks
            ],
            "outputSummaryPath": str(final_summary_path),
            "outputSummarySha256": _file_sha256(final_summary_path),
            "outputJournalPath": str(final_journal_path),
            "outputJournalSha256": journal_sha256,
            "physicalGraphLayerCount": requested_page_count,
            "physicalParameterElementsInitialized": (
                requested_page_count * page_parameter_elements
            ),
            "objectBytesReusedByHardlink": int(
                cold_bank.total_object_bytes_t
            ),
            "payloadBytesCopied": 0,
            "trainedPageCount": 0,
            "acceptedGenerationCommitted": False,
            "coldDiscoveryPassed": True,
        }
        _atomic_json(resolved_output / "composition_receipt.json", receipt)
        return receipt
    except Exception:
        if not renamed and temporary_root.exists():
            shutil.rmtree(temporary_root)
        raise


def plan_compact_page_bank_admission(
    bank: CompactPageBankBinding,
    scale_cohort: NoNEScaleCohortPacket,
    *,
    admitted_page_ids_t: torch.Tensor,
) -> CompactPageAdmissionPacket:
    """Bind available physical pages to a model-owned functional cohort."""

    family_page_ids_t = (
        scale_cohort.selected_family_page_ids_t.detach().cpu().long().reshape(-1)
    )
    if (
        bank.ready_t.numel() != 1
        or not bool(bank.ready_t)
        or scale_cohort.ready_t.numel() != 1
        or not bool(scale_cohort.ready_t)
        or scale_cohort.selected_page_count_t.numel() != 1
        or int(scale_cohort.selected_page_count_t) != family_page_ids_t.numel()
        or family_page_ids_t.numel() < 1
        or bank.page_ids_t.ndim != 1
        or bank.object_sha256s_t.shape != (bank.page_ids_t.numel(), 32)
        or bank.object_bytes_t.shape != bank.page_ids_t.shape
        or admitted_page_ids_t.dtype != torch.long
    ):
        raise RuntimeError("compact page admission tensor authority differs")
    admitted_t = admitted_page_ids_t.detach().cpu().reshape(-1)
    if torch.unique(admitted_t).numel() != admitted_t.numel():
        raise RuntimeError("compact page admission history contains duplicates")
    available_indexes_t = (~torch.isin(bank.page_ids_t, admitted_t)).nonzero(
        as_tuple=False
    ).reshape(-1)
    selected_count = family_page_ids_t.numel()
    if available_indexes_t.numel() < selected_count:
        raise RuntimeError("compact page bank has insufficient unadmitted capacity")
    selected_indexes_t = available_indexes_t[:selected_count]
    selected_page_ids_t = bank.page_ids_t.index_select(0, selected_indexes_t)
    selected_sha256s_t = bank.object_sha256s_t.index_select(
        0,
        selected_indexes_t,
    )
    selected_bytes_t = bank.object_bytes_t.index_select(0, selected_indexes_t)

    # Discovery seals the complete bank mapping once. Admission must still
    # recheck the selected physical inputs before they cross from storage into
    # a model-owned training transaction: that is a bounded cohort read, not a
    # full-bank rescan. Template-reference banks have one physical object, so
    # validate that exact object once before virtual selected pages are derived.
    if bank.storage_format == SCALED_FLOAT8_TRANSFER_STORAGE:
        objects_root = bank.store_root / "objects" / "sha256"
        verification_rows = tuple(
            (
                int(selected_page_ids_t[index]),
                bytes(selected_sha256s_t[index].tolist()).hex(),
                int(selected_bytes_t[index]),
            )
            for index in range(selected_page_ids_t.numel())
        )

        object_authorities: list[tuple[Path, str]] = []
        for page_id, expected_sha256, expected_bytes in verification_rows:
            object_path = objects_root / f"{expected_sha256}.safetensors"
            if (
                not object_path.is_file()
                or object_path.stat().st_size != expected_bytes
            ):
                raise RuntimeError("compact page object differs at admission")
            object_authorities.append((object_path, expected_sha256))
        verified_sha256s = file_sha256_authority_batch_boundary(
            tuple(object_authorities),
            identity_cache_root=(
                bank.store_root / "compact_admission_sha256_identity_cache"
            ),
        )
        if verified_sha256s != tuple(
            expected_sha256
            for _page_id, expected_sha256, _expected_bytes in verification_rows
        ):
            raise RuntimeError("compact page object differs at admission")
    elif bank.storage_format == TEMPLATE_REFERENCE_TRANSFER_STORAGE:
        if bank.template_object_path is None:
            raise RuntimeError("compact template admission object is absent")
        _template_reference_payload_boundary(
            template_path=bank.template_object_path,
            template_sha256=bytes(bank.template_object_sha256_t.tolist()).hex(),
            template_bytes=int(bank.physical_archive_object_bytes_t),
            template_page_id=int(bank.template_page_id_t),
            page_id_offset=int(bank.template_page_id_offset_t),
        )
    else:
        raise RuntimeError("compact page admission storage format is unsupported")

    return CompactPageAdmissionPacket(
        bank_summary_path=bank.summary_path,
        bank_journal_path=bank.journal_path,
        bank_store_root=bank.store_root,
        bank_storage_format=bank.storage_format,
        bank_template_object_path=bank.template_object_path,
        bank_summary_sha256_t=bank.summary_sha256_t.clone(),
        bank_journal_sha256_t=bank.journal_sha256_t.clone(),
        bank_session_id_t=bank.session_id_t.clone(),
        bank_source_checkpoint_sha256_t=(
            bank.source_checkpoint_sha256_t.clone()
        ),
        bank_writer_source_sha256s_t=(
            bank.writer_source_sha256s_t.clone()
        ),
        selected_family_page_ids_t=family_page_ids_t,
        selected_page_ids_t=selected_page_ids_t.clone(),
        selected_object_sha256s_t=selected_sha256s_t.clone(),
        selected_object_bytes_t=selected_bytes_t.clone(),
        bank_page_count_t=torch.tensor(bank.page_ids_t.numel(), dtype=torch.long),
        page_parameter_elements_t=bank.page_parameter_elements_t.clone(),
        bank_physical_parameter_elements_t=(
            bank.physical_parameter_elements_t.clone()
        ),
        bank_total_object_bytes_t=bank.total_object_bytes_t.clone(),
        bank_template_object_sha256_t=(
            bank.template_object_sha256_t.clone()
        ),
        bank_template_page_id_t=bank.template_page_id_t.clone(),
        bank_template_page_id_offset_t=(
            bank.template_page_id_offset_t.clone()
        ),
        bank_physical_archive_object_bytes_t=(
            bank.physical_archive_object_bytes_t.clone()
        ),
        selected_page_count_t=torch.tensor(selected_count, dtype=torch.long),
        ready_t=torch.ones((), dtype=torch.bool),
    )


def _compact_summary_candidates(store_root: Path) -> tuple[Path, ...]:
    """Find compact summaries below an externally supplied store root."""

    resolved_root = store_root.expanduser().resolve()
    if resolved_root.is_file():
        return (resolved_root,)
    candidates = {
        resolved_root / "summary.json",
        *resolved_root.glob("*.summary.json"),
    }
    return tuple(sorted((path for path in candidates if path.is_file()), key=str))


def _compact_transfer_registry_roots_boundary(
    *,
    registry_roots: tuple[Path, ...] | None,
) -> tuple[Path, ...]:
    """Map mount-local page registries to transfer-bank-only registries.

    A transfer-initialized bank has no accepted generation and must never be
    advertised through ``page-stores``.  Its sibling registry remains
    filesystem-discoverable across mounts while keeping generation, route, and
    acceptance authority outside this metadata boundary.
    """

    page_roots = (
        _default_page_store_registry_roots_boundary()
        if registry_roots is None
        else tuple(root.expanduser().resolve() for root in registry_roots)
    )
    return tuple(
        sorted(
            {
                root.parent / COMPACT_TRANSFER_REGISTRY_DIRECTORY
                for root in page_roots
            },
            key=str,
        )
    )


def publish_compact_transfer_page_bank_locator_boundary(
    summary_path: Path,
    *,
    registry_roots: tuple[Path, ...] | None = None,
) -> tuple[Path, ...]:
    """Publish one sealed untrained bank without claiming a page generation.

    The locator identifies immutable compact-bank metadata only.  It cannot
    select routes, mutate an accepted pointer, or make an initialization a
    trained capability.  Discovery revalidates the complete summary/journal
    contract before admitting a model-selected subset.
    """

    bank = discover_compact_transfer_page_bank(summary_path)
    session_id_t = bank.session_id_t.detach().cpu().long().clone()
    session_key = _session_key(session_id_t)
    summary_sha256 = _digest_hex(bank.summary_sha256_t)
    journal_sha256 = _digest_hex(bank.journal_sha256_t)
    payload: dict[str, Any] = {
        "schema": COMPACT_TRANSFER_LOCATOR_SCHEMA,
        "sessionKey": session_key,
        "sessionId": session_id_t.tolist(),
        "root": str(bank.store_root),
        "summaryPath": str(bank.summary_path),
        "summarySha256": summary_sha256,
        "journalPath": str(bank.journal_path),
        "journalSha256": journal_sha256,
        "storageFormat": bank.storage_format,
        "completePageMappingRetained": True,
        "trainedPageCount": 0,
        "acceptedGenerationCommitted": False,
        "routingAuthority": False,
        "acceptedPointerMutationAuthority": False,
    }
    locator_id = hashlib.sha256(
        json.dumps(
            {
                "root": payload["root"],
                "summarySha256": summary_sha256,
            },
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    ).hexdigest()
    roots = {
        bank.store_root
        / ".nnf-resynthesis"
        / COMPACT_TRANSFER_REGISTRY_DIRECTORY,
        _page_store_registry_root_for_boundary(bank.store_root).parent
        / COMPACT_TRANSFER_REGISTRY_DIRECTORY,
    }
    if registry_roots is not None:
        roots.update(
            root.parent / COMPACT_TRANSFER_REGISTRY_DIRECTORY
            for root in registry_roots
        )
    written: list[Path] = []
    for root in sorted(roots, key=str):
        path = root / session_key / f"{locator_id}.json"
        _atomic_json(path, payload)
        written.append(path)
    return tuple(written)


def _compact_transfer_locator_candidates_boundary(
    *,
    session_id_t: torch.Tensor,
    registry_roots: tuple[Path, ...] | None,
) -> tuple[tuple[Path, Path], ...]:
    """Resolve only sealed dedicated compact-transfer locator records."""

    session = session_id_t.detach().cpu().long().reshape(-1)
    session_key = _session_key(session)
    expected_session = session.tolist()
    candidates: dict[Path, Path] = {}
    for registry_root in _compact_transfer_registry_roots_boundary(
        registry_roots=registry_roots
    ):
        session_root = registry_root / session_key
        try:
            session_root_available = session_root.is_dir()
            locator_paths = (
                sorted(session_root.glob("*.json"), key=str)
                if session_root_available
                else ()
            )
        except PermissionError:
            # Registry roots are discovery hints. An optional kernel/BPF
            # registry may be mounted but intentionally unreadable to this
            # process; skip that root and retain strict validation for every
            # locator that is actually discovered.
            continue
        for locator_path in locator_paths:
            payload = json.loads(locator_path.read_text(encoding="utf-8"))
            if not isinstance(payload, dict):
                raise RuntimeError("compact transfer locator is not an object")
            root_value = payload.get("root")
            summary_value = payload.get("summaryPath")
            summary_sha256 = payload.get("summarySha256")
            journal_value = payload.get("journalPath")
            journal_sha256 = payload.get("journalSha256")
            if (
                payload.get("schema") != COMPACT_TRANSFER_LOCATOR_SCHEMA
                or payload.get("sessionKey") != session_key
                or payload.get("sessionId") != expected_session
                or not isinstance(root_value, str)
                or not isinstance(summary_value, str)
                or not isinstance(summary_sha256, str)
                or len(summary_sha256) != 64
                or not isinstance(journal_value, str)
                or not isinstance(journal_sha256, str)
                or len(journal_sha256) != 64
                or payload.get("completePageMappingRetained") is not True
                or payload.get("trainedPageCount") != 0
                or payload.get("acceptedGenerationCommitted") is not False
                or payload.get("routingAuthority") is not False
                or payload.get("acceptedPointerMutationAuthority") is not False
            ):
                raise RuntimeError("compact transfer locator authority differs")
            store_root = Path(root_value).expanduser().resolve()
            summary_path = Path(summary_value).expanduser().resolve()
            journal_path = Path(journal_value).expanduser().resolve()
            if (
                not store_root.is_dir()
                or summary_path.parent != store_root
                or not summary_path.is_file()
                or not journal_path.is_file()
                or _file_sha256(summary_path) != summary_sha256
                or _file_sha256(journal_path) != journal_sha256
            ):
                raise RuntimeError("compact transfer locator payload differs")
            previous = candidates.get(summary_path)
            if previous is not None and previous != store_root:
                raise RuntimeError("compact transfer locators disagree on a bank")
            candidates[summary_path] = store_root
    return tuple((root, summary) for summary, root in sorted(candidates.items()))


def _legacy_compact_transfer_locator_candidates_boundary(
    *,
    session_id_t: torch.Tensor,
    registry_roots: tuple[Path, ...] | None,
) -> tuple[tuple[Path, Path], ...]:
    """Read legacy transfer-bank metadata without granting page-store status.

    Older writers published transfer candidates in ``page-stores`` before an
    accepted pointer existed.  Their roots are valid only if a compact summary
    independently passes the current immutable bank validation below; stale
    page-store records with no compact summary contribute no capacity.
    """

    session = session_id_t.detach().cpu().long().reshape(-1)
    session_key = _session_key(session)
    expected_session = session.tolist()
    page_roots = (
        _default_page_store_registry_roots_boundary()
        if registry_roots is None
        else tuple(root.expanduser().resolve() for root in registry_roots)
    )
    candidates: dict[Path, Path] = {}
    for registry_root in page_roots:
        session_root = registry_root / session_key
        try:
            session_root_exists = session_root.is_dir()
        except OSError:
            continue
        if not session_root_exists:
            continue
        try:
            locator_paths = sorted(session_root.glob("*.json"), key=str)
        except OSError:
            continue
        for locator_path in locator_paths:
            try:
                payload = json.loads(locator_path.read_text(encoding="utf-8"))
            except OSError:
                continue
            if not isinstance(payload, dict):
                continue
            if payload.get("schema") != PAGE_STORE_LOCATOR_SCHEMA:
                continue
            root_value = payload.get("root")
            if (
                payload.get("sessionKey") != session_key
                or payload.get("sessionId") != expected_session
                or not isinstance(root_value, str)
            ):
                raise RuntimeError("legacy compact locator authority differs")
            store_root = Path(root_value).expanduser().resolve()
            try:
                store_root_exists = store_root.is_dir()
            except OSError:
                continue
            if not store_root_exists:
                continue
            try:
                summary_paths = _compact_summary_candidates(store_root)
            except OSError:
                continue
            for summary_path in summary_paths:
                candidates.setdefault(summary_path, store_root)
    return tuple((root, summary) for summary, root in sorted(candidates.items()))


def _compact_stream_from_summary_boundary(
    *,
    store_root: Path,
    session_id_t: torch.Tensor,
    summary_path: Path,
) -> CompactPageStreamPacket | None:
    summary = json.loads(summary_path.read_text(encoding="utf-8"))
    if not isinstance(summary, dict):
        raise RuntimeError("compact page summary is not an object")
    if summary.get("schema") != COMPACT_TRANSFER_SUMMARY_SCHEMA:
        return None
    # A disjoint compact-bank builder publishes progress in its summary.  It
    # is discoverable metadata, but it is not admission authority until the
    # complete page mapping has been sealed.
    if (
        summary.get("passed") is not True
        or summary.get("completePageMappingRetained") is not True
    ):
        return None
    session_id = session_id_t.detach().cpu().long().reshape(-1).tolist()
    journal_value = summary.get("journalPath")
    journal_sha256 = summary.get("journalSha256")
    source_sha256 = summary.get("sourceCheckpointSha256")
    journaled_page_count = summary.get("journaledPageCount")
    transfer_initialized_page_count = summary.get(
        "transferInitializedPageCount"
    )
    requested_page_count = summary.get("requestedPageCount")
    start_page_id = summary.get("startPageId")
    if (
        summary.get("sessionId") != session_id
        or summary.get("storageFormat")
        not in {
            SCALED_FLOAT8_TRANSFER_STORAGE,
            TEMPLATE_REFERENCE_TRANSFER_STORAGE,
        }
        or summary.get("trainedPageCount") != 0
        or summary.get("acceptedGenerationCommitted") is not False
        or not isinstance(journal_value, str)
        or not isinstance(journal_sha256, str)
        or len(journal_sha256) != 64
        or not isinstance(source_sha256, str)
        or len(source_sha256) != 64
        or not isinstance(journaled_page_count, int)
        or isinstance(journaled_page_count, bool)
        or journaled_page_count < 0
        or not isinstance(transfer_initialized_page_count, int)
        or isinstance(transfer_initialized_page_count, bool)
        or not isinstance(requested_page_count, int)
        or isinstance(requested_page_count, bool)
        or requested_page_count < 1
        or journaled_page_count != requested_page_count
        or transfer_initialized_page_count != requested_page_count
        or not isinstance(start_page_id, int)
        or isinstance(start_page_id, bool)
        or start_page_id < 0
    ):
        raise RuntimeError("compact page summary authority differs")
    recorded_journal = Path(journal_value).expanduser()
    journal_path = (
        recorded_journal.resolve()
        if recorded_journal.is_file()
        else (store_root / recorded_journal.name).resolve()
    )
    if not journal_path.is_file():
        raise RuntimeError("compact page journal is absent during discovery")
    if _file_sha256(journal_path) != journal_sha256:
        refreshed = json.loads(summary_path.read_text(encoding="utf-8"))
        if (
            isinstance(refreshed, dict)
            and refreshed.get("journalSha256") != journal_sha256
        ):
            return _compact_stream_from_summary_boundary(
                store_root=store_root,
                session_id_t=session_id_t,
                summary_path=summary_path,
            )
        raise RuntimeError("compact page journal changed during discovery")
    records = _read_compact_journal(
        journal_path,
        source_sha256=source_sha256,
        session_id=session_id,
    )
    ordered_ids = sorted(records)
    if (
        len(ordered_ids) != journaled_page_count
        or ordered_ids
        != list(range(start_page_id, start_page_id + journaled_page_count))
    ):
        raise RuntimeError("compact page journal mapping is incomplete")
    if not (store_root / "objects/sha256").is_dir():
        raise RuntimeError("discovered compact page object root is absent")
    page_ids_t = torch.tensor(ordered_ids, dtype=torch.long)
    object_sha256_t = (
        torch.stack(
            tuple(
                digest_tensor(str(records[page_id]["objectSha256"]))
                for page_id in ordered_ids
            ),
            dim=0,
        )
        if ordered_ids
        else torch.zeros(0, 32, dtype=torch.uint8)
    )
    object_bytes_t = torch.tensor(
        [int(records[page_id]["objectBytes"]) for page_id in ordered_ids],
        dtype=torch.long,
    )
    return CompactPageStreamPacket(
        store_root=store_root,
        summary_path=summary_path.resolve(),
        journal_path=journal_path,
        storage_format=str(summary["storageFormat"]),
        session_id_t=session_id_t.detach().cpu().long().clone(),
        page_ids_t=page_ids_t,
        object_sha256_t=object_sha256_t,
        object_bytes_t=object_bytes_t,
        trained_t=torch.zeros(len(ordered_ids), dtype=torch.bool),
        accepted_generation_committed_t=torch.zeros(
            len(ordered_ids),
            dtype=torch.bool,
        ),
    )


def discover_compact_page_streams_boundary(
    *,
    session_id_t: torch.Tensor,
    anchor_roots: tuple[Path, ...] = (),
    registry_roots: tuple[Path, ...] | None = None,
) -> tuple[CompactPageStreamPacket, ...]:
    """Auto-discover every stable compact stream for one NoNE session."""

    candidates: dict[Path, tuple[Path, bool]] = {}
    for store_root, summary_path in _compact_transfer_locator_candidates_boundary(
        session_id_t=session_id_t,
        registry_roots=registry_roots,
    ):
        candidates[summary_path] = (store_root, True)
    for store_root, summary_path in _legacy_compact_transfer_locator_candidates_boundary(
        session_id_t=session_id_t,
        registry_roots=registry_roots,
    ):
        candidates.setdefault(summary_path, (store_root, False))
    for anchor_root in anchor_roots:
        resolved_anchor = anchor_root.expanduser().resolve()
        for summary_path in _compact_summary_candidates(resolved_anchor):
            candidates.setdefault(summary_path, (resolved_anchor, False))
    streams: list[CompactPageStreamPacket] = []
    observed_pages: dict[int, bytes] = {}
    for summary_path, (store_root, dedicated_locator) in sorted(
        candidates.items(),
        key=lambda entry: str(entry[0]),
    ):
        stream = _compact_stream_from_summary_boundary(
            store_root=store_root,
            session_id_t=session_id_t,
            summary_path=summary_path,
        )
        if stream is None:
            if dedicated_locator:
                raise RuntimeError("compact transfer locator summary differs")
            continue
        for index, page_id in enumerate(stream.page_ids_t.tolist()):
            digest = bytes(stream.object_sha256_t[index].tolist())
            previous = observed_pages.get(page_id)
            if previous is not None and previous != digest:
                raise RuntimeError(
                    "compact page discovery found conflicting page identities"
                )
            observed_pages[page_id] = digest
        streams.append(stream)
    return tuple(
        sorted(
            streams,
            key=lambda stream: (str(stream.store_root), str(stream.summary_path)),
        )
    )


def none_scale_evidence_from_record_boundary(
    record: Mapping[str, Any],
) -> NoNEScaleEvidencePacket:
    """Reconstruct and revalidate persisted model-owned scale evidence."""

    family_ids = record.get("familyRootPageIds")
    retained = record.get("retainedFamilyMask")
    gap = record.get("unresolvedGapPressure")
    route = record.get("routePressure")
    distinct = record.get("distinctGradientMask")
    updates = record.get("freshGradientUpdateCounts")
    gradient_norms = record.get("freshGradientNorms")
    parameter_deltas = record.get("freshParameterDeltaNorms")
    signatures = record.get("freshGradientSignatures")
    if (
        record.get("schema") != "nnf.resynthesis.none_scale_evidence.v1"
        or record.get("modelOwned") is not True
        or record.get("targetFree") is not True
        or record.get("storageAuthority") is not False
        or record.get("graphAdmissionAuthority") is not False
        or record.get("trainingClaimed") is not False
        or type(record.get("evidenceReady")) is not bool
        or not isinstance(family_ids, list)
        or not family_ids
        or any(
            not isinstance(value, int) or isinstance(value, bool) or value < 0
            for value in family_ids
        )
        or len(set(family_ids)) != len(family_ids)
    ):
        raise RuntimeError("NoNE scale evidence authority differs")
    family_count = len(family_ids)
    vector_values = (
        retained,
        gap,
        route,
        distinct,
        updates,
        gradient_norms,
        parameter_deltas,
    )
    if any(
        not isinstance(value, list) or len(value) != family_count
        for value in vector_values
    ):
        raise RuntimeError("NoNE scale evidence vector geometry differs")
    retained_values = cast(list[Any], retained)
    gap_values = cast(list[Any], gap)
    route_values = cast(list[Any], route)
    distinct_values = cast(list[Any], distinct)
    update_values = cast(list[Any], updates)
    gradient_norm_values = cast(list[Any], gradient_norms)
    parameter_delta_values = cast(list[Any], parameter_deltas)
    if (
        any(type(value) is not bool for value in retained_values)
        or any(type(value) is not bool for value in distinct_values)
        or any(
            not isinstance(value, int) or isinstance(value, bool) or value < 0
            for value in update_values
        )
        or any(
            not isinstance(value, (int, float))
            or isinstance(value, bool)
            or not math.isfinite(float(value))
            or float(value) < 0.0
            for values in (
                gap_values,
                route_values,
                gradient_norm_values,
                parameter_delta_values,
            )
            for value in values
        )
        or not isinstance(signatures, list)
        or len(signatures) != family_count
        or not signatures
        or not isinstance(signatures[0], list)
        or not signatures[0]
    ):
        raise RuntimeError("NoNE scale evidence values differ")
    signature_values = signatures
    signature_width = len(signature_values[0])
    if any(
        not isinstance(row, list)
        or len(row) != signature_width
        or any(
            not isinstance(value, (int, float))
            or isinstance(value, bool)
            or not math.isfinite(float(value))
            for value in row
        )
        for row in signature_values
    ):
        raise RuntimeError("NoNE scale evidence signature geometry differs")

    packet = NoNEScaleEvidencePacket(
        family_page_ids_t=torch.tensor(family_ids, dtype=torch.long),
        retained_family_mask_t=torch.tensor(retained_values, dtype=torch.bool),
        unresolved_gap_pressure_t=torch.tensor(gap_values, dtype=torch.float32),
        route_pressure_t=torch.tensor(route_values, dtype=torch.float32),
        distinct_gradient_mask_t=torch.tensor(
            distinct_values,
            dtype=torch.bool,
        ),
        fresh_gradient_update_count_t=torch.tensor(
            update_values,
            dtype=torch.long,
        ),
        fresh_gradient_norm_t=torch.tensor(
            gradient_norm_values,
            dtype=torch.float32,
        ),
        fresh_parameter_delta_norm_t=torch.tensor(
            parameter_delta_values,
            dtype=torch.float32,
        ),
        fresh_gradient_signature_t=torch.tensor(
            signature_values,
            dtype=torch.float32,
        ),
        evidence_ready_t=torch.tensor(
            bool(record.get("evidenceReady")),
            dtype=torch.bool,
        ),
    )
    eligible_t = _qualified_scale_evidence_mask(packet)
    eligible_count = record.get("eligibleFamilyCount")
    if (
        not isinstance(eligible_count, int)
        or isinstance(eligible_count, bool)
        or eligible_count != int(eligible_t.long().sum())
        or record.get("evidenceReady") != bool(eligible_t.any())
    ):
        raise RuntimeError("NoNE scale evidence readiness differs")
    return packet


def _qualified_scale_evidence_mask(
    packet: NoNEScaleEvidencePacket,
) -> torch.Tensor:
    """Require fresh routing, gradients, and parameter change for page growth."""

    family_count = packet.family_page_ids_t.reshape(-1).shape[0]
    vector_shapes = (
        packet.retained_family_mask_t.reshape(-1).shape,
        packet.unresolved_gap_pressure_t.reshape(-1).shape,
        packet.route_pressure_t.reshape(-1).shape,
        packet.distinct_gradient_mask_t.reshape(-1).shape,
        packet.fresh_gradient_update_count_t.reshape(-1).shape,
        packet.fresh_gradient_norm_t.reshape(-1).shape,
        packet.fresh_parameter_delta_norm_t.reshape(-1).shape,
    )
    if any(shape != (family_count,) for shape in vector_shapes):
        raise RuntimeError("NoNE scale evidence vector geometry differs")
    gap_t = packet.unresolved_gap_pressure_t.reshape(-1)
    route_t = packet.route_pressure_t.reshape(-1)
    gradient_norm_t = packet.fresh_gradient_norm_t.reshape(-1)
    parameter_delta_t = packet.fresh_parameter_delta_norm_t.reshape(-1)
    finite_t = (
        torch.isfinite(gap_t)
        & torch.isfinite(route_t)
        & torch.isfinite(gradient_norm_t)
        & torch.isfinite(parameter_delta_t)
    )
    return (
        packet.retained_family_mask_t.reshape(-1).bool()
        & packet.distinct_gradient_mask_t.reshape(-1).bool()
        & finite_t
        & gap_t.gt(0)
        & route_t.gt(0)
        & packet.fresh_gradient_update_count_t.reshape(-1).gt(0)
        & gradient_norm_t.gt(0)
        & parameter_delta_t.gt(0)
    )


def _template_reference_selected_equivalent_bank_boundary(
    *,
    streams: tuple[CompactPageStreamPacket, ...],
    admitted_page_ids_t: torch.Tensor,
    source_bank: CompactPageBankBinding,
    selected_page_ids_t: torch.Tensor,
    selected_object_sha256s_t: torch.Tensor,
    selected_object_bytes_t: torch.Tensor,
) -> CompactPageBankBinding:
    """Prefer an exact template representation for an already selected cohort.

    A shorter conventional bank can rank before a wider template-reference bank
    even when their next selected pages are byte-identical.  Storage extent is
    not model authority, so preserve the source cohort exactly and substitute
    only a template bank with the same selected IDs, object digests, sizes,
    geometry, session, source checkpoint, and writer provenance.
    """

    admitted_t = admitted_page_ids_t.detach().cpu().long().reshape(-1)
    selected_ids = selected_page_ids_t.detach().cpu().long().reshape(-1)
    selected_sha256s = selected_object_sha256s_t.detach().cpu().to(
        dtype=torch.uint8
    )
    selected_bytes = selected_object_bytes_t.detach().cpu().long().reshape(-1)
    selected_count = selected_ids.numel()
    if (
        selected_count < 1
        or selected_sha256s.shape != (selected_count, 32)
        or selected_bytes.numel() != selected_count
    ):
        raise RuntimeError("compact selected cohort geometry differs")
    for stream in streams:
        if stream.storage_format != TEMPLATE_REFERENCE_TRANSFER_STORAGE:
            continue
        available_indexes_t = (~torch.isin(stream.page_ids_t, admitted_t)).nonzero(
            as_tuple=False
        ).reshape(-1)
        if available_indexes_t.numel() < selected_count:
            continue
        indexes_t = available_indexes_t[:selected_count]
        if (
            not torch.equal(
                stream.page_ids_t.index_select(0, indexes_t),
                selected_ids,
            )
            or not torch.equal(
                stream.object_sha256_t.index_select(0, indexes_t),
                selected_sha256s,
            )
            or not torch.equal(
                stream.object_bytes_t.index_select(0, indexes_t),
                selected_bytes,
            )
        ):
            continue
        template_bank = discover_compact_transfer_page_bank(stream.summary_path)
        template_available_indexes_t = (
            ~torch.isin(template_bank.page_ids_t, admitted_t)
        ).nonzero(as_tuple=False).reshape(-1)
        template_indexes_t = template_available_indexes_t[:selected_count]
        if (
            template_indexes_t.numel() != selected_count
            or not torch.equal(
                template_bank.page_ids_t.index_select(0, template_indexes_t),
                selected_ids,
            )
            or not torch.equal(
                template_bank.object_sha256s_t.index_select(0, template_indexes_t),
                selected_sha256s,
            )
            or not torch.equal(
                template_bank.object_bytes_t.index_select(0, template_indexes_t),
                selected_bytes,
            )
            or not torch.equal(
                template_bank.session_id_t,
                source_bank.session_id_t,
            )
            or not torch.equal(
                template_bank.source_checkpoint_sha256_t,
                source_bank.source_checkpoint_sha256_t,
            )
            or not torch.equal(
                template_bank.page_parameter_elements_t,
                source_bank.page_parameter_elements_t,
            )
        ):
            continue
        return template_bank
    return source_bank


def plan_discovered_compact_page_admission_boundary(
    *,
    session_id_t: torch.Tensor,
    scale_evidence: NoNEScaleEvidencePacket,
    accepted_training_saturation: NoNEAcceptedTrainingSaturationPacket,
    admitted_page_ids_t: torch.Tensor,
    replica_free_bytes_t: torch.Tensor,
    reserve_bytes_t: torch.Tensor,
    current_physical_parameter_elements_t: torch.Tensor,
    capacity_envelope_parameter_elements_t: torch.Tensor,
    maximum_page_count_t: torch.Tensor | None = None,
    objective_compatible_family_mask_t: torch.Tensor | None = None,
    anchor_roots: tuple[Path, ...] = (),
    registry_roots: tuple[Path, ...] | None = None,
) -> DiscoveredCompactAdmissionPlan:
    """Bind retained model evidence to auto-discovered physical capacity.

    Discovery resolves a movable session-owned store and chooses the earliest
    unadmitted physical identities. It does not mutate a catalog or accepted
    pointer; the returned packet must still pass the immutable migration.
    """

    if session_id_t.reshape(-1).shape != (4,):
        raise ValueError("NoNE compact admission session geometry differs")
    saturation_ready_t = (
        validate_none_accepted_training_saturation_boundary(
            accepted_training_saturation
        )
    )
    if (
        saturation_ready_t.numel() != 1
        or not bool(saturation_ready_t)
        or accepted_training_saturation.remaining_unproven_page_ids_t.numel()
        != 0
    ):
        raise RuntimeError(
            "NoNE compact admission requires a saturated accepted page bank"
        )
    maximum_count_t: torch.Tensor | None = None
    if maximum_page_count_t is not None:
        maximum_count_t = maximum_page_count_t.reshape(()).long()
        if maximum_count_t.numel() != 1 or not bool(maximum_count_t.gt(0)):
            raise ValueError("NoNE compact admission maximum must be positive")
    qualified_family_mask_t = _qualified_scale_evidence_mask(scale_evidence)
    if (
        scale_evidence.evidence_ready_t.numel() != 1
        or not bool(scale_evidence.evidence_ready_t)
        or not bool(qualified_family_mask_t.any())
    ):
        raise RuntimeError("NoNE compact admission has no retained scale evidence")
    admitted_t = admitted_page_ids_t.detach().cpu().long().reshape(-1)
    if torch.unique(admitted_t).numel() != admitted_t.numel():
        raise RuntimeError("NoNE compact admission history contains duplicates")
    if not torch.equal(
        torch.sort(admitted_t).values,
        accepted_training_saturation.training_eligible_page_ids_t.detach()
        .cpu()
        .long(),
    ):
        raise RuntimeError(
            "NoNE compact admission saturation page identity differs"
        )
    streams = discover_compact_page_streams_boundary(
        session_id_t=session_id_t,
        anchor_roots=anchor_roots,
        registry_roots=registry_roots,
    )
    candidates: list[
        tuple[
            tuple[int, int, int, int],
            str,
            tuple[tuple[int, bytes, int], ...],
            CompactPageStreamPacket,
        ]
    ] = []
    for stream in streams:
        available_mask_t = ~torch.isin(stream.page_ids_t, admitted_t)
        available_indexes_t = available_mask_t.nonzero(
            as_tuple=False
        ).reshape(-1)
        available_count = int(available_indexes_t.numel())
        if available_count > 0:
            total_count = int(stream.page_ids_t.numel())
            available_ids_t = stream.page_ids_t.index_select(
                0,
                available_indexes_t,
            )
            fully_unadmitted = available_count == total_count
            identity = tuple(
                (
                    int(stream.page_ids_t[index]),
                    bytes(stream.object_sha256_t[index].tolist()),
                    int(stream.object_bytes_t[index]),
                )
                for index in available_indexes_t.tolist()
            )
            candidates.append(
                (
                    (
                        int(available_ids_t.amin()),
                        0 if fully_unadmitted else 1,
                        total_count,
                        available_count,
                    ),
                    str(stream.summary_path),
                    identity,
                    stream,
                )
            )
    selected_candidates: list[
        tuple[
            tuple[int, int, int, int],
            str,
            tuple[tuple[int, bytes, int], ...],
            CompactPageStreamPacket,
        ]
    ] = []
    for rank in sorted({candidate[0] for candidate in candidates}):
        ranked = sorted(
            (candidate for candidate in candidates if candidate[0] == rank),
            # Equal-rank candidates must expose the exact same virtual page
            # identities below.  Prefer the template-reference representation
            # in that case: it revalidates one immutable template plus every
            # selected virtual identity instead of rereading every equivalent
            # scaled-float8 object before the same admission transaction.
            key=lambda candidate: (
                0
                if candidate[3].storage_format
                == TEMPLATE_REFERENCE_TRANSFER_STORAGE
                else 1,
                candidate[1],
            ),
        )
        identities = {candidate[2] for candidate in ranked}
        if len(identities) != 1:
            raise RuntimeError(
                "compact page discovery found conflicting equal-rank banks"
            )
        selected_candidates.append(ranked[0])
    for _rank, _summary_name, _identity, stream in selected_candidates:
        bank = discover_compact_transfer_page_bank(stream.summary_path)
        if (
            not torch.equal(bank.session_id_t, session_id_t.detach().cpu().long())
            or not torch.equal(bank.page_ids_t, stream.page_ids_t)
            or not torch.equal(bank.object_sha256s_t, stream.object_sha256_t)
            or not torch.equal(bank.object_bytes_t, stream.object_bytes_t)
        ):
            raise RuntimeError("auto-discovered compact bank identity differs")
        available_mask_t = ~torch.isin(bank.page_ids_t, admitted_t)
        available_count_t = available_mask_t.long().sum()
        if not bool(available_count_t.gt(0)):
            continue
        measured_page_bytes_t = bank.object_bytes_t.masked_select(
            available_mask_t
        ).amax()
        free_bytes_t = replica_free_bytes_t.reshape(-1).long()
        reserve_t = reserve_bytes_t.reshape(()).long()
        bank_capacity_bytes_t = available_count_t * measured_page_bytes_t
        effective_free_bytes_t = torch.minimum(
            (free_bytes_t - reserve_t).clamp_min(0),
            bank_capacity_bytes_t.expand_as(free_bytes_t),
        ) + reserve_t
        effective_envelope_t = capacity_envelope_parameter_elements_t
        if maximum_count_t is not None:
            effective_envelope_t = torch.minimum(
                capacity_envelope_parameter_elements_t.reshape(()).long(),
                current_physical_parameter_elements_t.reshape(()).long()
                + maximum_count_t * bank.page_parameter_elements_t,
            )
        cohort = plan_none_scale_cohort(
            family_page_ids_t=scale_evidence.family_page_ids_t,
            retained_family_mask_t=qualified_family_mask_t,
            unresolved_gap_pressure_t=(
                scale_evidence.unresolved_gap_pressure_t
            ),
            route_pressure_t=scale_evidence.route_pressure_t,
            distinct_gradient_mask_t=(
                scale_evidence.distinct_gradient_mask_t
            ),
            replica_free_bytes_t=effective_free_bytes_t,
            reserve_bytes_t=reserve_t,
            measured_compact_page_bytes_t=measured_page_bytes_t,
            page_parameter_elements_t=bank.page_parameter_elements_t,
            current_physical_parameter_elements_t=(
                current_physical_parameter_elements_t
            ),
            capacity_envelope_parameter_elements_t=(
                effective_envelope_t
            ),
            objective_compatible_family_mask_t=(
                objective_compatible_family_mask_t
            ),
        )
        if not bool(cohort.ready_t):
            continue
        selected_count = int(cohort.selected_page_count_t)
        selected_indexes_t = available_mask_t.nonzero(
            as_tuple=False
        ).reshape(-1)[:selected_count]
        bank = _template_reference_selected_equivalent_bank_boundary(
            streams=streams,
            admitted_page_ids_t=admitted_t,
            source_bank=bank,
            selected_page_ids_t=bank.page_ids_t.index_select(
                0,
                selected_indexes_t,
            ),
            selected_object_sha256s_t=bank.object_sha256s_t.index_select(
                0,
                selected_indexes_t,
            ),
            selected_object_bytes_t=bank.object_bytes_t.index_select(
                0,
                selected_indexes_t,
            ),
        )
        admission = plan_compact_page_bank_admission(
            bank,
            cohort,
            admitted_page_ids_t=admitted_t,
        )
        return DiscoveredCompactAdmissionPlan(
            bank=bank,
            scale_cohort=cohort,
            admission=admission,
        )
    raise RuntimeError(
        "NoNE compact admission found no storage-safe unadmitted page cohort"
    )


def compact_training_eligible_page_ids_boundary(
    *,
    streams: tuple[CompactPageStreamPacket, ...],
    catalog_page_ids_t: torch.Tensor,
    catalog_untrained_page_ids_t: torch.Tensor,
) -> torch.Tensor:
    """Intersect discovered capacity with explicit untrained catalog authority."""

    catalog_ids = catalog_page_ids_t.detach().cpu().long().reshape(-1)
    untrained_ids = catalog_untrained_page_ids_t.detach().cpu().long().reshape(-1)
    if (
        torch.unique(catalog_ids).numel() != catalog_ids.numel()
        or torch.unique(untrained_ids).numel() != untrained_ids.numel()
        or not untrained_ids.unsqueeze(1).eq(catalog_ids.unsqueeze(0)).any(dim=1).all()
    ):
        raise RuntimeError("compact training catalog authority differs")
    discovered_ids = (
        torch.cat(tuple(stream.page_ids_t for stream in streams), dim=0)
        if streams
        else torch.zeros(0, dtype=torch.long)
    )
    if discovered_ids.numel() == 0:
        return discovered_ids
    discovered_ids = torch.unique(discovered_ids, sorted=True)
    return discovered_ids[
        discovered_ids.unsqueeze(1).eq(untrained_ids.unsqueeze(0)).any(dim=1)
    ]


def _append_journal_rows(
    path: Path,
    rows: tuple[Mapping[str, Any], ...],
) -> None:
    if not rows:
        raise ValueError("compact page journal chunk is empty")
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        for row in rows:
            handle.write(
                json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n"
            )
        handle.flush()
        os.fsync(handle.fileno())


def _write_compact_summary(
    *,
    summary_path: Path,
    journal_path: Path,
    records: Mapping[int, Mapping[str, Any]],
    store_root: Path,
    session_id: list[int],
    source_checkpoint_sha256: str,
    start_page_id: int,
    requested_page_count: int,
    page_parameter_elements: int,
) -> dict[str, Any]:
    ordered = [records[page_id] for page_id in sorted(records)]
    payload: dict[str, Any] = {
        "schema": COMPACT_TRANSFER_SUMMARY_SCHEMA,
        "passed": len(ordered) == requested_page_count,
        "storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
        "storeRoot": str(store_root),
        "sessionId": session_id,
        "sourceCheckpointSha256": source_checkpoint_sha256,
        "startPageId": start_page_id,
        "requestedPageCount": requested_page_count,
        "journaledPageCount": len(ordered),
        "journalPath": str(journal_path),
        "journalSha256": _file_sha256(journal_path),
        "pageParameterElements": page_parameter_elements,
        "physicalParameterElementsInitialized": (
            len(ordered) * page_parameter_elements
        ),
        "objectBytes": sum(int(row["objectBytes"]) for row in ordered),
        "completePageMappingRetained": True,
        "trainedPageCount": 0,
        "transferInitializedPageCount": len(ordered),
        "acceptedGenerationCommitted": False,
        "promotionRequiresTrainingAndHeldoutProof": True,
    }
    _atomic_json(summary_path, payload)
    return payload


def write_compact_transfer_pages(
    *,
    source_gate_t: torch.Tensor,
    source_up_t: torch.Tensor,
    source_down_t: torch.Tensor,
    source_glyph_down_t: torch.Tensor,
    source_glyph_up_t: torch.Tensor,
    source_checkpoint_sha256: str,
    store_root: Path,
    session_id_t: torch.Tensor,
    journal_path: Path,
    summary_path: Path,
    start_page_id: int,
    page_count: int,
    router_size: int,
    writer_source_sha256: str | None = None,
    receipt_every: int = 100,
    registry_roots: tuple[Path, ...] | None = None,
) -> dict[str, Any]:
    """Create and verify every compact candidate with crash-safe full mapping."""

    if len(source_checkpoint_sha256) != 64:
        raise ValueError("source checkpoint SHA-256 is malformed")
    writer_source_diagnostic = (
        writer_source_sha256
        if _diagnostic_sha256_bytes(writer_source_sha256) is not None
        else None
    )
    if start_page_id < 0 or page_count < 1 or receipt_every < 1:
        raise ValueError("compact page range and receipt cadence must be positive")
    session_id = session_id_t.detach().cpu().long().reshape(-1).tolist()
    if not session_id:
        raise ValueError("compact page growth requires a session identity")
    store = NoNEImmutablePageStore(store_root)
    store.begin_session(session_id_t)
    template_weights = compact_transfer_weights_from_source(
        page_id=0,
        source_gate_t=source_gate_t,
        source_up_t=source_up_t,
        source_down_t=source_down_t,
        source_glyph_down_t=source_glyph_down_t,
        source_glyph_up_t=source_glyph_up_t,
        router_size=router_size,
    )
    transfer_template = store.build_compact_transfer_template_boundary(
        template_weights
    )
    records = _read_compact_journal(
        journal_path,
        source_sha256=source_checkpoint_sha256,
        session_id=session_id,
    )
    requested_ids = range(start_page_id, start_page_id + page_count)
    unexpected = set(records).difference(requested_ids)
    if unexpected:
        raise RuntimeError("compact page journal contains out-of-range pages")
    page_parameter_elements = sum(
        int(tensor.numel())
        for tensor in (
            source_gate_t,
            source_up_t,
            source_down_t,
            source_glyph_down_t,
            source_glyph_up_t,
        )
    ) + 1 + 3 * router_size
    pending_bindings: list[NoNEPageObjectBinding] = []
    pending_records: list[dict[str, Any]] = []

    def commit_pending_chunk() -> None:
        if not pending_bindings:
            return
        store.commit_compact_transfer_chunk_boundary(tuple(pending_bindings))
        _append_journal_rows(journal_path, tuple(pending_records))
        for record in pending_records:
            records[int(record["pageId"])] = record
        pending_bindings.clear()
        pending_records.clear()
        _write_compact_summary(
            summary_path=summary_path,
            journal_path=journal_path,
            records=records,
            store_root=store.root,
            session_id=session_id,
            source_checkpoint_sha256=source_checkpoint_sha256,
            start_page_id=start_page_id,
            requested_page_count=page_count,
            page_parameter_elements=page_parameter_elements,
        )

    for page_id in requested_ids:
        existing = records.get(page_id)
        if existing is not None:
            store.verify_compact_transfer_page_object_boundary(
                _binding_from_record(existing)
            )
        else:
            binding = store.write_compact_transfer_template_page_boundary(
                transfer_template,
                page_id=page_id,
            )
            record: dict[str, Any] = {
                "schema": COMPACT_TRANSFER_JOURNAL_SCHEMA,
                "pageId": page_id,
                "objectSha256": _digest_hex(binding.object_sha256_t),
                "objectBytes": int(binding.object_bytes_t.detach().cpu()),
                "storageFormat": SCALED_FLOAT8_TRANSFER_STORAGE,
                "sourceCheckpointSha256": source_checkpoint_sha256,
                "sessionId": session_id,
                "trained": False,
                "acceptedGenerationCommitted": False,
            }
            pending_bindings.append(binding)
            pending_records.append(record)
            if len(pending_bindings) >= receipt_every:
                commit_pending_chunk()
    commit_pending_chunk()
    summary = _write_compact_summary(
        summary_path=summary_path,
        journal_path=journal_path,
        records=records,
        store_root=store.root,
        session_id=session_id,
        source_checkpoint_sha256=source_checkpoint_sha256,
        start_page_id=start_page_id,
        requested_page_count=page_count,
        page_parameter_elements=page_parameter_elements,
    )
    publish_compact_transfer_page_bank_locator_boundary(
        summary_path,
        registry_roots=registry_roots,
    )
    if writer_source_diagnostic is None:
        return summary
    return {
        **summary,
        "writerSourceSha256Diagnostic": writer_source_diagnostic,
        "writerSourceSha256DiagnosticOnly": True,
    }