runtime/src/resynthesis/parent_bridge.py

"""Parent NoNE bridge for Resynthesis.

Binds Resynthesis to the master_tim_context InternalModelFabric (Pillar 16).
The fabric participates once per phase in every decode via ``step_phase``
(not once per decode). Expert/layer coordination, partial-construction slots,
traversal tracking, and self-improvement signals are model-owned tensors.

Parent absence = construction failure for NoNE (not optional on the hot path).
The ``ResynthesisNoNEFabric`` is the tensor-native bridge packet that injects
expert/layer biases, intent indices, and decision surfaces into the parent fabric.
"""
from __future__ import annotations

import weakref
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, cast

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

from resynthesis.config import RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS
from resynthesis.science_layers import online_softmax_last_token_pool

if TYPE_CHECKING:
    from resynthesis.rbo import RBOResult


# STACK+COMPOSE: fabric long pool tile matches Dual-Chunk successive window.
LONG_CONTEXT_FABRIC_CHUNK_TOKENS = RESYNTHESIS_ONLINE_SOFTMAX_TILE_TOKENS
PARENT_ROUTE_GATE_PARAMETERIZATION_VERSION = 2


def _fabric_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 Fabric hidden position.

    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 fabric pool.
    """

    return online_softmax_last_token_pool(
        hidden,
        context_intent=context_intent,
        context_action=context_action,
        chunk_tokens=LONG_CONTEXT_FABRIC_CHUNK_TOKENS,
        intent_additive_gate=intent_additive_gate,
        action_additive_gate=action_additive_gate,
        intent_multiplicative_gate=intent_multiplicative_gate,
    )


DRAFTING_LIFECYCLE_NAMES: tuple[str, ...] = (
    "reproduction",
    "triage",
    "drafting",
    "experiment",
    "peer_critique",
    "revision",
    "disclosure",
    "patching",
    "verification",
    "submission",
)


@dataclass(frozen=True)
class NoNEDraftingLifecyclePacket:
    """Tensor-only multi-expert drafting, critique, and verification state.

    The packet is produced from the same trained NoNE phase and expert tensors
    that route the active model.  Independently executed verifier evidence is
    supplied only through the explicit outcome boundary; absent or failed
    evidence cannot become a positive verification signal.
    """

    lifecycle_scores: torch.Tensor
    worker_assignments: torch.Tensor
    reproduction_states: torch.Tensor
    triage_weights: torch.Tensor
    worker_draft_states: torch.Tensor
    experiment_plan_state: torch.Tensor
    peer_review_routes: torch.Tensor
    critique_state: torch.Tensor
    revised_draft_state: torch.Tensor
    disclosure_state: torch.Tensor
    patch_state: torch.Tensor
    verification_state: torch.Tensor
    revision_delta: torch.Tensor
    draft_residual_scale: torch.Tensor
    draft_proposal_gate: torch.Tensor
    draft_proposal_delta: torch.Tensor
    experiment_requirement: torch.Tensor
    experiment_observed: torch.Tensor
    experiment_passed: torch.Tensor
    revision_gate: torch.Tensor
    verified_revision_delta: torch.Tensor
    submission_proposal: torch.Tensor
    submission_readiness: torch.Tensor


@dataclass(frozen=True)
class NoNEFabricRoutePacket:
    """Tensor-native route packet injected into the parent fabric.

    All fields are tensors — no string-key dicts, no Python scalars.
    This is the boundary contract between Resynthesis and the parent fabric.
    """

    expert_bias: torch.Tensor          # [batch, num_experts] additive bias
    layer_bias: torch.Tensor           # [batch, num_layers] additive bias
    intent_index: torch.Tensor         # [num_experts, glyph_dim] intent glyphs
    decision_surface: torch.Tensor     # [batch, num_experts] routing decision
    phase_surface: torch.Tensor        # [batch, num_phases] phase participation
    graph_surface: torch.Tensor        # [num_domains] domain residency
    residency_prediction: torch.Tensor # [batch, num_layers] residency prediction
    transfer_active: torch.Tensor      # [num_clusters] transfer activation
    parent_expert_route: torch.Tensor  # [batch, num_experts] executed parent route
    parent_layer_route: torch.Tensor   # [batch, num_layers] executed parent route
    parent_route_conditioning: torch.Tensor  # scalar causal-conditioning mass
    parent_fabric_connected: torch.Tensor    # scalar real-object attachment proof


class ResynthesisNoNEFabric(nn.Module):
    """Tensor-native NoNE fabric bridge for Resynthesis.

    Wraps the science stack + RBO and produces ``NoNEFabricRoutePacket`` tensors
    for the parent InternalModelFabric. The fabric participates once per phase
    via ``step_phase``.

    Session-owned: state resumes without cross-session mixing.
    """

    _session_step: torch.Tensor
    _parent_route_uses: torch.Tensor
    _last_parent_route_conditioning: torch.Tensor
    parent_route_gate_parameterization_t: torch.Tensor
    parent_connected: torch.Tensor

    def __init__(
        self,
        *,
        hidden_size: int,
        num_experts: int,
        num_layers: int,
        glyph_dim: int,
        num_domains: int = 8,
        num_clusters: int = 10,
        num_phases: int = 10,
    ) -> None:
        super().__init__()
        for name, value in (
            ("hidden_size", hidden_size),
            ("num_experts", num_experts),
            ("num_layers", num_layers),
            ("glyph_dim", glyph_dim),
            ("num_domains", num_domains),
            ("num_clusters", num_clusters),
            ("num_phases", num_phases),
        ):
            if (
                not isinstance(value, int)
                or isinstance(value, bool)
                or value < 1
            ):
                raise ValueError(f"{name} must be a positive runtime geometry")
        self.hidden_size = int(hidden_size)
        self.num_experts = int(num_experts)
        self.num_layers = int(num_layers)
        self.glyph_dim = int(glyph_dim)
        self.num_domains = int(num_domains)
        self.num_clusters = int(num_clusters)
        self.num_phases = int(num_phases)
        # Learned fabric surfaces (all tensors)
        self.expert_bias_table = nn.Parameter(torch.zeros(self.num_experts))
        self.layer_bias_table = nn.Parameter(torch.zeros(self.num_layers))
        self.intent_table = nn.Parameter(torch.randn(self.num_experts, self.glyph_dim) * 0.02)
        self.domain_residency = nn.Parameter(torch.zeros(self.num_domains))
        self.transfer_table = nn.Parameter(torch.zeros(self.num_clusters))
        self.phase_proj = nn.Linear(self.hidden_size, self.num_phases, bias=False)
        self.decision_proj = nn.Linear(self.hidden_size, self.num_experts, bias=False)
        self.expert_hidden_table = nn.Parameter(torch.randn(self.num_experts, self.hidden_size) * 0.02)
        self.layer_hidden_table = nn.Parameter(torch.randn(self.num_layers, self.hidden_size) * 0.02)
        self.phase_hidden_table = nn.Parameter(torch.randn(self.num_phases, self.hidden_size) * 0.02)
        self.residual_gate = nn.Parameter(torch.tensor(-4.0))
        self.parent_expert_route_gate = nn.Parameter(torch.zeros(()))
        self.parent_layer_route_gate = nn.Parameter(torch.zeros(()))
        # Session state (resumes without cross-session mixing)
        self.register_buffer("_session_step", torch.tensor(0, dtype=torch.long), persistent=True)
        self.register_buffer("_parent_route_uses", torch.tensor(0, dtype=torch.long), persistent=True)
        self.register_buffer(
            "_last_parent_route_conditioning",
            torch.zeros(()),
            persistent=True,
        )
        # v1 interpreted a zero route-gate logit through sigmoid as 0.5, which
        # silently made the frozen vocabulary parent a dominant prior.  v2
        # centers that same learned logit at sigmoid(0), preserving parameter
        # bytes and optimizer moments while making zero exactly neutral.
        self.register_buffer(
            "parent_route_gate_parameterization_t",
            torch.tensor(
                PARENT_ROUTE_GATE_PARAMETERIZATION_VERSION,
                dtype=torch.long,
            ),
            persistent=True,
        )
        self.register_buffer("parent_connected", torch.zeros((), dtype=torch.bool), persistent=True)
        self._parent_fabric_ref: weakref.ReferenceType[nn.Module] | None = None

    def connect_parent(self, parent_fabric: nn.Module) -> None:
        """Attach the exact loaded parent fabric without registering it twice.

        The parent is already a child of ``FrozenResynthesisParent``.  A weak reference
        preserves that single ownership path and prevents its frozen tensors
        from being duplicated in the additive state dict.
        """

        if not isinstance(parent_fabric, nn.Module):
            raise TypeError("parent NoNE fabric must be an nn.Module")
        if not callable(getattr(parent_fabric, "step_phase", None)):
            raise RuntimeError("parent NoNE fabric exposes no step_phase contract")
        attached = self._parent_fabric_ref() if self._parent_fabric_ref is not None else None
        if attached is not None and attached is not parent_fabric:
            raise RuntimeError("parent NoNE fabric identity changed inside the active graph")
        self._parent_fabric_ref = weakref.ref(parent_fabric)
        self.parent_connected.fill_(True)

    def reset_session(self) -> None:
        """Reset caller-local traversal evidence while retaining graph identity."""

        self._session_step.zero_()
        self._parent_route_uses.zero_()
        self._last_parent_route_conditioning.zero_()

    def _require_parent(self) -> nn.Module:
        parent = self._parent_fabric_ref() if self._parent_fabric_ref is not None else None
        if parent is None:
            self.parent_connected.zero_()
            raise RuntimeError("Resynthesis NoNE has no live parent NoNE fabric attachment")
        return parent

    @staticmethod
    def _condition_parent_route(
        route_history: torch.Tensor,
        *,
        batch_size: int,
        target_width: int,
        reference: torch.Tensor,
    ) -> torch.Tensor:
        """Pool an executed parent route history without host scalar extraction."""

        if route_history.ndim < 2 or route_history.numel() == 0:
            raise RuntimeError("parent NoNE produced no executed route history")
        if route_history.shape[-2] != batch_size:
            raise ValueError("parent NoNE route batch geometry differs from hidden state")
        history = route_history.to(device=reference.device, dtype=reference.dtype)
        route = history.reshape(-1, batch_size, history.shape[-1]).mean(dim=0)
        pooled = F.adaptive_avg_pool1d(route.unsqueeze(1), target_width).squeeze(1)
        nonnegative = pooled.clamp_min(0.0)
        return nonnegative / nonnegative.sum(dim=-1, keepdim=True).clamp_min(1.0e-9)

    def step_phase(
        self,
        hidden: torch.Tensor,
        parent_expert_routes: torch.Tensor,
        parent_layer_routes: torch.Tensor,
    ) -> NoNEFabricRoutePacket:
        """Participate once per phase. Produces the route packet for the parent.

        Pillar 16: this is called once per phase (not once per decode).
        Parent NoNE has already participated in the frozen parent forward.  Its
        executed route tensors condition this additive phase; its ``step_phase``
        is deliberately not invoked again.
        """
        self._require_parent()
        if hidden.ndim != 3:
            raise ValueError("fabric hidden state must have shape [batch, sequence, hidden]")
        parent_expert_route = self._condition_parent_route(
            parent_expert_routes,
            batch_size=hidden.shape[0],
            target_width=self.num_experts,
            reference=hidden,
        )
        parent_layer_route = self._condition_parent_route(
            parent_layer_routes,
            batch_size=hidden.shape[0],
            target_width=self.num_layers,
            reference=hidden,
        )
        self._session_step += 1
        self._parent_route_uses += 1
        expert_table = self.expert_hidden_table.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        route_intent_c = torch.matmul(parent_expert_route, expert_table)
        context_intent = route_intent_c.unsqueeze(1).expand(-1, hidden.shape[1], -1)
        layer_table = self.layer_hidden_table.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        route_action_c = torch.matmul(parent_layer_route, layer_table)
        context_action = route_action_c.unsqueeze(1).expand(-1, hidden.shape[1], -1)
        # The existing parameters remain logits so accepted checkpoints and
        # their Adam moments retain exact coordinates.  Centering the sigmoid
        # keeps its derivative unchanged but removes the accidental 0.5 prior.
        expert_gate = (
            torch.sigmoid(self.parent_expert_route_gate)
            - self.parent_expert_route_gate.new_tensor(0.5)
        ).to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        layer_gate = (
            torch.sigmoid(self.parent_layer_route_gate)
            - self.parent_layer_route_gate.new_tensor(0.5)
        ).to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        pooled = _fabric_hidden_context(
            hidden,
            context_intent=context_intent,
            context_action=context_action,
            intent_additive_gate=expert_gate,
            action_additive_gate=layer_gate,
        )
        expert_log_prior = parent_expert_route.clamp_min(1.0e-9).log()
        expert_log_prior = expert_log_prior - expert_log_prior.mean(dim=-1, keepdim=True)
        decision_surface = self.decision_proj(pooled) + expert_gate * expert_log_prior
        phase_surface = self.phase_proj(pooled)        # [batch, num_phases]
        expert_bias = (
            torch.sigmoid(self.expert_bias_table).unsqueeze(0)
            + expert_gate * parent_expert_route
        )
        layer_bias = (
            torch.sigmoid(self.layer_bias_table).unsqueeze(0)
            + layer_gate * parent_layer_route
        )
        parent_route_conditioning = (
            expert_gate * parent_expert_route.square().mean().sqrt()
            + layer_gate * parent_layer_route.square().mean().sqrt()
        )
        self._last_parent_route_conditioning.copy_(
            parent_route_conditioning.detach().to(
                device=self._last_parent_route_conditioning.device,
                dtype=self._last_parent_route_conditioning.dtype,
            )
        )
        return NoNEFabricRoutePacket(
            expert_bias=expert_bias,
            layer_bias=layer_bias,
            intent_index=torch.nn.functional.normalize(self.intent_table, dim=-1),
            decision_surface=decision_surface,
            phase_surface=phase_surface,
            graph_surface=torch.sigmoid(self.domain_residency),
            residency_prediction=layer_bias,
            transfer_active=torch.sigmoid(self.transfer_table),
            parent_expert_route=parent_expert_route,
            parent_layer_route=parent_layer_route,
            parent_route_conditioning=parent_route_conditioning,
            parent_fabric_connected=self.parent_connected.to(device=hidden.device),
        )

    def draft_lifecycle(
        self,
        hidden: torch.Tensor,
        outcome_features: torch.Tensor,
        outcome_present: torch.Tensor,
    ) -> NoNEDraftingLifecyclePacket:
        """Run parallel expert drafts and peer revision inside the model graph.

        The ten lifecycle scores reuse the trained NoNE phase surface in the
        order declared by :data:`DRAFTING_LIFECYCLE_NAMES`.  Each adaptive
        expert owns one draft state, every expert reviews every other expert,
        and only a real verifier observation can activate correction revision
        or satisfy experiment verification.  No target or gold answer enters
        this method.
        """

        if hidden.ndim != 3 or hidden.shape[-1] != self.hidden_size:
            raise ValueError(
                "draft lifecycle hidden state must have shape [batch, sequence, hidden]"
            )
        if outcome_features.shape != (hidden.shape[0], 8):
            raise ValueError("draft lifecycle outcome geometry must be [batch, 8]")
        if outcome_present.shape != (hidden.shape[0], 1):
            raise ValueError("draft lifecycle outcome-presence geometry must be [batch, 1]")
        if self.num_phases != len(DRAFTING_LIFECYCLE_NAMES):
            raise RuntimeError("NoNE phase geometry differs from the drafting lifecycle")

        pooled = _fabric_hidden_context(hidden)
        if self.transfer_table.shape != (self.num_phases,):
            raise RuntimeError("NoNE transfer and drafting phase geometry differs")
        lifecycle_scores = torch.sigmoid(
            self.phase_proj(pooled)
            + self.transfer_table.to(
                device=hidden.device,
                dtype=hidden.dtype,
            ).unsqueeze(0)
        )
        worker_assignments = torch.softmax(self.decision_proj(pooled), dim=-1)
        expert_table = self.expert_hidden_table.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        phase_table = self.phase_hidden_table.to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        reproduction_states = torch.tanh(
            pooled.unsqueeze(1)
            + expert_table.unsqueeze(0)
            + lifecycle_scores[:, 0:1].unsqueeze(-1)
            * phase_table[0].reshape(1, 1, -1)
        )
        triage_query = torch.tanh(
            pooled
            + lifecycle_scores[:, 1:2] * phase_table[1].reshape(1, -1)
        )
        triage_affinity = torch.sum(
            F.normalize(reproduction_states.float(), dim=-1)
            * F.normalize(triage_query.float(), dim=-1).unsqueeze(1),
            dim=-1,
        ).to(dtype=hidden.dtype)
        triage_weights = torch.softmax(
            worker_assignments.clamp_min(1.0e-9).log() + triage_affinity,
            dim=-1,
        )
        worker_draft_states = torch.tanh(
            reproduction_states
            + lifecycle_scores[:, 2:3].unsqueeze(-1)
            * phase_table[2].reshape(1, 1, -1)
            + lifecycle_scores[:, 3:4].unsqueeze(-1)
            * phase_table[3].reshape(1, 1, -1)
        )
        experiment_plan_state = torch.sum(
            triage_weights.unsqueeze(-1) * worker_draft_states,
            dim=1,
        )
        normalized_workers = F.normalize(worker_draft_states.float(), dim=-1).to(
            dtype=hidden.dtype
        )
        peer_similarity = torch.matmul(
            normalized_workers,
            normalized_workers.transpose(-1, -2),
        )
        self_penalty = torch.eye(
            self.num_experts,
            device=hidden.device,
            dtype=hidden.dtype,
        ).unsqueeze(0)
        peer_review_routes = torch.softmax(
            peer_similarity - self_penalty * 2.0,
            dim=-1,
        )
        peer_context = torch.matmul(peer_review_routes, worker_draft_states)
        critique_strength = lifecycle_scores[:, 4:5].unsqueeze(-1)
        revised_worker_states = worker_draft_states + critique_strength * torch.tanh(
            peer_context - worker_draft_states
        )
        critique_state = torch.sum(
            triage_weights.unsqueeze(-1) * revised_worker_states,
            dim=1,
        )
        revised_draft_state = critique_state + lifecycle_scores[:, 5:6] * torch.tanh(
            phase_table[5].reshape(1, -1) + critique_state - pooled
        )
        disclosure_state = revised_draft_state + lifecycle_scores[:, 6:7] * torch.tanh(
            phase_table[6].reshape(1, -1) + revised_draft_state - pooled
        )
        patch_state = disclosure_state + lifecycle_scores[:, 7:8] * torch.tanh(
            phase_table[7].reshape(1, -1) + disclosure_state - revised_draft_state
        )
        verification_state = patch_state + lifecycle_scores[:, 8:9] * torch.tanh(
            phase_table[8].reshape(1, -1) + patch_state - revised_draft_state
        )
        revision_delta = torch.tanh(revised_draft_state - pooled)
        draft_residual_scale = torch.sigmoid(self.residual_gate).to(
            device=hidden.device,
            dtype=hidden.dtype,
        )
        draft_proposal_gate = lifecycle_scores[:, :5].mean(dim=-1, keepdim=True)
        draft_proposal_delta = (
            draft_residual_scale * draft_proposal_gate * revision_delta
        )

        observed = outcome_features.to(device=hidden.device, dtype=hidden.dtype)
        present = outcome_present.to(device=hidden.device, dtype=hidden.dtype)
        verifier_executed = observed[:, 4:5].clamp(0.0, 1.0)
        infrastructure_available = 1.0 - observed[:, 6:7].clamp(0.0, 1.0)
        experiment_observed = present * verifier_executed * infrastructure_available
        experiment_passed = experiment_observed * observed[:, 0:1].clamp(0.0, 1.0)
        experiment_failed = experiment_observed * observed[:, 1:2].clamp(0.0, 1.0)
        experiment_requirement = lifecycle_scores[:, 3:4]
        revision_process = lifecycle_scores[:, 5:9].mean(dim=-1, keepdim=True)
        revision_gate = experiment_requirement * revision_process * experiment_failed
        verified_revision_delta = (
            draft_residual_scale
            * revision_gate
            * torch.tanh(verification_state - revised_draft_state)
        )
        submission_proposal = lifecycle_scores[:, 9:10]
        verification_gate = (
            1.0
            - experiment_requirement
            + experiment_requirement * experiment_passed
        )
        submission_readiness = torch.minimum(
            submission_proposal,
            verification_gate,
        )
        return NoNEDraftingLifecyclePacket(
            lifecycle_scores=lifecycle_scores,
            worker_assignments=worker_assignments,
            reproduction_states=reproduction_states,
            triage_weights=triage_weights,
            worker_draft_states=worker_draft_states,
            experiment_plan_state=experiment_plan_state,
            peer_review_routes=peer_review_routes,
            critique_state=critique_state,
            revised_draft_state=revised_draft_state,
            disclosure_state=disclosure_state,
            patch_state=patch_state,
            verification_state=verification_state,
            revision_delta=revision_delta,
            draft_residual_scale=draft_residual_scale,
            draft_proposal_gate=draft_proposal_gate,
            draft_proposal_delta=draft_proposal_delta,
            experiment_requirement=experiment_requirement,
            experiment_observed=experiment_observed,
            experiment_passed=experiment_passed,
            revision_gate=revision_gate,
            verified_revision_delta=verified_revision_delta,
            submission_proposal=submission_proposal,
            submission_readiness=submission_readiness,
        )

    def apply_packet(self, packet: NoNEFabricRoutePacket, hidden: torch.Tensor) -> torch.Tensor:
        """Apply the fabric route packet to hidden states (additive residual)."""
        if hidden.ndim != 3:
            raise ValueError("fabric hidden states must have shape [B, S, H]")
        batch_size = hidden.shape[0]
        if (
            packet.decision_surface.ndim != 2
            or packet.decision_surface.shape[0] != batch_size
            or packet.layer_bias.ndim != 2
            or packet.layer_bias.shape[0] != batch_size
            or packet.phase_surface.ndim != 2
            or packet.phase_surface.shape[0] != batch_size
        ):
            raise ValueError("fabric route packet tensors must preserve the hidden batch")
        expert_weights = torch.softmax(packet.decision_surface, dim=-1)
        layer_weights = torch.softmax(packet.layer_bias, dim=-1)
        phase_weights = torch.softmax(packet.phase_surface, dim=-1)
        expert_hidden = torch.matmul(
            expert_weights,
            self.expert_hidden_table.to(device=hidden.device, dtype=hidden.dtype),
        )
        layer_hidden = torch.matmul(
            layer_weights,
            self.layer_hidden_table.to(device=hidden.device, dtype=hidden.dtype),
        )
        phase_hidden = torch.matmul(
            phase_weights,
            self.phase_hidden_table.to(device=hidden.device, dtype=hidden.dtype),
        )
        residual = torch.tanh(expert_hidden + layer_hidden + phase_hidden).unsqueeze(1)
        return hidden + torch.sigmoid(self.residual_gate).to(dtype=hidden.dtype) * residual


class FabricAwareRBO(nn.Module):
    """Wraps ResynthesisRBO so the fabric's step_phase drives every macro-step.

    Pillar 16: the fabric's ``_build_macro_step_fn`` install drives ``step_phase``
    every macro-step. This wrapper ensures the fabric participates once per phase.
    """

    rbo: Any

    def __init__(self, rbo: Any, fabric: ResynthesisNoNEFabric | None = None) -> None:
        super().__init__()
        if not isinstance(rbo, nn.Module):
            raise TypeError("FabricAwareRBO requires an nn.Module RBO")
        self.rbo = rbo
        model_cfg = getattr(rbo, "model_cfg", None)
        science_stack = getattr(rbo, "science_stack", None)
        hidden_size = getattr(model_cfg, "hidden_size", None)
        num_experts = getattr(science_stack, "num_experts", None)
        num_layers = getattr(science_stack, "num_layers", None)
        glyph_dim = getattr(model_cfg, "glyph_input_dim", None)
        if (
            not isinstance(hidden_size, int)
            or isinstance(hidden_size, bool)
            or hidden_size < 1
            or not isinstance(num_experts, int)
            or isinstance(num_experts, bool)
            or num_experts < 1
            or not isinstance(num_layers, int)
            or isinstance(num_layers, bool)
            or num_layers < 1
            or not isinstance(glyph_dim, int)
            or isinstance(glyph_dim, bool)
            or glyph_dim < 1
        ):
            raise RuntimeError(
                "FabricAwareRBO requires exact live additive graph geometry"
            )
        attached_fabric = fabric or ResynthesisNoNEFabric(
            hidden_size=hidden_size,
            num_experts=num_experts,
            num_layers=num_layers,
            glyph_dim=glyph_dim,
        )
        attach_fabric = getattr(self.rbo, "attach_fabric", None)
        if not callable(attach_fabric):
            raise RuntimeError("Resynthesis RBO has no in-graph Fabric attachment seam")
        attach_fabric(attached_fabric)

    @property
    def fabric(self) -> ResynthesisNoNEFabric:
        """Return the fabric owned by the wrapped RBO without double registration."""

        fabric = getattr(self.rbo, "fabric", None)
        if not isinstance(fabric, ResynthesisNoNEFabric):
            raise RuntimeError("wrapped RBO has no Resynthesis NoNE fabric")
        return fabric

    def forward_thinking(self, input_ids: torch.Tensor, **kwargs: Any) -> "RBOResult":
        """Forward with fabric step_phase participation."""
        return cast("RBOResult", self.rbo.forward_thinking(input_ids, **kwargs))