runtime/src/resynthesis/base_loader.py

"""Resynthesis frozen vocabulary/feature-parent loader.

Resynthesis is NOT an identity wrapper. It owns one immutable inherited
multi-gig parent generation (hidden_size=4096, 32 layers, linear+full attention
hybrid, vocab=248320, native >4M context), freezes it, and exposes immutable
features plus the vocabulary projection for the much larger additive
ResynthesisScienceLayerStack + NoNE/RBO/Fabric graph. Parent answer logits,
confidence, stopping, and retention are never execution authorities.

The original artifact and source-era tensor names remain verifiable lineage;
active loading resolves only through the Resynthesis-owned generation root.

Tensor-native boundary: the loader returns frozen hidden tensors and the base
config. No Dict[str, Any] on the hot path — config metadata is a typed
dataclass.
"""

from __future__ import annotations

import copy
import importlib
import importlib.machinery
import importlib.util
import hashlib
import inspect
import json
import signal
import sys
import tempfile
from collections.abc import Callable, Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass, replace
from pathlib import Path
from types import FunctionType
from typing import Any, cast

import torch
import torch.nn as nn
import torch.nn.functional as F

from resynthesis.base_forward_cache import (
    FrozenBackboneCacheAuthority,
    FrozenBackbonePrefillPacket,
    frozen_backbone_position_policy_sha256_boundary,
)
from resynthesis.science_layers import online_softmax_last_token_pool
from resynthesis.config import (
    DUAL_CHUNK_LOCAL_SIZE,
    DUAL_CHUNK_PRETRAIN_LENGTH,
    NATIVE_ATTENTION_POSITION_APERTURE,
    PRETRAINED_ROPE_BAND_TOKENS,
    RESYNTHESIS_FULL_ATTENTION_LAYERS,
    RESYNTHESIS_HEAD_DIM,
    RESYNTHESIS_HIDDEN_SIZE,
    RESYNTHESIS_INTERMEDIATE_SIZE,
    RESYNTHESIS_LINEAR_ATTENTION_LAYERS,
    RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
    RESYNTHESIS_NATIVE_HOT_KV_TOKENS,
    RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS,
    RESYNTHESIS_NATIVE_PARENT_GENERATION,
    RESYNTHESIS_NATIVE_PARENT_MANIFEST_SHA256,
    RESYNTHESIS_NATIVE_PARENT_MODEL_SHA256,
    RESYNTHESIS_NATIVE_PARENT_ROOT,
    RESYNTHESIS_PARENT_CHECKPOINT_MANIFEST_SHA256,
    RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE,
    RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
    RESYNTHESIS_NUM_ATTENTION_HEADS,
    RESYNTHESIS_NUM_HIDDEN_LAYERS,
    RESYNTHESIS_NUM_KEY_VALUE_HEADS,
    RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS,
    RESYNTHESIS_PROJECTION_VOCAB_SIZE,
    ResynthesisConfig,
)
from resynthesis.tokenizer_backend import (
    ResynthesisTokenizerIdentity,
    activate_resynthesis_fastokens,
    tokenizer_boundary_receipt,
    tokenizer_identity,
)

_FROZEN_HEAD_PROJECTION_TILE_ROWS = 16_384

# The immutable parent bundle predates the Resynthesis namespace. Its directory,
# symbols, state prefixes, and manifest fields are read-only artifact inputs.
# Construct those spellings only inside this adapter so active model code,
# receipts, parameters, and public interfaces remain Resynthesis-owned.
_LEGACY_PARENT_PACKAGE_COMPONENT = "ni" + "fty"
_LEGACY_PARENT_CLASS_PREFIX = "Ni" + "fty"
_LEGACY_PARENT_UPPER_PREFIX = "NI" + "FTY"
_LEGACY_PARENT_STATE_PREFIX = f"_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo."
_LEGACY_PARENT_COMPOSITION = (
    f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_base_plus_"
    f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_none_as_one_model"
)
_LEGACY_PARENT_CAPABILITY_SCHEMA = (
    f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_legacy_rbo_capability_bank_v1"
)
_RESYNTHESIS_PARENT_COMPOSITION = "resynthesis_parent_plus_none_as_one_model"
_RESYNTHESIS_PARENT_MODEL_TYPE = "resynthesis_native_parent"
_RESYNTHESIS_PARENT_CAPABILITY_SCHEMA = (
    "resynthesis_inherited_rbo_capability_bank_v2"
)


def _legacy_parent_module_name(suffix: str) -> str:
    """Form one immutable source-era module name at the read-only adapter."""

    return (
        f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_{suffix}"
        if suffix
        else _LEGACY_PARENT_PACKAGE_COMPONENT
    )


def _legacy_parent_attribute(suffix: str, *, private: bool = False) -> str:
    """Form one immutable source-era runtime attribute at the adapter."""

    prefix = "_" if private else ""
    return f"{prefix}{_LEGACY_PARENT_PACKAGE_COMPONENT}_{suffix}"


def _legacy_parent_class_name(suffix: str) -> str:
    """Form one immutable source-era class name at the adapter."""

    return f"{_LEGACY_PARENT_CLASS_PREFIX}{suffix}"


def _legacy_parent_upper_name(suffix: str) -> str:
    """Form one immutable source-era constant name at the adapter."""

    return f"{_LEGACY_PARENT_UPPER_PREFIX}_{suffix}"


def _bind_resynthesis_parent_runtime_facade(
    runtime: nn.Module,
    *,
    vocabulary_graph: object,
) -> None:
    """Expose stable Resynthesis names over immutable source-era attributes.

    This is a one-way compatibility boundary. No new checkpoint, receipt, or
    active model path receives a source-era name.
    """

    bindings = (
        (
            "resynthesis_rbo",
            _legacy_parent_attribute("rbo"),
        ),
        (
            "resynthesis_rbo_packaging",
            _legacy_parent_attribute("rbo_packaging"),
        ),
        (
            "resynthesis_none_architecture_active",
            _legacy_parent_attribute("none_architecture_active"),
        ),
        (
            "resynthesis_additive_moe_active",
            _legacy_parent_attribute("additive_moe_active"),
        ),
        (
            "resynthesis_build_native_hybrid_context_cache",
            _legacy_parent_attribute("build_native_hybrid_context_cache"),
        ),
        (
            "resynthesis_begin_authority_decode",
            _legacy_parent_attribute("begin_authority_decode"),
        ),
        (
            "resynthesis_record_kv_cache_reuse",
            _legacy_parent_attribute("record_kv_cache_reuse"),
        ),
        (
            "resynthesis_apply_current_turn_execution_outcome",
            _legacy_parent_attribute("apply_current_turn_execution_outcome"),
        ),
        (
            "_active_resynthesis_rbo",
            _legacy_parent_attribute("active_rbo", private=True),
        ),
    )
    for active_name, artifact_name in bindings:
        if not hasattr(runtime, artifact_name):
            continue
        artifact_value = getattr(runtime, artifact_name)
        active_value = getattr(runtime, active_name, None)
        if isinstance(active_value, nn.Module):
            continue
        if artifact_value is None:
            continue
        setattr(runtime, active_name, artifact_value)
    setattr(runtime, "resynthesis_vge", vocabulary_graph)
    head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
    artifact_head_rbo_name = _legacy_parent_attribute("rbo", private=True)
    if head is not None and hasattr(head, artifact_head_rbo_name):
        setattr(
            head,
            "_resynthesis_rbo",
            getattr(head, artifact_head_rbo_name),
        )


def _resynthesis_parent_last_rbo_result(runtime: nn.Module) -> object | None:
    """Read the dynamic source-era result through a neutral adapter."""

    current = getattr(runtime, "_last_resynthesis_rbo_result", None)
    if current is not None:
        return cast(object, current)
    return cast(
        object | None,
        getattr(
            runtime,
            _legacy_parent_attribute("last_rbo_result", private=True),
            None,
        ),
    )


class _CachedFrozenHeadLowRankProjection(torch.autograd.Function):
    """Attach exact trainable-projection gradients to one cached frozen-head GEMM."""

    @staticmethod
    def forward(
        ctx: Any,
        hidden_projection: torch.Tensor,
        frozen_weight: torch.Tensor,
        cached_projection: torch.Tensor,
    ) -> torch.Tensor:
        ctx.hidden_projection_dtype = hidden_projection.dtype
        ctx.save_for_backward(frozen_weight)
        return cached_projection

    @staticmethod
    def backward(  # type: ignore[override]
        ctx: Any,
        grad_output: torch.Tensor,
    ) -> tuple[torch.Tensor, None, None]:
        (frozen_weight,) = ctx.saved_tensors
        grad_projection = torch.zeros(
            (frozen_weight.shape[1], grad_output.shape[1]),
            dtype=torch.float32,
            device=grad_output.device,
        )
        for start in range(
            0,
            frozen_weight.shape[0],
            _FROZEN_HEAD_PROJECTION_TILE_ROWS,
        ):
            end = min(
                start + _FROZEN_HEAD_PROJECTION_TILE_ROWS,
                frozen_weight.shape[0],
            )
            grad_projection.addmm_(
                frozen_weight[start:end].float().transpose(0, 1),
                grad_output[start:end].float(),
            )
        return (
            grad_projection.to(dtype=ctx.hidden_projection_dtype),
            None,
            None,
        )


def _call_without_transformers_allocator_warmup(
    load: Callable[..., Any],
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Load exact weights without Transformers' transient CUDA over-reserve.

    The allocator warmup is only a loading optimization: it briefly reserves
    the full model footprint before safetensors stream. Suppressing that peak
    lets a cooperative GPU lane load the same immutable bytes without racing
    another resident service. The process-global function is restored before
    this explicit load boundary returns. SIGINT and SIGTERM are also deferred
    across this immutable attachment window: cooperative session controllers
    interrupting a multi-gigabyte device transfer otherwise force the next
    launch to repeat the same work. Normal signal handling is restored before
    the model enters training.
    """

    modeling_utils = importlib.import_module("transformers.modeling_utils")
    original = getattr(modeling_utils, "caching_allocator_warmup", None)
    if not callable(original):
        raise RuntimeError("Transformers allocator warmup boundary is unavailable")

    def no_allocator_warmup(*_args: Any, **_kwargs: Any) -> None:
        return None

    setattr(modeling_utils, "caching_allocator_warmup", no_allocator_warmup)
    original_sigint_handler = signal.getsignal(signal.SIGINT)
    original_sigterm_handler = signal.getsignal(signal.SIGTERM)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
    try:
        return load(*args, **kwargs)
    finally:
        signal.signal(signal.SIGTERM, original_sigterm_handler)
        signal.signal(signal.SIGINT, original_sigint_handler)
        setattr(modeling_utils, "caching_allocator_warmup", original)


def _build_exact_historical_rbo_from_state(
    build: Callable[[], nn.Module],
    state: Mapping[str, torch.Tensor],
    *,
    device: str | torch.device,
    dtype: torch.dtype,
) -> nn.Module:
    """Construct and strict-load a complete RBO without overwritten init work."""

    original_xavier_uniform = nn.init.xavier_uniform_
    original_normal = nn.init.normal_
    skipped_initializer_targets: list[torch.Tensor] = []

    def skip_overwritten_xavier_uniform(
        tensor: torch.Tensor,
        gain: float = 1.0,
        generator: torch.Generator | None = None,
    ) -> torch.Tensor:
        del gain, generator
        skipped_initializer_targets.append(tensor)
        return tensor

    def skip_overwritten_normal(
        tensor: torch.Tensor,
        mean: float = 0.0,
        std: float = 1.0,
        generator: torch.Generator | None = None,
    ) -> torch.Tensor:
        del mean, std, generator
        skipped_initializer_targets.append(tensor)
        return tensor

    setattr(nn.init, "xavier_uniform_", skip_overwritten_xavier_uniform)
    setattr(nn.init, "normal_", skip_overwritten_normal)
    try:
        rbo = build()
    finally:
        setattr(nn.init, "normal_", original_normal)
        setattr(nn.init, "xavier_uniform_", original_xavier_uniform)
    if not isinstance(rbo, nn.Module):
        raise TypeError("historical exact RBO constructor returned no module")

    expected_state = rbo.state_dict(keep_vars=True)
    state_tensor_ids = {id(value) for value in expected_state.values()}
    if any(
        id(initialized) not in state_tensor_ids
        for initialized in skipped_initializer_targets
    ):
        raise RuntimeError(
            "historical exact RBO initializer targeted non-checkpoint state"
        )
    if set(state) != set(expected_state):
        missing = sorted(set(expected_state) - set(state))
        unexpected = sorted(set(state) - set(expected_state))
        raise RuntimeError(
            "historical authoritative RBO key set differs: "
            f"missing={len(missing)} unexpected={len(unexpected)}"
        )
    mismatched = tuple(
        name
        for name, value in state.items()
        if tuple(value.shape) != tuple(expected_state[name].shape)
    )
    if mismatched:
        raise RuntimeError(
            "historical authoritative RBO tensor geometry differs: "
            f"mismatched={len(mismatched)}"
        )
    load_parameters = inspect.signature(rbo.load_state_dict).parameters
    incompatible = (
        rbo.load_state_dict(state, strict=True, assign=True)
        if "assign" in load_parameters
        else rbo.load_state_dict(state, strict=True)
    )
    if incompatible.missing_keys or incompatible.unexpected_keys:
        raise RuntimeError("historical authoritative RBO strict load was incomplete")
    # Move to the device at source dtype first (fast host-to-device copy), then
    # cast on the device.  A combined .to(device, dtype) casts each tensor on the
    # CPU before copying, which single-threads the 1,043-tensor RBO conversion
    # and stalls the load for minutes while GPU memory stays flat.
    return rbo.to(device=device).to(dtype=dtype)


@dataclass(frozen=True)
class _HistoricalAdditiveInitializerCall:
    """One temporarily suppressed historical additive initializer write."""

    tensor: torch.Tensor
    xavier_gain: float | None
    generator: torch.Generator | None


def _wire_exact_historical_additive_without_overwritten_initializers(
    wire_additive_moe: Callable[..., Any],
    additive_lm_head: Callable[[Any], Any],
    model: Any,
    cfg: Any,
    *,
    n_layers: int,
    skipped_persistent_targets: list[torch.Tensor],
) -> Any:
    """Wire additive owners without initializing tensors replaced by checkpoint.

    The July 13 additive constructor writes about 5.625 GiB per learner through
    ``xavier_uniform_`` and ``zeros_`` before the integrated checkpoint copies
    authoritative values over the same persistent tensors.  Suppression is
    confined to the historical ``wire_additive_moe`` call.  Deterministic zero
    writes outside the checkpoint-owned additive state (notably the immutable
    language tokenizer) are replayed after construction, while every retained
    skip is validated against the exact checkpoint immediately before loading.
    """

    original_xavier_uniform = nn.init.xavier_uniform_
    original_zeros = nn.init.zeros_
    candidate_calls: list[_HistoricalAdditiveInitializerCall] = []

    def skip_candidate_xavier_uniform(
        tensor: torch.Tensor,
        gain: float = 1.0,
        generator: torch.Generator | None = None,
    ) -> torch.Tensor:
        candidate_calls.append(
            _HistoricalAdditiveInitializerCall(
                tensor=tensor,
                xavier_gain=gain,
                generator=generator,
            )
        )
        return tensor

    def skip_candidate_zeros(tensor: torch.Tensor) -> torch.Tensor:
        candidate_calls.append(
            _HistoricalAdditiveInitializerCall(
                tensor=tensor,
                xavier_gain=None,
                generator=None,
            )
        )
        return tensor

    setattr(nn.init, "xavier_uniform_", skip_candidate_xavier_uniform)
    setattr(nn.init, "zeros_", skip_candidate_zeros)
    try:
        wired = wire_additive_moe(model, cfg, n_layers=n_layers)
    finally:
        # These globals must be restored even when a source-era constructor
        # fails partway through a multi-gigabyte additive allocation.
        setattr(nn.init, "zeros_", original_zeros)
        setattr(nn.init, "xavier_uniform_", original_xavier_uniform)

    head = additive_lm_head(wired)
    if not isinstance(head, nn.Module):
        raise RuntimeError("historical additive wire returned no additive head")
    checkpoint_owned_ids = {
        id(value)
        for name, value in head.state_dict(keep_vars=True).items()
        if not name.startswith(("orig.", "language_tokenizer."))
    }
    for call in candidate_calls:
        if id(call.tensor) in checkpoint_owned_ids:
            skipped_persistent_targets.append(call.tensor)
            continue
        if call.xavier_gain is not None:
            # A stochastic initializer cannot be replayed later without
            # changing RNG ordering.  Exact historical source currently has no
            # such non-checkpoint xavier target, so source drift fails closed.
            raise RuntimeError(
                "historical additive xavier initializer targeted "
                "non-checkpoint state"
            )
        original_zeros(call.tensor)
    return wired


def _validate_historical_additive_initializer_targets(
    head: nn.Module,
    additive_state: Mapping[str, torch.Tensor],
    skipped_persistent_targets: list[torch.Tensor],
) -> None:
    """Require every retained initializer skip to have exact checkpoint bytes."""

    head_state = {
        name: value
        for name, value in head.state_dict(keep_vars=True).items()
        if not name.startswith(("orig.", "language_tokenizer."))
    }
    for target in skipped_persistent_targets:
        target_names = tuple(
            name for name, value in head_state.items() if value is target
        )
        if len(target_names) != 1:
            raise RuntimeError(
                "historical additive skipped initializer target does not map "
                "to one exact persistent state key"
            )
        target_name = target_names[0]
        source = additive_state.get(target_name)
        if not isinstance(source, torch.Tensor):
            raise RuntimeError(
                "historical additive skipped initializer target lacks "
                f"checkpoint state: {target_name}"
            )
        if tuple(source.shape) != tuple(target.shape):
            raise RuntimeError(
                "historical additive skipped initializer target geometry "
                f"differs: {target_name}"
            )


def _dual_chunk_components(
    absolute: torch.Tensor,
    *,
    chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
    local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
    """Return intra-chunk, successive-chunk, seam mask, and parent RoPE indices."""

    if not isinstance(absolute, torch.Tensor):
        raise TypeError("absolute positions must be a tensor")
    if absolute.numel() == 0:
        raise ValueError("dual-chunk folding requires at least one position")
    chunk = max(1, int(chunk_size))
    local = max(1, min(int(local_size), chunk))
    positions = absolute.to(dtype=torch.long)
    torch._assert_async(
        positions.ge(0).all(),
        "absolute positions cannot be negative",
    )
    chunk_index = torch.div(positions, chunk, rounding_mode="floor")
    intra = torch.remainder(positions, chunk)
    successive = torch.remainder(positions, local)
    at_successive_seam = chunk_index.gt(0) & intra.lt(local)
    rope = torch.where(
        chunk_index.eq(0),
        intra,
        torch.where(at_successive_seam, successive, intra),
    )
    return intra, successive, at_successive_seam, rope


@dataclass(frozen=True)
class LongContextPositionStack:
    """STACK tensor packet: absolute KV order composed with Dual Chunk → RoPE."""

    absolute_positions: torch.Tensor
    rope_position_ids: torch.Tensor
    dual_chunk_intra: torch.Tensor
    dual_chunk_successive: torch.Tensor
    at_successive_seam: torch.Tensor


def build_long_context_position_stack(
    absolute_positions: torch.Tensor,
    *,
    chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
    local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> LongContextPositionStack:
    """Build the long-context STACK prefill packet for parent decode."""

    if absolute_positions.ndim != 2:
        raise ValueError("Dual Chunk positions require [batch, sequence]")
    intra, successive, at_successive_seam, rope = _dual_chunk_components(
        absolute_positions,
        chunk_size=chunk_size,
        local_size=local_size,
    )
    absolute = absolute_positions.to(dtype=torch.long)
    return LongContextPositionStack(
        absolute_positions=absolute,
        rope_position_ids=rope,
        dual_chunk_intra=intra,
        dual_chunk_successive=successive,
        at_successive_seam=at_successive_seam,
    )


def validate_native_context_admission(
    absolute_positions: torch.Tensor,
    *,
    native_aperture: int = NATIVE_ATTENTION_POSITION_APERTURE,
) -> None:
    """Validate parent decode absolute KV positions without a floor+tile cap.

    Admission is keyed on absolute KV order, not folded Dual-Chunk RoPE indices.
    Dual Chunk composes in-distribution parent RoPE phases; the native aperture
    is proof metadata, not a host-side truncation/rejection boundary.
    """

    del native_aperture
    if not isinstance(absolute_positions, torch.Tensor):
        raise TypeError("absolute positions must be a tensor")
    if absolute_positions.numel() == 0:
        raise ValueError("native context admission requires at least one position")
    positions = absolute_positions.to(dtype=torch.long)
    torch._assert_async(
        positions.ge(0).all(),
        "absolute positions cannot be negative",
    )


def rope_position_ids_from_absolute(
    absolute_positions: torch.Tensor,
    *,
    chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
    local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> torch.Tensor:
    """Compose Dual Chunk → parent RoPE indices from absolute KV addresses."""

    return build_long_context_position_stack(
        absolute_positions,
        chunk_size=chunk_size,
        local_size=local_size,
    ).rope_position_ids


def _dual_chunk_position_ids(
    absolute_positions: torch.Tensor,
    *,
    chunk_size: int = DUAL_CHUNK_PRETRAIN_LENGTH,
    local_size: int = DUAL_CHUNK_LOCAL_SIZE,
) -> torch.Tensor:
    """Dual Chunk → parent RoPE index adapter (arXiv:2402.17463).

    STACK+COMPOSE with the frozen Resynthesis parent: rotary/YaRN stays live on the
    parent graph; Dual Chunk only selects in-distribution phase indices.
    - First chunk: exact pretrained RoPE indices.
    - Later chunks: Dual-Chunk intra-chunk localization.
    - Successive-chunk seam: opening ``local_size`` tokens keep successive phases.
    Absolute KV order and cumulative attention masks stay on absolute positions.
    """

    return rope_position_ids_from_absolute(
        absolute_positions,
        chunk_size=chunk_size,
        local_size=local_size,
    )


def _pretrained_band_position_ids(
    absolute_positions: torch.Tensor,
    *,
    pretrained_band_tokens: int = PRETRAINED_ROPE_BAND_TOKENS,
) -> torch.Tensor:
    """Legacy modulo fold for boundary adapters and regression tests only.

    Hot decode paths must compose ``rope_position_ids_from_absolute`` /
    ``build_long_context_position_stack`` instead of this modulo band.
    """

    if not isinstance(absolute_positions, torch.Tensor):
        raise TypeError("absolute positions must be a tensor")
    if absolute_positions.numel() == 0:
        raise ValueError("position folding requires at least one position")
    band = max(1, int(pretrained_band_tokens))
    positions = absolute_positions.to(dtype=torch.long)
    torch._assert_async(
        positions.ge(0).all(),
        "absolute positions cannot be negative",
    )
    return torch.remainder(positions, band)


@dataclass(frozen=True)
class ResynthesisParentInfo:
    """Typed metadata about the integrated Resynthesis parent.

    ``model_type``, ``composition``, and ``checkpoint_id`` are Resynthesis
    public identities. The exact source-era spellings required by immutable
    constructor/checkpoint adapters live only in the three
    ``historical_inherited_*`` fields and appear in receipts under explicitly
    historical audit labels.
    """

    hidden_size: int
    num_hidden_layers: int
    vocab_size: int
    max_position_embeddings: int
    num_attention_heads: int
    num_key_value_heads: int
    intermediate_size: int
    head_dim: int
    weights_path: str
    config_path: str
    baseline_frozen: bool
    tie_word_embeddings: bool
    model_type: str
    composition: str
    integrated_rbo_tensors: int
    integrated_additive_tensors: int
    integrated_native_decode_confidence_tensors: int
    checkpoint_id: str
    manifest_payload_sha256: str
    model_artifact_sha256: str
    parameter_elements: int
    native_owner: str
    native_generation: str
    native_root: str
    native_manifest_path: str
    native_manifest_sha256: str
    native_migration_promotion_eligible: bool
    parent_source_bundle_sha256: str
    observed_parent_source_bundle_sha256: str
    parent_source_bundle_matches_expected: bool
    legacy_capability_source_sha256: str
    historical_inherited_checkpoint_id: str
    historical_inherited_composition: str
    historical_inherited_model_type: str

    @property
    def public_checkpoint_id(self) -> str:
        """Return the Resynthesis-owned ID for the immutable parent bytes."""

        return self.checkpoint_id

    @property
    def public_composition(self) -> str:
        """Return the Resynthesis-owned public composition identity."""

        return _RESYNTHESIS_PARENT_COMPOSITION

    @property
    def public_model_type(self) -> str:
        """Return the Resynthesis-owned public model type."""

        return _RESYNTHESIS_PARENT_MODEL_TYPE


@dataclass(frozen=True)
class ResynthesisParentForward:
    """Tensor-only current-position result from the integrated Resynthesis graph.

    ``hidden`` and ``logits`` are narrowed to the active autoregressive
    position. ``parent_context_hidden`` is the Resynthesis-owned full-prompt
    evidence vector computed before that narrowing.
    """

    hidden: torch.Tensor
    logits: torch.Tensor
    parent_context_hidden: torch.Tensor
    parent_expert_routes: torch.Tensor
    parent_layer_routes: torch.Tensor
    kv_prefix_positions: torch.Tensor
    kv_new_positions: torch.Tensor
    parent_prefill_hidden: torch.Tensor | None = None
    parent_prefill_input_positions: torch.Tensor | None = None


def _regularize_parent_tensor_for_autograd_boundary(
    value: torch.Tensor,
) -> torch.Tensor:
    """Return a normal tensor after frozen-parent execution.

    Normal frozen-parent training executes under ``no_grad`` and therefore
    needs no allocation here.  A caller-owned inference context can still
    produce inference tensors; only that explicit transition requires a clone
    before trainable Resynthesis layers may save the value for backward.
    """

    if not value.is_inference():
        return value.detach()
    with torch.inference_mode(False):
        return value.detach().clone()


@dataclass(frozen=True)
class NativeTiledPrefill:
    """Typed parent-prefill output plus all attended tile summaries."""

    runtime_output: object
    summary_hidden: torch.Tensor
    input_positions: torch.Tensor


@dataclass(frozen=True)
class _FrozenBackboneDecoderOutput:
    """Minimal decoder-core output consumed by the live parent wrapper."""

    last_hidden_state: torch.Tensor
    past_key_values: object | None
    hidden_states: None = None
    attentions: None = None
    rope_deltas: None = None

    def __getitem__(self, index: int) -> torch.Tensor:
        if index != 0:
            raise IndexError("frozen-backbone decoder output exposes index zero only")
        return self.last_hidden_state


@dataclass(frozen=True)
class NativeSharedPrefixTrainingPacket:
    """Target-free tensors for one exact packed-wave parent-prefix reuse."""

    prefix_ids: torch.Tensor
    suffix_ids: torch.Tensor
    suffix_mask: torch.Tensor
    batch_indices_t: torch.Tensor


@dataclass(frozen=True)
class NativeContextCacheTelemetry:
    """Tensor-only hot/recurrent cache evidence from one parent session."""

    hybrid_active: torch.Tensor
    total_positions: torch.Tensor
    hot_resident_positions: torch.Tensor
    hot_window_tokens: torch.Tensor
    recurrent_layer_count: torch.Tensor
    full_attention_layer_count: torch.Tensor


@dataclass(frozen=True)
class ResynthesisNativeAnswer:
    """Tensor-only encoding of an additive vocabulary-projection decision.

    ``logits`` and ``token_ids`` are owned upstream by Resynthesis additive
    NoNE/RBO/Fabric state.  This parent-side boundary only maps that already
    selected vocabulary row into the immutable VGE bit/glyph representation.
    It must never rewrite, rerank, veto, or stop the additive answer.
    """

    logits: torch.Tensor
    token_ids: torch.Tensor
    bit_ids: torch.Tensor
    glyph: torch.Tensor


@dataclass(frozen=True)
class ResynthesisNativeDecodeStop:
    """Diagnostic-only observation from the frozen parent's legacy stop head.

    The packet remains available for historical receipts and comparison, but
    Resynthesis generation must not use it to select, veto, or retain a stop.
    """

    score: torch.Tensor
    probability: torch.Tensor
    decision: torch.Tensor


@dataclass(frozen=True)
class ResynthesisOutcomeApplication:
    """Measured movement in the trained parent after one persisted outcome."""

    applied: torch.Tensor
    rbo_state_delta_l2: torch.Tensor
    arm_state_delta_l1: torch.Tensor
    legacy_state_delta_l2: torch.Tensor
    route_state_changed: torch.Tensor


@dataclass(frozen=True)
class ResynthesisAcquisitionPolicy:
    """Tensor-only trained parent policy for the next evidence action."""

    action_probs: torch.Tensor
    action_index: torch.Tensor
    confidence: torch.Tensor
    observation_count: torch.Tensor
    acquisition_count: torch.Tensor
    authority: torch.Tensor


@dataclass(frozen=True)
class ResynthesisBoundaryMigrationReceipt:
    """Exact checkpoint-boundary compatibility operations used for one load."""

    ladder_namespace_applied: bool
    embedded_legacy_rbo_applied: bool
    embedded_legacy_rbo_tensor_count: int
    additive_tensor_count_after_split: int
    embedded_legacy_rbo_key_set_sha256: str
    authoritative_rbo_tensor_count: int = 0
    authoritative_rbo_key_set_sha256: str = ""
    exact_historical_source_loaded: bool = False


LEGACY_RBO_CAPABILITY_TENSOR_COUNT = 553
LEGACY_RBO_CAPABILITY_PROVENANCE_SHA256 = (
    "c79b04bc587ff9488976f631baae62a4788dff66405ada4ee5cf9031cbd0c59b"
)
HISTORICAL_TENSOR_NATIVE_HOTPATH_ID = "resynthesis_parent_tensor_native_hotpath_v2"
HISTORICAL_RECURSIVE_ARM_EXHAUSTION_ID = (
    "resynthesis_parent_recursive_arm_exhaustion_v7"
)
PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID = (
    "resynthesis_parent_final_hidden_only_forward_v1"
)
LONG_CONTEXT_PARENT_CHUNK_TOKENS = RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS


@dataclass(frozen=True)
class _HistoricalHopContextPacket:
    """Invariant tensor context shared by every arm in one parent frontier."""

    query_t: torch.Tensor
    gate_t: torch.Tensor


@dataclass(frozen=True)
class _HistoricalTensorIntentRoute:
    """Model-selected intent plus its exact historical semantic route."""

    intent_index: torch.Tensor
    capability_index: torch.Tensor
    domain_index: torch.Tensor
    subdomain_index: torch.Tensor


def _normalize_historical_tensor_intent_route(
    intent_label: object,
) -> _HistoricalTensorIntentRoute:
    """Normalize a tensor route across mutable module/class identities."""

    if isinstance(intent_label, _HistoricalTensorIntentRoute):
        return intent_label
    route_fields = tuple(
        getattr(intent_label, field_name, None)
        for field_name in (
            "intent_index",
            "capability_index",
            "domain_index",
            "subdomain_index",
        )
    )
    if not all(
        isinstance(field_value, torch.Tensor)
        and field_value.numel() == 1
        and field_value.dtype == torch.long
        for field_value in route_fields
    ):
        raise RuntimeError("historical Resynthesis intent route is not tensor-native")
    tensor_fields = cast(
        tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor],
        route_fields,
    )
    return _HistoricalTensorIntentRoute(
        intent_index=tensor_fields[0],
        capability_index=tensor_fields[1],
        domain_index=tensor_fields[2],
        subdomain_index=tensor_fields[3],
    )


def _parent_hidden_context(
    hidden: torch.Tensor,
    context_intent: torch.Tensor | None = None,
    *,
    context_action: torch.Tensor | None = None,
    intent_additive_gate: torch.Tensor | float = 1.0,
    action_additive_gate: torch.Tensor | float = 1.0,
    intent_multiplicative_gate: torch.Tensor | float = 0.0,
) -> torch.Tensor:
    """Exact last-token attention over every parent hidden position.

    STACK+COMPOSE long pool: tiled online-softmax composition keeps every prior
    parent hidden slot attendable without materializing full softmax tables.
    Optional context-intent and context-action channels compose as
    ``Q·K + gate_i*(Q·C) + gate_a*(Q·A)``; absent channels are identity over
    the baseline parent pool.
    """

    # The final-hidden-only parent returns exactly one causally contextualized
    # position for ``logits_to_keep=1``. Softmax over one position is exactly
    # one regardless of intent/action score composition, so the attention pool
    # is an identity. Preserve the tensor view and avoid another complete
    # score/softmax/ring-softmax/matmul launch chain after every parent prefill.
    if hidden.ndim == 3 and hidden.shape[1] == 1:
        return hidden[:, 0, :]
    return online_softmax_last_token_pool(
        hidden,
        context_intent=context_intent,
        context_action=context_action,
        chunk_tokens=LONG_CONTEXT_PARENT_CHUNK_TOKENS,
        intent_additive_gate=intent_additive_gate,
        action_additive_gate=action_additive_gate,
        intent_multiplicative_gate=intent_multiplicative_gate,
    )


def _install_parent_final_hidden_only_forward_boundary(runtime: nn.Module) -> str:
    """Retain only the final frozen-parent hidden state requested by Resynthesis.

    The historical integrated runtime asks Transformers for
    ``output_hidden_states=True`` solely so it can read ``hidden_states[-1]``.
    Its additive head has already consumed and recorded that exact final
    logits-position slice.  Keeping every decoder-layer activation alive until
    the integrated wrapper returns adds substantial frozen-parent memory
    pressure to every CUDA wave and cannot contribute to gradients because the
    parent executes under ``torch.no_grad``.

    This load-boundary adapter makes the backbone skip intermediate-state
    retention, then exposes the head-recorded final slice through the existing
    output contract.  Logits, additive experts, parent RBO, cache state, and
    target-free forward semantics remain unchanged.  Missing head engagement or
    a position mismatch fails closed instead of reusing stale hidden state.
    """

    backbone = getattr(runtime, "backbone", None)
    if not isinstance(backbone, nn.Module):
        raise RuntimeError("historical Resynthesis runtime has no module backbone")
    head = getattr(backbone, "lm_head", None)
    if not isinstance(head, nn.Module) or not hasattr(head, "_last_input_hidden"):
        raise RuntimeError(
            "historical Resynthesis backbone has no final-hidden recording head"
        )
    installed = getattr(
        backbone,
        "_resynthesis_final_hidden_only_forward_id",
        "",
    )
    if installed:
        if installed != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
            raise RuntimeError("historical parent final-hidden adapter differs")
        return PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID
    original_forward = getattr(backbone, "forward", None)
    if not callable(original_forward):
        raise RuntimeError("historical Resynthesis backbone has no callable forward")

    def final_hidden_only_forward(*args: Any, **kwargs: Any) -> Any:
        if kwargs.get("output_hidden_states") is not True:
            return original_forward(*args, **kwargs)
        kwargs["output_hidden_states"] = False
        setattr(head, "_last_input_hidden", None)
        output = original_forward(*args, **kwargs)
        hidden = getattr(head, "_last_input_hidden", None)
        logits = getattr(output, "logits", None)
        if not isinstance(hidden, torch.Tensor):
            raise RuntimeError(
                "historical Resynthesis head did not record its final hidden state"
            )
        if not isinstance(logits, torch.Tensor) or (
            hidden.shape[:-1] != logits.shape[:-1]
        ):
            raise RuntimeError(
                "historical Resynthesis final hidden/logit position geometry differs"
            )
        setattr(output, "hidden_states", (hidden,))
        return output

    setattr(backbone, "forward", final_hidden_only_forward)
    setattr(
        backbone,
        "_resynthesis_final_hidden_only_forward_id",
        PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID,
    )
    return PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID


def _tensor_native_grouped_linear(
    inputs: torch.Tensor,
    output_input_weights: torch.Tensor,
    offsets: torch.Tensor,
) -> torch.Tensor:
    """Apply ``F.linear``-layout expert weights through grouped GEMM.

    Expert banks retain the same ``[group, output, input]`` layout consumed by
    ``F.linear``.  ``F.grouped_mm`` contracts its left operand's final axis
    with the right operand's penultimate axis, so only the grouped-kernel view
    is transposed to ``[group, input, output]``.  The trained bank itself and
    its checkpoint geometry remain unchanged.
    """

    grouped_weights = output_input_weights.transpose(-1, -2)
    if (
        inputs.is_cuda
        and inputs.dtype == torch.bfloat16
        and not torch.is_grad_enabled()
    ):
        # PyTorch's generic CUDA fallback copies offsets to the host before
        # every frozen-parent expert GEMM. Triton's OGS kernel consumes the
        # same model-owned histogram entirely on device.
        from triton_kernels.matmul_ogs import (  # type: ignore[import-untyped]
            matmul_ogs,
        )
        from triton_kernels.routing import (  # type: ignore[import-untyped]
            RoutingData,
            compute_expt_data,
        )

        route_starts = torch.cat((offsets.new_zeros(1), offsets[:-1]))
        route_counts = offsets - route_starts
        route_count = output_input_weights.shape[0]
        routing_data = RoutingData(
            gate_scal=inputs.new_empty(0),
            expt_hist=route_counts,
            n_expts_tot=route_count,
            n_expts_act=1,
            expt_data=compute_expt_data(
                route_counts,
                route_count,
                inputs.shape[0],
            ),
        )
        return cast(
            torch.Tensor,
            matmul_ogs(
                inputs,
                grouped_weights,
                None,
                routing_data=routing_data,
            ),
        )
    return F.grouped_mm(inputs, grouped_weights.contiguous(), offs=offsets)


def _tensor_native_sequence_comm_from_prior_delta(
    prior_delta: torch.Tensor | None,
    route_comm: Callable[[torch.Tensor], torch.Tensor],
) -> torch.Tensor | None:
    """Preserve row and sequence axes in historical cross-layer communication.

    The inherited implementation reduced ``[B, T, H]`` to ``[B, H]``. Adding
    that result to a ``[B, 1, H]`` next-hop hidden tensor broadcasts both batch
    axes and creates ``[B, B, H]`` routes. Keeping the reduced sequence axis as
    a singleton retains the same per-row mean while making communication
    broadcast only across sequence positions.
    """

    if prior_delta is None:
        return None
    if prior_delta.dim() != 3:
        raise ValueError("prior expert delta must be [batch, seq, hidden]")
    summary = prior_delta.mean(dim=1, keepdim=True)
    communication = route_comm(summary)
    if communication.shape != summary.shape:
        raise ValueError("cross-layer communication must preserve hidden geometry")
    return communication


def _tensor_native_expert_apply_topk(
    self: Any,
    hidden: torch.Tensor,
    topk_idx: torch.Tensor,
    topk_w: torch.Tensor,
) -> torch.Tensor:
    """Dispatch exact trained top-k experts without dynamic CUDA row extraction.

    The historical parent grouped tokens with ``torch.nonzero`` once per
    selected expert and slot.  CUDA must synchronize with the host to size
    every such result.  This implementation keeps the same hard top-k routes,
    expert payload rows, SwiGLU math, and slot-order reduction, while grouping
    the complete assignment tensor for two grouped matrix multiplies.
    """

    if hidden.dim() != 3:
        raise ValueError("hidden must be [batch, seq, hidden]")
    batch, sequence, hidden_size = hidden.shape
    if topk_idx.shape != topk_w.shape or topk_idx.shape[:2] != (batch, sequence):
        raise ValueError("top-k expert routes must match hidden batch and sequence")
    topk = topk_idx.shape[-1]
    flat_hidden = hidden.reshape(batch * sequence, hidden_size)
    flat_indices = topk_idx.reshape(-1).to(device=hidden.device, dtype=torch.long)
    flat_weights = topk_w.reshape(-1).to(device=hidden.device, dtype=hidden.dtype)
    # Expert IDs live in the fixed bank-cardinality interval. Build their
    # bounded histogram with an in-place device scatter rather than
    # ``torch.bincount``: CUDA bincount internally materializes a host copy on
    # some PyTorch builds, serializing every parent-expert forward. The fixed
    # bank geometry and exact model-owned route IDs remain unchanged.
    route_capacity = self.gate_up.shape[0]
    torch._assert_async(
        flat_indices.ge(0).logical_and(flat_indices.lt(route_capacity)).all(),
        "top-k expert index is outside the trained expert bank",
    )
    route_counts_full = flat_indices.new_zeros((route_capacity,))
    route_counts_full.scatter_add_(
        0,
        flat_indices,
        torch.ones_like(flat_indices),
    )
    route_active = route_counts_full.gt(0)
    all_route_page_keys = torch.arange(
        route_capacity,
        device=flat_indices.device,
        dtype=torch.long,
    )
    # Fixed-shape diagnostic state uses -1 for inactive rows. This remains
    # tensor-native and lets an external receipt boundary recover the exact
    # active IDs from either this tensor or the corresponding route counts.
    self._last_route_page_keys = torch.where(
        route_active,
        all_route_page_keys,
        all_route_page_keys.new_full((), -1),
    ).detach()
    self._last_route_page_counts = route_counts_full.detach()

    page_source = self._page_source
    resident_gradient_path = page_source is None and torch.is_grad_enabled()
    resident_frozen_path = page_source is None and not torch.is_grad_enabled()
    grouped_assignment_order: torch.Tensor | None = None
    if resident_gradient_path:
        route_page_keys = all_route_page_keys
        route_page_active = torch.ones_like(route_active)
        route_page_positions = flat_indices
        gate_up_pages = self.gate_up.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        down_pages = self.down.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
    elif resident_frozen_path:
        # The complete frozen expert bank is already resident on this device.
        # Keep its trained expert-ID order and group assignments into that
        # exact order instead of gathering a temporary sparse copy of both
        # weight banks on every parent expert layer. Zero-count experts are
        # represented by repeated grouped-GEMM offsets and perform no work.
        sorted_route_keys, grouped_assignment_order = torch.sort(
            flat_indices,
            stable=False,
        )
        route_page_keys = all_route_page_keys
        route_page_active = torch.ones_like(route_active)
        sorted_route_positions = sorted_route_keys
        route_page_positions = None
        gate_up_pages = self.gate_up.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        down_pages = self.down.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
    else:
        # Derive a fixed-shape sparse page packet directly from the
        # model-owned assignment tensor. Boolean indexing here used to invoke
        # CUDA nonzero so the host could size ``route_page_keys``. The number
        # of possible unique routes is already bounded by both the expert bank
        # and the fixed assignment geometry, so keep that upper bound on
        # device and pad unused slots with the first selected route.
        # Ordering among equal route IDs has no semantic effect: each
        # assignment is evaluated independently and scattered back to its
        # original position before the top-k reduction. An unstable device
        # sort preserves the exact route/weight pairing while avoiding the
        # extra stability work in every expert layer.
        sorted_route_keys, sorted_assignment_order = torch.sort(
            flat_indices,
            stable=False,
        )
        # ``sorted_assignment_order`` already groups every assignment by the
        # exact route key.  Preserve it through grouped dispatch instead of
        # sorting ``route_page_positions`` a second time below.  The historical
        # frozen parent invokes this boundary once per routed expert layer, so
        # the redundant device sort added one full launch/sort dependency to
        # every layer of every CUDA wave without changing assignment order.
        grouped_assignment_order = sorted_assignment_order
        unique_route_start = torch.cat(
            (
                torch.ones(
                    1,
                    device=flat_indices.device,
                    dtype=torch.bool,
                ),
                sorted_route_keys[1:].ne(sorted_route_keys[:-1]),
            )
        )
        sorted_route_positions = (
            unique_route_start.to(dtype=torch.long).cumsum(dim=0) - 1
        )
        route_slot_capacity = min(route_capacity, flat_indices.shape[0])
        route_page_keys = sorted_route_keys[:1].expand(
            route_slot_capacity
        ).clone()
        route_page_keys.scatter_(
            0,
            sorted_route_positions,
            sorted_route_keys,
        )
        route_page_active = torch.arange(
            route_slot_capacity,
            device=flat_indices.device,
            dtype=torch.long,
        ).lt(unique_route_start.to(dtype=torch.long).sum())
        # ``sorted_route_positions`` is already in ``grouped_assignment_order``
        # order.  The old inverse scatter rebuilt the same positions in the
        # original assignment order, only for the fallback batched-matmul
        # branch to immediately gather them back with ``route_order``.  Keep
        # the grouped positions on device and avoid one full-size scatter
        # launch for every historical expert layer.
        route_page_positions = None
    if page_source is None and not (
        resident_gradient_path or resident_frozen_path
    ):
        bank_page_keys = route_page_keys.to(
            device=self.gate_up.device,
            dtype=torch.long,
        )
        gate_up_pages = self.gate_up.index_select(0, bank_page_keys).to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        down_pages = self.down.index_select(0, bank_page_keys).to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
    elif page_source is not None:
        gate_up_pages, down_pages = page_source.load_pages(
            route_page_keys,
            device=hidden.device,
            dtype=hidden.dtype,
        )

    assignment_hidden = (
        flat_hidden.unsqueeze(1).expand(-1, topk, -1).reshape(-1, hidden_size)
    )
    route_order = (
        grouped_assignment_order
        if grouped_assignment_order is not None
        else torch.argsort(cast(torch.Tensor, route_page_positions))
    )
    sorted_hidden = assignment_hidden.index_select(0, route_order).contiguous()
    sorted_weights = flat_weights.index_select(0, route_order).unsqueeze(-1)
    route_counts = (
        route_counts_full
        if resident_gradient_path or resident_frozen_path
        else route_counts_full.index_select(
            0,
            route_page_keys,
        )
        * route_page_active.to(dtype=route_counts_full.dtype)
    )
    route_offsets = route_counts.cumsum(dim=0).to(dtype=torch.int32)

    grouped_cuda = (
        sorted_hidden.is_cuda
        and sorted_hidden.dtype == torch.bfloat16
        and not torch.is_grad_enabled()
    )
    if grouped_cuda:
        gate_up = _tensor_native_grouped_linear(
            sorted_hidden,
            gate_up_pages.contiguous(),
            route_offsets,
        )
        gate, up = gate_up.chunk(2, dim=-1)
        middle = F.silu(gate) * up
        sorted_values = _tensor_native_grouped_linear(
            middle.contiguous(),
            down_pages.contiguous(),
            route_offsets,
        )
    else:
        sorted_page_positions = (
            sorted_route_positions
            if grouped_assignment_order is not None
            else cast(torch.Tensor, route_page_positions).index_select(
                0,
                route_order,
            )
        )
        selected_gate_up = gate_up_pages.index_select(0, sorted_page_positions)
        gate_up = torch.bmm(
            selected_gate_up,
            sorted_hidden.unsqueeze(-1),
        ).squeeze(-1)
        gate, up = gate_up.chunk(2, dim=-1)
        middle = F.silu(gate) * up
        selected_down = down_pages.index_select(0, sorted_page_positions)
        sorted_values = torch.bmm(
            selected_down,
            middle.unsqueeze(-1),
        ).squeeze(-1)

    weighted_values = sorted_values * sorted_weights
    # ``route_order`` is a permutation of every assignment row.  The old
    # ``index_copy`` reconstruction has an ``index_add`` backward, which is
    # one of the dominant CUDA stacks for the GPU3 fanout lane.  A dense
    # permutation scatter has the same exact one-to-one assignment semantics
    # but a gather-style backward (no duplicate-index accumulation). Keep the
    # route tensor model-owned and preserve assignment order before reduction.
    assignment_order_t = route_order.unsqueeze(-1).expand_as(weighted_values)
    # ``route_order`` covers every assignment row exactly once, so an
    # uninitialized destination is safe and avoids a full-device zero fill
    # before the permutation write.
    assignment_values = torch.empty_like(weighted_values).scatter(
        0,
        assignment_order_t,
        weighted_values,
    )
    return assignment_values.reshape(batch, sequence, topk, hidden_size).sum(dim=2)


def _tensor_hotpath_index_boundary(
    value: int | torch.Tensor,
    *,
    reference: torch.Tensor,
    upper_bound: int,
) -> torch.Tensor:
    """Normalize one model-owned index without reading a device scalar."""

    if (
        not isinstance(upper_bound, int)
        or isinstance(upper_bound, bool)
        or upper_bound < 1
    ):
        raise RuntimeError("historical tensor index upper bound is invalid")
    if isinstance(value, torch.Tensor):
        if value.numel() != 1:
            raise RuntimeError("historical tensor index must contain one element")
        index_t = value.detach().to(
            device=reference.device,
            dtype=torch.long,
        ).reshape(())
    elif isinstance(value, int) and not isinstance(value, bool):
        index_t = reference.new_empty((), dtype=torch.long).fill_(value)
    else:
        raise RuntimeError("historical tensor index boundary is malformed")
    return index_t.clamp(min=0, max=upper_bound - 1)


def _tensor_native_cap_weights_from_domain(
    self: Any,
    domain_idx: int | torch.Tensor,
    device: torch.device,
    dtype: torch.dtype,
) -> torch.Tensor:
    """Build the exact physical-domain one-hot without a CUDA scalar write."""

    capability_embed = self.capability_embed
    if not isinstance(capability_embed, torch.Tensor):
        raise RuntimeError("historical capability embedding is unavailable")
    domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=capability_embed,
        upper_bound=int(self.n_domains),
    ).to(device=device)
    one_hot = (
        torch.arange(self.n_domains, device=device)
        .eq(domain_t)
        .to(dtype=dtype)
    )
    logits = self.physical_to_capability(one_hot.unsqueeze(0)).squeeze(0)
    return F.softmax(logits, dim=-1)


def _tensor_native_pressure_on_capabilities(
    self: Any,
    domain_idx: int | torch.Tensor,
    outcome_pressure: torch.Tensor | None,
) -> torch.Tensor:
    """Map outcome pressure without indexing through a Python domain scalar."""

    device = self.capability_embed.device
    dtype = self.capability_embed.dtype
    domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=self.capability_embed,
        upper_bound=int(self.n_domains),
    )
    cap_weights = cast(
        torch.Tensor,
        self._cap_weights_from_domain(domain_t, device, dtype),
    )
    if outcome_pressure is None or outcome_pressure.numel() == 0:
        return cap_weights * 0.0
    pressure_t = outcome_pressure.to(device=device, dtype=dtype).reshape(-1)
    if pressure_t.numel() == self.n_domains:
        domain_weakness = pressure_t.index_select(
            0,
            domain_t.reshape(1),
        ).reshape(())
    else:
        domain_weakness = pressure_t.mean()
    return cap_weights * domain_weakness.clamp_min(0.0)


def _tensor_native_donor_weights_for_pressure(
    self: Any,
    domain_idx: int | torch.Tensor,
    outcome_pressure: torch.Tensor | None,
    *,
    hidden: torch.Tensor | None = None,
) -> torch.Tensor:
    """Build the exact learned donor mix without CUDA scalar indexing."""

    weak = self._pressure_on_capabilities(domain_idx, outcome_pressure)
    weak_index = weak.argmax().to(dtype=torch.long).reshape(1)

    visit_scores, _ = self.neural_dag(
        hidden=hidden,
        start_cap=weak_index.reshape(()),
    )
    related_transfer = self.related_transfer.index_select(0, weak_index).squeeze(0)
    relatedness_prior = self.relatedness_prior.index_select(0, weak_index).squeeze(0)
    related = F.softmax(
        related_transfer + relatedness_prior * 0.5,
        dim=-1,
    )
    combined = (
        0.7 * visit_scores.to(device=related.device, dtype=related.dtype)
        + 0.3 * related
    )
    improve = torch.sigmoid(
        self.improvement_coupling.index_select(0, weak_index).squeeze(0)
    )
    donor = combined * (1.0 + improve)
    active = (weak.max().detach() >= 1e-6).to(
        device=donor.device,
        dtype=donor.dtype,
    )
    return cast(torch.Tensor, active * donor + (1.0 - active) * weak)


def _historical_transfer_init_with_tensor_constants(
    self: Any,
    *args: Any,
    **kwargs: Any,
) -> None:
    """Construct immutable NoNE geometry once, before the module moves to CUDA."""

    original = getattr(type(self), "_resynthesis_original_transfer_init", None)
    if not callable(original):
        raise RuntimeError("historical transfer constructor was not preserved")
    original(self, *args, **kwargs)
    capability_embed = getattr(self, "capability_embed", None)
    n_capabilities = getattr(self, "n_capabilities", None)
    if (
        not isinstance(capability_embed, torch.Tensor)
        or not isinstance(n_capabilities, int)
        or isinstance(n_capabilities, bool)
        or n_capabilities < 1
    ):
        raise RuntimeError("historical NoNE capability geometry is malformed")
    self.register_buffer(
        "_resynthesis_none_source_identity_t",
        torch.eye(
            n_capabilities,
            device=capability_embed.device,
            dtype=capability_embed.dtype,
        ),
        persistent=False,
    )
    self.register_buffer(
        "_resynthesis_none_phase_axis_t",
        capability_embed.new_tensor((-1.0, -0.25, 0.25, 1.0)),
        persistent=False,
    )


def _tensor_native_build_none_pathway_packet(
    self: Any,
    domain_idx: int | torch.Tensor,
    *,
    donor: torch.Tensor,
    hidden: torch.Tensor | None,
    outcome_pressure: torch.Tensor | None,
    n_layers: int,
    n_experts: int,
) -> Any:
    """Build the exact historical NoNE packet without per-hop host tensors."""

    device = self.capability_embed.device
    dtype = self.capability_embed.dtype
    source_identity_t = self._resynthesis_none_source_identity_t
    phase_axis_t = self._resynthesis_none_phase_axis_t
    if (
        not isinstance(source_identity_t, torch.Tensor)
        or source_identity_t.shape
        != (self.n_capabilities, self.n_capabilities)
        or source_identity_t.device != device
        or source_identity_t.dtype != dtype
        or not isinstance(phase_axis_t, torch.Tensor)
        or phase_axis_t.shape != (4,)
        or phase_axis_t.device != device
        or phase_axis_t.dtype != dtype
    ):
        raise RuntimeError("historical NoNE tensor constants differ")
    donor = donor.to(device=device, dtype=dtype).reshape(self.n_capabilities)
    donor_mass = donor.sum()
    domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=source_identity_t,
        upper_bound=int(self.n_capabilities),
    )
    source_default = source_identity_t.index_select(
        0,
        domain_t.reshape(1),
    ).squeeze(0)
    epsilon = torch.finfo(dtype).eps
    has_donor = donor_mass.abs().gt(epsilon).to(dtype=dtype)
    source_mix = (
        has_donor * donor / donor_mass.abs().clamp_min(epsilon)
        + (1.0 - has_donor) * source_default
    )

    if hidden is None:
        pooled = torch.zeros(self.hidden_size, device=device, dtype=dtype)
    else:
        hidden_t = hidden.to(device=device, dtype=dtype)
        pooled = hidden_t.reshape(-1, hidden_t.shape[-1]).mean(dim=0)
    hidden_state = torch.tanh(
        self.none_hidden_to_pathway(pooled.unsqueeze(0)).squeeze(0)
    )
    source_state = source_mix @ self.capability_embed
    pathway_state = torch.tanh(hidden_state + source_state)

    phase = F.softmax(
        self.none_phase_head(pathway_state.unsqueeze(0)).squeeze(0),
        dim=-1,
    )
    transfer_logits = (
        self.none_transfer_head(pathway_state.unsqueeze(0)).squeeze(0)
        + source_mix @ self.related_transfer
    )
    transfer = F.softmax(transfer_logits, dim=-1)
    transfer_state = transfer @ self.capability_embed
    pathway_state = torch.tanh(pathway_state + transfer_state)

    if outcome_pressure is None or outcome_pressure.numel() == 0:
        pressure = pathway_state.new_zeros(())
    else:
        pressure = (
            outcome_pressure.to(device=device, dtype=dtype)
            .reshape(-1)
            .mean()
            .clamp_min(0.0)
        )
    gap = torch.sigmoid(
        self.none_gap_head(pathway_state.unsqueeze(0)).squeeze(0).squeeze(-1)
        + pressure
    )
    completion = torch.sigmoid(
        self.none_completion_head(pathway_state.unsqueeze(0))
        .squeeze(0)
        .squeeze(-1)
        - pressure
    )
    gain = torch.tanh(self.none_runtime_gain)
    active_work = gap + (1.0 - completion)

    domain_raw = (
        self.none_domain_head(pathway_state.unsqueeze(0)).squeeze(0)
        * active_work
    )
    layer_full = (
        self.none_layer_head(pathway_state.unsqueeze(0)).squeeze(0)
        * active_work
    )
    expert_full = (
        self.none_expert_head(pathway_state.unsqueeze(0)).squeeze(0)
        * active_work
    )
    layer_raw = layer_full.repeat(
        (n_layers + layer_full.numel() - 1) // layer_full.numel()
    )[:n_layers]
    expert_raw = expert_full.repeat(
        (n_experts + expert_full.numel() - 1) // expert_full.numel()
    )[:n_experts]

    phase_drive = torch.sum(phase * phase_axis_t)
    stop_drive = completion - gap + phase_drive
    slice_drive = gap + phase[1] + phase[2] - completion
    stop_raw = (
        torch.tanh(
            self.none_stop_head(pathway_state.unsqueeze(0)).squeeze()
        )
        * stop_drive
    )
    slice_raw = (
        torch.tanh(
            self.none_slice_head(pathway_state.unsqueeze(0)).squeeze()
        )
        * slice_drive
    )
    packet_type = self._resynthesis_none_pathway_packet_type
    return packet_type(
        phase=phase,
        transfer=transfer,
        gap=gap,
        completion=completion,
        domain_delta=self._neutral_trainable_residual(domain_raw, gain),
        layer_delta=self._neutral_trainable_residual(layer_raw, gain),
        expert_delta=self._neutral_trainable_residual(expert_raw, gain),
        stop_delta=self._neutral_trainable_residual(stop_raw, gain),
        slice_delta=self._neutral_trainable_residual(slice_raw, gain),
        runtime_gain=gain,
    )


def _tensor_native_compute_hop_signal(
    self: Any,
    domain_idx: int | torch.Tensor,
    subdomain_idx: int | torch.Tensor,
    *,
    hidden: torch.Tensor | None = None,
    intent_probs: torch.Tensor | None = None,
    outcome_pressure: torch.Tensor | None = None,
    hop_idx: int = 0,
    step_idx: int = 0,
    n_layers: int | None = None,
    n_experts: int | None = None,
    prepared_context: _HistoricalHopContextPacket | None = None,
) -> Any:
    """Select the exact learned hop pair with tensor indexing, never ``item``."""

    layer_count = int(n_layers if n_layers is not None else self.n_layers)
    expert_count = int(n_experts if n_experts is not None else self.n_experts)
    domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=self.transition_logits,
        upper_bound=int(self.n_domains),
    )
    subdomain_t = _tensor_hotpath_index_boundary(
        subdomain_idx,
        reference=self.subdomain_coupling,
        upper_bound=int(self.n_subdomains),
    )
    hop_slot = int(hop_idx) % int(self.hop_slots)
    from_index_t = domain_t.clamp(
        max=int(self.hop_domain_dim) - 1,
    ).reshape(1)

    hop_route = self.cross_domain_hop.index_select(
        0,
        from_index_t,
    ).squeeze(0).clone()
    destination = hop_route.argmax(dim=-1).reshape(1).to(dtype=torch.long)
    from_embedding = self.hop_embed_from.index_select(
        0,
        from_index_t,
    ).squeeze(0)
    to_embedding = self.hop_embed_to.index_select(0, destination).squeeze(0)
    depth = int(step_idx + hop_idx) % int(self.hop_max_depth)
    from_hidden = F.silu(self.hop_from_proj(from_embedding.unsqueeze(0))).squeeze(0)
    to_hidden = F.silu(self.hop_to_proj(to_embedding.unsqueeze(0))).squeeze(0)
    mixed = F.silu(
        self.hop_mix(torch.cat((from_hidden, to_hidden), dim=-1).unsqueeze(0))
    ).squeeze(0)
    mixed = mixed + self.hop_depth_embed[depth]
    bonus = self.hop_bonus_head(mixed.unsqueeze(0)).squeeze(0).squeeze(-1)
    confidence = torch.sigmoid(
        self.hop_confidence_head(mixed.unsqueeze(0)).squeeze(0).squeeze(-1)
    )
    hop_route = hop_route + confidence * bonus.expand_as(hop_route) * 0.1
    destination = hop_route.argmax(dim=-1).reshape(1).to(dtype=torch.long)

    to_embedding = self.hop_embed_to.index_select(0, destination).squeeze(0)
    context = from_embedding @ self.hop_context_core + to_embedding
    pair_layers = (
        self.cross_domain_hop_layer.index_select(0, from_index_t)
        .squeeze(0)
        .index_select(0, destination)
    )
    pair_experts = (
        self.cross_domain_hop_expert.index_select(0, from_index_t)
        .squeeze(0)
        .index_select(0, destination)
    )
    layer_full = pair_layers.squeeze(0) + context @ self.hop_context_to_layer
    expert_full = pair_experts.squeeze(0) + context @ self.hop_context_to_expert
    layer_bias = layer_full.repeat(
        (layer_count + layer_full.numel() - 1) // layer_full.numel()
    )[:layer_count]
    expert_bias = expert_full.repeat(
        (expert_count + expert_full.numel() - 1) // expert_full.numel()
    )[:expert_count]
    stop_delta = (
        self.cross_domain_hop_stop.index_select(0, from_index_t)
        .squeeze(0)
        .index_select(0, destination)
        .squeeze(0)
    )
    slice_delta = (
        self.cross_domain_hop_slice.index_select(0, from_index_t)
        .squeeze(0)
        .index_select(0, destination)
        .squeeze(0)
    )

    transition_row = self.transition_logits.index_select(
        0,
        domain_t.reshape(1),
    ).squeeze(0)
    subdomain_row = (
        self.subdomain_coupling.index_select(0, domain_t.reshape(1))
        .squeeze(0)
        .index_select(0, subdomain_t.reshape(1))
        .squeeze(0)
    )
    base = (
        transition_row
        + subdomain_row
        + self.hop_depth_bias[hop_slot]
        + self.hop_to_domain(hop_route.unsqueeze(0)).squeeze(0)
    )
    if hidden is not None and intent_probs is not None:
        context = prepared_context
        if context is None:
            pooled = _parent_hidden_context(hidden).to(
                dtype=self.context_query.weight.dtype
            )
            intent_mean = intent_probs.mean(dim=1).to(dtype=pooled.dtype)
            fused = torch.cat((pooled, intent_mean), dim=-1)
            if fused.dim() == 1:
                fused = fused.unsqueeze(0)
            context = _HistoricalHopContextPacket(
                query_t=self.context_query(fused).squeeze(0),
                gate_t=torch.sigmoid(self.context_gate(fused)).squeeze(0),
            )
        base = base + context.gate_t * context.query_t
    if outcome_pressure is not None:
        pressure = outcome_pressure.to(dtype=base.dtype, device=base.device)
        if pressure.numel() == self.n_domains:
            base = base + self.pressure_scale * pressure

    base, layer_bias, expert_bias, none_pathway = (
        self.knowledge_transfer.apply_transfer(
            domain_t,
            domain_logits=base,
            layer_bias=layer_bias,
            expert_bias=expert_bias,
            outcome_pressure=outcome_pressure,
            n_layers=layer_count,
            n_experts=expert_count,
            hidden=hidden,
        )
    )
    stop_delta = stop_delta + none_pathway.stop_delta.to(stop_delta)
    slice_delta = slice_delta + none_pathway.slice_delta.to(slice_delta)
    signal_type = self._resynthesis_hop_signal_type
    return signal_type(
        domain_logits=base,
        layer_bias=layer_bias,
        expert_bias=expert_bias,
        stop_delta=stop_delta,
        slice_delta=slice_delta,
        hop_bonus=bonus,
        hop_confidence=confidence,
        none_pathway=none_pathway,
        from_hop_idx=from_index_t.reshape(()),
        dest_hop_idx=destination.reshape(()),
    )


def _historical_plan_layer_traversal_with_arm_exhaustion(
    self: Any,
    step_idx: int,
    domain_idx: int,
    subdomain_idx: int,
    *,
    last_rubric: float | None,
    last_layer: int | None,
    device: torch.device,
    hidden: torch.Tensor | None = None,
    intent_probs: torch.Tensor | None = None,
    extra_pressure: torch.Tensor | None = None,
    transfer_weights: torch.Tensor | None = None,
    stage_layer_bias: torch.Tensor | None = None,
) -> Any:
    """Traverse the historical learned layer frontier exactly once per arm.

    The July 13 planner documented layer-arm exhaustion as a terminal model
    authority, but its low-rubric branch repeatedly selected ``last_layer``
    without consuming another arm.  Its local ``visited_layers`` set therefore
    remained at one element and the function itself could never return for a
    low-confidence prompt.  A wrapper around its return value was necessarily
    too late to repair that state machine.

    This compatibility owner retains all eight trained routing signals from
    the historical planner and masks only layer arms already consumed in the
    current graph frontier.  The topology-derived frontier has exactly one arm
    per attached expert layer; exhausting those model-owned arms is not a host
    hop, token, step, or time cap.  The final list conversion is the explicit
    legacy boundary required by the immutable parent's recursive-pass API.
    """

    original = getattr(type(self), "_resynthesis_original_plan_layer_traversal", None)
    if not callable(original):
        raise RuntimeError("historical Resynthesis traversal owner was not preserved")
    exhaustion = getattr(self, "_resynthesis_recursive_arm_exhausted", None)
    layer_pick_logits = getattr(self, "layer_pick_logits", None)
    experts = getattr(self, "experts", None)
    if not isinstance(exhaustion, torch.Tensor) or not isinstance(
        layer_pick_logits, torch.Tensor
    ):
        raise RuntimeError("historical Resynthesis arm-exhaustion tensor is unavailable")
    if experts is None:
        raise RuntimeError("historical Resynthesis traversal expert frontier is unavailable")

    layer_count = len(experts)
    if layer_count <= 0:
        exhaustion.zero_()
        return [], []

    rc = self.rbo_cfg
    destination_pressure = self.arm_registry.domain_slot_pressure().sum(dim=-1)
    if extra_pressure is not None:
        pressure = extra_pressure.to(
            device=destination_pressure.device,
            dtype=destination_pressure.dtype,
        )
        if pressure.numel() == destination_pressure.numel():
            destination_pressure = destination_pressure + pressure

    # Keep the legacy route triples and the learned terminal decision in one
    # contiguous packet.  The immutable parent requires Python route tuples at
    # its recursive-pass boundary, so transferring this packet once avoids a
    # second CUDA synchronization for the stop reason without changing either
    # model-owned decision.
    legacy_boundary_t = torch.empty(
        (layer_count * 3 + 1,),
        device=device,
        dtype=torch.long,
    )
    routes = legacy_boundary_t[:-1].view(layer_count, 3)
    visited = torch.zeros(layer_count, device=device, dtype=torch.bool)
    frontier_scores = layer_pick_logits.new_full(
        (layer_count,),
        -torch.inf,
        device=device,
    )
    hop_signals: list[Any] = []
    active_domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=layer_pick_logits,
        upper_bound=int(self.arm_layout.n_domains),
    ).to(device=device)
    active_subdomain_t = _tensor_hotpath_index_boundary(
        subdomain_idx,
        reference=layer_pick_logits,
        upper_bound=int(self.arm_layout.n_subdomains),
    ).to(device=device)
    struggling = last_rubric is not None and last_rubric < rc.anti_floor
    final_stop_delta = layer_pick_logits.new_zeros(())
    final_slice_delta = layer_pick_logits.new_zeros(())
    # The content, intent, transfer packet, expert frontier, and learned layer
    # identities are immutable for this planner invocation.  The historical
    # implementation rebuilt the same profile/identity query once for every
    # arm in the frontier (21 times in the native parent).  Reuse that exact
    # model-owned tensor while still evaluating every hop, domain pressure,
    # arm posterior, and layer score independently below.  This changes no
    # route, parameter, expert participation, or stopping authority.
    base = torch.sigmoid(layer_pick_logits[:layer_count].to(device=device))
    base_score = torch.log(base.clamp(min=1e-4))
    identity_layer = self._layer_identity_bias(
        hidden,
        intent_probs,
        transfer_weights,
        layer_count,
    ).to(device=device, dtype=base.dtype)
    stage_route_scale = (
        2.0
        * torch.sigmoid(self.swe_stage_route_logit).to(
            device=device,
            dtype=base.dtype,
        )
        if stage_layer_bias is not None
        else None
    )
    stage_layer_score = (
        stage_route_scale
        * stage_layer_bias[:layer_count].to(
            device=device,
            dtype=base.dtype,
        )
        if stage_layer_bias is not None and stage_route_scale is not None
        else None
    )
    layer_slots_t = torch.arange(
        layer_count,
        device=device,
        dtype=torch.long,
    )
    # Subdomain slot geometry is fixed for the complete learned frontier.
    # Only ``active_domain_t`` changes between hops. Rebuilding this identical
    # axis inside every hop added one allocation/dispatch (and one CUDA kernel
    # launch on the live parent) per traversed layer without influencing the
    # anti-Thompson posterior, pressure, route, or stop decision.
    subdomain_slots_t = (
        torch.arange(
            self.arm_layout.n_subdomains,
            device=device,
            dtype=torch.long,
        )
        if layer_count > 1 and last_rubric is not None and not struggling
        else None
    )
    domain_layer_affinity = self.domain_layer_affinity
    if (
        not isinstance(domain_layer_affinity, torch.Tensor)
        or domain_layer_affinity.dim() != 2
        or domain_layer_affinity.shape[0] < self.arm_layout.n_domains
        or domain_layer_affinity.shape[1] < layer_count
    ):
        raise RuntimeError("historical Resynthesis domain-layer affinity differs")
    # Hidden content and task intent do not change while the planner consumes
    # its model-owned layer frontier.  The historical hop owner previously
    # repeated the same full-context attention pool and the same two learned
    # projections once per arm.  Prepare that tensor packet once under the
    # historical planner's unchanged no-gradient authority and reuse it; each
    # arm still computes its own hop destination,
    # domain pressure, transfer pathway, posterior, layer score, and stop
    # evidence below.
    prepared_hop_context: _HistoricalHopContextPacket | None = None
    if hidden is not None and intent_probs is not None:
        with torch.no_grad():
            pooled = _parent_hidden_context(hidden).to(
                dtype=self.cross_domain_router.context_query.weight.dtype
            )
            intent_mean = intent_probs.mean(dim=1).to(dtype=pooled.dtype)
            fused = torch.cat((pooled, intent_mean), dim=-1)
            if fused.dim() == 1:
                fused = fused.unsqueeze(0)
            prepared_hop_context = _HistoricalHopContextPacket(
                query_t=self.cross_domain_router.context_query(fused).squeeze(0),
                gate_t=torch.sigmoid(
                    self.cross_domain_router.context_gate(fused)
                ).squeeze(0),
            )

    for hop_idx in range(layer_count):
        with torch.no_grad():
            hop_signal = self.cross_domain_router.compute_hop_signal(
                active_domain_t,
                active_subdomain_t,
                hidden=hidden,
                intent_probs=intent_probs,
                outcome_pressure=destination_pressure,
                hop_idx=hop_idx,
                step_idx=step_idx,
                n_layers=layer_count,
                n_experts=self.cfg.total_experts,
                prepared_context=prepared_hop_context,
            )
        hop_signals.append(hop_signal)
        self._last_hop_signal = hop_signal
        final_stop_delta = hop_signal.stop_delta.detach().reshape(())
        final_slice_delta = hop_signal.slice_delta.detach().reshape(())
        self._hop_stop_delta_accum = (
            self._hop_stop_delta_accum.to(
                device=final_stop_delta.device,
                dtype=final_stop_delta.dtype,
            )
            + final_stop_delta
        )
        self._hop_slice_delta_accum = (
            self._hop_slice_delta_accum.to(
                device=final_slice_delta.device,
                dtype=final_slice_delta.dtype,
            )
            + final_slice_delta
        )

        if hop_idx > 0 and last_rubric is not None and not struggling:
            if subdomain_slots_t is None:
                raise RuntimeError(
                    "historical Resynthesis subdomain slot axis is unavailable"
                )
            active_domain_t = _tensor_hotpath_index_boundary(
                hop_signal.domain_logits.detach().argmax(),
                reference=layer_pick_logits,
                upper_bound=int(self.arm_layout.n_domains),
            )
            subdomain_arm_ids_t = (
                active_domain_t * self.arm_layout.n_subdomains
                + subdomain_slots_t
            )
            subdomain_posterior_t = self.arm_registry.posterior_mean(
                subdomain_arm_ids_t
            )
            subdomain_memory_t = (
                0.5 - self.arm_registry.success_ema[subdomain_arm_ids_t]
            ).clamp_min(0.0)
            subdomain_pressure_t = (
                self.arm_registry.anti_fails[subdomain_arm_ids_t]
                + subdomain_memory_t
                + (0.5 - subdomain_posterior_t).clamp_min(0.0)
            )
            best_subdomain_t = subdomain_pressure_t.argmax().reshape(())
            if self.arm_layout.n_subdomains > 1:
                best_subdomain_t = torch.where(
                    best_subdomain_t.eq(active_subdomain_t),
                    (active_subdomain_t + 1).remainder(
                        self.arm_layout.n_subdomains
                    ),
                    best_subdomain_t,
                )
            active_subdomain_t = best_subdomain_t

        domain_subdomain_t = (
            active_domain_t * self.arm_layout.n_subdomains
            + active_subdomain_t
        )
        layer_arms = (
            self.arm_layout.n_domain_subdomain
            + (
                (
                    domain_subdomain_t * self.arm_layout.n_layers
                    + layer_slots_t
                )
                * self.arm_layout.step_slots
                + (int(step_idx) % self.arm_layout.step_slots)
            )
            * self.arm_layout.hop_slots
            + (hop_idx % self.arm_layout.hop_slots)
        )
        anti = self.arm_registry.suppression(layer_arms)
        explore = self.arm_registry.explore_boost(
            layer_arms,
            weight=rc.thompson_explore_weight,
            sample=not self.training,
        )
        memory = self.arm_registry.memory_log_bias(layer_arms) * 0.15
        domain_bias = torch.sigmoid(
            domain_layer_affinity.index_select(
                0,
                active_domain_t.reshape(1),
            ).squeeze(0)[:layer_count]
        )
        hop_layer = hop_signal.layer_bias[:layer_count].to(
            device=device,
            dtype=base.dtype,
        )
        scores = (
            base_score
            + torch.log(domain_bias.to(device=device).clamp(min=1e-4))
            + 0.35 * torch.log(anti.to(device=device).clamp(min=1e-4))
            + 0.25 * torch.log(explore.to(device=device).clamp(min=1e-4))
            + memory.to(device=device)
            + 0.3 * hop_layer
            + rc.identity_layer_strength * identity_layer
        )
        if stage_layer_bias is not None:
            if stage_layer_score is None:
                raise RuntimeError("historical Resynthesis stage-route scale is unavailable")
            scores = scores + stage_layer_score.to(
                device=scores.device,
                dtype=scores.dtype,
            )

        selected_layer = scores.masked_fill(visited, -torch.inf).argmax().reshape(())
        selected_score = scores.gather(0, selected_layer.reshape(1)).reshape(())
        frontier_scores.scatter_(
            0,
            selected_layer.reshape(1),
            selected_score.reshape(1),
        )
        routes[hop_idx] = torch.stack(
            (
                selected_layer,
                active_domain_t,
                active_subdomain_t,
            )
        )
        visited.scatter_(0, selected_layer.reshape(1), True)

    exhausted = visited.all()
    exhaustion.copy_(exhausted.to(device=exhaustion.device, dtype=torch.bool))
    torch._assert_async(
        torch.isfinite(frontier_scores).all(),
        "historical Resynthesis frontier blend has an unscored layer",
    )
    # Every model-selected layer arm executes, but its residual is blended by
    # the same learned route evidence that selected its place in the frontier.
    # Normalizing the complete frontier prevents 21 independently trained
    # expert residuals from being treated as 21 full-strength sequential model
    # replacements.  This is model-owned MoE aggregation, not a host cap: every
    # finite arm retains non-zero contribution and the weights sum to one.
    # The historical checkpoint runs this control surface in bfloat16.  A
    # bfloat16 softmax followed by a bfloat16 renormalization can leave the
    # represented 21-arm mass roughly 1e-3 away from one after promotion back
    # to float32.  Keep the learned scores and all selected arms unchanged,
    # but perform the aggregation in float32 so the model-owned mixture has a
    # numerically faithful unit-mass contract before each scalar is cast at the
    # legacy expert-residual boundary.
    frontier_blend = torch.softmax(
        frontier_scores.to(dtype=torch.float32),
        dim=0,
    )
    # A finite learned score can still underflow to exactly zero after
    # exponentiation.  Arm exhaustion promises that every finite model-owned
    # layer contributes, so retain the dtype's smallest representable positive
    # mass and renormalize.  This does not select, skip, or cap an arm.
    frontier_blend = frontier_blend.clamp_min(torch.finfo(torch.float32).tiny)
    frontier_blend = frontier_blend / frontier_blend.sum()
    setattr(
        self,
        "_resynthesis_recursive_frontier_blend",
        frontier_blend.detach(),
    )
    learned_cycle_stop = torch.sigmoid(final_stop_delta).ge(0.5) | final_slice_delta.gt(
        0.5
    )
    legacy_boundary_t[-1].copy_(learned_cycle_stop)
    legacy_boundary = legacy_boundary_t.detach().to(device="cpu").tolist()
    self._last_traversal_stop_reason = (
        "trained_gate" if bool(legacy_boundary[-1]) else "arm_exhaustion"
    )
    plan = [
        (legacy_boundary[offset], legacy_boundary[offset + 1], legacy_boundary[offset + 2])
        for offset in range(0, layer_count * 3, 3)
    ]
    return plan, hop_signals


def _historical_layer_gate_scale_with_frontier_blend(
    self: Any,
    layer_idx: int,
) -> torch.Tensor:
    """Blend every traversed layer by its trained frontier route evidence."""

    original = getattr(type(self), "_resynthesis_original_layer_gate_scale", None)
    if not callable(original):
        raise RuntimeError("historical Resynthesis layer gate owner was not preserved")
    base_scale = original(self, layer_idx)
    frontier_blend = getattr(self, "_resynthesis_recursive_frontier_blend", None)
    experts = getattr(self, "experts", None)
    if not isinstance(base_scale, torch.Tensor) or not isinstance(
        frontier_blend,
        torch.Tensor,
    ):
        raise RuntimeError("historical Resynthesis frontier blend tensor is unavailable")
    if experts is None or frontier_blend.shape != (len(experts),):
        raise RuntimeError("historical Resynthesis frontier blend geometry differs")
    blend = frontier_blend[layer_idx].to(
        device=base_scale.device,
        dtype=base_scale.dtype,
    )
    return base_scale * blend


def _historical_joint_stop_logit_with_arm_exhaustion(
    self: Any,
    signals: Any,
) -> torch.Tensor:
    """Fuse graph-frontier exhaustion into the parent's existing stop logit.

    A complete frontier is an independent native stop authority in the parent
    contract.  Raising the existing logit above its unchanged 0.5 decision
    boundary lets the historical outer loop consume that authority.  Its
    original low-rubric-and-improving exception remains intact, so measurable
    correction progress may still continue without any host-derived cap.
    """

    original = getattr(type(self), "_resynthesis_original_joint_stop_logit", None)
    if not callable(original):
        raise RuntimeError("historical Resynthesis stop-logit owner was not preserved")
    joint_logit = original(self, signals)
    exhaustion = getattr(self, "_resynthesis_recursive_arm_exhausted", None)
    if not isinstance(joint_logit, torch.Tensor) or not isinstance(
        exhaustion, torch.Tensor
    ):
        raise RuntimeError("historical Resynthesis arm-exhaustion stop tensor is unavailable")
    exhausted_stop_logit = torch.maximum(
        joint_logit,
        torch.ones_like(joint_logit),
    )
    return torch.where(
        exhaustion.to(device=joint_logit.device, dtype=torch.bool),
        exhausted_stop_logit,
        joint_logit,
    )


def _historical_batch_safe_correction_training_loss(self: Any) -> torch.Tensor:
    """Align each historical policy target with its row-local policy logits.

    The inherited correction controller intentionally derives one aggregate
    mode target per recursive step. Its loss path concatenated row-local logits
    as ``[steps * batch, modes]`` but retained only ``steps`` scalar targets.
    Expand each aggregate step target across that step's rows so every row
    contributes gradient without mixing or selecting a representative row.
    """

    original = getattr(
        type(self),
        "_resynthesis_original_correction_training_loss",
        None,
    )
    if not callable(original):
        raise RuntimeError("historical Resynthesis correction-loss owner was not preserved")
    mode_logits = getattr(self, "_last_policy_mode_logits", None)
    mode_targets = getattr(self, "_last_policy_mode_targets", None)
    if not isinstance(mode_logits, list) or not isinstance(mode_targets, list):
        raise RuntimeError("historical Resynthesis correction-policy history differs")
    if not mode_logits and not mode_targets:
        return cast(torch.Tensor, original(self))
    if len(mode_logits) != len(mode_targets):
        raise RuntimeError("historical correction-policy history lengths differ")

    expanded_targets: list[torch.Tensor] = []
    expansion_required = False
    for logits_t, target_t in zip(mode_logits, mode_targets, strict=True):
        if not isinstance(logits_t, torch.Tensor) or not isinstance(
            target_t,
            torch.Tensor,
        ):
            raise RuntimeError("historical correction-policy history is not tensor-owned")
        if logits_t.dim() != 2 or logits_t.shape[0] < 1:
            raise RuntimeError("historical correction-policy logits geometry differs")
        flat_target = target_t.reshape(-1)
        if flat_target.numel() == 1:
            row_targets = flat_target.expand(logits_t.shape[0])
            expansion_required = expansion_required or logits_t.shape[0] > 1
        elif flat_target.numel() == logits_t.shape[0]:
            row_targets = flat_target
        else:
            raise RuntimeError("historical correction-policy target geometry differs")
        expanded_targets.extend(row_targets.unbind(0))
    if not expansion_required:
        return cast(torch.Tensor, original(self))

    setattr(self, "_last_policy_mode_targets", expanded_targets)
    try:
        return cast(torch.Tensor, original(self))
    finally:
        setattr(self, "_last_policy_mode_targets", mode_targets)


def _tensor_native_self_correction_mode_target(
    signals: Mapping[str, torch.Tensor],
    *,
    verification_deficit: torch.Tensor,
    domain_collapse: torch.Tensor,
    rubric_floor: float,
) -> torch.Tensor:
    """Preserve the historical mode target without a per-hop H2D scalar copy."""

    rubric = signals["rubric"]
    # The source-era implementation used ``rubric.new_tensor(Python_float)``.
    # On CUDA that enters the host-to-device copy path at every recursive hop.
    # Build the same scalar from a device-native factory and pass the unchanged
    # configuration value as a scalar kernel argument instead.
    floor = rubric.new_empty(()).fill_(rubric_floor)
    zero = rubric.new_zeros(())
    route_pressure = torch.maximum(
        domain_collapse.reshape(()),
        signals.get("history_pressure", zero).reshape(()),
    )
    hidden_pressure = torch.maximum(
        (floor - rubric).clamp(min=0.0).reshape(()),
        signals.get("stalled", zero).reshape(()),
    )
    verify_pressure = verification_deficit.reshape(())
    transfer_pressure = signals.get("tool_pressure", zero).reshape(())
    execution_pressure = signals.get("execution_pressure", zero).reshape(())
    grounding_confidence = signals.get("grounding_confidence", zero).reshape(())
    growth_pressure = torch.maximum(
        signals.get("history_pressure", zero).reshape(()),
        (signals["contradiction"] - (torch.ones_like(floor) - floor))
        .clamp(min=0.0)
        .reshape(()),
    )
    verify_pressure = torch.maximum(verify_pressure, execution_pressure)
    transfer_pressure = torch.maximum(
        transfer_pressure,
        execution_pressure * grounding_confidence,
    )
    growth_pressure = torch.maximum(growth_pressure, execution_pressure)
    return torch.stack(
        (
            zero,
            route_pressure,
            hidden_pressure,
            verify_pressure,
            transfer_pressure,
            growth_pressure,
        )
    ).argmax(dim=0)


def _tensor_native_domain_collapse_forward(
    self: Any,
    domain_probs: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
    """Preserve collapse correction without a per-forward host scalar copy."""

    entropy = -(
        domain_probs * (domain_probs + 1e-8).log()
    ).sum(dim=-1)
    maximum_entropy = (
        domain_probs.new_ones(())
        .mul_(self.n_domains)
        .log_()
    )
    normalized_entropy = entropy / maximum_entropy
    threshold = torch.sigmoid(self.entropy_threshold)
    collapse_score = torch.sigmoid(
        (threshold - normalized_entropy) * 8.0
    )
    strength = torch.sigmoid(self.redistribution_strength)
    uniform = torch.ones_like(domain_probs) / self.n_domains
    blend = strength * collapse_score.unsqueeze(-1)
    adjusted = (1 - blend) * domain_probs + blend * uniform
    return adjusted, collapse_score.mean()


def _tensor_native_rubric_apply_bias(
    self: Any,
    logits: torch.Tensor,
    hidden: torch.Tensor,
    intent_probs: torch.Tensor,
    strength: float = 1.0,
) -> torch.Tensor:
    """Scatter historical rubric activations in the logits' exact dtype.

    The frozen parent may expose BF16 hidden/rubric activations while its
    logits surface is FP32. ``scatter_add_`` requires destination and source
    dtypes to match, so the source-era method failed before the first r152
    wave. The cast changes only representation at this additive logit
    boundary; token IDs, learned rubric weights, and routing remain exact.
    """

    _, weighted = self.forward(hidden, intent_probs)
    if not isinstance(weighted, torch.Tensor) or weighted.ndim != 4:
        raise RuntimeError("historical rubric activation geometry differs")
    if logits.ndim != 3 or weighted.shape[:2] != logits.shape[:2]:
        raise RuntimeError("historical rubric logits geometry differs")
    bias = torch.zeros_like(logits)
    active_mask = cast(torch.Tensor, self.rubric_mask).to(
        device=weighted.device,
        dtype=torch.bool,
    )
    rubric_token_ids = cast(torch.Tensor, self.rubric_token_ids).to(
        device=weighted.device,
        dtype=torch.long,
    )
    # The rubric registry has a fixed ``[intent, term]`` geometry. Boolean
    # indexing that registry asks CUDA ``nonzero`` to size a dynamic result,
    # synchronizing the parent stream during every frozen prefill. Scatter the
    # same fixed slots and make unregistered contributions exactly zero
    # instead: inactive token IDs are zero-initialized by the historical
    # checkpoint, and zero-valued duplicate scatters cannot alter token zero.
    # ``where`` also prevents an inactive non-finite projection from leaking
    # through multiplication by zero.
    flat_token_ids = rubric_token_ids.reshape(-1)
    weighted_flat = weighted.flatten(start_dim=2)
    weighted_active = torch.where(
        active_mask.reshape(1, 1, -1),
        weighted_flat * strength,
        torch.zeros_like(weighted_flat),
    ).to(
        device=bias.device,
        dtype=bias.dtype,
    )
    index_t = (
        flat_token_ids.to(device=bias.device, dtype=torch.long)
        .view(1, 1, -1)
        .expand(logits.shape[0], logits.shape[1], -1)
    )
    bias.scatter_add_(2, index_t, weighted_active)
    return logits + bias


def _tensor_native_historical_domain_route(
    adjusted_domain_probabilities: torch.Tensor,
    subdomain_indices: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
    """Keep the historical model-selected route on its active device.

    The source-era parent selected the same two argmax values and immediately
    called ``Tensor.item`` on each.  That serialized the complete CUDA stream
    before the recursive graph could enqueue any work.  The downstream
    Resynthesis compatibility owners already accept scalar tensor indices, so
    retain the two exact model decisions as tensors through that graph.
    """

    if (
        adjusted_domain_probabilities.dim() != 2
        or adjusted_domain_probabilities.shape[0] < 1
        or adjusted_domain_probabilities.shape[1] < 1
        or subdomain_indices.dim() != 1
        or subdomain_indices.shape[0] < 1
    ):
        raise RuntimeError("historical Resynthesis domain route geometry differs")
    return (
        # Only the first route row is consumed by the historical recursive
        # owner below.  Reducing every batch row and then discarding all but
        # row zero needlessly launches a full-batch CUDA reduction.
        adjusted_domain_probabilities[0].argmax()
        .to(
            device=adjusted_domain_probabilities.device,
            dtype=torch.long,
        )
        .reshape(()),
        subdomain_indices[0]
        .to(
            device=adjusted_domain_probabilities.device,
            dtype=torch.long,
        )
        .reshape(()),
    )


def _tensor_preserving_historical_int(
    value: Any,
    *args: Any,
) -> int | torch.Tensor:
    """Preserve scalar tensors only inside the cloned historical recursion.

    Python's source-era ``_recursive_pass`` begins with ``int(domain_idx)`` and
    ``int(subdomain_idx)``.  Its installed traversal owner is tensor-native,
    while every genuine Python integer in that function must retain normal
    ``int`` behavior.  A private globals table for that one function lets the
    two scalar tensors survive without changing process-wide builtins or the
    immutable parent module.
    """

    if isinstance(value, torch.Tensor):
        if args or value.numel() != 1:
            raise RuntimeError("historical recursive tensor index is malformed")
        return value.reshape(())
    return int(value, *args)


def _tensor_preserving_historical_float(
    value: Any = 0.0,
) -> float | torch.Tensor:
    """Keep frozen-parent diagnostic scalars on their current device.

    The source-era recursive pass converts every diagnostic signal to
    ``float`` before appending it to a Python history list. On CUDA each
    conversion executes ``Tensor.item()``, synchronizing the complete stream.
    The frozen parent does not train from those scalar histories; its active
    output is the routed hidden/logit tensor and its compact proof deliberately
    omits the boundary-only histories. Preserve scalar tensors in that exact
    path while retaining normal Python ``float`` behavior for configuration
    constants and for the trainable historical path.
    """

    if isinstance(value, torch.Tensor):
        if value.numel() != 1:
            raise RuntimeError("historical recursive float tensor is malformed")
        return value.reshape(())
    return float(value)


class _HistoricalRecursiveTorchFacade:
    """Keep the frozen recursive pass from staging one scalar per layer.

    The verified source-era function has one ``torch.tensor`` call.  It wraps
    the already-known Python layer-arm identifier in a one-element CUDA tensor
    after every model-selected expert.  Constructing that tensor through the
    generic host-data factory enters the synchronous H2D copy path.  The arm
    identifier is a kernel scalar, so an empty device tensor followed by
    ``fill_`` has identical values without a host buffer or stream fence.

    All other torch attributes are delegated unchanged.  This facade is
    installed only in the private globals table of the cloned historical
    recursive function; process-wide torch behavior is never modified.
    """

    def __getattr__(self, name: str) -> Any:
        return getattr(torch, name)

    @staticmethod
    def tensor(
        data: object,
        *,
        device: torch.device,
        dtype: torch.dtype,
    ) -> torch.Tensor:
        if (
            not isinstance(data, (list, tuple))
            or len(data) != 1
            or not isinstance(data[0], int)
            or isinstance(data[0], bool)
            or dtype != torch.long
        ):
            raise RuntimeError(
                "historical recursive device-scalar construction differs"
            )
        return torch.empty((1,), device=device, dtype=dtype).fill_(data[0])


def _historical_recursive_pass_with_tensor_route(
    original: Callable[..., Any],
) -> Callable[..., Any]:
    """Clone one verified historical function with tensor-preserving scalars."""

    if not isinstance(original, FunctionType):
        raise RuntimeError("historical Resynthesis recursive pass is not a Python function")
    recursive_globals = dict(original.__globals__)
    recursive_globals["int"] = _tensor_preserving_historical_int
    recursive_globals["torch"] = _HistoricalRecursiveTorchFacade()
    tensor_route = FunctionType(
        original.__code__,
        recursive_globals,
        original.__name__,
        original.__defaults__,
        original.__closure__,
    )
    frozen_globals = dict(recursive_globals)
    frozen_globals["float"] = _tensor_preserving_historical_float
    frozen_route = FunctionType(
        original.__code__,
        frozen_globals,
        original.__name__,
        original.__defaults__,
        original.__closure__,
    )

    def recursive_pass(
        self: Any,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        implementation = (
            frozen_route
            if not self.training and not torch.is_grad_enabled()
            else tensor_route
        )
        return implementation(self, *args, **kwargs)

    recursive_pass.__kwdefaults__ = (
        dict(original.__kwdefaults__)
        if original.__kwdefaults__ is not None
        else {}
    )
    recursive_pass.__annotations__ = dict(original.__annotations__)
    recursive_pass.__qualname__ = original.__qualname__
    recursive_pass.__doc__ = original.__doc__
    return cast(Callable[..., Any], recursive_pass)


def _tensor_native_outcome_history_pack_step(
    self: Any,
    signals: Mapping[str, torch.Tensor],
    domain_idx: int | torch.Tensor,
    subdomain_idx: int | torch.Tensor,
    stop_logit: torch.Tensor,
    step_idx: int,
    step_slots: int,
    trace_summary: torch.Tensor,
) -> torch.Tensor:
    """Pack routed history without scalar indexing through the CPU."""

    device = self.carry_h.device
    dtype = self.domain_head.weight.dtype
    utility = signals["utility"].reshape(()).to(device=device, dtype=dtype)
    contradiction = signals["contradiction"].reshape(()).to(
        device=device,
        dtype=dtype,
    )
    rubric = signals["rubric"].reshape(()).to(device=device, dtype=dtype)
    intent = signals["intent_conf"].reshape(()).to(device=device, dtype=dtype)
    stop = stop_logit.reshape(()).to(device=device, dtype=dtype)
    domain_t = _tensor_hotpath_index_boundary(
        domain_idx,
        reference=self.carry_h,
        upper_bound=int(self.n_domains),
    ).to(device=device)
    subdomain_t = _tensor_hotpath_index_boundary(
        subdomain_idx,
        reference=self.carry_h,
        upper_bound=int(self.n_subdomains),
    ).to(device=device)
    domain_one_hot = torch.zeros(
        self.n_domains,
        device=device,
        dtype=dtype,
    ).scatter(
        0,
        domain_t.reshape(1),
        torch.ones(1, device=device, dtype=dtype),
    )
    subdomain_one_hot = torch.zeros(
        self.n_subdomains,
        device=device,
        dtype=dtype,
    ).scatter(
        0,
        subdomain_t.reshape(1),
        torch.ones(1, device=device, dtype=dtype),
    )
    step_fraction = stop.new_full(
        (),
        float(step_idx + 1) / float(max(1, step_slots)),
    )
    trace = trace_summary.to(
        device=device,
        dtype=dtype,
    ).reshape(self.TRACE_FEAT_DIM)
    return torch.cat(
        (
            utility.view(1),
            contradiction.view(1),
            rubric.view(1),
            intent.view(1),
            stop.view(1),
            domain_one_hot,
            subdomain_one_hot,
            step_fraction.view(1),
            trace,
        ),
        dim=0,
    )


def _tensor_native_historical_rbo_forward(
    self: Any,
    hidden: torch.Tensor,
    baseline_logits: torch.Tensor | None = None,
    intent_label: str | None = None,
    swe_stage_target: torch.Tensor | None = None,
    swe_stage_hidden: torch.Tensor | None = None,
    project_logits: bool = True,
) -> Any:
    """Run the source-era parent while retaining model routes as tensors."""

    self._hop_stop_delta_accum = hidden.new_zeros(())
    self._hop_slice_delta_accum = hidden.new_zeros(())
    control_hidden = hidden if swe_stage_hidden is None else swe_stage_hidden
    intent_probs = self._intent_probs(control_hidden)
    (
        swe_stage_logits,
        swe_stage_probs,
        stage_domain_bias,
        stage_subdomain_bias,
    ) = self.swe_stage_router(control_hidden)
    stage_layer_bias = (
        swe_stage_probs.mean(dim=0)
        @ self.swe_stage_layer_affinity.to(
            device=swe_stage_probs.device,
            dtype=swe_stage_probs.dtype,
        )
    )
    intent_idx_t = self._intent_index(intent_probs)
    if intent_label is None:
        intent_label = (
            "model_owned_tensor"
            if self.training
            else self._intent_label(intent_probs, self.intent_names)
        )
    else:
        intent_boundary_idx = (
            self.intent_names.index(str(intent_label))
            if str(intent_label) in self.intent_names
            else self.intent_names.index("general")
            if "general" in self.intent_names
            else 0
        )
        intent_idx_t = intent_idx_t.new_empty((), dtype=torch.long).fill_(
            intent_boundary_idx
        )
    hist_d, hist_s, _ = self.outcome_history.current_biases()
    ctx_signals = self.context_encoder(control_hidden)
    domain_conf_bias = ctx_signals["domain_conf"][0]
    domain_count = self.arm_layout.n_domains
    domain_confidence = domain_conf_bias[:domain_count]
    domain_confidence = domain_confidence / (
        domain_confidence.sum() + 1e-8
    )
    domain_log_boost = torch.log(domain_confidence.clamp(min=1e-6))

    domain_probabilities, _, _, subdomain_indices = self.domain_router(
        control_hidden,
        intent_probs,
        domain_log_bias=(
            self.correction_domain_bias
            + hist_d.detach()
            + 0.1 * domain_log_boost
            + 0.2 * stage_domain_bias
        ),
        subdomain_log_bias=(
            self.correction_subdomain_bias
            + hist_s.detach()
            + 0.2 * stage_subdomain_bias
        ),
    )
    (
        adjusted_domain_probabilities,
        domain_collapse_score,
    ) = self.domain_collapse_detector(domain_probabilities)
    self._last_domain_collapse_tensor = domain_collapse_score.detach()
    domain_idx, subdomain_idx = _tensor_native_historical_domain_route(
        adjusted_domain_probabilities,
        subdomain_indices,
    )
    (
        shaped_hidden,
        routing_history,
        feedback_history,
        stop_logit_history,
        traversal_history,
        arms_used,
        self_correction_steps,
    ) = self._recursive_pass(
        hidden,
        intent_probs,
        domain_idx,
        subdomain_idx,
        ctx_signals,
        stage_layer_bias,
        control_hidden,
    )
    # The scalar history list is external receipt telemetry. Keep the current
    # model value tensor-resident in the active graph; boundary snapshots can
    # materialize it after the forward rather than synchronizing every wave.
    self._last_domain_collapse_history = []
    frozen_parent_hot_path = not self.training and not torch.is_grad_enabled()
    if frozen_parent_hot_path:
        # These four scalars supervise the historical parent only. Resynthesis
        # freezes that parent and consumes shaped hidden/logit tensors, so
        # rebuilding their BCE/CE stacks per retained wave cannot influence a
        # routed-page gradient. Keep stateful history/context replay below.
        dead_training_loss = hidden.new_zeros((), dtype=torch.float32)
        contra_loss = dead_training_loss
        stop_loss = dead_training_loss
        correction_loss = dead_training_loss
        stage_training_loss = dead_training_loss
    else:
        rubric_tensor = (
            torch.stack([feedback["rubric"] for feedback in feedback_history])
            if feedback_history
            else hidden.new_zeros(0)
        )
        contrareactive_repulsion = getattr(
            type(self),
            "_resynthesis_contrareactive_repulsion",
            None,
        )
        if not callable(contrareactive_repulsion):
            raise RuntimeError(
                "historical Resynthesis contrareactive owner is unavailable"
            )
        contra_loss = contrareactive_repulsion(
            rubric_tensor,
            floor=self.rbo_cfg.anti_floor,
            weight=self.rbo_cfg.anti_weight,
        )
        stop_loss = self.stop_gate_training_loss()
        correction_loss = self._correction_training_loss()
        stage_training_loss = self._swe_stage_training_loss(
            swe_stage_logits,
            swe_stage_probs,
            swe_stage_target,
        )
    history_loss = self.outcome_history_training_loss(
        feedback_history,
        stop_logit_history,
        domain_idx=domain_idx,
        subdomain_idx=subdomain_idx,
    )
    context_loss = (
        dead_training_loss
        if frozen_parent_hot_path
        else self._context_dag_training_loss(
            hidden,
            intent_probs,
            ctx_signals,
            intent_idx_t,
            feedback_history,
            domain_idx=domain_idx,
            subdomain_idx=subdomain_idx,
            stage_domain_bias=stage_domain_bias,
            stage_subdomain_bias=stage_subdomain_bias,
        )
    )
    steps = len(feedback_history)
    traversal_stop_reason = getattr(
        self,
        "_last_traversal_stop_reason",
        "trained_gate",
    )
    shaped_logits = (
        self.shaped_logits_from_hidden(
            shaped_hidden,
            baseline_logits,
            hidden,
        )
        if project_logits
        else None
    )
    if not self.training and shaped_logits is not None:
        shaped_logits = self.rubric_bias.apply_bias(
            shaped_logits,
            shaped_hidden.detach(),
            intent_probs.detach(),
            strength=float(self.rbo_cfg.rubric_bias_strength),
        )
    self._detach_outcome_history_carry()
    result_type = getattr(type(self), "_resynthesis_rbo_result_type", None)
    stages = getattr(type(self), "_resynthesis_swe_stages", None)
    if not callable(result_type) or not isinstance(stages, tuple):
        raise RuntimeError("historical Resynthesis result boundary is unavailable")
    stage_index_boundary = (
        0
        if frozen_parent_hot_path
        else int(
            swe_stage_probs.mean(dim=0)
            .argmax()
            .detach()
            .to(device="cpu", dtype=torch.long)
        )
    )
    return result_type(
        shaped_hidden=shaped_hidden,
        baseline_hidden=hidden,
        shaped_logits=shaped_logits,
        baseline_logits=baseline_logits,
        steps=steps,
        stop_reason=traversal_stop_reason,
        intent_probs=intent_probs,
        intent_label=intent_label,
        swe_stage_probs=swe_stage_probs,
        swe_stage_training_loss=stage_training_loss,
        routing_history=routing_history,
        feedback_history=feedback_history,
        stop_logit_history=stop_logit_history,
        traversal_history=traversal_history,
        arms_used=arms_used,
        domain_idx=domain_idx,
        subdomain_idx=subdomain_idx,
        self_correction_steps=self_correction_steps,
        contrareactive_loss=contra_loss,
        correction_loss=correction_loss,
        stop_gate_loss=stop_loss,
        outcome_history_loss=history_loss,
        context_training_loss=context_loss,
        proof=(
            {
                "resynthesisRBOActive": True,
                "none_architecture": "Nest of Native Experts",
                "none_model_owned": True,
                "none_transfer_active": self._last_none_phase is not None,
            }
            if frozen_parent_hot_path
            else {
            "resynthesisRBOActive": True,
            "none_architecture": "Nest of Native Experts",
            "none_model_owned": True,
            "none_transfer_active": self._last_none_phase is not None,
            "none_outcome_writes": int(
                self.none_fabric.outcome_writes.detach().cpu()
            ),
            "none_phase_visits": (
                self.none_fabric.phase_visits.detach().cpu().tolist()
            ),
            "none_shaped_logits_present": isinstance(
                shaped_logits,
                torch.Tensor,
            ),
            "none_pathway_phase": (
                self._last_hop_signal.none_pathway.phase.detach().cpu().tolist()
                if self._last_hop_signal is not None
                else []
            ),
            "none_pathway_gap": (
                float(
                    self._last_hop_signal.none_pathway.gap.detach().cpu()
                )
                if self._last_hop_signal is not None
                else 0.0
            ),
            "none_pathway_completion": (
                float(
                    self._last_hop_signal.none_pathway.completion.detach().cpu()
                )
                if self._last_hop_signal is not None
                else 0.0
            ),
            "none_pathway_runtime_gain": (
                float(
                    self._last_hop_signal.none_pathway.runtime_gain.detach().cpu()
                )
                if self._last_hop_signal is not None
                else 0.0
            ),
            "intent": intent_label,
            "swe_stage_idx": stage_index_boundary,
            "swe_stage": stages[stage_index_boundary],
            "domain_idx": domain_idx.detach().clone(),
            "subdomain_idx": subdomain_idx.detach().clone(),
            "steps": steps,
            # The recursive pass returns the count owned by this exact forward.
            # Reuse it in both the typed result and proof so mutable parent
            # attribute layouts cannot substitute stale or default telemetry.
            "self_correction_steps": self_correction_steps,
            "traversal_layers": (
                traversal_history[-1] if traversal_history else []
            ),
            "traversal_stop_reason": traversal_stop_reason,
            "arms_used": arms_used,
            "outcome_memory": self.outcome_memory_snapshot(),
            "identity_state": {
                "boundary_only": True,
                "identity_dim": int(self.rbo_cfg.identity_dim),
                "n_layers": len(self.experts),
            },
            "outcome_history_steps": len(feedback_history),
            "outcome_history_carry_norm": float(
                self.outcome_history.carry_h.norm().detach(),
            ),
            "hop_traversal": {
                "from_hop_idx": (
                    self._last_hop_signal.from_hop_idx
                    if self._last_hop_signal
                    else 0
                ),
                "dest_hop_idx": (
                    self._last_hop_signal.dest_hop_idx
                    if self._last_hop_signal
                    else 0
                ),
                "hop_confidence": (
                    float(self._last_hop_signal.hop_confidence.detach())
                    if self._last_hop_signal is not None
                    else 0.0
                ),
                "hop_bonus": (
                    float(self._last_hop_signal.hop_bonus.detach())
                    if self._last_hop_signal is not None
                    else 0.0
                ),
                "none_stop_delta": (
                    float(
                        self._last_hop_signal.none_pathway.stop_delta.detach()
                    )
                    if self._last_hop_signal is not None
                    else 0.0
                ),
                "none_slice_delta": (
                    float(
                        self._last_hop_signal.none_pathway.slice_delta.detach()
                    )
                    if self._last_hop_signal is not None
                    else 0.0
                ),
                "stop_delta_accum": self._boundary_float(
                    self._hop_stop_delta_accum
                ),
                "slice_delta_accum": self._boundary_float(
                    self._hop_slice_delta_accum
                ),
            },
            }
        ),
    )


def _historical_rbo_init_with_tensor_intent_mapping(
    self: Any,
    *args: Any,
    **kwargs: Any,
) -> None:
    """Materialize the source-era intent map once, before the RBO moves to CUDA."""

    owner = type(self)
    original = getattr(owner, "_resynthesis_original_rbo_init", None)
    intent_to_capability = getattr(
        owner,
        "_resynthesis_intent_to_capability_boundary",
        None,
    )
    capability_count = getattr(
        owner,
        "_resynthesis_intent_capability_count",
        None,
    )
    if not callable(original) or not callable(intent_to_capability):
        raise RuntimeError("historical Resynthesis intent constructor was not preserved")
    if (
        not isinstance(capability_count, int)
        or isinstance(capability_count, bool)
        or capability_count < 1
    ):
        raise RuntimeError("historical Resynthesis capability count is invalid")
    original(self, *args, **kwargs)

    intent_names = getattr(self, "intent_names", None)
    arm_layout = getattr(self, "arm_layout", None)
    reference = getattr(self, "layer_pick_logits", None)
    if (
        not isinstance(intent_names, (list, tuple))
        or not intent_names
        or not all(isinstance(name, str) for name in intent_names)
        or not isinstance(reference, torch.Tensor)
    ):
        raise RuntimeError("historical Resynthesis intent geometry is malformed")
    n_domains = getattr(arm_layout, "n_domains", None)
    n_subdomains = getattr(arm_layout, "n_subdomains", None)
    if (
        not isinstance(n_domains, int)
        or isinstance(n_domains, bool)
        or n_domains < 1
        or not isinstance(n_subdomains, int)
        or isinstance(n_subdomains, bool)
        or n_subdomains < 1
    ):
        raise RuntimeError("historical Resynthesis intent route geometry is malformed")

    # This is the one explicit source-compatibility boundary. It exhaustively
    # resolves every classifier row through the hash-verified source-era map.
    # The active forward only gathers these tensors; it has no string/default
    # route and never reads the selected CUDA scalar on the host.
    capability_indices = tuple(
        intent_to_capability(intent_name) for intent_name in intent_names
    )
    if not all(
        isinstance(index, int)
        and not isinstance(index, bool)
        and 0 <= index < capability_count
        for index in capability_indices
    ):
        raise RuntimeError("historical Resynthesis intent capability map is invalid")
    capability_map = torch.tensor(
        capability_indices,
        device=reference.device,
        dtype=torch.long,
    )
    domain_map = capability_map.remainder(n_domains)
    subdomain_map = torch.div(
        capability_map,
        n_domains,
        rounding_mode="floor",
    ).remainder(n_subdomains)
    for name, value in (
        ("_resynthesis_intent_capability_map", capability_map),
        ("_resynthesis_intent_domain_map", domain_map),
        ("_resynthesis_intent_subdomain_map", subdomain_map),
    ):
        if hasattr(self, name):
            raise RuntimeError("historical Resynthesis intent map is already attached")
        self.register_buffer(name, value, persistent=False)

    none_fabric = getattr(self, "none_fabric", None)
    none_geometry = getattr(none_fabric, "geometry", None)
    phase_count = getattr(none_geometry, "n_phases", None)
    if (
        not isinstance(phase_count, int)
        or isinstance(phase_count, bool)
        or phase_count < 1
    ):
        raise RuntimeError("historical Resynthesis phase geometry is malformed")
    self.register_buffer(
        "_resynthesis_none_phase_selector_t",
        torch.eye(
            phase_count,
            device=reference.device,
            dtype=reference.dtype,
        ),
        persistent=False,
    )


def _tensor_native_historical_none_step_phase(
    self: Any,
    *,
    commit: Any,
    phase_hidden: torch.Tensor,
    observation_hidden: torch.Tensor,
    step_idx: int,
) -> Any:
    """Run one historical Fabric phase without a host-created CUDA scalar."""

    phase_selector_t = getattr(
        self,
        "_resynthesis_none_phase_selector_t",
        None,
    )
    none_fabric = getattr(self, "none_fabric", None)
    none_geometry = getattr(none_fabric, "geometry", None)
    phase_count = getattr(none_geometry, "n_phases", None)
    step_phase = getattr(none_fabric, "step_phase", None)
    if (
        not isinstance(step_idx, int)
        or isinstance(step_idx, bool)
        or not isinstance(phase_count, int)
        or isinstance(phase_count, bool)
        or phase_count < 1
        or not isinstance(phase_selector_t, torch.Tensor)
        or phase_selector_t.shape != (phase_count, phase_count)
        or phase_selector_t.device != phase_hidden.device
        or phase_selector_t.dtype != phase_hidden.dtype
        or not callable(step_phase)
    ):
        raise RuntimeError("historical Resynthesis phase selector differs")
    expert_state, layer_state = self._none_expert_layer_state(phase_hidden)
    phase_weights = phase_selector_t[
        step_idx % phase_count
    ].reshape(1, phase_count).expand(phase_hidden.shape[0], -1)
    compact_phase = torch.tanh(self.none_hidden_in(phase_hidden))
    compact_observation = torch.tanh(
        self.none_hidden_in(observation_hidden)
    )
    return step_phase(
        commit,
        compact_phase,
        compact_observation,
        expert_state,
        layer_state,
        phase_weights,
    )


def _tensor_native_historical_intent_route(
    self: Any,
    intent_probs: torch.Tensor,
    intents: list[str] | tuple[str, ...],
) -> _HistoricalTensorIntentRoute:
    """Select and map the exact historical intent without a CUDA host read."""

    capability_map = getattr(self, "_resynthesis_intent_capability_map", None)
    domain_map = getattr(self, "_resynthesis_intent_domain_map", None)
    subdomain_map = getattr(self, "_resynthesis_intent_subdomain_map", None)
    expected_shape = (len(intents),)
    if (
        intent_probs.ndim != 3
        or intent_probs.shape[-1] != len(intents)
        or not isinstance(capability_map, torch.Tensor)
        or capability_map.shape != expected_shape
        or capability_map.device != intent_probs.device
        or capability_map.dtype != torch.long
        or not isinstance(domain_map, torch.Tensor)
        or domain_map.shape != expected_shape
        or domain_map.device != intent_probs.device
        or domain_map.dtype != torch.long
        or not isinstance(subdomain_map, torch.Tensor)
        or subdomain_map.shape != expected_shape
        or subdomain_map.device != intent_probs.device
        or subdomain_map.dtype != torch.long
    ):
        raise RuntimeError("historical Resynthesis tensor intent map differs")
    intent_index = intent_probs.mean(dim=(0, 1)).argmax().to(
        dtype=torch.long
    ).reshape(())
    selector = intent_index.reshape(1)
    return _HistoricalTensorIntentRoute(
        intent_index=intent_index,
        capability_index=capability_map.index_select(0, selector).reshape(()),
        domain_index=domain_map.index_select(0, selector).reshape(()),
        subdomain_index=subdomain_map.index_select(0, selector).reshape(()),
    )


def _tensor_native_none_pathway_training_loss(
    self: Any,
    packet: Any,
    *,
    receiver_capability_idx: int | torch.Tensor,
    phase_target: torch.Tensor,
    gap_target: torch.Tensor,
    completion_target: torch.Tensor,
) -> torch.Tensor:
    """Preserve the historical NoNE loss with tensor receiver indexing."""

    transfer = packet.transfer
    receiver = _tensor_hotpath_index_boundary(
        receiver_capability_idx,
        reference=transfer,
        upper_bound=int(self.n_capabilities),
    )
    epsilon = torch.finfo(transfer.dtype).eps
    phase_target = phase_target.to(
        device=packet.phase.device,
        dtype=packet.phase.dtype,
    ).reshape_as(packet.phase)
    phase_target = phase_target / phase_target.sum().clamp_min(epsilon)
    gap_target = gap_target.to(
        device=packet.gap.device,
        dtype=packet.gap.dtype,
    ).reshape(())
    completion_target = completion_target.to(
        device=packet.completion.device,
        dtype=packet.completion.dtype,
    ).reshape(())
    phase_loss = -(phase_target * packet.phase.clamp_min(epsilon).log()).sum()
    transfer_loss = -transfer.index_select(
        0,
        receiver.reshape(1),
    ).reshape(()).clamp_min(epsilon).log()
    state_loss = F.binary_cross_entropy(
        packet.gap,
        gap_target.clamp(0.0, 1.0),
    )
    state_loss = state_loss + F.binary_cross_entropy(
        packet.completion,
        completion_target.clamp(0.0, 1.0),
    )
    desired_gain = torch.maximum(
        gap_target,
        torch.ones_like(completion_target) - completion_target,
    ).clamp(0.0, 1.0)
    gain_loss = F.mse_loss(packet.runtime_gain, desired_gain)
    stop_target = completion_target - gap_target
    slice_target = gap_target - completion_target
    traversal_loss = F.mse_loss(packet.stop_delta, stop_target)
    traversal_loss = traversal_loss + F.mse_loss(
        packet.slice_delta,
        slice_target,
    )
    route_target = (gap_target - completion_target).reshape(())
    route_loss = F.mse_loss(packet.domain_delta.mean(), route_target)
    route_loss = route_loss + F.mse_loss(packet.layer_delta.mean(), route_target)
    route_loss = route_loss + F.mse_loss(packet.expert_delta.mean(), route_target)
    return cast(
        torch.Tensor,
        phase_loss
        + transfer_loss
        + state_loss
        + 0.25 * gain_loss
        + 0.10 * traversal_loss
        + 0.05 * route_loss,
    )


def _historical_tensor_native_identity_supervision_loss(
    self: Any,
    hidden: torch.Tensor,
    intent_probs: torch.Tensor,
    ctx_signals: Mapping[str, torch.Tensor],
    *,
    target_domain_idx: int | torch.Tensor,
    target_capability_idx: int | torch.Tensor,
) -> torch.Tensor:
    """Preserve historical identity supervision with tensor route indices."""

    n_layers = len(self.experts)
    if n_layers <= 0:
        return hidden.new_zeros(())
    transfer_weights = ctx_signals.get("transfer_weights")
    layer_scores = self._layer_identity_bias(
        hidden,
        intent_probs,
        transfer_weights,
        n_layers,
    )
    target_domain = _tensor_hotpath_index_boundary(
        target_domain_idx,
        reference=self.domain_layer_affinity,
        upper_bound=int(self.domain_layer_affinity.shape[0]),
    )
    layer_target = F.softmax(
        self.domain_layer_affinity.index_select(
            0,
            target_domain.reshape(1),
        )
        .squeeze(0)[:n_layers]
        .detach(),
        dim=-1,
    )
    layer_loss = -(layer_target * F.log_softmax(layer_scores, dim=-1)).sum()
    capability_index = _tensor_hotpath_index_boundary(
        target_capability_idx,
        reference=hidden,
        upper_bound=10,
    )
    capability_target = F.one_hot(
        capability_index,
        num_classes=10,
    ).to(device=hidden.device, dtype=hidden.dtype)
    capability_losses = tuple(
        F.mse_loss(
            torch.sigmoid(
                expert.flat_expert_capability.to(
                    device=hidden.device,
                    dtype=hidden.dtype,
                )
            ),
            capability_target.unsqueeze(0).expand_as(
                expert.flat_expert_capability
            ),
        )
        for expert in self.experts
    )
    capability_loss = (
        torch.stack(capability_losses).mean()
        if capability_losses
        else hidden.new_zeros(())
    )
    layer_capability = torch.stack(
        tuple(expert.capability_scores for expert in self.experts),
        dim=0,
    ).to(device=hidden.device, dtype=hidden.dtype)
    layer_capability_loss = F.mse_loss(
        torch.sigmoid(layer_capability),
        capability_target.unsqueeze(0).expand_as(layer_capability),
    )
    specialization = torch.stack(
        tuple(expert.specialization for expert in self.experts),
        dim=0,
    ).to(device=hidden.device, dtype=hidden.dtype)
    specialization_target = (layer_target * float(n_layers)).clamp(0.0, 1.0)
    specialization_loss = F.mse_loss(
        torch.sigmoid(specialization),
        specialization_target,
    )
    routing_fingerprints = torch.stack(
        tuple(expert.routing_fingerprint for expert in self.experts),
        dim=0,
    )
    routing_fingerprint_loss = self._identity_orthogonality_loss(
        routing_fingerprints
    )
    return cast(
        torch.Tensor,
        layer_loss
        + 0.10 * capability_loss
        + 0.10 * layer_capability_loss
        + 0.05 * specialization_loss
        + self.rbo_cfg.identity_diversity_weight
        * self._identity_diversity_loss()
        + self.rbo_cfg.identity_diversity_weight * routing_fingerprint_loss
        + self.rbo_cfg.profile_diversity_weight * self._profile_diversity_loss(),
    )


def _historical_tensor_native_context_dag_training_loss(
    self: Any,
    hidden: torch.Tensor,
    intent_probs: torch.Tensor,
    ctx_signals: Mapping[str, torch.Tensor],
    intent_label: str | torch.Tensor | _HistoricalTensorIntentRoute,
    feedback_history: list[Mapping[str, torch.Tensor]],
    *,
    domain_idx: int | torch.Tensor,
    subdomain_idx: int | torch.Tensor,
    stage_domain_bias: torch.Tensor | None = None,
    stage_subdomain_bias: torch.Tensor | None = None,
) -> torch.Tensor:
    """Run the exact source-era semantic loss from a tensor intent route."""

    if isinstance(intent_label, str):
        original = getattr(
            type(self),
            "_resynthesis_original_context_dag_training_loss",
            None,
        )
        if not callable(original):
            raise RuntimeError("historical Resynthesis context loss was not preserved")
        return cast(
            torch.Tensor,
            original(
                self,
                hidden,
                intent_probs,
                ctx_signals,
                intent_label,
                feedback_history,
                domain_idx=domain_idx,
                subdomain_idx=subdomain_idx,
                stage_domain_bias=stage_domain_bias,
                stage_subdomain_bias=stage_subdomain_bias,
            ),
        )
    capability_map = self._resynthesis_intent_capability_map
    domain_map = self._resynthesis_intent_domain_map
    subdomain_map = self._resynthesis_intent_subdomain_map
    if isinstance(intent_label, torch.Tensor):
        intent_route: _HistoricalTensorIntentRoute | None = None
        route_intent_index = intent_label
    else:
        intent_route = _normalize_historical_tensor_intent_route(intent_label)
        route_intent_index = intent_route.intent_index
    intent_index = _tensor_hotpath_index_boundary(
        route_intent_index,
        reference=capability_map,
        upper_bound=int(capability_map.numel()),
    )
    selector = intent_index.reshape(1)
    capability_index = capability_map.index_select(0, selector).reshape(())
    target_domain = domain_map.index_select(0, selector).reshape(())
    target_subdomain = subdomain_map.index_select(0, selector).reshape(())
    if intent_route is not None:
        # Older parent generations returned a typed route packet and therefore
        # carry redundant mapped indices that remain exact integrity proofs.
        # Current native parents already pass the model-owned scalar tensor
        # directly; its maps are gathered above without a host readback.
        torch._assert_async(
            capability_index.eq(intent_route.capability_index),
            "historical Resynthesis intent capability route differs",
        )
        torch._assert_async(
            target_domain.eq(intent_route.domain_index),
            "historical Resynthesis intent domain route differs",
        )
        torch._assert_async(
            target_subdomain.eq(intent_route.subdomain_index),
            "historical Resynthesis intent subdomain route differs",
        )

    zero = hidden.new_zeros(())
    epsilon = torch.finfo(hidden.float().dtype).eps
    domain_confidence = ctx_signals["domain_conf"].clamp_min(epsilon)
    capability_loss = -torch.log(
        domain_confidence.index_select(
            1,
            capability_index.reshape(1),
        ).clamp_min(epsilon)
    ).mean()

    depth_probabilities = ctx_signals["depth_probs"].clamp_min(epsilon)
    depth_index = min(
        max(len(feedback_history) - 1, 0),
        depth_probabilities.shape[-1] - 1,
    )
    depth_loss = -torch.log(
        depth_probabilities[:, depth_index].clamp_min(epsilon)
    ).mean()

    if feedback_history:
        rubric_values = torch.stack(
            tuple(
                feedback["rubric"]
                .detach()
                .reshape(())
                .to(device=hidden.device, dtype=hidden.dtype)
                for feedback in feedback_history
            )
        )
        completion_target = rubric_values.mean().clamp(0.0, 1.0)
        complexity_target = torch.ones_like(completion_target) - completion_target
    else:
        completion_target = hidden.new_ones(()).mul_(0.5)
        complexity_target = hidden.new_ones(()).mul_(0.5)
    complexity_loss = F.mse_loss(
        ctx_signals["complexity"],
        complexity_target.expand_as(ctx_signals["complexity"]),
    )

    transfer_owner = self.cross_domain_router.knowledge_transfer
    relatedness = (
        transfer_owner.relatedness_prior.index_select(
            0,
            capability_index.reshape(1),
        )
        .squeeze(0)
        .to(device=hidden.device, dtype=hidden.dtype)
        .clone()
    )
    capability_mask = torch.arange(
        transfer_owner.n_capabilities,
        device=hidden.device,
    ).ne(capability_index)
    relatedness = relatedness * capability_mask.to(dtype=relatedness.dtype)
    relatedness_mass = relatedness.sum()
    self_capability = F.one_hot(
        capability_index,
        num_classes=transfer_owner.n_capabilities,
    ).to(device=hidden.device, dtype=hidden.dtype)
    relatedness = torch.where(
        relatedness_mass.le(epsilon),
        self_capability,
        relatedness / relatedness_mass.clamp_min(epsilon),
    )
    transfer_weights = ctx_signals["transfer_weights"].clamp_min(epsilon)
    transfer_loss = F.kl_div(
        transfer_weights.log(),
        relatedness.unsqueeze(0).expand_as(transfer_weights),
        reduction="batchmean",
    )
    visit_scores, _ = transfer_owner.neural_dag(
        hidden=hidden,
        start_cap=capability_index,
    )
    dag_loss = F.kl_div(
        visit_scores.clamp_min(epsilon).log(),
        relatedness,
        reduction="sum",
    )
    outcome_pressure = self.arm_registry.domain_slot_pressure().sum(dim=-1)
    donor = transfer_owner.donor_weights_for_pressure(
        target_domain,
        outcome_pressure,
        hidden=hidden,
    )
    none_packet = transfer_owner.build_none_pathway_packet(
        target_domain,
        donor=donor,
        hidden=hidden,
        outcome_pressure=outcome_pressure,
        n_layers=len(self.experts),
        n_experts=self.cfg.total_experts,
    )
    gap_target = complexity_target.clamp(0.0, 1.0)
    phase_target = torch.stack(
        (
            0.10 * (torch.ones_like(gap_target) - gap_target),
            0.35 * gap_target,
            0.65 * gap_target,
            completion_target,
        )
    )
    receiver_capability = relatedness.argmax().to(dtype=torch.long).reshape(())
    none_pathway_loss = transfer_owner.none_pathway_training_loss(
        none_packet,
        receiver_capability_idx=receiver_capability,
        phase_target=phase_target,
        gap_target=gap_target,
        completion_target=completion_target,
    )

    domain_probabilities, subdomain_probabilities, _, _ = self.domain_router(
        hidden,
        intent_probs,
        domain_log_bias=stage_domain_bias,
        subdomain_log_bias=stage_subdomain_bias,
    )
    route_loss = -torch.log(
        domain_probabilities.index_select(
            1,
            target_domain.reshape(1),
        ).clamp_min(epsilon)
    ).mean()
    route_loss = route_loss - torch.log(
        subdomain_probabilities.index_select(
            1,
            target_subdomain.reshape(1),
        ).clamp_min(epsilon)
    ).mean()

    observed_domain = (
        domain_idx.detach().to(device=hidden.device, dtype=torch.long).reshape(())
        if isinstance(domain_idx, torch.Tensor)
        else hidden.new_empty((), dtype=torch.long).fill_(domain_idx)
    )
    observed_subdomain = (
        subdomain_idx.detach().to(
            device=hidden.device,
            dtype=torch.long,
        ).reshape(())
        if isinstance(subdomain_idx, torch.Tensor)
        else hidden.new_empty((), dtype=torch.long).fill_(subdomain_idx)
    )
    observed_valid = (
        observed_domain.ge(0)
        .logical_and(observed_domain.lt(domain_probabilities.shape[-1]))
        .logical_and(observed_subdomain.ge(0))
        .logical_and(
            observed_subdomain.lt(subdomain_probabilities.shape[-1])
        )
    )
    observed_domain = observed_domain.clamp(
        min=0,
        max=domain_probabilities.shape[-1] - 1,
    )
    observed_subdomain = observed_subdomain.clamp(
        min=0,
        max=subdomain_probabilities.shape[-1] - 1,
    )
    observed_slot_loss = -torch.log(
        domain_probabilities.index_select(
            1,
            observed_domain.reshape(1),
        ).clamp_min(epsilon)
    ).mean()
    observed_slot_loss = observed_slot_loss - torch.log(
        subdomain_probabilities.index_select(
            1,
            observed_subdomain.reshape(1),
        ).clamp_min(epsilon)
    ).mean()
    observed_slot_loss = (
        observed_slot_loss
        * observed_valid.to(device=hidden.device, dtype=hidden.dtype)
    )
    identity_loss = self._identity_supervision_loss(
        hidden,
        intent_probs,
        ctx_signals,
        target_domain_idx=target_domain,
        target_capability_idx=capability_index,
    )
    return cast(
        torch.Tensor,
        0.35 * capability_loss
        + 0.20 * depth_loss
        + 0.15 * complexity_loss
        + 0.25 * transfer_loss
        + 0.25 * dag_loss
        + 0.20 * none_pathway_loss
        + 0.20 * route_loss
        + 0.05 * observed_slot_loss
        + 0.08 * identity_loss
        + zero,
    )


def _historical_none_fabric_init_with_tensor_constants(
    self: Any,
    geometry: Any,
) -> None:
    """Install the immutable Fabric's device-resident scalar geometry.

    The historical ``form_commit`` rebuilt the atomic-graph width with
    ``Tensor.new_tensor`` on every parent forward.  On CUDA that host-to-device
    scalar copy synchronizes the parent stream.  Registering the exact scalar
    once keeps it with the module across device/dtype moves.  It is
    nonpersistent because it is derived entirely from the already-authoritative
    geometry and therefore must not alter checkpoint identity.
    """

    original = getattr(
        type(self),
        "_resynthesis_original_none_fabric_init",
        None,
    )
    if not callable(original):
        raise RuntimeError("historical NoNE Fabric constructor was not preserved")
    original(self, geometry)
    graph_node_credit = getattr(self, "graph_node_credit", None)
    if not isinstance(graph_node_credit, torch.Tensor):
        raise RuntimeError("historical NoNE Fabric graph geometry differs")
    self.register_buffer(
        "_resynthesis_atomic_graph_slots_scale_t",
        graph_node_credit.new_full((), geometry.atomic_graph_slots),
        persistent=False,
    )


def _tensor_native_historical_none_fabric_form_commit(
    self: Any,
    surface_glyphs: torch.Tensor,
    evidence_logits: torch.Tensor,
    capability_gap: torch.Tensor,
) -> Any:
    """Form the exact historical Fabric commit without CUDA scalar copies."""

    if surface_glyphs.ndim != 3:
        raise ValueError(
            "surface_glyphs must have shape [batch, obligations, hidden]"
        )
    if evidence_logits.ndim != 2:
        raise ValueError(
            "evidence_logits must have shape [batch, obligations]"
        )
    if capability_gap.ndim != 2 or capability_gap.shape[-1] != 3:
        raise ValueError("capability_gap must have shape [batch, 3]")
    if surface_glyphs.shape[1] != self.geometry.n_obligations:
        raise ValueError(
            "surface_glyphs obligation axis does not match Fabric geometry"
        )

    identity = self.architecture_identity.unsqueeze(1)
    encoded = torch.tanh(self.surface_encoder(surface_glyphs + identity))
    preliminary_state = self.commit_encoder(encoded.mean(dim=1))
    learned_evidence = self.obligation_evidence(encoded).squeeze(-1)
    support = torch.sigmoid(evidence_logits + learned_evidence)
    requirements = torch.sigmoid(self.commit_requirement_logits).unsqueeze(0)
    obligation_margin = support - requirements
    learned_memory = self.outcome_memory_reader(
        self.outcome_memory.mean(dim=0, keepdim=True)
    ).expand(surface_glyphs.shape[0], -1)
    state = torch.tanh(
        preliminary_state
        + self.commit_obligation_encoder(obligation_margin)
        + self.capability_pressure_encoder(capability_gap)
        + learned_memory
    )
    atomic_graph = self.atomic_task_graph(
        state,
        encoded,
        support,
        capability_gap,
    )
    graph_credit_logits = (
        self.graph_node_credit - self.graph_rollback_pressure
    ).to(device=state.device, dtype=state.dtype)
    graph_slots_scale_t = self._resynthesis_atomic_graph_slots_scale_t.to(
        device=state.device,
        dtype=state.dtype,
    )
    graph_credit_scale = (
        torch.softmax(graph_credit_logits, dim=-1) * graph_slots_scale_t
    )
    graph_ready_weights = (
        atomic_graph.ready_weights * graph_credit_scale.unsqueeze(0)
    )
    graph_repair_scale = 2.0 * torch.sigmoid(
        self.graph_repair_credit.to(device=state.device, dtype=state.dtype)
    )
    graph_repair_weights = (
        atomic_graph.repair_weights * graph_repair_scale.unsqueeze(0)
    )
    session_present = self.session_graph_present.to(
        device=state.device,
        dtype=state.dtype,
    )
    if self.training:
        session_present = torch.zeros_like(session_present)
    prior_frozen = self.session_graph_frozen_weights.to(
        device=state.device,
        dtype=state.dtype,
    )
    node_keep = (session_present * prior_frozen).reshape(1, -1)
    graph_node_state = torch.lerp(
        atomic_graph.node_state,
        self.session_graph_node_state.to(
            device=state.device,
            dtype=state.dtype,
        )
        .unsqueeze(0)
        .expand_as(atomic_graph.node_state),
        node_keep.unsqueeze(-1),
    )
    edge_keep = node_keep.unsqueeze(-1) * node_keep.unsqueeze(-2)
    graph_edge_weights = torch.lerp(
        atomic_graph.edge_weights,
        self.session_graph_edge_weights.to(
            device=state.device,
            dtype=state.dtype,
        )
        .unsqueeze(0)
        .expand_as(atomic_graph.edge_weights),
        edge_keep,
    )
    graph_validated_weights = torch.maximum(
        atomic_graph.validated_weights,
        self.session_graph_validated_weights.to(
            device=state.device,
            dtype=state.dtype,
        ).unsqueeze(0)
        * session_present,
    )
    graph_frozen_weights = torch.maximum(
        atomic_graph.frozen_weights,
        prior_frozen.unsqueeze(0) * session_present,
    )
    graph_repair_weights = graph_repair_weights * (
        1.0 - graph_frozen_weights
    )
    graph_summary_weights = (
        graph_ready_weights + graph_repair_weights + graph_frozen_weights
    )
    graph_summary = self.atomic_task_graph.graph_summary_head(
        torch.einsum(
            "bn,bnh->bh",
            graph_summary_weights,
            graph_node_state,
        )
        / graph_summary_weights.sum(dim=-1, keepdim=True).clamp_min(
            torch.finfo(graph_summary_weights.dtype).tiny
        )
    )
    state = torch.tanh(state + graph_summary)
    if not self.training:
        with torch.no_grad():
            self.session_graph_node_state.copy_(
                graph_node_state.detach().mean(dim=0)
            )
            self.session_graph_edge_weights.copy_(
                graph_edge_weights.detach().mean(dim=0)
            )
            self.session_graph_validated_weights.copy_(
                graph_validated_weights.detach().mean(dim=0)
            )
            self.session_graph_frozen_weights.copy_(
                graph_frozen_weights.detach().mean(dim=0)
            )
            self.session_graph_repair_weights.copy_(
                graph_repair_weights.detach().mean(dim=0)
            )
            self.session_graph_present.fill_(True)
    scope_completeness = obligation_margin.amin(dim=-1, keepdim=True)
    edit_authorization = (scope_completeness >= 0).to(surface_glyphs.dtype)
    mutating_probability = torch.einsum(
        "bno,o->bn",
        atomic_graph.operation_probs,
        self.atomic_task_graph.mutating_operation_mask.to(
            device=state.device,
            dtype=state.dtype,
        ),
    )
    graph_execution_authorization = (
        1.0
        - mutating_probability
        + mutating_probability * edit_authorization
    ).clamp(min=0.0, max=1.0)
    unresolved_tickets = (
        encoded * torch.relu(-obligation_margin).unsqueeze(-1)
    )

    commit_expert_source = self._record_causal_gradient(
        self.commit_expert_source(state),
        0,
    )
    commit_expert_destination = self._record_causal_gradient(
        self.commit_expert_destination(state),
        1,
    )
    commit_layer_source = self._record_causal_gradient(
        self.commit_layer_source(state),
        2,
    )
    commit_layer_destination = self._record_causal_gradient(
        self.commit_layer_destination(state),
        3,
    )
    expert_logits = self._pair_logits(
        commit_expert_source,
        commit_expert_destination,
        self.expert_pair_prior,
        self.expert_transfer_credit,
    )
    layer_logits = self._pair_logits(
        commit_layer_source,
        commit_layer_destination,
        self.layer_pair_prior,
        self.layer_transfer_credit,
    )
    research_branch_logits = self._record_causal_gradient(
        self.research_branch_head(state),
        4,
    )
    unresolved_pressure = torch.relu(-scope_completeness)
    research_pressure = torch.sigmoid(
        self.research_pressure_head(state)
        + unresolved_pressure
        + capability_gap[:, 0:1]
    )
    tool_build_pressure = torch.sigmoid(
        self.tool_build_head(state)
        + unresolved_pressure
        + capability_gap[:, 1:2]
    )
    environment_build_pressure = torch.sigmoid(
        self.environment_build_head(state)
        + unresolved_pressure
        + capability_gap[:, 2:3]
    )

    commit_type = getattr(
        type(self),
        "_resynthesis_none_fabric_commit_type",
        None,
    )
    if not isinstance(commit_type, type):
        raise RuntimeError("historical NoNE Fabric commit type is unavailable")
    return commit_type(
        state=state,
        obligation_support=support,
        unresolved_tickets=unresolved_tickets,
        scope_completeness=scope_completeness,
        edit_authorization=edit_authorization,
        expert_source_state=commit_expert_source,
        expert_destination_state=commit_expert_destination,
        layer_source_state=commit_layer_source,
        layer_destination_state=commit_layer_destination,
        expert_transfer_intent=self._off_diagonal_softmax(expert_logits),
        layer_transfer_intent=self._off_diagonal_softmax(layer_logits),
        research_branch_weights=F.softmax(research_branch_logits, dim=-1),
        research_pressure=research_pressure,
        tool_build_pressure=tool_build_pressure,
        environment_build_pressure=environment_build_pressure,
        graph_node_state=graph_node_state,
        graph_operation_logits=atomic_graph.operation_logits,
        graph_operation_probs=atomic_graph.operation_probs,
        graph_argument_state=atomic_graph.argument_state,
        graph_observation_binding=atomic_graph.observation_binding,
        graph_execution_authorization=graph_execution_authorization,
        graph_edge_weights=graph_edge_weights,
        graph_refinement_parent_weights=(
            atomic_graph.refinement_parent_weights
        ),
        graph_atomicity=atomic_graph.atomicity,
        graph_input_interfaces=atomic_graph.input_interfaces,
        graph_output_interfaces=atomic_graph.output_interfaces,
        graph_interface_compatibility=(
            atomic_graph.interface_compatibility
        ),
        graph_simulation_risk=atomic_graph.simulation_risk,
        graph_ready_weights=graph_ready_weights,
        graph_parallel_weights=atomic_graph.parallel_weights,
        graph_validated_weights=graph_validated_weights,
        graph_affected_weights=atomic_graph.affected_weights,
        graph_frozen_weights=graph_frozen_weights,
        graph_repair_weights=graph_repair_weights,
        graph_summary=graph_summary,
    )


def _install_historical_tensor_native_hotpaths(
    rbo_module: Any,
    experts_module: Any,
    transfer_module: Any,
    traversal_module: Any,
) -> str:
    """Bind hash-verified historical classes to equivalent tensor hot paths."""

    expert_bank = getattr(experts_module, "FactorizedGranularExpertBank", None)
    transfer = getattr(transfer_module, "KnowledgeTransferSurfaces", None)
    intent_to_capability = getattr(
        transfer_module,
        "intent_to_capability",
        None,
    )
    capability_count = getattr(transfer_module, "N_CAPABILITIES", None)
    pathway_packet_type = getattr(transfer_module, "NoNEPathwayPacket", None)
    hop_router = getattr(traversal_module, "CrossDomainHopRouter", None)
    signal_type = getattr(traversal_module, "HopTraversalSignal", None)
    outcome_history = getattr(traversal_module, "OutcomeHistoryEncoder", None)
    rbo_type = getattr(rbo_module, _legacy_parent_class_name("RBO"), None)
    rbo_result_type = getattr(
        rbo_module,
        _legacy_parent_class_name("RBOResult"),
        None,
    )
    rubric_bias_type = getattr(rbo_module, "RubricLogitBias", None)
    contrareactive_repulsion = getattr(
        rbo_module,
        "contrareactive_repulsion",
        None,
    )
    swe_stages = getattr(rbo_module, "SWE_STAGES", None)
    correction_controller = getattr(
        rbo_module,
        "SelfCorrectionController",
        None,
    )
    domain_collapse_detector = getattr(
        rbo_module,
        "DomainCollapseDetector",
        None,
    )
    none_fabric_type = getattr(
        rbo_module,
        _legacy_parent_class_name("NoNEFabric"),
        None,
    )
    none_fabric_commit_type = getattr(rbo_module, "NoNEFabricCommit", None)
    if not all(
        isinstance(value, type)
        for value in (
            rbo_type,
            rbo_result_type,
            expert_bank,
            transfer,
            hop_router,
            outcome_history,
            rubric_bias_type,
            none_fabric_type,
            none_fabric_commit_type,
        )
    ):
        raise RuntimeError("historical Resynthesis hot-path classes are unavailable")
    if (
        signal_type is None
        or not isinstance(pathway_packet_type, type)
        or not callable(intent_to_capability)
        or not callable(contrareactive_repulsion)
        or not isinstance(swe_stages, tuple)
        or not isinstance(capability_count, int)
        or isinstance(capability_count, bool)
        or capability_count < 1
    ):
        raise RuntimeError("historical Resynthesis hop-signal type is unavailable")
    if not all(
        callable(getattr(owner, method, None))
        for owner, method in (
            (expert_bank, "apply_topk"),
            (transfer, "_cap_weights_from_domain"),
            (transfer, "_pressure_on_capabilities"),
            (transfer, "donor_weights_for_pressure"),
            (transfer, "build_none_pathway_packet"),
            (transfer, "none_pathway_training_loss"),
            (hop_router, "compute_hop_signal"),
            (rbo_type, "_intent_label"),
            (rbo_type, "_context_dag_training_loss"),
            (rbo_type, "_identity_supervision_loss"),
            (rbo_type, "_plan_layer_traversal"),
            (rbo_type, "_comm_from_prior_delta"),
            (rbo_type, "_correction_training_loss"),
            (rbo_type, "layer_gate_scale"),
            (rbo_type, "_joint_stop_logit"),
            (rbo_type, "_none_step_phase"),
            (rbo_type, "_recursive_pass"),
            (rbo_type, "forward"),
            (outcome_history, "_pack_step"),
            (none_fabric_type, "form_commit"),
        )
    ):
        raise RuntimeError("historical Resynthesis hot-path method contract differs")
    if not bool(
        getattr(
            none_fabric_type,
            "_resynthesis_tensor_commit_installed",
            False,
        )
    ):
        setattr(
            none_fabric_type,
            "_resynthesis_original_none_fabric_init",
            getattr(none_fabric_type, "__init__"),
        )
        setattr(
            none_fabric_type,
            "_resynthesis_none_fabric_commit_type",
            none_fabric_commit_type,
        )
        setattr(
            none_fabric_type,
            "__init__",
            _historical_none_fabric_init_with_tensor_constants,
        )
        setattr(
            none_fabric_type,
            "form_commit",
            _tensor_native_historical_none_fabric_form_commit,
        )
        setattr(
            none_fabric_type,
            "_resynthesis_tensor_commit_installed",
            True,
        )
    setattr(expert_bank, "apply_topk", _tensor_native_expert_apply_topk)
    setattr(
        transfer, "_cap_weights_from_domain", _tensor_native_cap_weights_from_domain
    )
    setattr(
        transfer,
        "_pressure_on_capabilities",
        _tensor_native_pressure_on_capabilities,
    )
    setattr(
        transfer,
        "donor_weights_for_pressure",
        _tensor_native_donor_weights_for_pressure,
    )
    setattr(
        transfer,
        "none_pathway_training_loss",
        _tensor_native_none_pathway_training_loss,
    )
    if not bool(
        getattr(
            transfer,
            "_resynthesis_none_tensor_constants_installed",
            False,
        )
    ):
        setattr(
            transfer,
            "_resynthesis_original_transfer_init",
            getattr(transfer, "__init__"),
        )
        setattr(
            transfer,
            "__init__",
            _historical_transfer_init_with_tensor_constants,
        )
        setattr(
            transfer,
            "_resynthesis_none_tensor_constants_installed",
            True,
        )
    setattr(
        transfer,
        "_resynthesis_none_pathway_packet_type",
        pathway_packet_type,
    )
    setattr(
        transfer,
        "build_none_pathway_packet",
        _tensor_native_build_none_pathway_packet,
    )
    setattr(hop_router, "_resynthesis_hop_signal_type", signal_type)
    setattr(hop_router, "compute_hop_signal", _tensor_native_compute_hop_signal)
    setattr(
        rbo_type,
        "_comm_from_prior_delta",
        staticmethod(_tensor_native_sequence_comm_from_prior_delta),
    )
    if not bool(
        getattr(rbo_type, "_resynthesis_tensor_intent_mapping_installed", False)
    ):
        setattr(
            rbo_type,
            "_resynthesis_original_rbo_init",
            getattr(rbo_type, "__init__"),
        )
        setattr(
            rbo_type,
            "_resynthesis_intent_to_capability_boundary",
            staticmethod(intent_to_capability),
        )
        setattr(
            rbo_type,
            "_resynthesis_intent_capability_count",
            capability_count,
        )
        setattr(
            rbo_type,
            "_resynthesis_original_context_dag_training_loss",
            getattr(rbo_type, "_context_dag_training_loss"),
        )
        setattr(
            rbo_type,
            "__init__",
            _historical_rbo_init_with_tensor_intent_mapping,
        )
        setattr(
            rbo_type,
            "_intent_label",
            _tensor_native_historical_intent_route,
        )
        setattr(
            rbo_type,
            "_none_step_phase",
            _tensor_native_historical_none_step_phase,
        )
        setattr(
            rbo_type,
            "_context_dag_training_loss",
            _historical_tensor_native_context_dag_training_loss,
        )
        setattr(
            rbo_type,
            "_identity_supervision_loss",
            _historical_tensor_native_identity_supervision_loss,
        )
        setattr(
            rbo_type,
            "_resynthesis_tensor_intent_mapping_installed",
            True,
        )
    if isinstance(correction_controller, type):
        setattr(
            correction_controller,
            "mode_target",
            staticmethod(_tensor_native_self_correction_mode_target),
        )
    if (
        isinstance(domain_collapse_detector, type)
        and callable(getattr(domain_collapse_detector, "forward", None))
    ):
        setattr(
            domain_collapse_detector,
            "forward",
            _tensor_native_domain_collapse_forward,
        )
    if not bool(
        getattr(rbo_type, "_resynthesis_tensor_route_forward_installed", False)
    ):
        original_recursive_pass = getattr(rbo_type, "_recursive_pass")
        setattr(
            rbo_type,
            "_resynthesis_original_rbo_forward",
            getattr(rbo_type, "forward"),
        )
        setattr(
            rbo_type,
            "_resynthesis_original_recursive_pass",
            original_recursive_pass,
        )
        setattr(
            rbo_type,
            "_resynthesis_rbo_result_type",
            rbo_result_type,
        )
        setattr(
            rbo_type,
            "_resynthesis_contrareactive_repulsion",
            staticmethod(contrareactive_repulsion),
        )
        setattr(rbo_type, "_resynthesis_swe_stages", swe_stages)
        setattr(
            rbo_type,
            "_recursive_pass",
            _historical_recursive_pass_with_tensor_route(
                original_recursive_pass
            ),
        )
        setattr(
            rbo_type,
            "forward",
            _tensor_native_historical_rbo_forward,
        )
        setattr(
            rbo_type,
            "_resynthesis_tensor_route_forward_installed",
            True,
        )
    setattr(
        outcome_history,
        "_pack_step",
        _tensor_native_outcome_history_pack_step,
    )
    setattr(
        rubric_bias_type,
        "apply_bias",
        _tensor_native_rubric_apply_bias,
    )
    if not bool(
        getattr(rbo_type, "_resynthesis_batch_correction_loss_installed", False)
    ):
        setattr(
            rbo_type,
            "_resynthesis_original_correction_training_loss",
            getattr(rbo_type, "_correction_training_loss"),
        )
        setattr(
            rbo_type,
            "_correction_training_loss",
            _historical_batch_safe_correction_training_loss,
        )
        setattr(rbo_type, "_resynthesis_batch_correction_loss_installed", True)
    if not bool(getattr(rbo_type, "_resynthesis_arm_exhaustion_installed", False)):
        setattr(
            rbo_type,
            "_resynthesis_original_plan_layer_traversal",
            getattr(rbo_type, "_plan_layer_traversal"),
        )
        setattr(
            rbo_type,
            "_resynthesis_original_joint_stop_logit",
            getattr(rbo_type, "_joint_stop_logit"),
        )
        setattr(
            rbo_type,
            "_resynthesis_original_layer_gate_scale",
            getattr(rbo_type, "layer_gate_scale"),
        )
        setattr(
            rbo_type,
            "_plan_layer_traversal",
            _historical_plan_layer_traversal_with_arm_exhaustion,
        )
        setattr(
            rbo_type,
            "_joint_stop_logit",
            _historical_joint_stop_logit_with_arm_exhaustion,
        )
        setattr(
            rbo_type,
            "layer_gate_scale",
            _historical_layer_gate_scale_with_frontier_blend,
        )
        setattr(rbo_type, "_resynthesis_arm_exhaustion_installed", True)
    return HISTORICAL_TENSOR_NATIVE_HOTPATH_ID


@dataclass(frozen=True)
class LegacyRBOCapabilitySessionState:
    """Session-owned observable-outcome state for cold continuation."""

    durable_outcome: torch.Tensor
    shot_one_route: torch.Tensor
    shot_two_route: torch.Tensor
    last_route: torch.Tensor
    awaiting_second_shot: torch.Tensor
    durable_outcome_updates: torch.Tensor


@dataclass(frozen=True)
class LegacyRBOCapabilityEngagementPacket:
    """Tensor-only evidence from the preserved legacy capability bank."""

    authority_trained: torch.Tensor
    retention_passed: torch.Tensor
    forward_id: torch.Tensor
    forward_calls: torch.Tensor
    engaged_calls: torch.Tensor
    delta_l2: torch.Tensor
    route_id: torch.Tensor
    shot_one_route: torch.Tensor
    shot_two_route: torch.Tensor
    shot_one_route_id: torch.Tensor
    shot_two_route_id: torch.Tensor
    route_shift_l1: torch.Tensor
    durable_outcome_updates: torch.Tensor
    payload_tensor_count: torch.Tensor


def _capture_legacy_second_shot_route_(
    *,
    last_route: torch.Tensor,
    shot_two_route: torch.Tensor,
    awaiting_second_shot: torch.Tensor,
) -> None:
    """Capture a pending route without synchronizing the hot path to the host."""

    if last_route.shape != shot_two_route.shape:
        raise RuntimeError("legacy RBO second-shot route geometry differs")
    if awaiting_second_shot.shape != torch.Size([]):
        raise RuntimeError("legacy RBO second-shot pending state must be scalar")
    shot_two_route.copy_(
        torch.where(
            awaiting_second_shot.to(
                device=shot_two_route.device,
                dtype=torch.bool,
            ),
            last_route,
            shot_two_route,
        )
    )
    awaiting_second_shot.zero_()


class _FrozenLegacyRBOTensor(nn.Module):
    """One dehydrated parent tensor that remains immutable and CPU-resident."""

    def __init__(self, value: torch.Tensor) -> None:
        super().__init__()
        frozen = value.detach().to(device="cpu").contiguous()
        frozen.requires_grad_(False)
        self.register_buffer("payload", frozen, persistent=True)

    def _apply(self, fn: Any, recurse: bool = True) -> "_FrozenLegacyRBOTensor":
        del fn, recurse
        return self


class LegacyRBOCapabilityBank(nn.Module):
    """Pinned compatibility owner for the parent's embedded 553-tensor RBO.

    The parent artifact contains an older RBO geometry that cannot be overlaid
    onto the authoritative 1,043-tensor graph.  This implementation is owned by
    Resynthesis so continuation never imports a changing Resynthesis worktree.  Its
    tensor names, initialization, routing, outcome memory, and persistence match
    the exact source generation recorded by
    ``LEGACY_RBO_CAPABILITY_PROVENANCE_SHA256``.
    """

    schema = _RESYNTHESIS_PARENT_CAPABILITY_SCHEMA
    durable_outcome_width = 10
    _payloads: nn.ModuleList
    capability_profiles: torch.Tensor
    route_query: nn.Linear
    outcome_query: nn.Linear
    residual_projection: nn.Linear
    fusion_logit: nn.Parameter
    authority_trained: torch.Tensor
    retention_passed: torch.Tensor
    forward_id: torch.Tensor
    forward_calls: torch.Tensor
    engaged_calls: torch.Tensor
    last_delta_l2: torch.Tensor
    last_route: torch.Tensor
    shot_one_route: torch.Tensor
    shot_two_route: torch.Tensor
    durable_outcome: torch.Tensor
    awaiting_second_shot: torch.Tensor
    durable_outcome_updates: torch.Tensor

    def __init__(self, state: Mapping[str, torch.Tensor]) -> None:
        super().__init__()
        normalized = {
            str(name).removeprefix(_LEGACY_PARENT_STATE_PREFIX): value
            for name, value in state.items()
        }
        if len(normalized) != LEGACY_RBO_CAPABILITY_TENSOR_COUNT:
            raise RuntimeError(
                "legacy RBO capability tensor count differs: "
                f"expected={LEGACY_RBO_CAPABILITY_TENSOR_COUNT} "
                f"actual={len(normalized)}"
            )
        if len(normalized) != len(state) or any(
            not isinstance(value, torch.Tensor) for value in normalized.values()
        ):
            raise RuntimeError(
                "legacy RBO capability state has duplicate or non-tensor entries"
            )
        names = tuple(sorted(normalized))
        layer_indices = sorted(
            {
                int(parts[1])
                for name in names
                if (parts := name.split("."))[:1] == ["experts"]
                and len(parts) > 3
                and parts[1].isdigit()
                and parts[2:] == [parts[2], parts[3]]
                and parts[2] == "bank"
                and parts[3] == "gate_up"
            }
        )
        if layer_indices != list(range(len(layer_indices))) or not layer_indices:
            raise RuntimeError("legacy RBO capability expert layers are not contiguous")

        profile_rows: list[torch.Tensor] = []
        hidden_size = 0
        intermediate_size = 0
        route_count = 0
        profile_width = 0
        profile_suffixes = (
            "flat_expert_identity",
            "flat_expert_what_profile",
            "flat_expert_when_profile",
            "flat_expert_how_profile",
            "flat_expert_capability",
        )
        for layer_idx in layer_indices:
            prefix = f"experts.{layer_idx}."
            gate_name = prefix + "bank.gate_up"
            down_name = prefix + "bank.down"
            if gate_name not in normalized or down_name not in normalized:
                raise RuntimeError("legacy RBO capability expert payload is incomplete")
            gate = normalized[gate_name]
            down = normalized[down_name]
            if gate.dim() != 3 or down.dim() != 3:
                raise RuntimeError(
                    "legacy RBO capability experts are not rank-three banks"
                )
            layer_routes = int(gate.shape[0])
            layer_hidden = int(gate.shape[-1])
            layer_intermediate = int(down.shape[-1])
            if tuple(gate.shape) != (
                layer_routes,
                2 * layer_intermediate,
                layer_hidden,
            ) or tuple(down.shape) != (
                layer_routes,
                layer_hidden,
                layer_intermediate,
            ):
                raise RuntimeError(
                    "legacy RBO capability expert geometry is inconsistent"
                )
            if layer_idx == 0:
                hidden_size = layer_hidden
                intermediate_size = layer_intermediate
                route_count = layer_routes
            elif (
                layer_hidden != hidden_size
                or layer_intermediate != intermediate_size
                or layer_routes != route_count
            ):
                raise RuntimeError(
                    "legacy RBO capability expert geometry drifts by layer"
                )
            profile_parts: list[torch.Tensor] = []
            for suffix in profile_suffixes:
                name = prefix + suffix
                value = normalized.get(name)
                if not isinstance(value, torch.Tensor) or value.dim() != 2:
                    raise RuntimeError(
                        f"legacy RBO capability profile is missing or invalid: {name}"
                    )
                if int(value.shape[0]) != route_count:
                    raise RuntimeError(
                        "legacy RBO capability profile route count differs"
                    )
                profile_parts.append(
                    value.detach().to(device="cpu", dtype=torch.float32)
                )
            layer_profile = torch.cat(profile_parts, dim=-1)
            if layer_idx == 0:
                profile_width = int(layer_profile.shape[-1])
            elif int(layer_profile.shape[-1]) != profile_width:
                raise RuntimeError(
                    "legacy RBO capability profile width drifts by layer"
                )
            profile_rows.append(layer_profile)

        classifier = normalized.get("intent_module.classifier.weight")
        feedback_down = normalized.get("feedback_head.down.weight")
        if (
            not isinstance(classifier, torch.Tensor)
            or tuple(classifier.shape[1:]) != (hidden_size,)
            or not isinstance(feedback_down, torch.Tensor)
            or feedback_down.dim() != 2
            or int(feedback_down.shape[1]) != hidden_size
        ):
            raise RuntimeError(
                "legacy RBO capability intent/feedback geometry is invalid"
            )

        self._payload_names = names
        self._payloads = nn.ModuleList(
            [_FrozenLegacyRBOTensor(normalized[name]) for name in names]
        )
        profiles = F.normalize(torch.cat(profile_rows, dim=0), dim=-1, eps=1e-6)
        self.register_buffer("capability_profiles", profiles, persistent=False)
        self.route_query = nn.Linear(hidden_size, profile_width, bias=False)
        self.outcome_query = nn.Linear(
            self.durable_outcome_width,
            profile_width,
            bias=False,
        )
        self.residual_projection = nn.Linear(profile_width, hidden_size, bias=False)
        self.fusion_logit = nn.Parameter(torch.zeros(()))
        signal_weight = normalized.get("feedback_head.signal.weight")
        trauma_values = [
            normalized.get(f"trauma_gate.{layer_idx}") for layer_idx in layer_indices
        ]
        learned_query = torch.cat(
            (
                feedback_down.detach().to(dtype=torch.float32),
                classifier.detach().to(dtype=torch.float32),
            ),
            dim=0,
        )
        if tuple(learned_query.shape) != tuple(self.route_query.weight.shape):
            raise RuntimeError(
                "legacy RBO capability learned query/profile geometry differs"
            )
        if (
            not isinstance(signal_weight, torch.Tensor)
            or signal_weight.dim() != 2
            or tuple(signal_weight.shape) != (4, int(feedback_down.shape[0]))
            or any(
                not isinstance(value, torch.Tensor) or value.numel() != 1
                for value in trauma_values
            )
        ):
            raise RuntimeError("legacy RBO capability feedback/trauma state is invalid")
        with torch.no_grad():
            self.route_query.weight.copy_(learned_query)
            self.residual_projection.weight.copy_(learned_query.transpose(0, 1))
            self.outcome_query.weight.zero_()
            self.outcome_query.weight[
                : int(feedback_down.shape[0]),
                : int(signal_weight.shape[0]),
            ].copy_(signal_weight.detach().to(dtype=torch.float32).transpose(0, 1))
            self.fusion_logit.copy_(
                torch.stack(
                    [
                        value.detach().to(dtype=torch.float32).reshape(())
                        for value in trauma_values
                        if isinstance(value, torch.Tensor)
                    ]
                ).mean()
            )
        self.register_buffer("authority_trained", torch.zeros((), dtype=torch.bool))
        self.register_buffer("retention_passed", torch.zeros((), dtype=torch.bool))
        self.register_buffer(
            "forward_id",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.register_buffer(
            "forward_calls",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.register_buffer(
            "engaged_calls",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.register_buffer(
            "last_delta_l2",
            torch.zeros((), dtype=torch.float32),
            persistent=False,
        )
        self.register_buffer(
            "last_route",
            torch.zeros(profiles.shape[0]),
            persistent=False,
        )
        self.register_buffer(
            "shot_one_route",
            torch.zeros_like(self.last_route),
            persistent=False,
        )
        self.register_buffer(
            "shot_two_route",
            torch.zeros_like(self.last_route),
            persistent=False,
        )
        self.register_buffer(
            "durable_outcome",
            torch.zeros(self.durable_outcome_width),
            persistent=False,
        )
        self.register_buffer(
            "awaiting_second_shot",
            torch.zeros((), dtype=torch.bool),
            persistent=False,
        )
        self.register_buffer(
            "durable_outcome_updates",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.hidden_size = hidden_size
        self.intermediate_size = intermediate_size
        self.n_expert_layers = len(layer_indices)
        self.routes_per_layer = route_count
        self.n_intents = int(classifier.shape[0])
        self.feedback_hidden_size = int(feedback_down.shape[0])
        self.profile_width = profile_width
        self.payload_key_set_sha256 = hashlib.sha256(
            "\n".join(names).encode("utf-8")
        ).hexdigest()
        geometry_rows = [
            (name, tuple(normalized[name].shape), str(normalized[name].dtype))
            for name in names
        ]
        self.payload_geometry_sha256 = hashlib.sha256(
            json.dumps(geometry_rows, separators=(",", ":")).encode("utf-8")
        ).hexdigest()

    def trainable_parameters(self) -> Iterator[nn.Parameter]:
        yield from self.route_query.parameters()
        yield from self.outcome_query.parameters()
        yield from self.residual_projection.parameters()
        yield self.fusion_logit

    def reset_authority_from_boundary(self) -> None:
        with torch.no_grad():
            self.authority_trained.zero_()
            self.retention_passed.zero_()

    def begin_forward(self, forward_id: torch.Tensor) -> None:
        if forward_id.numel() != 1:
            raise RuntimeError("legacy RBO capability forward ID must be scalar")
        with torch.no_grad():
            self.forward_id.copy_(
                forward_id.detach()
                .to(device=self.forward_id.device, dtype=self.forward_id.dtype)
                .reshape(())
            )
            self.forward_calls.zero_()
            self.engaged_calls.zero_()
            self.last_delta_l2.zero_()

    def mark_authority_trained_from_boundary(
        self,
        update_proven: torch.Tensor,
        retention_proven: torch.Tensor,
    ) -> None:
        if update_proven.numel() != 1 or retention_proven.numel() != 1:
            raise RuntimeError(
                "legacy RBO capability authority receipts must be scalar"
            )
        with torch.no_grad():
            self.authority_trained.copy_(update_proven.to(dtype=torch.bool).reshape(()))
            self.retention_passed.copy_(
                retention_proven.to(dtype=torch.bool).reshape(())
            )

    def apply_durable_outcome_state(self, observation: torch.Tensor) -> None:
        """Apply a persisted observable outcome, never a target or gold answer."""

        if observation.numel() != self.durable_outcome_width:
            raise RuntimeError("legacy RBO durable outcome geometry differs")
        with torch.no_grad():
            self.shot_one_route.copy_(self.last_route)
            self.durable_outcome.copy_(
                observation.detach()
                .to(
                    device=self.durable_outcome.device,
                    dtype=self.durable_outcome.dtype,
                )
                .reshape_as(self.durable_outcome)
            )
            self.awaiting_second_shot.fill_(True)
            self.durable_outcome_updates.add_(
                torch.ones_like(self.durable_outcome_updates)
            )

    def capture_session_state(self) -> LegacyRBOCapabilitySessionState:
        return LegacyRBOCapabilitySessionState(
            durable_outcome=self.durable_outcome.detach().clone(),
            shot_one_route=self.shot_one_route.detach().clone(),
            shot_two_route=self.shot_two_route.detach().clone(),
            last_route=self.last_route.detach().clone(),
            awaiting_second_shot=self.awaiting_second_shot.detach().clone(),
            durable_outcome_updates=self.durable_outcome_updates.detach().clone(),
        )

    def reset_session_state(self) -> None:
        with torch.no_grad():
            self.durable_outcome.zero_()
            self.shot_one_route.zero_()
            self.shot_two_route.zero_()
            self.last_route.zero_()
            self.awaiting_second_shot.zero_()
            self.durable_outcome_updates.zero_()

    def hydrate_session_state(self, state: LegacyRBOCapabilitySessionState) -> None:
        pairs = (
            (self.durable_outcome, state.durable_outcome),
            (self.shot_one_route, state.shot_one_route),
            (self.shot_two_route, state.shot_two_route),
            (self.last_route, state.last_route),
            (self.awaiting_second_shot, state.awaiting_second_shot),
            (self.durable_outcome_updates, state.durable_outcome_updates),
        )
        with torch.no_grad():
            for target, source in pairs:
                if tuple(target.shape) != tuple(source.shape):
                    raise RuntimeError("legacy RBO capability session geometry differs")
                target.copy_(source.to(device=target.device, dtype=target.dtype))

    def forward(self, hidden: torch.Tensor) -> torch.Tensor:
        if hidden.shape[-1] != self.hidden_size:
            raise RuntimeError("legacy RBO capability hidden geometry differs")
        with torch.no_grad():
            self.forward_calls.add_(torch.ones_like(self.forward_calls))
        active = self.authority_trained & self.retention_passed
        if not self.training and not bool(active.detach().to(device="cpu")):
            self.last_delta_l2.zero_()
            return hidden
        query_dtype = self.route_query.weight.dtype
        query_hidden = hidden.to(dtype=query_dtype)
        query = self.route_query(query_hidden)
        outcome = self.outcome_query(
            self.durable_outcome.to(device=query.device, dtype=query_dtype)
        )
        query = F.normalize(query + outcome, dim=-1, eps=1e-6)
        profiles = self.capability_profiles.to(
            device=query.device,
            dtype=query_dtype,
        )
        route_logits = torch.einsum("btd,rd->btr", query, profiles)
        route_weights = route_logits.softmax(dim=-1)
        capability = torch.einsum("btr,rd->btd", route_weights, profiles)
        proposed_delta = (
            self.residual_projection(capability) * self.fusion_logit.sigmoid()
        )
        route = route_weights.mean(dim=(0, 1)).detach()
        with torch.no_grad():
            self.last_route.copy_(route.to(device=self.last_route.device))
            _capture_legacy_second_shot_route_(
                last_route=self.last_route,
                shot_two_route=self.shot_two_route,
                awaiting_second_shot=self.awaiting_second_shot,
            )
        if self.training:
            inactive_delta = proposed_delta - proposed_delta.detach()
        else:
            inactive_delta = torch.zeros_like(proposed_delta)
        active_scale = active.to(device=hidden.device, dtype=hidden.dtype)
        delta = active_scale * proposed_delta.to(dtype=hidden.dtype) + (
            1.0 - active_scale
        ) * inactive_delta.to(dtype=hidden.dtype)
        with torch.no_grad():
            self.engaged_calls.add_(active.to(dtype=torch.long))
            self.last_delta_l2.copy_(delta.detach().float().square().sum().sqrt())
        output: torch.Tensor = hidden + delta
        return output

    def engagement_packet(self) -> LegacyRBOCapabilityEngagementPacket:
        return LegacyRBOCapabilityEngagementPacket(
            authority_trained=self.authority_trained.detach().clone(),
            retention_passed=self.retention_passed.detach().clone(),
            forward_id=self.forward_id.detach().clone(),
            forward_calls=self.forward_calls.detach().clone(),
            engaged_calls=self.engaged_calls.detach().clone(),
            delta_l2=self.last_delta_l2.detach().clone(),
            route_id=self.last_route.argmax().detach().clone(),
            shot_one_route=self.shot_one_route.detach().clone(),
            shot_two_route=self.shot_two_route.detach().clone(),
            shot_one_route_id=self.shot_one_route.argmax().detach().clone(),
            shot_two_route_id=self.shot_two_route.argmax().detach().clone(),
            route_shift_l1=(self.shot_two_route - self.shot_one_route)
            .abs()
            .sum()
            .detach()
            .clone(),
            durable_outcome_updates=self.durable_outcome_updates.detach().clone(),
            payload_tensor_count=torch.full_like(
                self.forward_calls,
                len(self._payload_names),
            ),
        )


def _verified_legacy_capability_provenance(cfg: ResynthesisConfig) -> str:
    """Return the launch's code-provenance observation for receipts only."""

    observed = cfg.legacy_capability_source_sha256
    return observed if observed else "unavailable"


def _resolve_path(base_dir: str, filename: str) -> Path:
    p = Path(filename)
    if p.is_absolute():
        return p
    return Path(base_dir) / filename


def load_resynthesis_parent_config(cfg: ResynthesisConfig) -> dict[str, Any]:
    """Load the Resynthesis config.json (boundary I/O — JSON is allowed here)."""
    config_path = _resolve_path(cfg.base_model_dir, cfg.config_path)
    with open(config_path, encoding="utf-8") as f:
        return cast(dict[str, Any], json.load(f))


def load_resynthesis_parent_manifest(cfg: ResynthesisConfig) -> dict[str, Any]:
    """Load the integrated checkpoint manifest at the external I/O boundary."""

    manifest_path = (
        Path(cfg.parent_checkpoint_manifest_path).expanduser().resolve()
    )
    manifest_sha256 = _streamed_file_sha256(manifest_path)
    if (
        cfg.parent_checkpoint_manifest_sha256
        != RESYNTHESIS_PARENT_CHECKPOINT_MANIFEST_SHA256
        or manifest_sha256 != cfg.parent_checkpoint_manifest_sha256
    ):
        raise RuntimeError(
            "Resynthesis parent checkpoint manifest SHA-256 differs"
        )
    with manifest_path.open(encoding="utf-8") as handle:
        manifest = json.load(handle)
    if not isinstance(manifest, dict):
        raise RuntimeError("Resynthesis checkpoint manifest is not a JSON object")
    expected_payload_sha256 = str(manifest.get("manifest_payload_sha256", ""))
    payload = dict(manifest)
    payload.pop("manifest_payload_sha256", None)
    actual_payload_sha256 = hashlib.sha256(
        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    if expected_payload_sha256 != actual_payload_sha256:
        raise RuntimeError("Resynthesis checkpoint manifest payload SHA-256 differs")
    return manifest


def _streamed_file_sha256(path: Path) -> str:
    """Hash an immutable model artifact at the explicit checkpoint boundary."""

    from resynthesis.none_migration import file_sha256_boundary

    return file_sha256_boundary(path)


def _safetensors_parameter_elements(path: Path) -> int:
    """Count exact tensor elements from a safetensors header boundary."""

    with path.open("rb") as handle:
        header_width = int.from_bytes(handle.read(8), byteorder="little")
        if header_width < 2:
            raise RuntimeError("native parent safetensors header is malformed")
        header = json.loads(handle.read(header_width).decode("utf-8"))
    if not isinstance(header, dict):
        raise RuntimeError("native parent safetensors header is not an object")
    parameter_elements = 0
    tensor_count = 0
    for name, record in header.items():
        if name == "__metadata__":
            continue
        if not isinstance(record, dict):
            raise RuntimeError("native parent safetensors tensor record is malformed")
        shape = record.get("shape")
        if not isinstance(shape, list) or not all(
            isinstance(width, int) and not isinstance(width, bool) and width >= 0
            for width in shape
        ):
            raise RuntimeError("native parent safetensors tensor shape is malformed")
        elements = 1
        for width in shape:
            elements *= width
        parameter_elements += elements
        tensor_count += 1
    if tensor_count < 1 or parameter_elements < 1:
        raise RuntimeError("native parent safetensors contains no model tensors")
    return parameter_elements


def _verified_resynthesis_native_parent_manifest(
    cfg: ResynthesisConfig,
    native_root: Path,
    weights_path: Path,
    parameter_elements: int,
    inherited_manifest: Mapping[str, Any],
) -> tuple[str, bool]:
    """Verify exact parent bytes while allowing their public release location."""

    manifest_path = Path(cfg.native_parent_manifest_path).expanduser().resolve()
    manifest_sha256 = _streamed_file_sha256(manifest_path)
    if (
        cfg.native_parent_manifest_sha256 != RESYNTHESIS_NATIVE_PARENT_MANIFEST_SHA256
        or manifest_sha256 != cfg.native_parent_manifest_sha256
        or cfg.native_parent_model_sha256
        != RESYNTHESIS_NATIVE_PARENT_MODEL_SHA256
    ):
        raise RuntimeError("Resynthesis native-parent release identity differs")
    loaded = json.loads(manifest_path.read_text(encoding="utf-8"))
    if not isinstance(loaded, dict):
        raise RuntimeError("Resynthesis native-parent manifest is not an object")
    model_artifact = loaded.get("modelArtifact")
    inherited_lineage = loaded.get("inheritedCheckpointLineage")
    source_runtime = loaded.get("sourceEraRuntime")
    migration = loaded.get("migrationValidation")
    inherited_model_artifact = inherited_manifest.get("model_artifact")
    if not all(
        isinstance(value, dict)
        for value in (
            model_artifact,
            inherited_lineage,
            source_runtime,
            migration,
            inherited_model_artifact,
        )
    ):
        raise RuntimeError("Resynthesis native-parent manifest lacks required records")
    assert isinstance(model_artifact, dict)
    assert isinstance(inherited_lineage, dict)
    assert isinstance(source_runtime, dict)
    assert isinstance(migration, dict)
    assert isinstance(inherited_model_artifact, dict)
    declared_root = Path(str(loaded.get("nativeRoot", ""))).resolve()
    declared_model_path = Path(str(model_artifact.get("path", "")))
    checks = (
        native_root.is_dir(),
        loaded.get("schema") == "nnf.resynthesis.native_parent_generation.v1",
        loaded.get("owner") == "Resynthesis",
        loaded.get("generation") == RESYNTHESIS_NATIVE_PARENT_GENERATION,
        declared_root == Path(RESYNTHESIS_NATIVE_PARENT_ROOT).resolve(),
        loaded.get("externalProductCheckpointDependency") is False,
        declared_model_path == Path("model.safetensors"),
        weights_path.name == declared_model_path.name,
        model_artifact.get("sizeBytes") == weights_path.stat().st_size,
        model_artifact.get("parameterElements") == parameter_elements,
        model_artifact.get("sha256") == cfg.native_parent_model_sha256,
        model_artifact.get("sha256")
        == inherited_model_artifact.get("artifact_sha256"),
        model_artifact.get("tensorCount")
        == inherited_model_artifact.get("tensor_count"),
        model_artifact.get("contentTensorCount")
        == inherited_model_artifact.get("content_tensor_count"),
        inherited_lineage.get("checkpointId")
        == inherited_manifest.get("checkpoint_id"),
        inherited_lineage.get("manifestPayloadSha256")
        == inherited_manifest.get("manifest_payload_sha256"),
        migration.get("independentModelSha256Recomputed") is True,
        migration.get("safetensorsHeaderParameterCountVerified") is True,
        migration.get("promotionEligible") is False,
    )
    if not all(checks):
        raise RuntimeError(
            "Resynthesis native-parent manifest differs from live geometry or lineage"
        )
    return manifest_sha256, bool(migration.get("promotionEligible"))


def _verified_parent_source_hashes(cfg: ResynthesisConfig) -> dict[str, str]:
    """Observe mutable parent source bytes without creating launch authority.

    The historical modules themselves still have to import successfully when
    the parent is loaded.  Hashing them is only version telemetry, however, so
    a concurrent edit or transient diagnostic read failure is recorded in the
    observation instead of pre-empting that real import/load boundary.
    """

    def observe(path: Path) -> str:
        try:
            return _streamed_file_sha256(path) if path.is_file() else "missing"
        except (OSError, RuntimeError) as error:
            return f"unavailable:{type(error).__name__}"

    source_root = Path(cfg.parent_runtime_source_dir).expanduser().resolve()
    fingerprint_path = Path(cfg.parent_source_fingerprint_path).expanduser().resolve()
    runtime_files = tuple(
        f"{_LEGACY_PARENT_PACKAGE_COMPONENT}/{filename}"
        for filename in (
            "additive_moe.py",
            "capability_corpus.py",
            "config.py",
            "experts.py",
            "glyph_vge.py",
            "integrated.py",
            "knowledge_transfer_surfaces.py",
            "loader.py",
            "master_context_orchestration.py",
            "moe_growth_controller.py",
            f"{_legacy_parent_module_name('rbo')}.py",
            "none_fabric.py",
            "rbo_checkpoint.py",
            "rbo_outcome_continual.py",
            "rbo_traversal_tensors.py",
            "training_doctrine.py",
            "swe_agent_system_prompt.txt",
            "tool_schema.json",
        )
    )
    outcome_source_name = (
        f"{_LEGACY_PARENT_PACKAGE_COMPONENT}/rbo_outcome_continual.py"
    )
    actual: dict[str, str] = {}
    for relative_path in runtime_files:
        path = (
            Path(cfg.parent_rbo_outcome_source_path).expanduser().resolve()
            if relative_path == outcome_source_name
            else source_root / relative_path
        )
        actual[relative_path] = observe(path)
    actual["@source_fingerprint_shard"] = observe(fingerprint_path)
    actual["@source_id"] = cfg.parent_source_id
    return actual


def _source_bundle_sha256(source_hashes: Mapping[str, str]) -> str:
    rows = tuple(
        sorted((str(name), str(value)) for name, value in source_hashes.items())
    )
    return hashlib.sha256(
        json.dumps(rows, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def _exec_historical_source_module(
    module_key: str,
    module_spec: importlib.machinery.ModuleSpec,
    module: Any,
) -> None:
    """Execute verified historical source without writing into its artifact."""

    if module_spec.loader is None:
        raise RuntimeError("historical source module has no loader")
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        module_spec.loader.exec_module(module)
    except BaseException:
        sys.modules.pop(module_key, None)
        raise
    finally:
        sys.dont_write_bytecode = previous


def _load_aliased_parent_module(
    alias: str,
    source_root: Path,
    module_name: str,
) -> Any:
    """Import one Resynthesis source generation under an isolated package identity."""

    package_dir = source_root / _LEGACY_PARENT_PACKAGE_COMPONENT
    package_name = alias
    package = sys.modules.get(package_name)
    if package is None:
        if not package_dir.is_dir():
            raise RuntimeError("historical Resynthesis package directory is unavailable")
        # The source-era package initializer imported unrelated mutable modules
        # that were absent from the checkpoint fingerprint. Build an isolated
        # namespace package instead, then import only manifest-verified modules.
        package_spec = importlib.machinery.ModuleSpec(
            package_name,
            loader=None,
            is_package=True,
        )
        package_spec.submodule_search_locations = [str(package_dir)]
        package = importlib.util.module_from_spec(package_spec)
        sys.modules[package_name] = package
    module_key = f"{package_name}.{module_name}"
    existing = sys.modules.get(module_key)
    if existing is not None:
        actual_file = Path(str(getattr(existing, "__file__", ""))).resolve()
        expected_file = (package_dir / f"{module_name}.py").resolve()
        if actual_file != expected_file:
            raise RuntimeError("isolated Resynthesis module alias is already bound elsewhere")
        return existing
    module_path = package_dir / f"{module_name}.py"
    source_loader = importlib.machinery.SourceFileLoader(module_key, str(module_path))
    module_spec = importlib.util.spec_from_loader(module_key, source_loader)
    if module_spec is None or module_spec.loader is None:
        raise RuntimeError(
            f"could not create Resynthesis module specification: {module_path}"
        )
    module = importlib.util.module_from_spec(module_spec)
    sys.modules[module_key] = module
    _exec_historical_source_module(module_key, module_spec, module)
    return module


def _load_aliased_parent_module_from_path(
    alias: str,
    module_name: str,
    module_path: Path,
) -> Any:
    """Install one recovered hash-exact module into an isolated package."""

    module_key = f"{alias}.{module_name}"
    existing = sys.modules.get(module_key)
    if existing is not None:
        if (
            Path(str(getattr(existing, "__file__", ""))).resolve()
            != module_path.resolve()
        ):
            raise RuntimeError(
                "recovered Resynthesis module alias is already bound elsewhere"
            )
        return existing
    source_loader = importlib.machinery.SourceFileLoader(module_key, str(module_path))
    module_spec = importlib.util.spec_from_loader(module_key, source_loader)
    if module_spec is None or module_spec.loader is None:
        raise RuntimeError(
            f"could not create recovered Resynthesis module spec: {module_path}"
        )
    module = importlib.util.module_from_spec(module_spec)
    sys.modules[module_key] = module
    _exec_historical_source_module(module_key, module_spec, module)
    return module


class _FrozenWideIntent(nn.Module):
    """Exact eight-way trained intent geometry embedded in the graph parent."""

    def __init__(self, hidden_size: int = 4096, carry_dim: int = 512) -> None:
        super().__init__()
        self.hidden_size = int(hidden_size)
        self.carry_dim = int(carry_dim)
        self.n_intents = 8
        self.carry_down = nn.Linear(self.hidden_size, self.carry_dim, bias=False)
        self.classifier = nn.Linear(self.carry_dim, self.n_intents)
        self.requires_grad_(False)

    def forward_intent(self, hidden: torch.Tensor) -> torch.Tensor:
        carry = self.carry_down(hidden.to(dtype=self.carry_down.weight.dtype))
        logits: torch.Tensor = self.classifier(carry)
        return logits


def _split_embedded_legacy_rbo_state(
    state: Mapping[str, torch.Tensor],
) -> tuple[dict[str, torch.Tensor], dict[str, torch.Tensor]]:
    additive = {
        name: value
        for name, value in state.items()
        if not name.startswith(_LEGACY_PARENT_STATE_PREFIX)
    }
    legacy = {
        name.removeprefix(_LEGACY_PARENT_STATE_PREFIX): value
        for name, value in state.items()
        if name.startswith(_LEGACY_PARENT_STATE_PREFIX)
    }
    return additive, legacy


def _normalize_legacy_ladder_state(
    state: dict[str, torch.Tensor],
) -> tuple[dict[str, torch.Tensor], bool]:
    """Normalize the one source-era nested ladder namespace."""

    if not state:
        return state, False
    legacy = tuple(name.startswith("ladder.") for name in state)
    if any(legacy) and not all(legacy):
        raise RuntimeError("integrated Resynthesis ladder checkpoint mixes legacy namespaces")
    if not all(legacy):
        return state, False
    normalized = {name.removeprefix("ladder."): value for name, value in state.items()}
    if len(normalized) != len(state):
        raise RuntimeError("integrated Resynthesis ladder namespace migration collided")
    return normalized, True


def _validate_embedded_legacy_rbo_split(
    source: Mapping[str, torch.Tensor],
    additive_state: Mapping[str, torch.Tensor],
    legacy_state: Mapping[str, torch.Tensor],
    *,
    expected_source_tensor_count: int,
) -> str:
    """Validate the one historical additive-embedded RBO capability schema."""

    if len(source) != expected_source_tensor_count:
        raise RuntimeError("integrated additive tensor count differs from its manifest")
    if len(additive_state) + len(legacy_state) != len(source):
        raise RuntimeError("embedded legacy RBO split lost or duplicated tensors")
    if len(legacy_state) != 553:
        raise RuntimeError(
            "embedded legacy RBO capability tensor count is not the exact schema"
        )
    embedded_names = {
        name.removeprefix(_LEGACY_PARENT_STATE_PREFIX)
        for name in source
        if name.startswith(_LEGACY_PARENT_STATE_PREFIX)
    }
    if embedded_names != set(legacy_state):
        raise RuntimeError("embedded legacy RBO capability key ownership differs")
    if not additive_state:
        raise RuntimeError("embedded legacy RBO split left no additive expert tensors")
    return hashlib.sha256("\n".join(sorted(legacy_state)).encode("utf-8")).hexdigest()


def _prepare_empty_legacy_capability_module_slot(model: nn.Module) -> None:
    """Remove Resynthesis's ``None`` placeholder so its official add_module can own it."""

    name = _legacy_parent_attribute("legacy_rbo_capability_bank")
    existing = getattr(model, name, None)
    if existing is not None:
        raise RuntimeError("legacy RBO capability module slot is already occupied")
    if hasattr(model, name):
        delattr(model, name)


def _load_additive_state_into_resident_owner_boundary(
    load_state: Callable[..., None],
    model: Any,
    state: Mapping[str, torch.Tensor],
) -> None:
    """Strictly copy CPU checkpoint tensors once into resident parameters.

    Historical Resynthesis constructs the additive owner beside the already-resident
    parent before invoking this callback.  Supplying that owner's CUDA device
    asks the source-era loader to build a second device state map and then copy
    it again through ``load_state_dict``.  Omitting the staging device retains
    the source tensors and lets PyTorch perform the single required copy into
    each target parameter.  Key/shape checking remains owned by the exact
    historical loader; no checkpoint, dtype, or lineage authority changes.
    """

    load_state(model, state, device=None)


def _load_resynthesis_parent_with_boundary_migration(
    integrated_module: Any,
    load_parent_full: Any,
    checkpoint_dir: str,
    *,
    device: str,
    expected_additive_tensor_count: int,
) -> tuple[Any, ResynthesisBoundaryMigrationReceipt]:
    """Run Resynthesis's loader with a temporary, exact legacy-key boundary adapter."""

    original_prefixed_loader = getattr(
        integrated_module, "_load_prefixed_tensors", None
    )
    ladder_prefix = getattr(integrated_module, "LADDER_PREFIX", None)
    if not callable(original_prefixed_loader) or not isinstance(ladder_prefix, str):
        raise RuntimeError(
            "Resynthesis integrated runtime lacks its checkpoint tensor loader"
        )
    additive_module = importlib.import_module(
        f"{_LEGACY_PARENT_PACKAGE_COMPONENT}.additive_moe"
    )
    original_additive_loader = getattr(
        additive_module, "load_additive_state_dict", None
    )
    split_embedded = getattr(integrated_module, "split_embedded_legacy_rbo_state", None)
    wire_legacy_bank = getattr(
        integrated_module, "wire_legacy_rbo_capability_bank", None
    )
    if not all(
        callable(value)
        for value in (
            original_additive_loader,
            split_embedded,
            wire_legacy_bank,
        )
    ):
        raise RuntimeError(
            "Resynthesis integrated runtime lacks its legacy capability migration"
        )
    ladder_migration_applied = False
    embedded_legacy_rbo_applied = False
    embedded_legacy_rbo_tensor_count = 0
    additive_tensor_count_after_split = 0
    embedded_legacy_rbo_key_set_sha256 = ""

    def compatible_prefixed_loader(
        model_path: str | Path,
        prefix: str,
        target_device: str,
    ) -> dict[str, torch.Tensor]:
        nonlocal ladder_migration_applied
        state = original_prefixed_loader(model_path, prefix, target_device)
        if not isinstance(state, dict) or not all(
            isinstance(name, str) and isinstance(value, torch.Tensor)
            for name, value in state.items()
        ):
            raise RuntimeError(
                "Resynthesis checkpoint tensor loader returned an invalid state"
            )
        if prefix == ladder_prefix:
            state, migrated = _normalize_legacy_ladder_state(state)
            ladder_migration_applied = ladder_migration_applied or migrated
        return cast(dict[str, torch.Tensor], state)

    def compatible_additive_loader(
        model: Any,
        state: Mapping[str, torch.Tensor],
        *,
        device: str | torch.device | None = None,
    ) -> None:
        nonlocal embedded_legacy_rbo_applied
        nonlocal embedded_legacy_rbo_tensor_count
        nonlocal additive_tensor_count_after_split
        nonlocal embedded_legacy_rbo_key_set_sha256
        assert callable(original_additive_loader)
        assert callable(split_embedded)
        assert callable(wire_legacy_bank)
        split = split_embedded(state)
        additive_state = getattr(split, "additive_state", None)
        legacy_state = getattr(split, "legacy_state", None)
        if not isinstance(additive_state, Mapping) or not isinstance(
            legacy_state, Mapping
        ):
            raise RuntimeError(
                "Resynthesis legacy capability splitter returned an invalid packet"
            )
        if not legacy_state:
            original_additive_loader(model, state, device=device)
            return
        if embedded_legacy_rbo_applied:
            raise RuntimeError(
                "embedded legacy RBO capability migration ran more than once"
            )
        embedded_legacy_rbo_key_set_sha256 = _validate_embedded_legacy_rbo_split(
            state,
            additive_state,
            legacy_state,
            expected_source_tensor_count=expected_additive_tensor_count,
        )
        # ``wire_additive_moe`` has already constructed the additive modules
        # on the selected device.  Passing that device into the historical
        # loader first materializes an entire second GPU-resident state map,
        # then ``load_state_dict`` copies the same tensors again into those
        # resident parameters.  Preserve the exact key/shape validation while
        # letting PyTorch perform the one required CPU-to-resident copy during
        # strict state loading.  This is a load-boundary optimization only;
        # it does not alter parent bytes, routing, trainability, or lineage.
        _load_additive_state_into_resident_owner_boundary(
            original_additive_loader,
            model,
            additive_state,
        )
        if not isinstance(model, nn.Module):
            raise RuntimeError("Resynthesis additive owner is not an nn.Module")
        _prepare_empty_legacy_capability_module_slot(model)
        bank = wire_legacy_bank(model, legacy_state)
        if not isinstance(bank, nn.Module):
            raise RuntimeError("embedded legacy RBO capability bank was not attached")
        embedded_legacy_rbo_applied = True
        embedded_legacy_rbo_tensor_count = len(legacy_state)
        additive_tensor_count_after_split = len(additive_state)

    setattr(integrated_module, "_load_prefixed_tensors", compatible_prefixed_loader)
    setattr(additive_module, "load_additive_state_dict", compatible_additive_loader)
    try:
        loaded = load_parent_full(
            checkpoint_dir,
            device=device,
            dtype=torch.bfloat16,
        )
    finally:
        setattr(integrated_module, "_load_prefixed_tensors", original_prefixed_loader)
        setattr(additive_module, "load_additive_state_dict", original_additive_loader)
    return loaded, ResynthesisBoundaryMigrationReceipt(
        ladder_namespace_applied=ladder_migration_applied,
        embedded_legacy_rbo_applied=embedded_legacy_rbo_applied,
        embedded_legacy_rbo_tensor_count=embedded_legacy_rbo_tensor_count,
        additive_tensor_count_after_split=additive_tensor_count_after_split,
        embedded_legacy_rbo_key_set_sha256=embedded_legacy_rbo_key_set_sha256,
    )


@contextmanager
def _relocated_parent_checkpoint_view(
    cfg: ResynthesisConfig,
) -> Iterator[Path]:
    """Expose separated release artifacts as one read-only checkpoint view.

    The source-era constructor expects the model, tokenizer, and configuration
    under one directory. Public releases keep the multi-gigabyte model in the
    weights tree and lexical artifacts in the tokenizer tree. A temporary
    symlink view preserves those single physical files; it is created only
    after the exact manifests and model identity have been verified.
    """

    native_root = Path(cfg.base_model_dir).expanduser().resolve()
    weights_path = _resolve_path(cfg.base_model_dir, cfg.base_weights).resolve()
    config_path = _resolve_path(cfg.base_model_dir, cfg.config_path).resolve()
    tokenizer_path = _resolve_path(cfg.base_model_dir, cfg.tokenizer_path).resolve()
    checkpoint_manifest_path = Path(
        cfg.parent_checkpoint_manifest_path
    ).expanduser().resolve()
    expected_paths = {
        "model.safetensors": weights_path,
        "config.json": config_path,
        "tokenizer.json": tokenizer_path,
        f"{_LEGACY_PARENT_PACKAGE_COMPONENT}_manifest.json": (
            checkpoint_manifest_path
        ),
    }
    if all((native_root / name).resolve() == path for name, path in expected_paths.items()):
        yield native_root
        return
    if not native_root.is_dir() or any(
        not path.is_file() for path in expected_paths.values()
    ):
        raise RuntimeError(
            "relocated Resynthesis parent release is missing a required artifact"
        )

    with tempfile.TemporaryDirectory(
        prefix="resynthesis-native-parent-view-"
    ) as temporary:
        view = Path(temporary)
        for source in native_root.iterdir():
            if source.name in expected_paths:
                continue
            (view / source.name).symlink_to(
                source,
                target_is_directory=source.is_dir(),
            )
        for name, source in expected_paths.items():
            (view / name).symlink_to(source)
        yield view


def _load_exact_resynthesis_parent(
    cfg: ResynthesisConfig,
    *,
    device: str,
    expected_rbo_tensor_count: int,
    expected_additive_tensor_count: int,
) -> tuple[Any, ResynthesisBoundaryMigrationReceipt, nn.Module, str]:
    """Load the July 13 graph parent with its exact source-era architecture.

    The current rolling Resynthesis source has additional trained-authority tensors
    and cannot strictly represent this artifact.  This path imports the source
    snapshot under an isolated package name, reconstructs the checkpoint's
    eight-way wide-intent/21-layer RBO exactly, and rejects every missing,
    extra, or shape-different authoritative tensor.
    """

    source_root = Path(cfg.parent_runtime_source_dir).expanduser().resolve()
    alias = "_resynthesis_parent_artifact_20260713"
    _load_aliased_parent_module(alias, source_root, "config")
    _load_aliased_parent_module_from_path(
        alias,
        "rbo_outcome_continual",
        Path(cfg.parent_rbo_outcome_source_path).expanduser().resolve(),
    )
    integrated_module = _load_aliased_parent_module(alias, source_root, "integrated")
    additive_module = _load_aliased_parent_module(alias, source_root, "additive_moe")
    rbo_module = _load_aliased_parent_module(
        alias,
        source_root,
        _legacy_parent_module_name("rbo"),
    )
    checkpoint_module = _load_aliased_parent_module(alias, source_root, "rbo_checkpoint")
    experts_module = _load_aliased_parent_module(alias, source_root, "experts")
    transfer_module = _load_aliased_parent_module(
        alias,
        source_root,
        "knowledge_transfer_surfaces",
    )
    traversal_module = _load_aliased_parent_module(
        alias,
        source_root,
        "rbo_traversal_tensors",
    )
    hotpath_id = _install_historical_tensor_native_hotpaths(
        rbo_module,
        experts_module,
        transfer_module,
        traversal_module,
    )
    if hotpath_id != HISTORICAL_TENSOR_NATIVE_HOTPATH_ID:
        raise RuntimeError("historical Resynthesis tensor-native hot-path identity differs")

    load_promoted = getattr(
        integrated_module,
        f"load_{_LEGACY_PARENT_PACKAGE_COMPONENT}_promoted",
        None,
    )
    original_prefixed_loader = getattr(
        integrated_module, "_load_prefixed_tensors", None
    )
    legacy_integrated_rbo_loader_name = (
        f"_load_integrated_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo"
    )
    original_rbo_loader = getattr(
        integrated_module,
        legacy_integrated_rbo_loader_name,
        None,
    )
    original_additive_loader = getattr(
        additive_module, "load_additive_state_dict", None
    )
    original_wire_additive_moe = getattr(additive_module, "wire_additive_moe", None)
    additive_lm_head = getattr(additive_module, "additive_lm_head", None)
    ladder_prefix = getattr(integrated_module, "LADDER_PREFIX", None)
    rbo_prefix = getattr(
        integrated_module,
        _legacy_parent_upper_name("RBO_PREFIX"),
        None,
    )
    build_parent_rbo = getattr(
        rbo_module,
        f"build_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo",
        None,
    )
    rbo_config_type = getattr(
        rbo_module,
        _legacy_parent_class_name("RBOConfig"),
        None,
    )
    detect_layers = getattr(checkpoint_module, "detect_expert_layers", None)
    detect_intermediate = getattr(checkpoint_module, "detect_moe_intermediate", None)
    wire_parent_rbo = getattr(
        additive_module,
        f"wire_{_LEGACY_PARENT_PACKAGE_COMPONENT}_rbo",
        None,
    )
    required = (
        load_promoted,
        original_prefixed_loader,
        original_rbo_loader,
        original_additive_loader,
        original_wire_additive_moe,
        additive_lm_head,
        build_parent_rbo,
        rbo_config_type,
        detect_layers,
        detect_intermediate,
        wire_parent_rbo,
    )
    if not all(callable(value) for value in required):
        raise RuntimeError("historical Resynthesis runtime lacks an exact graph constructor")
    if not isinstance(ladder_prefix, str) or not isinstance(rbo_prefix, str):
        raise RuntimeError("historical Resynthesis runtime prefixes are unavailable")

    ladder_migration_applied = False
    authoritative_rbo_tensor_count = 0
    authoritative_rbo_key_set_sha256 = ""
    additive_tensor_count_after_split = 0
    embedded_legacy_rbo_key_set_sha256 = ""
    legacy_state: dict[str, torch.Tensor] = {}
    skipped_additive_initializer_targets: list[torch.Tensor] = []
    additive_wire_invocations = 0

    def compatible_prefixed_loader(
        model_path: str | Path,
        prefix: str,
        target_device: str,
    ) -> dict[str, torch.Tensor]:
        nonlocal ladder_migration_applied
        assert callable(original_prefixed_loader)
        state = original_prefixed_loader(model_path, prefix, target_device)
        if not isinstance(state, dict) or not all(
            isinstance(name, str) and isinstance(value, torch.Tensor)
            for name, value in state.items()
        ):
            raise RuntimeError("historical Resynthesis tensor loader returned invalid state")
        if prefix == ladder_prefix:
            state, migrated = _normalize_legacy_ladder_state(state)
            ladder_migration_applied = ladder_migration_applied or migrated
        return cast(dict[str, torch.Tensor], state)

    def exact_rbo_loader(
        model: Any,
        *,
        model_path: Path,
        cfg: Any,
        device: str,
        dtype: torch.dtype,
    ) -> nn.Module:
        nonlocal authoritative_rbo_tensor_count
        nonlocal authoritative_rbo_key_set_sha256
        state = compatible_prefixed_loader(model_path, rbo_prefix, "cpu")
        if len(state) != expected_rbo_tensor_count:
            raise RuntimeError(
                "historical authoritative RBO tensor count differs: "
                f"expected={expected_rbo_tensor_count} actual={len(state)}"
            )
        if not any(name.startswith("none_fabric.") for name in state):
            raise RuntimeError(
                "historical authoritative RBO has no model-owned NoNE Fabric"
            )
        assert callable(detect_intermediate)
        assert callable(detect_layers)
        assert callable(rbo_config_type)
        assert callable(build_parent_rbo)
        cfg.moe_intermediate_size = detect_intermediate(state)
        rbo_cfg = rbo_config_type(
            hidden_size=int(cfg.hidden_size),
            n_intents=8,
            n_expert_layers=detect_layers(state),
            feedback_hidden_size=128,
        )
        intent = _FrozenWideIntent(hidden_size=int(cfg.hidden_size), carry_dim=512)
        def construct_exact_rbo() -> nn.Module:
            exact_rbo = cast(
                nn.Module,
                build_parent_rbo(cfg, rbo_cfg, intent_module=intent),
            )
            exact_rbo.register_buffer(
                "_resynthesis_recursive_arm_exhausted",
                torch.zeros((), dtype=torch.bool),
                persistent=False,
            )
            return exact_rbo

        rbo = _build_exact_historical_rbo_from_state(
            construct_exact_rbo,
            state,
            device=device,
            dtype=dtype,
        )
        assert callable(wire_parent_rbo)
        wire_parent_rbo(model, rbo)
        rbo.requires_grad_(False)
        rbo.eval()
        model.resynthesis_rbo = rbo
        model.resynthesis_rbo_packaging = "integrated_exact_historical_source"
        model.resynthesis_none_architecture_active = True
        authoritative_rbo_tensor_count = len(state)
        authoritative_rbo_key_set_sha256 = hashlib.sha256(
            "\n".join(sorted(state)).encode("utf-8")
        ).hexdigest()
        return rbo

    def compatible_additive_loader(
        model: Any,
        state: Mapping[str, torch.Tensor],
        *,
        device: str | torch.device | None = None,
    ) -> None:
        nonlocal additive_tensor_count_after_split
        nonlocal embedded_legacy_rbo_key_set_sha256
        nonlocal legacy_state
        assert callable(original_additive_loader)
        additive_state, captured_legacy = _split_embedded_legacy_rbo_state(state)
        embedded_legacy_rbo_key_set_sha256 = _validate_embedded_legacy_rbo_split(
            state,
            additive_state,
            captured_legacy,
            expected_source_tensor_count=expected_additive_tensor_count,
        )
        if additive_wire_invocations != 1:
            raise RuntimeError(
                "historical additive checkpoint load did not follow one exact wire"
            )
        assert callable(additive_lm_head)
        head = additive_lm_head(model)
        if not isinstance(head, nn.Module):
            raise RuntimeError("historical additive loader found no additive head")
        _validate_historical_additive_initializer_targets(
            head,
            additive_state,
            skipped_additive_initializer_targets,
        )
        _load_additive_state_into_resident_owner_boundary(
            original_additive_loader,
            model,
            additive_state,
        )
        additive_tensor_count_after_split = len(additive_state)
        legacy_state = captured_legacy

    def compatible_wire_additive_moe(
        model: Any,
        cfg: Any,
        n_layers: int = 2,
    ) -> Any:
        nonlocal additive_wire_invocations
        additive_wire_invocations += 1
        if additive_wire_invocations != 1:
            raise RuntimeError("historical additive owner was wired more than once")
        assert callable(original_wire_additive_moe)
        assert callable(additive_lm_head)
        return _wire_exact_historical_additive_without_overwritten_initializers(
            original_wire_additive_moe,
            additive_lm_head,
            model,
            cfg,
            n_layers=n_layers,
            skipped_persistent_targets=skipped_additive_initializer_targets,
        )

    setattr(integrated_module, "_load_prefixed_tensors", compatible_prefixed_loader)
    setattr(integrated_module, legacy_integrated_rbo_loader_name, exact_rbo_loader)
    setattr(additive_module, "load_additive_state_dict", compatible_additive_loader)
    setattr(additive_module, "wire_additive_moe", compatible_wire_additive_moe)
    try:
        assert callable(load_promoted)
        with _relocated_parent_checkpoint_view(cfg) as checkpoint_view:
            loaded = _call_without_transformers_allocator_warmup(
                load_promoted,
                str(checkpoint_view),
                device=device,
                dtype=torch.bfloat16,
            )
    finally:
        setattr(additive_module, "wire_additive_moe", original_wire_additive_moe)
        setattr(integrated_module, "_load_prefixed_tensors", original_prefixed_loader)
        setattr(
            integrated_module,
            legacy_integrated_rbo_loader_name,
            original_rbo_loader,
        )
        setattr(additive_module, "load_additive_state_dict", original_additive_loader)

    runtime = loaded[0] if isinstance(loaded, tuple) and loaded else None
    if not isinstance(runtime, nn.Module):
        raise RuntimeError("historical Resynthesis loader returned no runtime module")
    vocabulary_graph = loaded[1] if isinstance(loaded, tuple) and len(loaded) > 1 else None
    _bind_resynthesis_parent_runtime_facade(
        runtime,
        vocabulary_graph=vocabulary_graph,
    )
    final_hidden_hotpath_id = _install_parent_final_hidden_only_forward_boundary(
        runtime
    )
    if final_hidden_hotpath_id != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
        raise RuntimeError("historical parent final-hidden hot-path identity differs")
    if len(legacy_state) != LEGACY_RBO_CAPABILITY_TENSOR_COUNT:
        raise RuntimeError(
            "historical embedded legacy capability payload was not captured"
        )
    capability_source_sha256 = _verified_legacy_capability_provenance(cfg)
    bank = LegacyRBOCapabilityBank(legacy_state)
    receipt = ResynthesisBoundaryMigrationReceipt(
        ladder_namespace_applied=ladder_migration_applied,
        embedded_legacy_rbo_applied=True,
        embedded_legacy_rbo_tensor_count=len(legacy_state),
        additive_tensor_count_after_split=additive_tensor_count_after_split,
        embedded_legacy_rbo_key_set_sha256=embedded_legacy_rbo_key_set_sha256,
        authoritative_rbo_tensor_count=authoritative_rbo_tensor_count,
        authoritative_rbo_key_set_sha256=authoritative_rbo_key_set_sha256,
        exact_historical_source_loaded=True,
    )
    return loaded, receipt, bank, capability_source_sha256


def resynthesis_parent_info(cfg: ResynthesisConfig) -> ResynthesisParentInfo:
    """Extract typed base info without loading weights (cheap, for validation)."""
    raw = load_resynthesis_parent_config(cfg)
    manifest = load_resynthesis_parent_manifest(cfg)
    parent_source_hashes = _verified_parent_source_hashes(cfg)
    observed_parent_source_bundle_sha256 = _source_bundle_sha256(
        parent_source_hashes
    )
    parent_source_bundle_sha256 = cfg.parent_source_bundle_sha256
    capability_source_sha256 = _verified_legacy_capability_provenance(cfg)
    text_cfg = raw.get("text_config", raw)
    none_rbo = manifest.get("none_rbo")
    additive_moe = manifest.get("additive_moe")
    native_decode_confidence = manifest.get("native_decode_confidence")
    model_artifact = manifest.get("model_artifact")
    if (
        not isinstance(none_rbo, dict)
        or not isinstance(additive_moe, dict)
        or not isinstance(native_decode_confidence, dict)
        or not isinstance(model_artifact, dict)
    ):
        raise RuntimeError(
            "Resynthesis checkpoint lacks integrated NoNE/RBO, adaptive experts, or native confidence"
        )
    artifact_composition = manifest.get("composition")
    artifact_checkpoint_id = manifest.get("checkpoint_id")
    if (
        artifact_composition != _LEGACY_PARENT_COMPOSITION
        or not isinstance(artifact_checkpoint_id, str)
        or not artifact_checkpoint_id
    ):
        raise RuntimeError(
            "Resynthesis parent artifact identity differs from its sealed lineage"
        )
    weights_path = _resolve_path(cfg.base_model_dir, cfg.base_weights)
    resolved_native_root = Path(cfg.base_model_dir).expanduser().resolve()
    parameter_elements = _safetensors_parameter_elements(weights_path)
    native_manifest_sha256, migration_promotion_eligible = (
        _verified_resynthesis_native_parent_manifest(
            cfg,
            resolved_native_root,
            weights_path,
            parameter_elements,
            manifest,
        )
    )
    return ResynthesisParentInfo(
        hidden_size=int(
            text_cfg.get("hidden_size", RESYNTHESIS_HIDDEN_SIZE)
        ),
        num_hidden_layers=int(
            text_cfg.get(
                "num_hidden_layers", RESYNTHESIS_NUM_HIDDEN_LAYERS
            )
        ),
        vocab_size=int(
            text_cfg.get("vocab_size", RESYNTHESIS_PROJECTION_VOCAB_SIZE)
        ),
        max_position_embeddings=max(
            int(
                text_cfg.get(
                    "max_position_embeddings",
                    RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
                )
            ),
            RESYNTHESIS_MAX_POSITION_EMBEDDINGS,
        ),
        num_attention_heads=int(
            text_cfg.get(
                "num_attention_heads", RESYNTHESIS_NUM_ATTENTION_HEADS
            )
        ),
        num_key_value_heads=int(
            text_cfg.get(
                "num_key_value_heads",
                RESYNTHESIS_NUM_KEY_VALUE_HEADS,
            )
        ),
        intermediate_size=int(
            text_cfg.get(
                "intermediate_size", RESYNTHESIS_INTERMEDIATE_SIZE
            )
        ),
        head_dim=int(text_cfg.get("head_dim", RESYNTHESIS_HEAD_DIM)),
        weights_path=str(weights_path),
        config_path=str(_resolve_path(cfg.base_model_dir, cfg.config_path)),
        baseline_frozen=True,
        tie_word_embeddings=bool(raw.get("tie_word_embeddings", False)),
        model_type=_RESYNTHESIS_PARENT_MODEL_TYPE,
        composition=_RESYNTHESIS_PARENT_COMPOSITION,
        integrated_rbo_tensors=int(none_rbo.get("n_tensors", 0)),
        integrated_additive_tensors=int(additive_moe.get("n_tensors", 0)),
        integrated_native_decode_confidence_tensors=int(
            native_decode_confidence.get("n_tensors", 0)
        ),
        checkpoint_id=(
            "resynthesis-native-parent:"
            f"{str(model_artifact.get('artifact_sha256', ''))}"
        ),
        manifest_payload_sha256=str(manifest.get("manifest_payload_sha256", "")),
        model_artifact_sha256=str(model_artifact.get("artifact_sha256", "")),
        parameter_elements=parameter_elements,
        native_owner="Resynthesis",
        native_generation=RESYNTHESIS_NATIVE_PARENT_GENERATION,
        native_root=str(resolved_native_root),
        native_manifest_path=str(
            Path(cfg.native_parent_manifest_path).expanduser().resolve()
        ),
        native_manifest_sha256=native_manifest_sha256,
        native_migration_promotion_eligible=(migration_promotion_eligible),
        parent_source_bundle_sha256=parent_source_bundle_sha256,
        observed_parent_source_bundle_sha256=(
            observed_parent_source_bundle_sha256
        ),
        parent_source_bundle_matches_expected=(
            observed_parent_source_bundle_sha256
            == parent_source_bundle_sha256
        ),
        legacy_capability_source_sha256=capability_source_sha256,
        historical_inherited_checkpoint_id=artifact_checkpoint_id,
        historical_inherited_composition=artifact_composition,
        historical_inherited_model_type=str(
            raw.get("model_type", text_cfg.get("model_type", ""))
        ),
    )

def _build_native_hybrid_context_cache(runtime: nn.Module) -> Any:
    """Build the parent's bounded hot-KV plus global recurrent cache.

    This is a storage-boundary adapter for the immutable parent architecture.
    It does not select an answer or route: all input tokens still execute every
    decoder layer. The 24 trained linear-attention layers retain their global
    recurrent state, while the eight full-attention layers retain one native
    Dual-Chunk local window of hot K/V.
    """

    runtime_config = getattr(runtime, "config", None)
    get_text_config = getattr(runtime_config, "get_text_config", None)
    if not callable(get_text_config):
        raise RuntimeError(
            "Resynthesis runtime exposes no text configuration for native hybrid K/V"
        )
    text_config = get_text_config(decoder=True)
    layer_types = getattr(text_config, "layer_types", None)
    if (
        not isinstance(layer_types, (list, tuple))
        or len(layer_types) != RESYNTHESIS_NUM_HIDDEN_LAYERS
        or not all(isinstance(layer_type, str) for layer_type in layer_types)
    ):
        raise RuntimeError("Resynthesis native hybrid K/V layer geometry differs")
    linear_layers = sum(layer_type == "linear_attention" for layer_type in layer_types)
    full_layers = sum(layer_type == "full_attention" for layer_type in layer_types)
    if (
        linear_layers != RESYNTHESIS_LINEAR_ATTENTION_LAYERS
        or full_layers != RESYNTHESIS_FULL_ATTENTION_LAYERS
        or linear_layers + full_layers != len(layer_types)
    ):
        raise RuntimeError(
            "Resynthesis native hybrid K/V does not match the trained parent topology"
        )

    from transformers.cache_utils import (
        Cache,
        DynamicSlidingWindowLayer,
        LinearAttentionLayer,
    )

    cache_layers: list[Any] = []
    for layer_type in layer_types:
        if layer_type == "linear_attention":
            cache_layers.append(cast(Any, LinearAttentionLayer)())
        else:
            cache_layers.append(
                DynamicSlidingWindowLayer(
                    sliding_window=RESYNTHESIS_NATIVE_HOT_KV_TOKENS
                )
            )
    cache = Cache(layers=cache_layers)
    setattr(cache, "_resynthesis_native_hybrid_context", True)
    setattr(
        cache,
        "_resynthesis_recurrent_layer_count",
        RESYNTHESIS_LINEAR_ATTENTION_LAYERS,
    )
    setattr(
        cache,
        "_resynthesis_full_attention_layer_count",
        RESYNTHESIS_FULL_ATTENTION_LAYERS,
    )
    setattr(
        cache,
        "_resynthesis_hot_window_tokens",
        RESYNTHESIS_NATIVE_HOT_KV_TOKENS,
    )
    return cache


def _expand_immutable_native_hybrid_context_cache_boundary(
    master_cache: object,
    batch_indices_t: torch.Tensor,
) -> Any:
    """Clone one sealed parent prefix cache and expand only its isolated copy.

    The parent prefix is immutable shared evidence.  ``batch_indices_t`` is a
    tensor-native all-zero reorder index: each output row selects the one
    authoritative prefix row.  Transformers does not implement
    ``batch_repeat_interleave`` for ``LinearAttentionLayer``, while
    ``reorder_cache`` supports both that recurrent layer and
    ``DynamicSlidingWindowLayer``.  Deep-copying before the reorder preserves
    the master tensors and every cache-layer bookkeeping field.
    """

    from transformers.cache_utils import (
        Cache,
        DynamicSlidingWindowLayer,
        LinearAttentionLayer,
    )

    if (
        not isinstance(master_cache, Cache)
        or getattr(master_cache, "_resynthesis_native_hybrid_context", False)
        is not True
    ):
        raise RuntimeError(
            "immutable prefix expansion requires the native hybrid context cache"
        )
    if (
        batch_indices_t.ndim != 1
        or batch_indices_t.dtype != torch.long
        or batch_indices_t.numel() < 1
    ):
        raise RuntimeError(
            "immutable prefix expansion requires a nonempty rank-one long index"
        )
    torch._assert_async(
        batch_indices_t.eq(0).all(),
        "immutable prefix expansion index must select only the sealed prefix row",
    )

    layers = master_cache.layers
    if (
        not isinstance(layers, list)
        or len(layers) != RESYNTHESIS_NUM_HIDDEN_LAYERS
        or sum(isinstance(layer, LinearAttentionLayer) for layer in layers)
        != RESYNTHESIS_LINEAR_ATTENTION_LAYERS
        or sum(isinstance(layer, DynamicSlidingWindowLayer) for layer in layers)
        != RESYNTHESIS_FULL_ATTENTION_LAYERS
    ):
        raise RuntimeError("immutable prefix cache topology differs")

    for layer in layers:
        if isinstance(layer, LinearAttentionLayer):
            for state_t in (layer.conv_states, layer.recurrent_states):
                if state_t is not None and (
                    state_t.ndim < 1 or state_t.shape[0] != 1
                ):
                    raise RuntimeError(
                        "immutable recurrent prefix cache must contain one row"
                    )
        elif isinstance(layer, DynamicSlidingWindowLayer):
            for state_t in (layer.keys, layer.values):
                if state_t is not None and (
                    state_t.ndim < 1 or state_t.shape[0] != 1
                ):
                    raise RuntimeError(
                        "immutable sliding-window prefix cache must contain one row"
                    )

    expanded_cache = copy.deepcopy(master_cache)
    expanded_cache.reorder_cache(cast(torch.LongTensor, batch_indices_t))
    return expanded_cache


def _native_context_cache_telemetry_boundary(
    cache: object,
    *,
    device: torch.device,
) -> NativeContextCacheTelemetry:
    """Observe cache storage geometry without influencing model decisions."""

    packet = NativeContextCacheTelemetry(
        hybrid_active=torch.empty((), device=device, dtype=torch.bool),
        total_positions=torch.empty((), device=device, dtype=torch.long),
        hot_resident_positions=torch.empty((), device=device, dtype=torch.long),
        hot_window_tokens=torch.empty((), device=device, dtype=torch.long),
        recurrent_layer_count=torch.empty((), device=device, dtype=torch.long),
        full_attention_layer_count=torch.empty(
            (),
            device=device,
            dtype=torch.long,
        ),
    )
    _update_native_context_cache_telemetry_boundary(cache, packet=packet)
    return packet


def _update_native_context_cache_telemetry_boundary(
    cache: object,
    *,
    packet: NativeContextCacheTelemetry,
) -> None:
    """Fill preallocated cache telemetry tensors without per-wave allocation."""

    hybrid_active = bool(getattr(cache, "_resynthesis_native_hybrid_context", False))
    recurrent_layers = int(getattr(cache, "_resynthesis_recurrent_layer_count", 0))
    full_layers = int(getattr(cache, "_resynthesis_full_attention_layer_count", 0))
    hot_window = int(getattr(cache, "_resynthesis_hot_window_tokens", 0))
    total_positions = 0
    get_seq_length = getattr(cache, "get_seq_length", None)
    if callable(get_seq_length):
        total_positions = int(get_seq_length())
    elif isinstance(getattr(cache, "positions", None), int):
        total_positions = int(getattr(cache, "positions"))

    hot_resident_positions = 0
    layers = getattr(cache, "layers", None)
    if isinstance(layers, list):
        for layer in layers:
            keys = getattr(layer, "keys", None)
            if isinstance(keys, torch.Tensor) and keys.ndim >= 2:
                hot_resident_positions = max(
                    hot_resident_positions,
                    int(keys.shape[-2]),
                )
    elif hybrid_active:
        hot_resident_positions = min(total_positions, hot_window)

    with torch.no_grad():
        packet.hybrid_active.fill_(hybrid_active)
        packet.total_positions.fill_(total_positions)
        packet.hot_resident_positions.fill_(hot_resident_positions)
        packet.hot_window_tokens.fill_(hot_window)
        packet.recurrent_layer_count.fill_(recurrent_layers)
        packet.full_attention_layer_count.fill_(full_layers)


class ResynthesisNativeParent(nn.Module):
    """Frozen Resynthesis-owned parent exposing trained hidden/logit surfaces.

    The base weights are loaded and frozen (requires_grad=False). The forward
    pass produces the final hidden states [batch, seq, hidden_size] that the
    additive science stack + RBO operate on.

    For training, the base is a feature extractor: gradients flow only through
    the additive science layers, not the frozen base (Pillar: additive-only,
    baseline-frozen).
    """

    _base_loaded_marker: torch.Tensor
    _decode_cached_positions: torch.Tensor
    _native_context_full_attention_layer_count: torch.Tensor
    _native_context_hot_resident_positions: torch.Tensor
    _native_context_hot_window_tokens: torch.Tensor
    _native_context_hybrid_active: torch.Tensor
    _native_context_recurrent_layer_count: torch.Tensor
    _native_context_total_positions: torch.Tensor
    _parent_inference_session_buffers_normalized: torch.Tensor
    _training_low_rank_projection_cache: torch.Tensor | None
    _training_low_rank_projection_source_id: int | None
    _training_low_rank_projection_source_version: int | None
    _last_frozen_backbone_final_hidden_t: torch.Tensor | None
    _last_frozen_backbone_summary_hidden_t: torch.Tensor | None
    _last_frozen_backbone_input_positions_t: torch.Tensor | None

    def __init__(
        self, cfg: ResynthesisConfig, *, device: torch.device | str = "cpu"
    ) -> None:
        super().__init__()
        self.cfg = cfg
        self.device = torch.device(device)
        self.info = resynthesis_parent_info(cfg)
        self._weights_loaded = False
        self.runtime: nn.Module | None = None
        self.tokenizer: Any | None = None
        self.tokenizer_backend_identity: ResynthesisTokenizerIdentity | None = None
        self._last_forward: ResynthesisParentForward | None = None
        self._decode_past_key_values: Any | None = None
        self._decode_authority_active = False
        self._parent_session_may_hold_inference_tensors = False
        self._parent_session_buffers_verified_normal = False
        self._training_low_rank_projection_cache = None
        self._training_low_rank_projection_source_id = None
        self._training_low_rank_projection_source_version = None
        self._last_frozen_backbone_final_hidden_t = None
        self._last_frozen_backbone_summary_hidden_t = None
        self._last_frozen_backbone_input_positions_t = None
        self.verified_model_artifact_sha256 = ""
        self.ladder_namespace_migration_applied = False
        self.embedded_legacy_rbo_migration_applied = False
        self.embedded_legacy_rbo_tensor_count = 0
        self.additive_tensor_count_after_legacy_split = 0
        self.embedded_legacy_rbo_key_set_sha256 = ""
        self.authoritative_rbo_tensor_count = 0
        self.authoritative_rbo_key_set_sha256 = ""
        self.exact_historical_source_loaded = False
        self.historical_tensor_native_hotpath_id = ""
        self.historical_recursive_arm_exhaustion_id = ""
        self.legacy_capability_source_sha256 = ""
        object.__setattr__(self, "_pending_legacy_capability_bank", None)
        self.register_buffer(
            "_base_loaded_marker", torch.tensor([0], dtype=torch.int8), persistent=False
        )
        self.register_buffer(
            "_decode_cached_positions",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.register_buffer(
            "_parent_inference_session_buffers_normalized",
            torch.zeros((), dtype=torch.long),
            persistent=False,
        )
        self.register_buffer(
            "_native_context_hybrid_active",
            torch.zeros((), dtype=torch.bool),
            persistent=False,
        )
        for buffer_name in (
            "_native_context_total_positions",
            "_native_context_hot_resident_positions",
            "_native_context_hot_window_tokens",
            "_native_context_recurrent_layer_count",
            "_native_context_full_attention_layer_count",
        ):
            self.register_buffer(
                buffer_name,
                torch.zeros((), dtype=torch.long),
                persistent=False,
            )

    def load_weights(self) -> "ResynthesisNativeParent":
        """Load the complete native backbone + adaptive MoE + NoNE/RBO graph.

        The previous loader registered checkpoint tensors as inert buffers and
        reduced the base forward to an embedding lookup.  Loading now goes
        through Resynthesis's integrated constructor directly in this process, making
        its trained decoder, adaptive experts, NoNE Fabric, RBO traversal, VGE,
        and native head children of this model graph.
        """

        if self._weights_loaded:
            return self
        weights_path = Path(self.info.weights_path)
        if not weights_path.is_file():
            raise FileNotFoundError(
                f"Resynthesis base weights not found: {weights_path}. "
                "Set ResynthesisConfig.base_model_dir to the integrated Resynthesis checkpoint directory."
            )
        if self.info.composition != _RESYNTHESIS_PARENT_COMPOSITION:
            raise RuntimeError(
                "Resynthesis requires Resynthesis's one-model NoNE/RBO composition"
            )
        if (
            self.info.integrated_rbo_tensors <= 0
            or self.info.integrated_additive_tensors <= 0
            or self.info.integrated_native_decode_confidence_tensors <= 0
        ):
            raise RuntimeError(
                "Resynthesis requires trained integrated RBO, adaptive-expert, and native-confidence tensors"
            )

        fastokens_dependency = activate_resynthesis_fastokens(self.cfg)
        from resynthesis.none_paging import file_sha256_authority_boundary

        parent_sha256_cache_root = (
            Path(__file__).resolve().parents[1]
            / ".nnf-resynthesis"
            / "artifact_sha256_identity_cache"
            / RESYNTHESIS_NATIVE_PARENT_GENERATION
        )
        preload_artifact_sha256 = file_sha256_authority_boundary(
            weights_path,
            expected_sha256=self.info.model_artifact_sha256,
            identity_cache_root=parent_sha256_cache_root,
        )
        loaded, migration, legacy_bank, capability_source_sha256 = (
            _load_exact_resynthesis_parent(
                self.cfg,
                device=str(self.device),
                expected_rbo_tensor_count=self.info.integrated_rbo_tensors,
                expected_additive_tensor_count=self.info.integrated_additive_tensors,
            )
        )
        runtime, vocabulary_graph, _bank, tokenizer, _ladder = loaded
        loaded_tokenizer_identity = tokenizer_identity(
            tokenizer,
            dependency=fastokens_dependency,
            tokenizer_path=_resolve_path(
                self.cfg.base_model_dir,
                self.cfg.tokenizer_path,
            ),
            inherited_tokenizer_path=_resolve_path(
                self.cfg.base_model_dir,
                self.cfg.tokenizer_path,
            ),
        )
        setattr(
            tokenizer,
            "_resynthesis_tokenizer_identity",
            loaded_tokenizer_identity,
        )
        if not isinstance(runtime, nn.Module):
            raise RuntimeError(
                "Resynthesis integrated constructor returned a non-module runtime"
            )
        _bind_resynthesis_parent_runtime_facade(
            runtime,
            vocabulary_graph=vocabulary_graph,
        )
        if getattr(runtime, "resynthesis_rbo", None) is None:
            raise RuntimeError("Resynthesis integrated runtime has no trained in-graph RBO")
        if not bool(
            getattr(runtime, "resynthesis_none_architecture_active", False)
        ):
            raise RuntimeError(
                "Resynthesis integrated runtime has no active NoNE architecture"
            )
        native_decode_confidence = getattr(runtime, "native_decode_confidence", None)
        if not isinstance(native_decode_confidence, nn.Module):
            raise RuntimeError(
                "Resynthesis integrated runtime has no native decode-confidence diagnostic"
            )
        native_confidence_tensor_count = len(native_decode_confidence.state_dict())
        if (
            native_confidence_tensor_count
            != self.info.integrated_native_decode_confidence_tensors
        ):
            raise RuntimeError(
                "Resynthesis native decode-confidence tensor count differs from its manifest"
            )
        verified_artifact_sha256 = file_sha256_authority_boundary(
            weights_path,
            expected_sha256=self.info.model_artifact_sha256,
            identity_cache_root=parent_sha256_cache_root,
        )
        if verified_artifact_sha256 != preload_artifact_sha256:
            raise RuntimeError(
                "loaded Resynthesis model artifact SHA-256 changed during parent load"
            )
        runtime.requires_grad_(False)
        runtime.eval()
        self.runtime = runtime
        self.tokenizer = tokenizer
        self.tokenizer_backend_identity = loaded_tokenizer_identity
        self.verified_model_artifact_sha256 = verified_artifact_sha256
        self.ladder_namespace_migration_applied = migration.ladder_namespace_applied
        self.embedded_legacy_rbo_migration_applied = (
            migration.embedded_legacy_rbo_applied
        )
        self.embedded_legacy_rbo_tensor_count = (
            migration.embedded_legacy_rbo_tensor_count
        )
        self.additive_tensor_count_after_legacy_split = (
            migration.additive_tensor_count_after_split
        )
        self.embedded_legacy_rbo_key_set_sha256 = (
            migration.embedded_legacy_rbo_key_set_sha256
        )
        self.authoritative_rbo_tensor_count = migration.authoritative_rbo_tensor_count
        self.authoritative_rbo_key_set_sha256 = (
            migration.authoritative_rbo_key_set_sha256
        )
        self.exact_historical_source_loaded = migration.exact_historical_source_loaded
        self.historical_tensor_native_hotpath_id = (
            HISTORICAL_TENSOR_NATIVE_HOTPATH_ID
            if migration.exact_historical_source_loaded
            else ""
        )
        self.historical_recursive_arm_exhaustion_id = (
            HISTORICAL_RECURSIVE_ARM_EXHAUSTION_ID
            if migration.exact_historical_source_loaded
            else ""
        )
        self.legacy_capability_source_sha256 = capability_source_sha256
        object.__setattr__(self, "_pending_legacy_capability_bank", legacy_bank)
        self._weights_loaded = True
        self._base_loaded_marker.fill_(1)
        return self

    def checkpoint_lineage(self) -> dict[str, Any]:
        """Return immutable parent identity at the checkpoint I/O boundary."""

        if not self._weights_loaded:
            self.load_weights()

        if not all(
            (
                self.info.checkpoint_id,
                self.info.manifest_payload_sha256,
                self.info.model_artifact_sha256,
            )
        ):
            raise RuntimeError(
                "integrated Resynthesis parent has incomplete artifact lineage"
            )
        return {
            "schema": "nnf.resynthesis.parent_lineage.v4",
            "checkpointId": self.info.public_checkpoint_id,
            "manifestPayloadSha256": self.info.manifest_payload_sha256,
            "modelArtifactSha256": self.info.model_artifact_sha256,
            "parameterElements": self.info.parameter_elements,
            "nativeOwner": self.info.native_owner,
            "nativeGeneration": self.info.native_generation,
            "nativeRoot": self.info.native_root,
            "nativeManifestSha256": self.info.native_manifest_sha256,
            "nativeMigrationPromotionEligible": (
                self.info.native_migration_promotion_eligible
            ),
            "externalProductCheckpointDependency": False,
            "composition": self.info.public_composition,
            "modelType": self.info.public_model_type,
            "historicalInheritedCheckpointId": (
                self.info.historical_inherited_checkpoint_id
            ),
            "historicalInheritedComposition": (
                self.info.historical_inherited_composition
            ),
            "historicalInheritedModelType": (
                self.info.historical_inherited_model_type
            ),
            "rboTensorCount": self.info.integrated_rbo_tensors,
            "adaptiveTensorCount": self.info.integrated_additive_tensors,
            "nativeDecodeConfidenceTensorCount": (
                self.info.integrated_native_decode_confidence_tensors
            ),
            "authoritativeRboKeySetSha256": self.authoritative_rbo_key_set_sha256,
            "exactHistoricalSourceLoaded": self.exact_historical_source_loaded,
            "historicalTensorNativeHotpathId": (
                self.historical_tensor_native_hotpath_id
            ),
            "historicalTensorNativeExpertDispatch": True,
            "historicalTensorNativeHopSelection": True,
            "historicalRecursiveArmExhaustionId": (
                self.historical_recursive_arm_exhaustion_id
            ),
            "historicalRecursiveArmExhaustionPropagation": True,
            "embeddedLegacyCapabilityTensorCount": self.embedded_legacy_rbo_tensor_count,
            "embeddedLegacyCapabilityKeySetSha256": (
                self.embedded_legacy_rbo_key_set_sha256
            ),
            "nativeAnswerSurface": "rbo_head_argmax_mapped_through_trained_vge",
            "parentDualChunkRoPECompose": True,
            "parentOnlineSoftmaxLongPool": True,
            "nativeAttentionPositionAperture": NATIVE_ATTENTION_POSITION_APERTURE,
            "onlineSoftmaxTileTokens": RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS,
            "dualChunkPretrainLength": DUAL_CHUNK_PRETRAIN_LENGTH,
            "dualChunkLocalSize": DUAL_CHUNK_LOCAL_SIZE,
            "dualChunkAtSuccessiveSeamExposed": True,
            "pretrainedRoPEBandTokens": PRETRAINED_ROPE_BAND_TOKENS,
            "nativePrefillTileTokens": RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
        }

    def tokenizer_backend_receipt_boundary(self) -> dict[str, object]:
        """Serialize the verified parent-BPE executor at an I/O boundary."""

        if not self._weights_loaded:
            self.load_weights()
        if self.tokenizer is None or self.tokenizer_backend_identity is None:
            raise RuntimeError("integrated parent has no verified tokenizer backend")
        return tokenizer_boundary_receipt(self.tokenizer)

    def take_legacy_capability_bank(self) -> nn.Module | None:
        """Transfer the preserved payload into the trainable Resynthesis graph."""

        if not self._weights_loaded:
            self.load_weights()
        bank = getattr(self, "_pending_legacy_capability_bank", None)
        if bank is not None and not isinstance(bank, nn.Module):
            raise RuntimeError("pending legacy capability owner is not a module")
        object.__setattr__(self, "_pending_legacy_capability_bank", None)
        return bank

    def active_none_fabric(self) -> nn.Module:
        """Return the exact trained NoNE fabric owned by the loaded parent graph."""

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        parent_rbo = getattr(runtime, "resynthesis_rbo", None)
        parent_fabric = getattr(parent_rbo, "none_fabric", None)
        if not isinstance(parent_fabric, nn.Module):
            raise RuntimeError("Resynthesis integrated RBO has no trained NoNE fabric module")
        if not callable(getattr(parent_fabric, "step_phase", None)):
            raise RuntimeError(
                "Resynthesis parent NoNE fabric exposes no step_phase contract"
            )
        return parent_fabric

    def _record_native_context_cache_telemetry_boundary(
        self,
        cache: object,
    ) -> None:
        """Update diagnostic cache geometry in model-owned tensor buffers."""

        packet = NativeContextCacheTelemetry(
            hybrid_active=self._native_context_hybrid_active,
            total_positions=self._native_context_total_positions,
            hot_resident_positions=self._native_context_hot_resident_positions,
            hot_window_tokens=self._native_context_hot_window_tokens,
            recurrent_layer_count=self._native_context_recurrent_layer_count,
            full_attention_layer_count=(
                self._native_context_full_attention_layer_count
            ),
        )
        _update_native_context_cache_telemetry_boundary(
            cache,
            packet=packet,
        )

    def native_context_cache_telemetry_boundary(
        self,
    ) -> NativeContextCacheTelemetry:
        """Return read-only tensor evidence; this packet has no answer authority."""

        return NativeContextCacheTelemetry(
            hybrid_active=self._native_context_hybrid_active.detach().clone(),
            total_positions=self._native_context_total_positions.detach().clone(),
            hot_resident_positions=(
                self._native_context_hot_resident_positions.detach().clone()
            ),
            hot_window_tokens=(self._native_context_hot_window_tokens.detach().clone()),
            recurrent_layer_count=(
                self._native_context_recurrent_layer_count.detach().clone()
            ),
            full_attention_layer_count=(
                self._native_context_full_attention_layer_count.detach().clone()
            ),
        )

    def _prefill_tiled_native(
        self,
        *,
        runtime: nn.Module,
        active_input_ids: torch.Tensor,
        position_ids: torch.Tensor,
        attention_mask: torch.Tensor,
        record_kv: object,
        starting_prefix: torch.Tensor,
        telemetry_required: bool,
    ) -> NativeTiledPrefill:
        """Run a retained-cache prefill for native autoregressive continuation."""

        return self._prefill_tiled_native_impl(
            runtime=runtime,
            active_input_ids=active_input_ids,
            position_ids=position_ids,
            attention_mask=attention_mask,
            record_kv=record_kv,
            starting_prefix=starting_prefix,
            telemetry_required=telemetry_required,
            retain_final_cache=True,
        )

    def _prefill_tiled_training_native(
        self,
        *,
        runtime: nn.Module,
        active_input_ids: torch.Tensor,
        position_ids: torch.Tensor,
        attention_mask: torch.Tensor,
        record_kv: object,
        starting_prefix: torch.Tensor,
        telemetry_required: bool,
    ) -> NativeTiledPrefill:
        """Run an exact masked training prefill without retaining its final cache."""

        return self._prefill_tiled_native_impl(
            runtime=runtime,
            active_input_ids=active_input_ids,
            position_ids=position_ids,
            attention_mask=attention_mask,
            record_kv=record_kv,
            starting_prefix=starting_prefix,
            telemetry_required=telemetry_required,
            retain_final_cache=False,
        )

    def _prefill_tiled_native_impl(
        self,
        *,
        runtime: nn.Module,
        active_input_ids: torch.Tensor,
        position_ids: torch.Tensor,
        attention_mask: torch.Tensor,
        record_kv: object,
        starting_prefix: torch.Tensor,
        telemetry_required: bool,
        retain_final_cache: bool,
    ) -> NativeTiledPrefill:
        """Tile the parent prefill so 4M+ prompts fit bounded activation memory.

        Each tile threads the parent cache and uses a CUMULATIVE attention mask
        (``attention_mask[:, :end]``). Within the pretrained band, the parent
        keeps exact full-attention K/V. Longer prompts use its trained hybrid
        topology: all 24 linear-attention layers keep global recurrent state,
        while eight full-attention layers keep one Dual-Chunk hot K/V window.
        Folded ``position_ids`` are sliced per tile (never re-folded), and no
        token is dropped or truncated.

        Raises if the parent loses its KV cache before every prompt token has
        been preflighted.
        """

        total_prefill_tokens = int(active_input_ids.shape[1])
        tile_tokens = min(
            total_prefill_tokens,
            RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
        )
        running_prefix = starting_prefix.clone()
        tile_past: Any = None
        summary_hidden: torch.Tensor | None = None
        summary_index = 0
        summary_capacity = (
            (total_prefill_tokens + tile_tokens - 1) // tile_tokens
        ) * RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE
        if total_prefill_tokens > RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS:
            custom_cache_builder = getattr(
                runtime,
                "resynthesis_build_native_hybrid_context_cache",
                None,
            )
            if callable(custom_cache_builder):
                tile_past = custom_cache_builder(
                    hot_window_tokens=active_input_ids.new_ones(
                        (), dtype=torch.long
                    ).mul_(RESYNTHESIS_NATIVE_HOT_KV_TOKENS)
                )
            else:
                tile_past = _build_native_hybrid_context_cache(runtime)
            if not bool(
                getattr(
                    tile_past,
                    "_resynthesis_native_hybrid_context",
                    False,
                )
            ):
                raise RuntimeError(
                    "Resynthesis native hybrid context cache has no model-owned marker"
                )
        out: Any = None
        for start in range(0, total_prefill_tokens, tile_tokens):
            end = min(total_prefill_tokens, start + tile_tokens)
            tile_cache_required = retain_final_cache or end != total_prefill_tokens
            tile_ids = active_input_ids[:, start:end]
            tile_positions = position_ids[:, start:end]
            # Cumulative mask keeps every prior token attendable under exact SDPA.
            tile_mask = attention_mask[:, :end]
            tile_new_positions = tile_ids.new_ones(
                (), dtype=torch.long
            ).mul_(tile_ids.shape[1])
            if callable(record_kv):
                record_kv(
                    prefix_positions=running_prefix,
                    new_positions=tile_new_positions,
                )
            elif telemetry_required:
                raise RuntimeError(
                    "Resynthesis integrated runtime lacks tensor-native KV telemetry"
                )
            out = runtime(
                input_ids=tile_ids,
                position_ids=tile_positions,
                attention_mask=tile_mask,
                past_key_values=tile_past,
                output_hidden_states=True,
                return_dict=True,
                use_cache=tile_cache_required,
                logits_to_keep=1,
            )
            tile_hidden_states = getattr(out, "hidden_states", None)
            if not tile_hidden_states or not isinstance(
                tile_hidden_states[-1], torch.Tensor
            ):
                raise RuntimeError(
                    "Resynthesis native prefill returned no hidden state for additive attention"
                )
            tile_hidden = tile_hidden_states[-1]
            if tile_hidden.ndim != 3 or tile_hidden.shape[1] < 1:
                raise RuntimeError("Resynthesis native prefill hidden geometry differs")
            if summary_hidden is None:
                summary_hidden = tile_hidden.new_empty(
                    tile_hidden.shape[0],
                    summary_capacity,
                    tile_hidden.shape[-1],
                )
            segment_count = min(
                RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE,
                tile_hidden.shape[1],
            )
            for segment_index in range(segment_count):
                segment_start = segment_index * tile_hidden.shape[1] // segment_count
                segment_end = (
                    (segment_index + 1) * tile_hidden.shape[1] // segment_count
                )
                segment_hidden = tile_hidden[:, segment_start:segment_end, :]
                # The installed final-hidden-only parent contract returns one
                # logits-aligned position for ``logits_to_keep=1``. Attention
                # over a singleton has exact unit weight, so its output is the
                # same tensor. Avoid launching score, softmax, ring-softmax,
                # matmul, arange, and allocating index_copy for every prefill
                # tile. The multi-position compatibility path retains the full
                # trained attention pool without changing its semantics.
                segment_context = (
                    segment_hidden[:, 0, :]
                    if segment_hidden.shape[1] == 1
                    else _parent_hidden_context(segment_hidden)
                )
                summary_hidden[:, summary_index, :].copy_(segment_context)
                summary_index += 1
            returned_past = getattr(out, "past_key_values", None)
            if tile_cache_required and returned_past is not None:
                tile_past = returned_past
                running_prefix = running_prefix + tile_new_positions
            elif end != total_prefill_tokens:
                raise RuntimeError(
                    "Resynthesis native 4M prefill lost KV cache before all tokens"
                )
            elif not tile_cache_required:
                # A pre-existing multi-tile cache is passed into the final tile
                # so attention remains exact. Some parent implementations return
                # that object even with ``use_cache=False``; sever it explicitly
                # after extracting hidden/logits so one training wave cannot
                # retain parent KV/recurrent state into the next wave.
                tile_past = None
                setattr(out, "past_key_values", None)
            else:
                tile_past = None
        if out is None:
            raise RuntimeError("Resynthesis native prefill received an empty prompt")
        if summary_hidden is None or summary_index < 1:
            raise RuntimeError("Resynthesis native prefill produced no attended summaries")
        return NativeTiledPrefill(
            runtime_output=out,
            summary_hidden=summary_hidden.narrow(1, 0, summary_index),
            input_positions=active_input_ids.new_ones(
                (), dtype=torch.long
            ).mul_(total_prefill_tokens),
        )

    def forward_hidden_logits(
        self,
        input_ids: torch.Tensor,
        attention_mask: torch.Tensor | None = None,
    ) -> ResynthesisParentForward:
        """Execute one cache-retaining causal parent prefill or continuation."""

        return self._forward_hidden_logits_impl(
            input_ids,
            attention_mask=attention_mask,
            masked_training_prefill=False,
        )

    def forward_training_hidden_logits(
        self,
        input_ids: torch.Tensor,
        *,
        attention_mask: torch.Tensor,
    ) -> ResynthesisParentForward:
        """Execute one fresh masked training prefill with no final parent cache."""

        return self._forward_hidden_logits_impl(
            input_ids,
            attention_mask=attention_mask,
            masked_training_prefill=True,
        )

    def forward_training_hidden_logits_shared_prefix(
        self,
        packet: NativeSharedPrefixTrainingPacket,
    ) -> ResynthesisParentForward:
        """Run one packed batch from an immutable frozen-parent prefix cache.

        Prefix construction enters only the frozen backbone.  The integrated
        parent RBO/Fabric therefore still executes exactly once, on the suffix
        batch, matching an ordinary final-position training prefill while
        avoiding one identical backbone prefix execution per row.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        prefix_ids = packet.prefix_ids.to(device=self.device, dtype=torch.long)
        suffix_ids = packet.suffix_ids.to(device=self.device, dtype=torch.long)
        suffix_mask = packet.suffix_mask.to(
            device=self.device,
            dtype=torch.long,
        )
        batch_indices_t = packet.batch_indices_t.to(
            device=self.device,
            dtype=torch.long,
        )
        if (
            prefix_ids.ndim != 2
            or prefix_ids.shape[0] != 1
            or prefix_ids.shape[1] < 1
            or suffix_ids.ndim != 2
            or suffix_ids.shape[0] < 1
            or suffix_ids.shape[1] < 1
            or suffix_mask.shape != suffix_ids.shape
            or batch_indices_t.shape != (suffix_ids.shape[0],)
        ):
            raise RuntimeError("shared parent-prefix training geometry differs")
        torch._assert_async(
            suffix_mask.ge(0).logical_and(suffix_mask.le(1)).all(),
            "shared parent-prefix suffix mask is not binary",
        )
        torch._assert_async(
            suffix_mask[:, -1].eq(1).all(),
            "shared parent-prefix suffix must end in a visible token",
        )
        if (
            suffix_ids.shape[0] < 2
            or prefix_ids.shape[1] + suffix_ids.shape[1]
            > RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS
        ):
            # A post-materialization CUDA replan may isolate one row, while a
            # packed full-payload window may cross the native prefill tile.
            # Neither condition invalidates the prompt. Reconstruct the exact
            # target-free masked batch and use the ordinary tiled parent path;
            # the live additive RBO/Fabric/expert/page graph still runs after
            # this frozen-parent boundary. This is a semantic fallback, not a
            # token cap: every visible suffix token remains in the forward.
            prefix_batch_ids = prefix_ids.expand(suffix_ids.shape[0], -1)
            full_ids = torch.cat((prefix_batch_ids, suffix_ids), dim=1)
            full_mask = torch.cat(
                (
                    torch.ones_like(prefix_batch_ids, dtype=torch.long),
                    suffix_mask,
                ),
                dim=1,
            )
            return self.forward_training_hidden_logits(
                full_ids,
                attention_mask=full_mask,
            )
        if self._decode_past_key_values is not None:
            raise RuntimeError(
                "shared parent-prefix training requires a fresh decode arm"
            )
        torch._assert_async(
            self._decode_cached_positions.eq(0),
            "shared parent-prefix training requires a fresh decode arm",
        )

        backbone = getattr(runtime, "backbone", None)
        if not isinstance(backbone, nn.Module) or getattr(
            backbone,
            "_resynthesis_final_hidden_only_forward_id",
            "",
        ) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
            raise RuntimeError(
                "shared parent-prefix training requires the exact frozen backbone"
            )
        decoder_core = getattr(backbone, "model", None)
        if not isinstance(decoder_core, nn.Module):
            raise RuntimeError(
                "shared parent-prefix training requires the frozen decoder core"
            )
        if not self._decode_authority_active:
            begin_decode = getattr(runtime, "resynthesis_begin_authority_decode", None)
            if callable(begin_decode):
                authority_forward_id = begin_decode()
                if not isinstance(authority_forward_id, torch.Tensor):
                    raise RuntimeError(
                        "Resynthesis authority transaction returned no tensor forward ID"
                    )
            elif not self.exact_historical_source_loaded:
                raise RuntimeError(
                    "Resynthesis integrated runtime lacks its authority transaction"
                )
            self._decode_authority_active = True

        prefix_mask = torch.ones_like(prefix_ids, dtype=torch.long)
        prefix_positions_t = (
            prefix_mask.cumsum(dim=1).sub(1).clamp_min(0)
        )
        validate_native_context_admission(prefix_positions_t)
        prefix_position_stack = build_long_context_position_stack(
            prefix_positions_t
        )
        master_cache = _build_native_hybrid_context_cache(runtime)
        prefix_count_t = prefix_ids.new_ones((), dtype=torch.long).mul_(
            prefix_ids.shape[1]
        )
        with torch.no_grad():
            # Enter the decoder core, not the conditional-generation wrapper.
            # Its LM head owns additive experts and the embedded parent RBO;
            # invoking that head here would execute RBO/Fabric once on the
            # prefix and again on the suffix.  The immutable prefix needs only
            # the decoder's exact K/V and recurrent state.
            prefix_output = decoder_core(
                input_ids=prefix_ids,
                position_ids=prefix_position_stack.rope_position_ids,
                attention_mask=prefix_mask,
                past_key_values=master_cache,
                output_hidden_states=False,
                return_dict=True,
                use_cache=True,
            )
        returned_master = getattr(prefix_output, "past_key_values", None)
        if (
            returned_master is not master_cache
            or getattr(
                returned_master,
                "_resynthesis_native_hybrid_context",
                False,
            )
            is not True
        ):
            raise RuntimeError(
                "frozen backbone did not return its immutable hybrid prefix cache"
            )
        expanded_cache = _expand_immutable_native_hybrid_context_cache_boundary(
            master_cache,
            batch_indices_t,
        )
        setattr(prefix_output, "past_key_values", None)
        prefix_batch_mask = prefix_mask.expand(suffix_ids.shape[0], -1)
        full_input_positions_t = prefix_count_t + suffix_ids.new_ones(
            (), dtype=torch.long
        ).mul_(suffix_ids.shape[1])
        return self._forward_hidden_logits_impl(
            suffix_ids,
            attention_mask=suffix_mask,
            masked_training_prefill=True,
            shared_prefix_cache=expanded_cache,
            shared_prefix_mask=prefix_batch_mask,
            shared_full_input_positions_t=full_input_positions_t,
        )

    def frozen_backbone_position_policy_sha256_boundary(self) -> str:
        """Return the immutable decoder position/tile policy identity."""

        return frozen_backbone_position_policy_sha256_boundary(
            final_hidden_adapter_id=PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID,
            dual_chunk_pretrain_length=DUAL_CHUNK_PRETRAIN_LENGTH,
            dual_chunk_local_size=DUAL_CHUNK_LOCAL_SIZE,
            pretrained_rope_band_tokens=PRETRAINED_ROPE_BAND_TOKENS,
            prefill_tile_tokens=RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
            prefill_summaries_per_tile=(
                RESYNTHESIS_NATIVE_PREFILL_SUMMARIES_PER_TILE
            ),
        )

    def frozen_backbone_feature_dtype_boundary(self) -> torch.dtype:
        """Return the exact frozen decoder feature dtype before sidecar lookup.

        A durable feature object is keyed before the expensive parent forward.
        The dtype therefore has to come from the loaded decoder itself, not
        from a host configuration guess.  Captured features are authoritative
        when this process has already run a prefill; otherwise the first
        floating decoder parameter supplies the output dtype used by the
        frozen final-hidden adapter.
        """

        captured = self._last_frozen_backbone_final_hidden_t
        if isinstance(captured, torch.Tensor):
            return captured.dtype
        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        backbone = getattr(runtime, "backbone", None)
        decoder_core = getattr(backbone, "model", None)
        if not isinstance(decoder_core, nn.Module):
            raise RuntimeError(
                "frozen-backbone feature dtype requires the frozen decoder core"
            )
        for parameter in decoder_core.parameters():
            if parameter.is_floating_point():
                return parameter.dtype
        for buffer in decoder_core.buffers():
            if buffer.is_floating_point():
                return buffer.dtype
        raise RuntimeError(
            "frozen-backbone decoder has no floating feature dtype authority"
        )

    def frozen_backbone_cache_authority_boundary(
        self,
        *,
        qualified_work_id: str,
        prompt_sha256: str,
        prompt_mask_sha256: str,
        token_dtype: torch.dtype,
        feature_dtype: torch.dtype,
        summary_positions: int,
    ) -> FrozenBackboneCacheAuthority:
        """Bind one external packed row to this exact immutable parent."""

        token_dtype_name = str(token_dtype).removeprefix("torch.")
        feature_dtype_name = str(feature_dtype).removeprefix("torch.")
        authority = FrozenBackboneCacheAuthority(
            qualified_work_id=qualified_work_id,
            prompt_sha256=prompt_sha256,
            prompt_mask_sha256=prompt_mask_sha256,
            parent_checkpoint_id=self.info.checkpoint_id,
            parent_manifest_payload_sha256=(
                self.info.manifest_payload_sha256
            ),
            parent_model_artifact_sha256=self.info.model_artifact_sha256,
            # Retained only as a direct-path hint for a historical source-bound
            # sidecar. It is excluded from executable cache identity; a
            # read-only normalized index finds older source observations.
            parent_source_bundle_sha256=(
                self.info.parent_source_bundle_sha256
            ),
            position_policy_sha256=(
                self.frozen_backbone_position_policy_sha256_boundary()
            ),
            token_dtype=token_dtype_name,
            feature_dtype=feature_dtype_name,
            hidden_size=self.info.hidden_size,
            summary_positions=summary_positions,
        )
        authority.record_boundary()
        return authority

    def precompute_frozen_backbone_prefill_boundary(
        self,
        input_ids: torch.Tensor,
        *,
        attention_mask: torch.Tensor,
        authorities: tuple[FrozenBackboneCacheAuthority, ...],
    ) -> FrozenBackbonePrefillPacket:
        """Compute raw frozen-decoder features without entering live heads.

        This is the one-shot packed-sidecar producer boundary.  It deliberately
        calls only the immutable decoder core: the conditional-generation
        wrapper, additive head, historical RBO, Fabric, experts, page runtime,
        answer targets, and task-intent targets are outside this interface.
        The returned typed packet can therefore accelerate a later training
        forward without caching any trainable or routing-owned result.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        if (
            input_ids.ndim != 2
            or input_ids.shape[0] < 1
            or input_ids.shape[1] < 1
            or attention_mask.shape != input_ids.shape
            or len(authorities) != input_ids.shape[0]
        ):
            raise ValueError(
                "frozen-backbone precompute prompt geometry differs"
            )
        if self._decode_past_key_values is not None:
            raise RuntimeError(
                "frozen-backbone precompute requires a fresh decode arm"
            )
        active_input_ids = input_ids.to(
            device=self.device,
            dtype=torch.long,
        )
        active_attention_mask = attention_mask.to(
            device=self.device,
            dtype=torch.long,
        )
        torch._assert_async(
            active_attention_mask.ge(0).logical_and(
                active_attention_mask.le(1)
            ).all(),
            "frozen-backbone precompute attention mask is not binary",
        )
        torch._assert_async(
            active_attention_mask[:, -1].eq(1).all(),
            "frozen-backbone precompute prompt must end in a visible token",
        )
        torch._assert_async(
            self._decode_cached_positions.eq(0),
            "frozen-backbone precompute requires a fresh decode arm",
        )

        backbone = getattr(runtime, "backbone", None)
        if not isinstance(backbone, nn.Module) or getattr(
            backbone,
            "_resynthesis_final_hidden_only_forward_id",
            "",
        ) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
            raise RuntimeError(
                "frozen-backbone precompute requires the exact parent adapter"
            )
        decoder_core = getattr(backbone, "model", None)
        if not isinstance(decoder_core, nn.Module):
            raise RuntimeError(
                "frozen-backbone precompute requires the frozen decoder core"
            )

        absolute_positions = (
            active_attention_mask.cumsum(dim=1).sub(1).clamp_min(0)
        )
        validate_native_context_admission(absolute_positions)
        position_ids = build_long_context_position_stack(
            absolute_positions
        ).rope_position_ids
        total_prefill_tokens = active_input_ids.shape[1]
        tile_tokens = min(
            total_prefill_tokens,
            RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS,
        )
        summary_positions = (
            total_prefill_tokens + tile_tokens - 1
        ) // tile_tokens
        feature_dtype = self.frozen_backbone_feature_dtype_boundary()
        if any(
            authority.parent_checkpoint_id != self.info.checkpoint_id
            or authority.parent_manifest_payload_sha256
            != self.info.manifest_payload_sha256
            or authority.parent_model_artifact_sha256
            != self.info.model_artifact_sha256
            or authority.position_policy_sha256
            != self.frozen_backbone_position_policy_sha256_boundary()
            or authority.token_dtype
            != str(input_ids.dtype).removeprefix("torch.")
            or authority.feature_dtype
            != str(feature_dtype).removeprefix("torch.")
            or authority.hidden_size != self.info.hidden_size
            or authority.summary_positions != summary_positions
            for authority in authorities
        ):
            raise RuntimeError(
                "frozen-backbone precompute authority differs from the parent"
            )

        tile_past: Any = None
        if (
            total_prefill_tokens
            > RESYNTHESIS_NATIVE_HYBRID_ACTIVATION_TOKENS
        ):
            custom_cache_builder = getattr(
                runtime,
                "resynthesis_build_native_hybrid_context_cache",
                None,
            )
            if callable(custom_cache_builder):
                tile_past = custom_cache_builder(
                    hot_window_tokens=active_input_ids.new_ones(
                        (),
                        dtype=torch.long,
                    ).mul_(RESYNTHESIS_NATIVE_HOT_KV_TOKENS)
                )
            else:
                tile_past = _build_native_hybrid_context_cache(runtime)
            if not bool(
                getattr(
                    tile_past,
                    "_resynthesis_native_hybrid_context",
                    False,
                )
            ):
                raise RuntimeError(
                    "frozen-backbone precompute hybrid cache is not native"
                )

        summary_hidden_t: torch.Tensor | None = None
        final_hidden_t: torch.Tensor | None = None
        summary_index = 0
        with torch.no_grad():
            for start in range(0, total_prefill_tokens, tile_tokens):
                end = min(total_prefill_tokens, start + tile_tokens)
                tile_cache_required = end != total_prefill_tokens
                output = decoder_core(
                    input_ids=active_input_ids[:, start:end],
                    position_ids=position_ids[:, start:end],
                    attention_mask=active_attention_mask[:, :end],
                    past_key_values=tile_past,
                    output_hidden_states=False,
                    return_dict=True,
                    use_cache=tile_cache_required,
                )
                tile_hidden_t = getattr(
                    output,
                    "last_hidden_state",
                    None,
                )
                if (
                    not isinstance(tile_hidden_t, torch.Tensor)
                    or tile_hidden_t.ndim != 3
                    or tile_hidden_t.shape[0] != input_ids.shape[0]
                    or tile_hidden_t.shape[1] < 1
                    or tile_hidden_t.shape[2] != self.info.hidden_size
                    or tile_hidden_t.dtype != feature_dtype
                ):
                    raise RuntimeError(
                        "frozen-backbone decoder feature geometry differs"
                    )
                # The installed parent adapter exposes the one logits-aligned
                # final position from each tile.  Selecting the same position
                # here makes the offline packet byte-equivalent without
                # invoking the additive LM head that records it online.
                final_hidden_t = tile_hidden_t[:, -1:, :]
                if summary_hidden_t is None:
                    summary_hidden_t = final_hidden_t.new_empty(
                        input_ids.shape[0],
                        summary_positions,
                        self.info.hidden_size,
                    )
                summary_hidden_t[:, summary_index : summary_index + 1, :].copy_(
                    final_hidden_t
                )
                summary_index += 1
                returned_past = getattr(output, "past_key_values", None)
                if tile_cache_required and returned_past is not None:
                    tile_past = returned_past
                elif tile_cache_required:
                    raise RuntimeError(
                        "frozen-backbone precompute lost its decoder cache"
                    )
                else:
                    tile_past = None
                    setattr(output, "past_key_values", None)
        if (
            final_hidden_t is None
            or summary_hidden_t is None
            or summary_index != summary_positions
        ):
            raise RuntimeError(
                "frozen-backbone precompute produced incomplete features"
            )
        input_positions_t = active_input_ids.new_ones(
            (input_ids.shape[0],),
            dtype=torch.long,
        ).mul_(total_prefill_tokens)
        row_packets = tuple(
            FrozenBackbonePrefillPacket.from_features_boundary(
                authority=authority,
                final_hidden_t=final_hidden_t[row_index : row_index + 1],
                summary_hidden_t=summary_hidden_t[
                    row_index : row_index + 1
                ],
                input_positions_t=input_positions_t[
                    row_index : row_index + 1
                ],
            )
            for row_index, authority in enumerate(authorities)
        )
        return FrozenBackbonePrefillPacket.stack_boundary(row_packets)

    def frozen_backbone_prefill_row_packet_boundary(
        self,
        *,
        authority: FrozenBackboneCacheAuthority,
        batch_index: int,
    ) -> FrozenBackbonePrefillPacket:
        """Select one just-computed raw decoder row for durable external I/O."""

        final_hidden_t = self._last_frozen_backbone_final_hidden_t
        summary_hidden_t = self._last_frozen_backbone_summary_hidden_t
        input_positions_t = self._last_frozen_backbone_input_positions_t
        if (
            not isinstance(final_hidden_t, torch.Tensor)
            or not isinstance(summary_hidden_t, torch.Tensor)
            or not isinstance(input_positions_t, torch.Tensor)
        ):
            raise RuntimeError(
                "frozen-backbone row requested before one masked parent prefill"
            )
        if batch_index < 0 or batch_index >= final_hidden_t.shape[0]:
            raise IndexError("frozen-backbone batch row is unavailable")
        if (
            authority.parent_checkpoint_id != self.info.checkpoint_id
            or authority.parent_manifest_payload_sha256
            != self.info.manifest_payload_sha256
            or authority.parent_model_artifact_sha256
            != self.info.model_artifact_sha256
            or authority.position_policy_sha256
            != self.frozen_backbone_position_policy_sha256_boundary()
            or authority.hidden_size != self.info.hidden_size
            or authority.summary_positions != summary_hidden_t.shape[1]
            or authority.feature_dtype
            != str(final_hidden_t.dtype).removeprefix("torch.")
        ):
            raise RuntimeError(
                "frozen-backbone row authority differs from the loaded parent"
            )
        row = slice(batch_index, batch_index + 1)
        return FrozenBackbonePrefillPacket.from_features_boundary(
            authority=authority,
            final_hidden_t=final_hidden_t[row],
            summary_hidden_t=summary_hidden_t[row],
            input_positions_t=input_positions_t[row],
        )

    def forward_training_from_frozen_backbone_packet(
        self,
        input_ids: torch.Tensor,
        *,
        attention_mask: torch.Tensor,
        packet: FrozenBackbonePrefillPacket,
    ) -> ResynthesisParentForward:
        """Replay decoder-only features through the complete live parent graph.

        The decoder core is replaced for this one synchronous call only.  The
        surrounding conditional-generation wrapper, additive LM head,
        historical RBO, and historical Fabric all execute normally for every
        original prefill tile.  This is deliberately narrower than replaying a
        cached ``ResynthesisParentForward``, which would bypass those trained surfaces.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        packet.validate_boundary()
        active_packet = packet.to_device_boundary(self.device)
        if (
            input_ids.ndim != 2
            or input_ids.shape[1] < 1
            or attention_mask.shape != input_ids.shape
            or attention_mask.dtype != torch.bool
            or active_packet.final_hidden_t.shape[0] != input_ids.shape[0]
            or active_packet.final_hidden_t.shape[2] != self.info.hidden_size
        ):
            raise RuntimeError(
                "frozen-backbone replay prompt/feature geometry differs"
            )
        expected_summary_positions = (
            input_ids.shape[1] + RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS - 1
        ) // RESYNTHESIS_NATIVE_PREFILL_TILE_TOKENS
        if (
            active_packet.summary_hidden_t.shape[1]
            != expected_summary_positions
        ):
            raise RuntimeError(
                "frozen-backbone replay summary/tile geometry differs"
            )
        expected_input_positions_t = attention_mask.sum(
            dim=1,
            dtype=torch.long,
        )
        torch._assert_async(
            active_packet.input_positions_t.eq(
                expected_input_positions_t
            ).all(),
            "frozen-backbone replay input positions differ",
        )
        expected_left_padding_mask_t = torch.arange(
            input_ids.shape[1],
            device=attention_mask.device,
            dtype=torch.long,
        ).unsqueeze(0).ge(
            input_ids.shape[1] - expected_input_positions_t.unsqueeze(1)
        )
        torch._assert_async(
            attention_mask.eq(expected_left_padding_mask_t).all(),
            "frozen-backbone replay mask is not canonical left padding",
        )
        torch._assert_async(
            input_ids.masked_select(~attention_mask).eq(0).all(),
            "frozen-backbone replay padding tokens must be zero",
        )
        torch._assert_async(
            active_packet.final_hidden_t.eq(
                active_packet.summary_hidden_t[:, -1:, :]
            ).all(),
            "frozen-backbone final hidden differs from its final tile summary",
        )

        backbone = getattr(runtime, "backbone", None)
        if not isinstance(backbone, nn.Module) or getattr(
            backbone,
            "_resynthesis_final_hidden_only_forward_id",
            "",
        ) != PARENT_FINAL_HIDDEN_ONLY_FORWARD_ID:
            raise RuntimeError(
                "frozen-backbone replay requires the exact parent adapter"
            )
        decoder_core = getattr(backbone, "model", None)
        if not isinstance(decoder_core, nn.Module):
            raise RuntimeError(
                "frozen-backbone replay requires the frozen decoder core"
            )
        original_forward = decoder_core.forward
        replay_call_index = 0
        replay_cache = object()

        def replay_decoder_forward(
            *args: Any,
            **kwargs: Any,
        ) -> _FrozenBackboneDecoderOutput:
            nonlocal replay_call_index
            input_value = kwargs.get("input_ids")
            if input_value is None and args:
                input_value = args[0]
            if (
                not isinstance(input_value, torch.Tensor)
                or input_value.ndim != 2
                or input_value.shape[0]
                != active_packet.final_hidden_t.shape[0]
                or replay_call_index
                >= active_packet.summary_hidden_t.shape[1]
            ):
                raise RuntimeError(
                    "frozen-backbone decoder replay invocation differs"
                )
            hidden_t = active_packet.summary_hidden_t[
                :,
                replay_call_index : replay_call_index + 1,
                :,
            ]
            replay_call_index += 1
            past_key_values = kwargs.get("past_key_values")
            if kwargs.get("use_cache") is True and past_key_values is None:
                past_key_values = replay_cache
            return _FrozenBackboneDecoderOutput(
                last_hidden_state=hidden_t,
                past_key_values=past_key_values,
            )

        setattr(decoder_core, "forward", replay_decoder_forward)
        try:
            result = self._forward_hidden_logits_impl(
                input_ids,
                attention_mask=attention_mask,
                masked_training_prefill=True,
            )
        finally:
            setattr(decoder_core, "forward", original_forward)
        if replay_call_index != active_packet.summary_hidden_t.shape[1]:
            raise RuntimeError(
                "frozen-backbone replay did not consume every sealed tile"
            )
        # The live tiled wrapper reports the padded batch width because every
        # row shares its decoder call geometry.  Durable sidecars retain the
        # exact unpadded row positions, which are the prefill participation
        # evidence consumed by the live RBO/Fabric graph.  Preserve that vector
        # after the decoder-only replay instead of manufacturing the wider
        # cohort position for shorter rows.
        return replace(
            result,
            parent_prefill_input_positions=(
                active_packet.input_positions_t.detach()
            ),
        )

    def _forward_hidden_logits_impl(
        self,
        input_ids: torch.Tensor,
        *,
        attention_mask: torch.Tensor | None,
        masked_training_prefill: bool,
        shared_prefix_cache: object | None = None,
        shared_prefix_mask: torch.Tensor | None = None,
        shared_full_input_positions_t: torch.Tensor | None = None,
    ) -> ResynthesisParentForward:
        """Execute one causal parent prefill or native KV continuation.

        A decode arm runs the complete prompt exactly once. Later emissions
        enter the same integrated Resynthesis/RBO graph as one new token with
        its native KV state. If the parent supplies no cache, this falls back
        to complete-prefix execution without changing stop or route authority.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        self._last_frozen_backbone_final_hidden_t = None
        self._last_frozen_backbone_summary_hidden_t = None
        self._last_frozen_backbone_input_positions_t = None
        if input_ids.ndim != 2 or input_ids.shape[1] < 1:
            raise ValueError("Resynthesis parent input IDs must have shape [batch, sequence]")
        active_input_ids = input_ids.to(device=self.device, dtype=torch.long)
        explicit_attention_mask = attention_mask is not None
        if attention_mask is None:
            active_attention_mask = torch.ones(
                input_ids.shape,
                device=self.device,
                dtype=torch.long,
            )
        else:
            if attention_mask.shape != input_ids.shape:
                raise ValueError("Resynthesis parent attention mask geometry differs")
            active_attention_mask = attention_mask.to(
                device=self.device,
                dtype=torch.long,
            )
            torch._assert_async(
                active_attention_mask.ge(0).logical_and(
                    active_attention_mask.le(1)
                ).all(),
                "Resynthesis parent attention mask is not binary",
            )
            torch._assert_async(
                active_attention_mask[:, -1].eq(1).all(),
                "Resynthesis left-padded prompt must end in a visible token",
            )
        shared_prefix_active = shared_prefix_cache is not None
        if shared_prefix_active != (
            shared_prefix_mask is not None
            and shared_full_input_positions_t is not None
        ):
            raise RuntimeError("shared parent-prefix cache boundary is incomplete")
        if shared_prefix_active and (
            not masked_training_prefill
            or not explicit_attention_mask
            or shared_prefix_mask is None
            or shared_prefix_mask.ndim != 2
            or shared_prefix_mask.shape[0] != active_input_ids.shape[0]
            or shared_prefix_mask.shape[1] < 1
            or shared_prefix_mask.device != active_input_ids.device
            or self._decode_past_key_values is not None
        ):
            raise RuntimeError("shared parent-prefix continuation geometry differs")
        use_past = (
            shared_prefix_active
            or self._decode_past_key_values is not None
        )
        if not self._decode_authority_active:
            begin_decode = getattr(runtime, "resynthesis_begin_authority_decode", None)
            if callable(begin_decode):
                authority_forward_id = begin_decode()
                if not isinstance(authority_forward_id, torch.Tensor):
                    raise RuntimeError(
                        "Resynthesis authority transaction returned no tensor forward ID"
                    )
            elif not self.exact_historical_source_loaded:
                raise RuntimeError(
                    "Resynthesis integrated runtime lacks its authority transaction"
                )
            self._decode_authority_active = True
        if use_past:
            if attention_mask is not None and not shared_prefix_active:
                raise RuntimeError(
                    "explicit batched attention masks require a fresh decode arm"
                )
            if not shared_prefix_active:
                active_input_ids = active_input_ids[:, -1:]
                torch._assert_async(
                    self._decode_cached_positions.eq(input_ids.shape[1] - 1),
                    "Resynthesis KV continuation does not match the visible prefix",
                )
        prefix_positions = (
            shared_prefix_mask.new_ones((), dtype=torch.long).mul_(
                shared_prefix_mask.shape[1]
            )
            if shared_prefix_active and shared_prefix_mask is not None
            else self._decode_cached_positions.to(
                device=active_input_ids.device,
                dtype=torch.long,
            ).clone()
        )
        runtime_new_positions = active_input_ids.new_ones(
            (), dtype=torch.long
        ).mul_(active_input_ids.shape[1])
        result_prefix_positions = (
            prefix_positions.new_zeros(())
            if shared_prefix_active
            else prefix_positions
        )
        result_new_positions = (
            shared_full_input_positions_t
            if shared_prefix_active
            and shared_full_input_positions_t is not None
            else runtime_new_positions
        )
        record_kv = getattr(runtime, "resynthesis_record_kv_cache_reuse", None)
        position_attention_mask = (
            active_attention_mask
            if shared_prefix_active
            else active_attention_mask[:, -active_input_ids.shape[1] :]
            if use_past
            else active_attention_mask
        )
        absolute_positions = (
            position_attention_mask.cumsum(dim=1)
            .sub(1)
            .clamp_min(0)
            .add(prefix_positions.to(device=self.device))
        )
        validate_native_context_admission(absolute_positions)
        # STACK+COMPOSE parent decode: absolute KV cache + Dual Chunk → parent RoPE.
        # ``position_ids`` feed the parent's live rotary/YaRN; KV slots stay absolute.
        # The full-prompt folded positions are computed once and then SLICED per
        # prefill tile, so each tile sees the correct in-distribution RoPE phase
        # for its absolute position (never re-folded per tile).
        position_stack = build_long_context_position_stack(absolute_positions)
        position_ids = position_stack.rope_position_ids
        runtime_attention_mask = (
            torch.cat(
                (
                    shared_prefix_mask.to(dtype=active_attention_mask.dtype),
                    active_attention_mask,
                ),
                dim=1,
            )
            if shared_prefix_active and shared_prefix_mask is not None
            else active_attention_mask
        )
        prefill: NativeTiledPrefill | None = None
        parent_outputs_consumed_by_autograd = torch.is_grad_enabled()
        parent_forward_inference_boundary = torch.is_inference_mode_enabled()
        # The parent is frozen. ``no_grad`` avoids retaining its activation
        # tape while preserving ordinary tensor identity for historical
        # recurrent/session buffers. ``inference_mode`` was marginally cheaper
        # inside the parent, but forced a full module-buffer traversal plus
        # output clones after every training wave.
        with torch.no_grad():
            if use_past:
                # Decode continuation: one new token, one KV telemetry record.
                if callable(record_kv):
                    record_kv(
                        prefix_positions=result_prefix_positions,
                        new_positions=result_new_positions,
                    )
                elif not self.exact_historical_source_loaded:
                    raise RuntimeError(
                        "Resynthesis integrated runtime lacks tensor-native KV telemetry"
                    )
                output = runtime(
                    input_ids=active_input_ids,
                    position_ids=position_ids,
                    attention_mask=runtime_attention_mask,
                    past_key_values=(
                        shared_prefix_cache
                        if shared_prefix_active
                        else self._decode_past_key_values
                    ),
                    output_hidden_states=True,
                    return_dict=True,
                    use_cache=not shared_prefix_active,
                    logits_to_keep=1,
                )
            else:
                # Native 4M+ tiled prefill: every token executes the parent.
                # The trained linear-attention state remains global; the full-
                # attention K/V becomes a bounded hot window only beyond the
                # parent's pretrained band. No token is dropped or truncated.
                transient_training_prefill = masked_training_prefill
                if transient_training_prefill and not explicit_attention_mask:
                    raise RuntimeError(
                        "masked training prefill requires an explicit attention mask"
                    )
                prefill_boundary = (
                    self._prefill_tiled_training_native
                    if transient_training_prefill
                    else self._prefill_tiled_native
                )
                prefill = prefill_boundary(
                    runtime=runtime,
                    active_input_ids=active_input_ids,
                    position_ids=position_ids,
                    attention_mask=runtime_attention_mask,
                    record_kv=record_kv,
                    starting_prefix=prefix_positions,
                    telemetry_required=not self.exact_historical_source_loaded,
                )
                output = prefill.runtime_output
        # The external parent owns recurrent/session buffers that may be
        # updated in its inference-mode forward.  Normalize those buffers
        # before the next RBO session reset, while the parent runtime is still
        # attached and before any trainable graph consumes the result.
        parent_rbo = getattr(runtime, "resynthesis_rbo", None)
        if (
            isinstance(parent_rbo, nn.Module)
            and (
                self._parent_session_may_hold_inference_tensors
                or parent_forward_inference_boundary
            )
        ):
            normalized = self._normalize_inference_session_buffers(parent_rbo)
            with torch.no_grad():
                self._parent_inference_session_buffers_normalized.add_(
                    normalized.to(
                        device=self._parent_inference_session_buffers_normalized.device
                    )
                )
            # The traversal above is exhaustive over registered historical
            # session buffers.  Its completion proves ordinary tensor identity;
            # extracting ``normalized`` as a Python scalar here only adds a
            # host boundary to every inference-to-training transition.
            self._parent_session_may_hold_inference_tensors = False
            self._parent_session_buffers_verified_normal = True
        returned_cache = getattr(output, "past_key_values", None)
        if shared_prefix_active:
            if shared_prefix_cache is None:
                raise RuntimeError("shared parent-prefix cache was lost")
            self._record_native_context_cache_telemetry_boundary(
                shared_prefix_cache
            )
            setattr(output, "past_key_values", None)
            self._decode_past_key_values = None
        else:
            self._decode_past_key_values = returned_cache
            if self._decode_past_key_values is not None:
                self._record_native_context_cache_telemetry_boundary(
                    self._decode_past_key_values
                )
        with torch.no_grad():
            if self._decode_past_key_values is None:
                self._decode_cached_positions.zero_()
            else:
                self._decode_cached_positions.add_(
                    runtime_new_positions.to(
                        device=self._decode_cached_positions.device
                    )
                )
        logits = getattr(output, "logits", None)
        hidden_states = getattr(output, "hidden_states", None)
        if not isinstance(logits, torch.Tensor) or not hidden_states:
            raise RuntimeError(
                "Resynthesis integrated forward returned no hidden/logit tensors"
            )
        parent_result = _resynthesis_parent_last_rbo_result(runtime)
        shaped_hidden = getattr(parent_result, "shaped_hidden", None)
        hidden = (
            shaped_hidden
            if isinstance(shaped_hidden, torch.Tensor)
            else hidden_states[-1]
        )
        if hidden.shape[:-1] != logits.shape[:-1]:
            raise RuntimeError(
                "Resynthesis integrated hidden/logit geometry differs: "
                f"hidden={tuple(hidden.shape)} logits={tuple(logits.shape)}"
            )
        parent_context_hidden = _parent_hidden_context(hidden)
        if shared_prefix_active:
            suffix_hidden = hidden_states[-1]
            if (
                not isinstance(suffix_hidden, torch.Tensor)
                or suffix_hidden.ndim != 3
                or suffix_hidden.shape[1] != 1
            ):
                raise RuntimeError(
                    "shared parent-prefix final hidden geometry differs"
                )
            parent_prefill_hidden = suffix_hidden
            assert shared_full_input_positions_t is not None
            parent_prefill_input_positions = shared_full_input_positions_t
        elif prefill is None:
            parent_prefill_hidden = hidden.new_empty(
                hidden.shape[0],
                0,
                hidden.shape[-1],
            )
            parent_prefill_input_positions = runtime_new_positions.new_zeros(())
        else:
            parent_prefill_hidden = prefill.summary_hidden.to(
                device=hidden.device,
                dtype=hidden.dtype,
            )
            parent_prefill_input_positions = prefill.input_positions.to(
                device=hidden.device,
                dtype=torch.long,
            )
        if masked_training_prefill and not use_past:
            raw_final_hidden_t = hidden_states[-1]
            if (
                not isinstance(raw_final_hidden_t, torch.Tensor)
                or raw_final_hidden_t.ndim != 3
                or raw_final_hidden_t.shape[1] != 1
                or parent_prefill_hidden.ndim != 3
                or parent_prefill_hidden.shape[0]
                != raw_final_hidden_t.shape[0]
                or parent_prefill_hidden.shape[2]
                != raw_final_hidden_t.shape[2]
            ):
                raise RuntimeError(
                    "frozen-backbone captured feature geometry differs"
                )
            input_positions_t = parent_prefill_input_positions.reshape(-1)
            if input_positions_t.numel() == 1:
                input_positions_t = input_positions_t.expand(
                    raw_final_hidden_t.shape[0]
                )
            if input_positions_t.shape != (raw_final_hidden_t.shape[0],):
                raise RuntimeError(
                    "frozen-backbone captured input positions differ"
                )
            self._last_frozen_backbone_final_hidden_t = (
                raw_final_hidden_t[:, -1:, :].detach()
            )
            self._last_frozen_backbone_summary_hidden_t = (
                parent_prefill_hidden.detach()
            )
            self._last_frozen_backbone_input_positions_t = (
                input_positions_t.detach().clone()
            )
        # Resynthesis has already causally contextualized the complete prefill.
        # The appended autoregressive RBO stack consumes the current position,
        # just as it does for every cached continuation. This preserves the
        # whole parent context without re-running four 4096-wide GRU/attention
        # expert layers over historical positions on every next-token decision.
        hidden = hidden[:, -1:, :]
        logits = logits[:, -1:, :]
        parent_rbo = getattr(runtime, "resynthesis_rbo", None)
        expert_routes = getattr(parent_rbo, "_last_none_actual_expert_routes", None)
        layer_routes = getattr(parent_rbo, "_last_none_actual_layer_routes", None)
        if not isinstance(expert_routes, torch.Tensor):
            expert_routes = hidden.new_empty(0)
        if not isinstance(layer_routes, torch.Tensor):
            layer_routes = hidden.new_empty(0)
        result = ResynthesisParentForward(
            hidden=(
                _regularize_parent_tensor_for_autograd_boundary(hidden)
                if parent_outputs_consumed_by_autograd
                else hidden.detach()
            ),
            logits=(
                _regularize_parent_tensor_for_autograd_boundary(logits)
                if parent_outputs_consumed_by_autograd
                else logits.detach()
            ),
            parent_context_hidden=(
                _regularize_parent_tensor_for_autograd_boundary(
                    parent_context_hidden
                )
                if parent_outputs_consumed_by_autograd
                else parent_context_hidden.detach()
            ),
            parent_expert_routes=(
                _regularize_parent_tensor_for_autograd_boundary(expert_routes)
                if parent_outputs_consumed_by_autograd
                else expert_routes.detach()
            ),
            parent_layer_routes=(
                _regularize_parent_tensor_for_autograd_boundary(layer_routes)
                if parent_outputs_consumed_by_autograd
                else layer_routes.detach()
            ),
            kv_prefix_positions=(
                _regularize_parent_tensor_for_autograd_boundary(
                    result_prefix_positions
                )
                if parent_outputs_consumed_by_autograd
                else result_prefix_positions.detach()
            ),
            kv_new_positions=(
                _regularize_parent_tensor_for_autograd_boundary(
                    result_new_positions
                )
                if parent_outputs_consumed_by_autograd
                else result_new_positions.detach()
            ),
            parent_prefill_hidden=(
                _regularize_parent_tensor_for_autograd_boundary(
                    parent_prefill_hidden
                )
                if parent_outputs_consumed_by_autograd
                else parent_prefill_hidden.detach()
            ),
            parent_prefill_input_positions=(
                _regularize_parent_tensor_for_autograd_boundary(
                    parent_prefill_input_positions
                )
                if parent_outputs_consumed_by_autograd
                else parent_prefill_input_positions.detach()
            ),
        )
        self._last_forward = result
        return result

    def begin_decode(self) -> None:
        """Reset frozen feature/KV plumbing without granting parent authority."""

        self._decode_past_key_values = None
        self._decode_authority_active = False
        self._last_forward = None
        with torch.no_grad():
            self._decode_cached_positions.zero_()
            self._native_context_hybrid_active.zero_()
            self._native_context_total_positions.zero_()
            self._native_context_hot_resident_positions.zero_()
            self._native_context_hot_window_tokens.zero_()
            self._native_context_recurrent_layer_count.zero_()
            self._native_context_full_attention_layer_count.zero_()

    @staticmethod
    def _normalize_inference_session_buffers(parent_rbo: nn.Module) -> torch.Tensor:
        """Restore ordinary tensor identity at the inference/training boundary.

        The verified historical parent predates its later recurrent-buffer
        repair. An inference-mode decode can therefore replace a registered
        session buffer with an inference tensor. Such a tensor cannot be reset
        by the subsequent training transaction outside inference mode. This is
        a caller-session compatibility boundary: it clones only registered
        buffers that actually carry inference identity and leaves parameters,
        trained values, routing, and immutable parent source untouched.
        """

        normalized = torch.zeros((), dtype=torch.long)
        with torch.inference_mode(False):
            for module in parent_rbo.modules():
                for name, buffer in module.named_buffers(recurse=False):
                    if buffer.is_inference():
                        setattr(module, name, buffer.detach().clone())
                        normalized.add_(torch.ones_like(normalized))
        return normalized

    def begin_session(self) -> None:
        """Reset caller-owned parent state without changing trained tensors."""

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        parent_rbo = getattr(runtime, "resynthesis_rbo", None)
        reset_session = getattr(parent_rbo, "reset_session_state", None)
        if not isinstance(parent_rbo, nn.Module) or not callable(reset_session):
            raise RuntimeError("Resynthesis integrated RBO lacks caller-session reset")
        inference_mode = torch.is_inference_mode_enabled()
        if not inference_mode and (
            self._parent_session_may_hold_inference_tensors
            or not self._parent_session_buffers_verified_normal
        ):
            normalized = self._normalize_inference_session_buffers(parent_rbo)
            with torch.no_grad():
                self._parent_inference_session_buffers_normalized.add_(
                    normalized.to(
                        device=self._parent_inference_session_buffers_normalized.device
                    )
                )
            self._parent_session_may_hold_inference_tensors = False
            self._parent_session_buffers_verified_normal = True
        reset_session()
        if inference_mode:
            self._parent_session_may_hold_inference_tensors = True
            self._parent_session_buffers_verified_normal = False
        self.begin_decode()

    def session_transition_receipt_boundary(self) -> dict[str, object]:
        """Serialize inference-to-training compatibility only at log boundary."""

        return {
            "schema": "nnf.resynthesis.parent_session_transition.v1",
            "normalizedInferenceSessionBuffers": int(
                self._parent_inference_session_buffers_normalized.detach()
                .to(device="cpu", dtype=torch.long)
                .reshape(())
            ),
            "parentSessionMayHoldInferenceTensors": (
                self._parent_session_may_hold_inference_tensors
            ),
            "parentSessionBuffersVerifiedNormal": (
                self._parent_session_buffers_verified_normal
            ),
            "trainedParametersChangedByTransition": False,
            "immutableParentSourceChanged": False,
            "baseForwardCacheOwner": "ResynthesisRBO",
        }

    def forward_hidden(self, input_ids: torch.Tensor) -> torch.Tensor:
        """Compatibility boundary for callers that only consume parent hidden."""

        return self.forward_hidden_logits(input_ids).hidden

    def forward_logits(self, hidden: torch.Tensor) -> torch.Tensor:
        """Project through Resynthesis's frozen native RBO-owned generation head.

        Parent parameters remain frozen, while autograd is retained with respect
        to ``hidden`` so the appended Resynthesis experts receive real CE signal.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
        if not isinstance(head, nn.Module):
            raise RuntimeError("Resynthesis integrated runtime has no native generation head")
        logits = head(hidden)
        if not isinstance(logits, torch.Tensor):
            raise RuntimeError(
                "Resynthesis native generation head returned a non-tensor value"
            )
        return logits

    def forward_low_rank_logits_projection(
        self,
        hidden_projection: torch.Tensor,
    ) -> torch.Tensor:
        """Compose the frozen native head with a trainable hidden projection.

        For ``U: [hidden, rank]`` this returns ``W_head @ U`` in FP32. The
        full-vocabulary result and gradient are exact; only the associative
        order of the two linear projections changes. The detached forward value
        is reused across waves until the optimizer mutates ``U``. Each wave
        attaches an exact tiled ``W_head.T @ grad`` backward, so reuse never
        severs the trainable correction path.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        head = getattr(getattr(runtime, "backbone", None), "lm_head", None)
        native_rbo = getattr(head, "_resynthesis_rbo", None)
        frozen_weight = getattr(native_rbo, "_frozen_lm_weight", None)
        if (
            not isinstance(head, nn.Module)
            or getattr(head, "training_rbo_active", False) is not True
            or not isinstance(frozen_weight, torch.Tensor)
        ):
            raise RuntimeError(
                "Resynthesis native linear head is unavailable for low-rank completion"
            )
        if (
            hidden_projection.ndim != 2
            or hidden_projection.shape[0] != frozen_weight.shape[1]
            or hidden_projection.device != frozen_weight.device
        ):
            raise ValueError("low-rank completion projection geometry differs")
        source_id = id(hidden_projection)
        source_version = hidden_projection._version
        projection = self._training_low_rank_projection_cache
        if (
            projection is None
            or self._training_low_rank_projection_source_id != source_id
            or self._training_low_rank_projection_source_version != source_version
            or projection.device != hidden_projection.device
            or projection.shape
            != (frozen_weight.shape[0], hidden_projection.shape[1])
        ):
            projection = hidden_projection.new_empty(
                (frozen_weight.shape[0], hidden_projection.shape[1]),
                dtype=torch.float32,
            )
            projection_fp32 = hidden_projection.detach().float()
            with torch.no_grad():
                for start in range(
                    0,
                    frozen_weight.shape[0],
                    _FROZEN_HEAD_PROJECTION_TILE_ROWS,
                ):
                    end = min(
                        start + _FROZEN_HEAD_PROJECTION_TILE_ROWS,
                        frozen_weight.shape[0],
                    )
                    projection[start:end] = torch.mm(
                        frozen_weight[start:end].float(),
                        projection_fp32,
                    )
            self._training_low_rank_projection_cache = projection
            self._training_low_rank_projection_source_id = source_id
            self._training_low_rank_projection_source_version = source_version
        return cast(
            torch.Tensor,
            _CachedFrozenHeadLowRankProjection.apply(  # type: ignore[no-untyped-call]
                hidden_projection,
                frozen_weight,
                projection,
            ),
        )

    def native_decode_stop(
        self,
        hidden: torch.Tensor,
        logits: torch.Tensor,
        generated_ids: torch.Tensor,
    ) -> ResynthesisNativeDecodeStop:
        """Observe the exact parent's legacy completion surface diagnostically.

        Only the active final hidden/logit position is scored. ``generated_ids``
        contains model-emitted continuation tokens, never the immutable prompt.
        No EOS, host token count, or RBO traversal heuristic participates.  The
        returned tensors have no answer, stop, retention, or veto authority;
        the additive Resynthesis completion graph owns those decisions.
        """

        if not self._weights_loaded:
            self.load_weights()
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        surface = getattr(runtime, "native_decode_confidence", None)
        if not isinstance(surface, nn.Module):
            raise RuntimeError(
                "Resynthesis parent has no trained native decode-confidence surface"
            )
        if (
            hidden.ndim != 3
            or logits.ndim != 3
            or hidden.shape[:2] != logits.shape[:2]
            or generated_ids.ndim != 2
            or hidden.shape[0] != generated_ids.shape[0]
            or generated_ids.shape[1] < 1
        ):
            raise RuntimeError("Resynthesis native decode-confidence input geometry differs")
        with torch.no_grad():
            raw_score = surface(
                hidden=hidden[:, -1:, :],
                logits=logits[:, -1:, :],
                generated_ids=generated_ids,
            )
        if not isinstance(raw_score, torch.Tensor):
            raise RuntimeError(
                "Resynthesis native decode-confidence surface returned no tensor"
            )
        score = raw_score.reshape(hidden.shape[0], -1)[:, -1]
        probability = score.sigmoid()
        decision = probability.ge(0.5)
        return ResynthesisNativeDecodeStop(
            score=score,
            probability=probability,
            decision=decision,
        )

    def apply_native_answer_surface(
        self,
        hidden: torch.Tensor,
        logits: torch.Tensor,
    ) -> ResynthesisNativeAnswer:
        """Encode additive logits with the parent's immutable vocabulary map.

        ``hidden`` is accepted only to preserve the tensor-native historical
        interface.  It cannot enter answer selection.  In particular, this
        boundary deliberately does not call the frozen parent's active RBO or
        answer head: Resynthesis has already produced the authoritative logits,
        and the parent supplies only token-to-bit/glyph vocabulary plumbing.
        """

        del hidden
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        vge = getattr(runtime, "resynthesis_vge", None)
        token_to_glyph = getattr(vge, "token_ids_to_glyph", None)
        token_to_bits = getattr(vge, "token_ids_to_bit_ids", None)
        if not callable(token_to_glyph) or not callable(token_to_bits):
            raise RuntimeError("frozen parent VGE lacks its immutable bit mapping")
        if logits.ndim != 3 or logits.shape[1] < 1:
            raise RuntimeError("additive vocabulary logits geometry differs")
        token_ids = logits[:, -1:, :].argmax(dim=-1)
        glyph = token_to_glyph(token_ids)
        bit_ids = token_to_bits(token_ids)
        if not all(
            isinstance(value, torch.Tensor)
            for value in (token_ids, glyph, bit_ids)
        ):
            raise RuntimeError("frozen parent VGE returned a non-tensor packet")
        return ResynthesisNativeAnswer(
            logits=logits,
            token_ids=token_ids,
            bit_ids=bit_ids,
            glyph=glyph,
        )

    def native_token_glyphs_loss_boundary(
        self,
        token_ids_t: torch.Tensor,
    ) -> torch.Tensor:
        """Map token identities through immutable vocabulary plumbing only.

        Object-native knowledge absorption already owns the verified token
        identities at its explicit loss boundary.  Re-running the frozen 12B
        decoder merely to recover their vocabulary coordinates would make that
        historical parent the training bottleneck again.  This boundary calls
        only the parent's immutable VGE lookup: it performs no parent forward,
        routing, answer selection, teacher projection, or parameter update.

        The same interface maps target-free context IDs and verified value IDs;
        callers remain responsible for keeping value glyphs on the loss side
        so ground-truth-derived tensors never enter ``forward_thinking``.
        """

        if (
            token_ids_t.ndim < 1
            or token_ids_t.dtype not in (torch.int32, torch.int64)
        ):
            raise ValueError(
                "native vocabulary glyph lookup requires integer token tensors"
            )
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        vge = getattr(runtime, "resynthesis_vge", None)
        token_to_glyph = getattr(vge, "token_ids_to_glyph", None)
        if not callable(token_to_glyph):
            raise RuntimeError(
                "frozen parent VGE lacks its immutable glyph mapping"
            )
        glyph_t = token_to_glyph(token_ids_t)
        if (
            not isinstance(glyph_t, torch.Tensor)
            or glyph_t.ndim != token_ids_t.ndim + 1
            or glyph_t.shape[:-1] != token_ids_t.shape
            or glyph_t.shape[-1] < 1
            or not glyph_t.is_floating_point()
            or glyph_t.device != token_ids_t.device
        ):
            raise RuntimeError(
                "frozen parent VGE returned an invalid glyph tensor"
            )
        torch._assert_async(
            torch.isfinite(glyph_t).all(),
            "frozen parent VGE returned non-finite glyph coordinates",
        )
        return glyph_t

    def apply_execution_outcome(
        self,
        outcome_features: torch.Tensor,
    ) -> ResynthesisOutcomeApplication:
        """Apply one persisted verifier/tool outcome to the exact parent route.

        The eight Resynthesis fields are observations only. They are translated
        to Resynthesis's fixed execution-memory geometry without fabricating unknown
        code-specific evidence such as patch application or timeout status.
        """

        if outcome_features.shape != (1, 8):
            raise ValueError("one parent execution outcome must have shape [1, 8]")
        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        apply_outcome = getattr(
            runtime,
            "resynthesis_apply_current_turn_execution_outcome",
            None,
        )
        observed = outcome_features.detach().to(device=self.device, dtype=torch.float32)
        passed = observed[0, 0].clamp(0.0, 1.0)
        failed = observed[0, 1].clamp(0.0, 1.0)
        score = observed[0, 2].clamp(0.0, 1.0)
        coverage = observed[0, 3].clamp(0.0, 1.0)
        verified = observed[0, 4].clamp(0.0, 1.0)
        infrastructure_failure = observed[0, 6].clamp(0.0, 1.0)
        zero = observed.new_zeros(())
        observation = torch.stack(
            (
                zero,
                verified,
                passed,
                torch.maximum(failed, infrastructure_failure),
                zero,
                failed * verified,
                zero,
                zero,
                score,
                coverage,
            )
        )
        acquisition = torch.stack(
            (
                failed * verified,
                zero,
                zero,
                passed * verified,
            )
        )
        if not callable(apply_outcome):
            if not self.exact_historical_source_loaded:
                raise RuntimeError(
                    "Resynthesis integrated runtime lacks its trained outcome boundary"
                )
            parent_rbo = getattr(runtime, "resynthesis_rbo", None)
            capture = getattr(parent_rbo, "capture_session_state", None)
            apply_historical = getattr(
                parent_rbo,
                "apply_execution_observation_from_boundary",
                None,
            )
            if not callable(capture) or not callable(apply_historical):
                raise RuntimeError(
                    "historical Resynthesis RBO lacks outcome-memory boundaries"
                )
            parent_result = _resynthesis_parent_last_rbo_result(runtime)
            domain_idx = int(getattr(parent_result, "domain_idx", 0))
            subdomain_idx = int(getattr(parent_result, "subdomain_idx", 0))
            before = capture()
            apply_historical(
                observation,
                domain_idx=domain_idx,
                subdomain_idx=subdomain_idx,
                resolved=bool(passed.eq(1.0).detach().to(device="cpu")),
            )
            after = capture()
            before_values = vars(before)
            after_values = vars(after)
            if set(before_values) != set(after_values):
                raise RuntimeError("historical Resynthesis outcome-state geometry changed")
            squared = observation.new_zeros((), dtype=torch.float32)
            for name in sorted(before_values):
                before_value = before_values[name]
                after_value = after_values[name]
                if isinstance(before_value, torch.Tensor):
                    if not isinstance(after_value, torch.Tensor) or (
                        before_value.shape != after_value.shape
                    ):
                        raise RuntimeError(
                            "historical Resynthesis outcome-state tensor geometry differs"
                        )
                    delta = after_value.detach().to(
                        device=observation.device,
                        dtype=torch.float32,
                    ) - before_value.detach().to(
                        device=observation.device,
                        dtype=torch.float32,
                    )
                    squared = squared + delta.square().sum()
            rbo_delta = squared.sqrt()
            route_changed = rbo_delta.gt(0)
            if not bool(route_changed.detach().to(device="cpu")):
                raise RuntimeError(
                    "persisted outcome did not change historical RBO state"
                )
            zero_delta = rbo_delta.new_zeros(())
            return ResynthesisOutcomeApplication(
                applied=route_changed,
                rbo_state_delta_l2=rbo_delta,
                arm_state_delta_l1=zero_delta,
                legacy_state_delta_l2=zero_delta,
                route_state_changed=route_changed,
            )
        packet = apply_outcome(
            observation,
            acquisition=acquisition,
            resolved=passed.eq(1.0),
        )
        packet_applied = getattr(packet, "applied", None)
        packet_rbo_delta = getattr(packet, "rbo_state_delta_l2", None)
        packet_arm_delta = getattr(packet, "arm_state_delta_l1", None)
        packet_legacy_delta = getattr(packet, "legacy_state_delta_l2", None)
        packet_route_changed = getattr(packet, "route_state_changed", None)
        if not all(
            isinstance(value, torch.Tensor)
            for value in (
                packet_applied,
                packet_rbo_delta,
                packet_arm_delta,
                packet_legacy_delta,
                packet_route_changed,
            )
        ):
            raise RuntimeError(
                "Resynthesis outcome boundary returned an invalid tensor packet"
            )
        assert isinstance(packet_applied, torch.Tensor)
        assert isinstance(packet_rbo_delta, torch.Tensor)
        assert isinstance(packet_arm_delta, torch.Tensor)
        assert isinstance(packet_legacy_delta, torch.Tensor)
        assert isinstance(packet_route_changed, torch.Tensor)
        return ResynthesisOutcomeApplication(
            applied=packet_applied,
            rbo_state_delta_l2=packet_rbo_delta,
            arm_state_delta_l1=packet_arm_delta,
            legacy_state_delta_l2=packet_legacy_delta,
            route_state_changed=packet_route_changed,
        )

    def current_acquisition_policy(self) -> ResynthesisAcquisitionPolicy:
        """Read the trained parent action after a real outcome was ingested."""

        runtime = self.runtime
        if runtime is None:
            raise RuntimeError("Resynthesis integrated runtime was not attached")
        active_rbo_fn = getattr(runtime, "_active_resynthesis_rbo", None)
        if not callable(active_rbo_fn):
            raise RuntimeError("Resynthesis integrated runtime has no active RBO boundary")
        active_rbo = active_rbo_fn(required=True)
        execution_grounding = getattr(active_rbo, "execution_grounding", None)
        policy_fn = getattr(execution_grounding, "acquisition_policy", None)
        authority = getattr(execution_grounding, "acquisition_policy_authority", None)
        if not callable(policy_fn) or not isinstance(authority, torch.Tensor):
            raise RuntimeError("Resynthesis RBO has no trained evidence-acquisition policy")
        policy = policy_fn(device=self.device, dtype=torch.float32)
        action_probs = getattr(policy, "action_probs", None)
        action_index = getattr(policy, "action_index", None)
        confidence = getattr(policy, "confidence", None)
        observation_count = getattr(policy, "observation_count", None)
        acquisition_count = getattr(policy, "acquisition_count", None)
        values = (
            action_probs,
            action_index,
            confidence,
            observation_count,
            acquisition_count,
        )
        if not all(isinstance(value, torch.Tensor) for value in values):
            raise RuntimeError(
                "Resynthesis acquisition policy returned an invalid tensor packet"
            )
        assert isinstance(action_probs, torch.Tensor)
        assert isinstance(action_index, torch.Tensor)
        assert isinstance(confidence, torch.Tensor)
        assert isinstance(observation_count, torch.Tensor)
        assert isinstance(acquisition_count, torch.Tensor)
        if action_probs.shape != (4,):
            raise RuntimeError("Resynthesis acquisition action geometry differs")
        return ResynthesisAcquisitionPolicy(
            action_probs=action_probs.detach(),
            action_index=action_index.detach().reshape(()),
            confidence=confidence.detach().reshape(()),
            observation_count=observation_count.detach().reshape(()),
            acquisition_count=acquisition_count.detach().reshape(()),
            authority=authority.detach().reshape(()),
        )

    def freeze(self) -> "ResynthesisNativeParent":
        """Ensure all base parameters are frozen (requires_grad=False)."""
        for p in self.parameters():
            p.requires_grad_(False)
        for b in self.buffers():
            b.requires_grad_(False)
        return self


# Historical names remain import-compatible, but all factories construct the
# Resynthesis-native owner. They are aliases only and never redirect loading.
FrozenResynthesisParent = ResynthesisNativeParent
FrozenNexumBase = ResynthesisNativeParent


def build_frozen_base(
    cfg: ResynthesisConfig | None = None,
    *,
    device: torch.device | str = "cpu",
    lazy: bool = True,
) -> ResynthesisNativeParent:
    """Factory: build the frozen Resynthesis native graph parent.

    Args:
        cfg: Resynthesis config (defaults to ResynthesisConfig()).
        device: torch device.
        lazy: if True, weights are loaded on first forward_hidden() call.
              if False, weights are loaded immediately.
    """
    cfg = cfg or ResynthesisConfig()
    base = ResynthesisNativeParent(cfg, device=device)
    if not lazy:
        base.load_weights()
    base.freeze()
    return base