runtime/src/resynthesis/causal_algebra.py

"""Persistent model-owned causal algebra and executable world graph.

The hot path is tensor-native. The model compiles a learned causal program,
executes it over competing world hypotheses, chooses the least-cost
high-disagreement intervention, updates hypothesis posteriors from observations,
and emits proof obligations. Human-readable notation exists only at the
explicit receipt boundary.

The DSL has the following boundary rendering:

```
H := OBSERVE |> CAUSES |> INHIBITS |> ENABLES |> DO |> TRANSFER
     |> COMPARE |> REVISE ; TEST do(A_i) ; PROOF p
```

Each opcode denotes a learned low-rank endomorphism over the world latent. A
program is therefore executable rather than a text-only rationale. Program
length and graph geometry are checkpoint seed geometry, not inference caps:
RBO phases can recurrently feed a returned ``CausalWorldState`` into the next
execution.
"""

from __future__ import annotations

import hashlib
from dataclasses import dataclass, field
from typing import Final, cast

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


CAUSAL_ALGEBRA_INITIALIZATION_SCHEME: Final[str] = (
    "sha256_role_seeded_low_rank_world_graph_identity_residual_v1"
)
CAUSAL_ALGEBRA_DSL_SCHEMA: Final[str] = (
    "nnf.resynthesis.causal_algebra.dsl.v1"
)
CAUSAL_ALGEBRA_PROOF_SCHEMA: Final[str] = (
    "nnf.resynthesis.causal_algebra.proof_packet.v1"
)
CAUSAL_PROMOTION_COUPLED_GAIN_COUNT: Final[int] = 4
CAUSAL_ALGEBRA_OPCODE_NAMES: Final[tuple[str, ...]] = (
    "OBSERVE",
    "CAUSES",
    "INHIBITS",
    "ENABLES",
    "DO",
    "TRANSFER",
    "COMPARE",
    "REVISE",
)


def _missing_coupled_gain_evidence() -> torch.Tensor:
    """Return fail-closed migration evidence for an older boundary caller."""

    return torch.zeros(
        CAUSAL_PROMOTION_COUPLED_GAIN_COUNT,
        dtype=torch.float32,
    )


def _missing_scalar_gain_evidence() -> torch.Tensor:
    """Return one absent measured gain without manufacturing promotion proof."""

    return torch.zeros((), dtype=torch.float32)


def _missing_route_coverage_evidence() -> torch.Tensor:
    """Return an empty route mask that can never satisfy promotion."""

    return torch.zeros((0, 0), dtype=torch.bool)


def _missing_promotion_identity() -> torch.Tensor:
    """Return an absent SHA-256 identity as a tensor-native digest."""

    return torch.zeros(32, dtype=torch.uint8)


def _role_seeded_xavier_uniform_(
    tensor_t: torch.Tensor,
    role: str,
) -> None:
    """Initialize one growth tensor without changing the process RNG.

    The role digest is model authority, so it must produce the same bytes when
    a mutable cold load constructs the graph on CPU or directly on CUDA.
    PyTorch's CPU and CUDA generators are different algorithms even with the
    same seed.  Generate the small additive growth surface once in canonical
    CPU float32, then copy it to the requested device/dtype.  This is an
    initialization boundary, not a hot-path host roundtrip.
    """

    if tensor_t.device.type == "meta":
        return
    seed = int.from_bytes(
        hashlib.sha256(
            f"resynthesis.causal_algebra.v1:{role}".encode("utf-8")
        ).digest()[:8],
        byteorder="little",
        signed=False,
    )
    generator = torch.Generator(device="cpu")
    generator.manual_seed(seed)
    canonical_t = torch.empty(
        tuple(tensor_t.shape),
        dtype=torch.float32,
        device="cpu",
    )
    nn.init.xavier_uniform_(canonical_t, generator=generator)
    with torch.no_grad():
        tensor_t.copy_(
            canonical_t.to(
                device=tensor_t.device,
                dtype=tensor_t.dtype,
            )
        )


def _stable_probability(logits_t: torch.Tensor, dim: int) -> torch.Tensor:
    """Return finite probabilities while preserving the caller dtype."""

    return torch.softmax(logits_t.float(), dim=dim).to(dtype=logits_t.dtype)


def _hill_climb_lerp(
    prior_t: torch.Tensor,
    candidate_t: torch.Tensor,
    accept_weight_t: torch.Tensor,
) -> torch.Tensor:
    """Select recurrent state rows with a live straight-through accept path."""

    if prior_t.shape != candidate_t.shape:
        raise ValueError("hill-climb candidate geometry differs from prior")
    if prior_t.shape[0] != accept_weight_t.shape[0]:
        raise ValueError("hill-climb accept batch geometry differs")
    row_weight_t = accept_weight_t.reshape(
        accept_weight_t.shape[0],
        *((1,) * (candidate_t.ndim - 1)),
    )
    return torch.lerp(prior_t, candidate_t, row_weight_t)


@dataclass(frozen=True)
class CausalAlgebraConfig:
    """Checkpoint seed geometry, not a reasoning or traversal cap."""

    hidden_size: int
    action_size: int
    pathway_size: int
    world_size: int = 64
    hypothesis_count: int = 4
    primitive_count: int = 8
    program_steps: int = 4
    domain_count: int = 8
    operator_rank: int = 16

    def validated(self) -> "CausalAlgebraConfig":
        if self.hidden_size < 1:
            raise ValueError("causal algebra hidden size must be positive")
        if self.action_size < 2:
            raise ValueError("causal algebra requires at least two interventions")
        if self.pathway_size < 1:
            raise ValueError("causal algebra pathway size must be positive")
        if self.world_size < 2:
            raise ValueError("causal algebra world size must be at least two")
        if self.hypothesis_count < 2:
            raise ValueError("causal algebra requires competing hypotheses")
        if self.primitive_count < 2:
            raise ValueError("causal algebra requires multiple primitives")
        if self.program_steps < 1:
            raise ValueError("causal algebra program seed must be non-empty")
        if self.domain_count < 2:
            raise ValueError("causal algebra requires multiple domains")
        if self.operator_rank < 1:
            raise ValueError("causal algebra operator rank must be positive")
        return self


@dataclass(frozen=True)
class CausalWorldState:
    """Caller-owned recurrent world state with no cross-session mutation.

    ``working_memory_t`` is the model's explicit compact task workspace.  The
    action policy and hill-climb tensors are carried with it so a resumed RBO
    phase continues the same exploration trajectory instead of reconstructing
    it from host counters or sampling a fresh Thompson arm.
    """

    ontology_t: torch.Tensor
    rule_t: torch.Tensor
    posterior_t: torch.Tensor
    contradiction_t: torch.Tensor
    commitment_t: torch.Tensor
    working_memory_t: torch.Tensor
    action_policy_t: torch.Tensor
    replay_error_t: torch.Tensor
    information_value_t: torch.Tensor
    exploration_counterweight_t: torch.Tensor
    hill_climb_accept_t: torch.Tensor

    def validated(
        self,
        cfg: CausalAlgebraConfig,
        batch_size: int,
    ) -> "CausalWorldState":
        world_shape = (
            batch_size,
            cfg.hypothesis_count,
            cfg.world_size,
        )
        hypothesis_shape = (batch_size, cfg.hypothesis_count)
        workspace_shape = (batch_size, cfg.world_size)
        action_shape = (batch_size, cfg.action_size)
        scalar_shape = (batch_size,)
        if self.ontology_t.shape != world_shape:
            raise ValueError("causal ontology state geometry differs")
        if self.rule_t.shape != world_shape:
            raise ValueError("causal rule state geometry differs")
        if self.posterior_t.shape != hypothesis_shape:
            raise ValueError("causal posterior geometry differs")
        if self.contradiction_t.shape != hypothesis_shape:
            raise ValueError("causal contradiction geometry differs")
        if self.commitment_t.shape != hypothesis_shape:
            raise ValueError("causal commitment geometry differs")
        if self.working_memory_t.shape != workspace_shape:
            raise ValueError("causal working-memory geometry differs")
        if self.action_policy_t.shape != action_shape:
            raise ValueError("causal action-policy geometry differs")
        if self.replay_error_t.shape != scalar_shape:
            raise ValueError("causal replay-error geometry differs")
        if self.information_value_t.shape != scalar_shape:
            raise ValueError("causal information-value geometry differs")
        if self.exploration_counterweight_t.shape != scalar_shape:
            raise ValueError("causal exploration-counterweight geometry differs")
        if self.hill_climb_accept_t.shape != scalar_shape:
            raise ValueError("causal hill-climb decision geometry differs")
        if self.hill_climb_accept_t.dtype != torch.bool:
            raise ValueError("causal hill-climb decision must be boolean")
        torch._assert_async(
            torch.isfinite(self.ontology_t).all()
            & torch.isfinite(self.rule_t).all()
            & torch.isfinite(self.posterior_t).all()
            & torch.isfinite(self.contradiction_t).all()
            & torch.isfinite(self.commitment_t).all()
            & torch.isfinite(self.working_memory_t).all()
            & torch.isfinite(self.action_policy_t).all()
            & torch.isfinite(self.replay_error_t).all()
            & torch.isfinite(self.information_value_t).all()
            & torch.isfinite(self.exploration_counterweight_t).all(),
            "causal world state must be finite",
        )
        return self

    def detached(self) -> "CausalWorldState":
        return CausalWorldState(
            ontology_t=self.ontology_t.detach(),
            rule_t=self.rule_t.detach(),
            posterior_t=self.posterior_t.detach(),
            contradiction_t=self.contradiction_t.detach(),
            commitment_t=self.commitment_t.detach(),
            working_memory_t=self.working_memory_t.detach(),
            action_policy_t=self.action_policy_t.detach(),
            replay_error_t=self.replay_error_t.detach(),
            information_value_t=self.information_value_t.detach(),
            exploration_counterweight_t=(
                self.exploration_counterweight_t.detach()
            ),
            hill_climb_accept_t=self.hill_climb_accept_t.detach(),
        )

    def select_batch_row_boundary(self, batch_index: int) -> "CausalWorldState":
        """Return one caller-owned world at an explicit batch boundary."""

        batch_size = self.ontology_t.shape[0]
        if batch_index < 0 or batch_index >= batch_size:
            raise IndexError("causal world-state batch index is out of range")
        row = slice(batch_index, batch_index + 1)
        return CausalWorldState(
            ontology_t=self.ontology_t[row],
            rule_t=self.rule_t[row],
            posterior_t=self.posterior_t[row],
            contradiction_t=self.contradiction_t[row],
            commitment_t=self.commitment_t[row],
            working_memory_t=self.working_memory_t[row],
            action_policy_t=self.action_policy_t[row],
            replay_error_t=self.replay_error_t[row],
            information_value_t=self.information_value_t[row],
            exploration_counterweight_t=self.exploration_counterweight_t[row],
            hill_climb_accept_t=self.hill_climb_accept_t[row],
        )


@dataclass(frozen=True)
class CausalProgramPacket:
    """Tensor-native proof-carrying compiler output."""

    notation_probability_t: torch.Tensor
    execution_trace_t: torch.Tensor
    compiled_state_t: torch.Tensor
    executable_consistency_t: torch.Tensor

    def detached(self) -> "CausalProgramPacket":
        return CausalProgramPacket(
            notation_probability_t=self.notation_probability_t.detach(),
            execution_trace_t=self.execution_trace_t.detach(),
            compiled_state_t=self.compiled_state_t.detach(),
            executable_consistency_t=self.executable_consistency_t.detach(),
        )

    def select_batch_row_boundary(self, batch_index: int) -> "CausalProgramPacket":
        """Return one compiled program at an explicit batch boundary."""

        batch_size = self.notation_probability_t.shape[0]
        if batch_index < 0 or batch_index >= batch_size:
            raise IndexError("causal program batch index is out of range")
        row = slice(batch_index, batch_index + 1)
        return CausalProgramPacket(
            notation_probability_t=self.notation_probability_t[row],
            execution_trace_t=self.execution_trace_t[row],
            compiled_state_t=self.compiled_state_t[row],
            executable_consistency_t=self.executable_consistency_t[row],
        )


@dataclass(frozen=True)
class FalsifyingExperimentPacket:
    """Model-owned least-cost falsification policy."""

    disagreement_t: torch.Tensor
    value_per_cost_t: torch.Tensor
    experiment_probability_t: torch.Tensor
    experiment_index_t: torch.Tensor
    selection_margin_t: torch.Tensor

    def detached(self) -> "FalsifyingExperimentPacket":
        return FalsifyingExperimentPacket(
            disagreement_t=self.disagreement_t.detach(),
            value_per_cost_t=self.value_per_cost_t.detach(),
            experiment_probability_t=self.experiment_probability_t.detach(),
            experiment_index_t=self.experiment_index_t.detach(),
            selection_margin_t=self.selection_margin_t.detach(),
        )

    def select_batch_row_boundary(
        self,
        batch_index: int,
    ) -> "FalsifyingExperimentPacket":
        """Return one intervention policy at an explicit batch boundary."""

        batch_size = self.disagreement_t.shape[0]
        if batch_index < 0 or batch_index >= batch_size:
            raise IndexError("falsifying-experiment batch index is out of range")
        row = slice(batch_index, batch_index + 1)
        return FalsifyingExperimentPacket(
            disagreement_t=self.disagreement_t[row],
            value_per_cost_t=self.value_per_cost_t[row],
            experiment_probability_t=self.experiment_probability_t[row],
            experiment_index_t=self.experiment_index_t[row],
            selection_margin_t=self.selection_margin_t[row],
        )


@dataclass(frozen=True)
class CounterThompsonHillClimbPacket:
    """Model-owned exploration and recurrent state-revision decision.

    The counter-Thompson probability is a learned allocation between the
    exploitative action policy and the most informative credible alternative.
    The hill-climb decision independently decides whether the newly executed
    world should replace the prior working world after replay.
    """

    exploitation_probability_t: torch.Tensor
    information_probability_t: torch.Tensor
    action_probability_t: torch.Tensor
    action_index_t: torch.Tensor
    counter_thompson_probability_t: torch.Tensor
    hill_climb_probability_t: torch.Tensor
    hill_climb_gain_t: torch.Tensor
    hill_climb_accept_t: torch.Tensor
    hill_climb_accept_weight_t: torch.Tensor
    working_memory_t: torch.Tensor
    working_memory_delta_l2_t: torch.Tensor
    replay_error_t: torch.Tensor
    successor_replay_error_t: torch.Tensor
    operator_commutator_error_t: torch.Tensor
    domain_cycle_error_t: torch.Tensor
    information_value_t: torch.Tensor
    exploration_counterweight_t: torch.Tensor

    def detached(self) -> "CounterThompsonHillClimbPacket":
        return CounterThompsonHillClimbPacket(
            exploitation_probability_t=(
                self.exploitation_probability_t.detach()
            ),
            information_probability_t=self.information_probability_t.detach(),
            action_probability_t=self.action_probability_t.detach(),
            action_index_t=self.action_index_t.detach(),
            counter_thompson_probability_t=(
                self.counter_thompson_probability_t.detach()
            ),
            hill_climb_probability_t=self.hill_climb_probability_t.detach(),
            hill_climb_gain_t=self.hill_climb_gain_t.detach(),
            hill_climb_accept_t=self.hill_climb_accept_t.detach(),
            hill_climb_accept_weight_t=(
                self.hill_climb_accept_weight_t.detach()
            ),
            working_memory_t=self.working_memory_t.detach(),
            working_memory_delta_l2_t=(
                self.working_memory_delta_l2_t.detach()
            ),
            replay_error_t=self.replay_error_t.detach(),
            successor_replay_error_t=(
                self.successor_replay_error_t.detach()
            ),
            operator_commutator_error_t=(
                self.operator_commutator_error_t.detach()
            ),
            domain_cycle_error_t=self.domain_cycle_error_t.detach(),
            information_value_t=self.information_value_t.detach(),
            exploration_counterweight_t=(
                self.exploration_counterweight_t.detach()
            ),
        )

    def select_batch_row_boundary(
        self,
        batch_index: int,
    ) -> "CounterThompsonHillClimbPacket":
        """Return one caller's exploration proof at an explicit boundary."""

        batch_size = self.action_probability_t.shape[0]
        if batch_index < 0 or batch_index >= batch_size:
            raise IndexError("counter-Thompson batch index is out of range")
        row = slice(batch_index, batch_index + 1)
        return CounterThompsonHillClimbPacket(
            exploitation_probability_t=self.exploitation_probability_t[row],
            information_probability_t=self.information_probability_t[row],
            action_probability_t=self.action_probability_t[row],
            action_index_t=self.action_index_t[row],
            counter_thompson_probability_t=(
                self.counter_thompson_probability_t[row]
            ),
            hill_climb_probability_t=self.hill_climb_probability_t[row],
            hill_climb_gain_t=self.hill_climb_gain_t[row],
            hill_climb_accept_t=self.hill_climb_accept_t[row],
            hill_climb_accept_weight_t=self.hill_climb_accept_weight_t[row],
            working_memory_t=self.working_memory_t[row],
            working_memory_delta_l2_t=self.working_memory_delta_l2_t[row],
            replay_error_t=self.replay_error_t[row],
            successor_replay_error_t=self.successor_replay_error_t[row],
            operator_commutator_error_t=(
                self.operator_commutator_error_t[row]
            ),
            domain_cycle_error_t=self.domain_cycle_error_t[row],
            information_value_t=self.information_value_t[row],
            exploration_counterweight_t=self.exploration_counterweight_t[row],
        )


@dataclass(frozen=True)
class CausalKnowledgeOwnershipPacket:
    """Active, tensor-native view of durably verified model knowledge."""

    domain_transfer_probability_t: torch.Tensor
    operator_signature_t: torch.Tensor
    operator_strength_t: torch.Tensor
    ownership_strength_t: torch.Tensor
    disagreement_t: torch.Tensor


@dataclass(frozen=True)
class CausalTheoryProofPacket:
    """Proof packet emitted by the executable world graph."""

    world_state: CausalWorldState
    program: CausalProgramPacket
    predicted_outcome_t: torch.Tensor
    observation_t: torch.Tensor
    observation_error_t: torch.Tensor
    falsifying_experiment: FalsifyingExperimentPacket
    exploration: CounterThompsonHillClimbPacket
    source_domain_probability_t: torch.Tensor
    target_domain_probability_t: torch.Tensor
    domain_transfer_t: torch.Tensor
    knowledge_ownership_t: torch.Tensor
    knowledge_disagreement_t: torch.Tensor
    reopen_probability_t: torch.Tensor
    proof_obligation_t: torch.Tensor
    proof_validity_t: torch.Tensor
    continuation_probability_t: torch.Tensor
    promotion_authority_t: torch.Tensor

    def detached(self) -> "CausalTheoryProofPacket":
        return CausalTheoryProofPacket(
            world_state=self.world_state.detached(),
            program=self.program.detached(),
            predicted_outcome_t=self.predicted_outcome_t.detach(),
            observation_t=self.observation_t.detach(),
            observation_error_t=self.observation_error_t.detach(),
            falsifying_experiment=self.falsifying_experiment.detached(),
            exploration=self.exploration.detached(),
            source_domain_probability_t=(
                self.source_domain_probability_t.detach()
            ),
            target_domain_probability_t=(
                self.target_domain_probability_t.detach()
            ),
            domain_transfer_t=self.domain_transfer_t.detach(),
            knowledge_ownership_t=self.knowledge_ownership_t.detach(),
            knowledge_disagreement_t=self.knowledge_disagreement_t.detach(),
            reopen_probability_t=self.reopen_probability_t.detach(),
            proof_obligation_t=self.proof_obligation_t.detach(),
            proof_validity_t=self.proof_validity_t.detach(),
            continuation_probability_t=(
                self.continuation_probability_t.detach()
            ),
            promotion_authority_t=self.promotion_authority_t.detach(),
        )

    def select_batch_row_boundary(
        self,
        batch_index: int,
    ) -> "CausalTheoryProofPacket":
        """Return one proof-carrying theory at an explicit batch boundary."""

        batch_size = self.predicted_outcome_t.shape[0]
        if batch_index < 0 or batch_index >= batch_size:
            raise IndexError("causal proof batch index is out of range")
        row = slice(batch_index, batch_index + 1)
        return CausalTheoryProofPacket(
            world_state=self.world_state.select_batch_row_boundary(batch_index),
            program=self.program.select_batch_row_boundary(batch_index),
            predicted_outcome_t=self.predicted_outcome_t[row],
            observation_t=self.observation_t[row],
            observation_error_t=self.observation_error_t[row],
            falsifying_experiment=(
                self.falsifying_experiment.select_batch_row_boundary(batch_index)
            ),
            exploration=self.exploration.select_batch_row_boundary(batch_index),
            source_domain_probability_t=self.source_domain_probability_t[row],
            target_domain_probability_t=self.target_domain_probability_t[row],
            domain_transfer_t=self.domain_transfer_t[row],
            knowledge_ownership_t=self.knowledge_ownership_t[row],
            knowledge_disagreement_t=self.knowledge_disagreement_t[row],
            reopen_probability_t=self.reopen_probability_t[row],
            proof_obligation_t=self.proof_obligation_t[row],
            proof_validity_t=self.proof_validity_t[row],
            continuation_probability_t=self.continuation_probability_t[row],
            promotion_authority_t=self.promotion_authority_t[row],
        )


@dataclass(frozen=True)
class CausalAlgebraResult:
    """Additive hidden result plus executable causal proof."""

    hidden_t: torch.Tensor
    proof: CausalTheoryProofPacket
    auxiliary_loss_t: torch.Tensor


@dataclass(frozen=True)
class CausalPromotionEvidence:
    """Tensor-only evidence accepted at the external promotion boundary.

    The coupled-gain vector order is expert specialization, layer/traversal
    capacity, retained domain knowledge, and intent-to-end-state reasoning.
    Route masks bind verification to the domain/action pairs the held-out
    cohort actually required and observed.  Default factories preserve older
    boundary construction while representing *missing* evidence, so they can
    never grant promotion authority.
    """

    retained_heldout_t: torch.Tensor
    falsification_verified_t: torch.Tensor
    cross_domain_transfer_verified_t: torch.Tensor
    cold_reload_verified_t: torch.Tensor
    coupled_gain_delta_t: torch.Tensor = field(
        default_factory=_missing_coupled_gain_evidence
    )
    native_knowledge_ownership_gain_t: torch.Tensor = field(
        default_factory=_missing_scalar_gain_evidence
    )
    self_correction_gain_t: torch.Tensor = field(
        default_factory=_missing_scalar_gain_evidence
    )
    required_domain_action_coverage_t: torch.Tensor = field(
        default_factory=_missing_route_coverage_evidence
    )
    observed_domain_action_coverage_t: torch.Tensor = field(
        default_factory=_missing_route_coverage_evidence
    )
    qualification_identity_t: torch.Tensor = field(
        default_factory=_missing_promotion_identity
    )
    candidate_lineage_t: torch.Tensor = field(
        default_factory=_missing_promotion_identity
    )


def cheapest_falsifying_experiment(
    predicted_outcome_t: torch.Tensor,
    posterior_t: torch.Tensor,
    action_cost_t: torch.Tensor,
    *,
    policy_bias_t: torch.Tensor | None = None,
) -> FalsifyingExperimentPacket:
    """Select maximum posterior disagreement per unit intervention cost."""

    if predicted_outcome_t.ndim != 4:
        raise ValueError(
            "predicted causal outcomes expect [batch, hypotheses, actions, world]"
        )
    batch_size, hypothesis_count, action_count, _world_size = (
        predicted_outcome_t.shape
    )
    if hypothesis_count < 2 or action_count < 2:
        raise ValueError(
            "falsification requires competing hypotheses and interventions"
        )
    if posterior_t.shape != (batch_size, hypothesis_count):
        raise ValueError("falsification posterior geometry differs")
    if action_cost_t.shape != (action_count,):
        raise ValueError("falsification action-cost geometry differs")
    active_bias_t = (
        predicted_outcome_t.new_zeros(batch_size, action_count)
        if policy_bias_t is None
        else policy_bias_t.to(
            device=predicted_outcome_t.device,
            dtype=predicted_outcome_t.dtype,
        )
    )
    if active_bias_t.shape != (batch_size, action_count):
        raise ValueError("falsification policy-bias geometry differs")
    torch._assert_async(
        torch.isfinite(predicted_outcome_t).all()
        & torch.isfinite(posterior_t).all()
        & torch.isfinite(action_cost_t).all()
        & action_cost_t.gt(0).all(),
        "falsification inputs and positive costs must be finite",
    )

    normalized_posterior_t = posterior_t / posterior_t.sum(
        dim=-1,
        keepdim=True,
    ).clamp_min(torch.finfo(posterior_t.dtype).tiny)
    expected_outcome_t = torch.einsum(
        "bk,bkaw->baw",
        normalized_posterior_t,
        predicted_outcome_t,
    )
    disagreement_t = torch.einsum(
        "bk,bkaw->baw",
        normalized_posterior_t,
        (predicted_outcome_t - expected_outcome_t.unsqueeze(1)).square(),
    ).mean(dim=-1)
    value_per_cost_t = disagreement_t / action_cost_t.to(
        device=disagreement_t.device,
        dtype=disagreement_t.dtype,
    ).unsqueeze(0)
    experiment_probability_t = _stable_probability(
        value_per_cost_t + active_bias_t,
        dim=-1,
    )
    experiment_index_t = experiment_probability_t.argmax(dim=-1)
    top_two_t = torch.topk(
        experiment_probability_t,
        k=2,
        dim=-1,
    ).values
    selection_margin_t = top_two_t[:, 0] - top_two_t[:, 1]
    return FalsifyingExperimentPacket(
        disagreement_t=disagreement_t,
        value_per_cost_t=value_per_cost_t,
        experiment_probability_t=experiment_probability_t,
        experiment_index_t=experiment_index_t,
        selection_margin_t=selection_margin_t,
    )


class ProofCarryingCausalTheoryCompiler(nn.Module):
    """Compile soft causal notation into an executable low-rank world program."""

    def __init__(self, cfg: CausalAlgebraConfig) -> None:
        super().__init__()
        self.cfg = cfg.validated()
        self.program_query = nn.Linear(
            2 * self.cfg.world_size,
            self.cfg.program_steps * self.cfg.primitive_count,
            bias=True,
        )
        self.primitive_read_t = nn.Parameter(
            torch.empty(
                self.cfg.primitive_count,
                self.cfg.world_size,
                self.cfg.operator_rank,
            )
        )
        self.primitive_write_t = nn.Parameter(
            torch.empty(
                self.cfg.primitive_count,
                self.cfg.operator_rank,
                self.cfg.world_size,
            )
        )
        self.primitive_bias_t = nn.Parameter(
            torch.empty(
                self.cfg.primitive_count,
                self.cfg.world_size,
            )
        )
        self.program_step_gate_t = nn.Parameter(
            torch.zeros(self.cfg.program_steps)
        )
        self.reset_parameters()

    def reset_parameters(self) -> None:
        _role_seeded_xavier_uniform_(
            self.program_query.weight,
            "compiler.program_query.weight",
        )
        nn.init.zeros_(self.program_query.bias)
        _role_seeded_xavier_uniform_(
            self.primitive_read_t,
            "compiler.primitive_read_t",
        )
        _role_seeded_xavier_uniform_(
            self.primitive_write_t,
            "compiler.primitive_write_t",
        )
        _role_seeded_xavier_uniform_(
            self.primitive_bias_t,
            "compiler.primitive_bias_t",
        )
        nn.init.zeros_(self.program_step_gate_t)

    def forward(
        self,
        ontology_t: torch.Tensor,
        rule_t: torch.Tensor,
    ) -> CausalProgramPacket:
        expected_shape = (
            ontology_t.shape[0],
            self.cfg.hypothesis_count,
            self.cfg.world_size,
        )
        if ontology_t.shape != expected_shape or rule_t.shape != expected_shape:
            raise ValueError("causal compiler world geometry differs")
        compiler_input_t = torch.cat((ontology_t, rule_t), dim=-1)
        notation_logits_t = self.program_query(compiler_input_t).reshape(
            ontology_t.shape[0],
            self.cfg.hypothesis_count,
            self.cfg.program_steps,
            self.cfg.primitive_count,
        )
        notation_probability_t = _stable_probability(
            notation_logits_t,
            dim=-1,
        )

        state_t = ontology_t
        execution_trace_t = ontology_t.unsqueeze(2)
        for step_index in range(self.cfg.program_steps):
            primitive_read_t = torch.einsum(
                "bkw,pwr->bkpr",
                state_t,
                self.primitive_read_t.to(dtype=state_t.dtype),
            )
            primitive_output_t = torch.einsum(
                "bkpr,prw->bkpw",
                torch.tanh(primitive_read_t),
                self.primitive_write_t.to(dtype=state_t.dtype),
            )
            primitive_output_t = torch.tanh(
                primitive_output_t
                + self.primitive_bias_t.to(dtype=state_t.dtype).view(
                    1,
                    1,
                    self.cfg.primitive_count,
                    self.cfg.world_size,
                )
                + rule_t.unsqueeze(2)
            )
            selected_t = torch.einsum(
                "bkp,bkpw->bkw",
                notation_probability_t[:, :, step_index, :],
                primitive_output_t,
            )
            step_gate_t = torch.sigmoid(
                self.program_step_gate_t[step_index]
            ).to(dtype=state_t.dtype)
            state_t = state_t + step_gate_t * (selected_t - state_t)
            execution_trace_t = torch.cat(
                (execution_trace_t, state_t.unsqueeze(2)),
                dim=2,
            )

        execution_trace_t = execution_trace_t.narrow(
            2,
            1,
            self.cfg.program_steps,
        )
        executable_consistency_t = (
            state_t - torch.tanh(ontology_t + rule_t)
        ).square().mean(dim=-1)
        return CausalProgramPacket(
            notation_probability_t=notation_probability_t,
            execution_trace_t=execution_trace_t,
            compiled_state_t=state_t,
            executable_consistency_t=executable_consistency_t,
        )


class CausalAlgebraWorldGraph(nn.Module):
    """Additive causal head joining experts, layers, actions, and domains."""

    accepted_gradient_update_count_t: torch.Tensor
    accepted_gradient_norm_t: torch.Tensor
    accepted_parameter_delta_norm_t: torch.Tensor
    retained_heldout_t: torch.Tensor
    falsification_verified_t: torch.Tensor
    cross_domain_transfer_verified_t: torch.Tensor
    cold_reload_verified_t: torch.Tensor
    verified_abstraction_transfer_t: torch.Tensor
    verified_abstraction_count_t: torch.Tensor
    verified_operator_signature_t: torch.Tensor
    verified_operator_count_t: torch.Tensor
    promotion_cohort_identity_t: torch.Tensor
    promotion_transaction_identity_t: torch.Tensor
    promotion_heldout_identity_t: torch.Tensor
    verified_abstraction_cohort_identity_t: torch.Tensor
    promotion_generation_t: torch.Tensor
    promotion_gradient_floor_t: torch.Tensor
    verified_abstraction_gradient_update_count_t: torch.Tensor
    verified_coupled_gain_delta_t: torch.Tensor
    verified_native_knowledge_ownership_gain_t: torch.Tensor
    verified_self_correction_gain_t: torch.Tensor
    required_domain_action_coverage_t: torch.Tensor
    verified_domain_action_coverage_t: torch.Tensor
    promotion_qualification_identity_t: torch.Tensor
    promotion_candidate_lineage_t: torch.Tensor

    def __init__(self, cfg: CausalAlgebraConfig) -> None:
        super().__init__()
        self.cfg = cfg.validated()
        world = self.cfg.world_size
        hypotheses = self.cfg.hypothesis_count
        actions = self.cfg.action_size
        domains = self.cfg.domain_count
        rank = self.cfg.operator_rank

        self.evidence_norm = nn.LayerNorm(self.cfg.hidden_size)
        self.ontology_seed = nn.Linear(
            self.cfg.hidden_size,
            hypotheses * world,
            bias=False,
        )
        self.rule_seed = nn.Linear(
            self.cfg.hidden_size,
            hypotheses * world,
            bias=False,
        )
        self.pathway_seed = nn.Linear(
            self.cfg.pathway_size,
            world,
            bias=False,
        )
        self.hypothesis_prior = nn.Linear(
            self.cfg.hidden_size,
            hypotheses,
            bias=True,
        )
        self.domain_source = nn.Linear(
            self.cfg.hidden_size,
            domains,
            bias=True,
        )
        self.domain_basis_t = nn.Parameter(torch.empty(domains, world))
        self.domain_transfer_logits_t = nn.Parameter(
            torch.empty(domains, domains)
        )
        self.domain_transfer_scale = nn.Parameter(torch.tensor(0.25))
        self.prior_carry_scale = nn.Parameter(torch.tensor(0.50))
        self.likelihood_scale = nn.Parameter(torch.zeros(()))

        self.compiler = ProofCarryingCausalTheoryCompiler(self.cfg)
        self.action_read_t = nn.Parameter(
            torch.empty(actions, world, rank)
        )
        self.action_write_t = nn.Parameter(
            torch.empty(actions, rank, world)
        )
        self.action_bias_t = nn.Parameter(torch.empty(actions, world))
        self.action_policy = nn.Linear(actions, actions, bias=True)
        self.observation_hidden = nn.Linear(
            self.cfg.hidden_size,
            world,
            bias=False,
        )
        self.observation_action = nn.Linear(actions, world, bias=False)
        proof_context_size = 2 * world + 1
        self.contradiction_head = nn.Linear(
            proof_context_size,
            1,
            bias=True,
        )
        self.commitment_head = nn.Linear(
            proof_context_size,
            1,
            bias=True,
        )
        self.reopen_head = nn.Linear(
            proof_context_size,
            1,
            bias=True,
        )
        self.experiment_policy = nn.Linear(
            self.cfg.hidden_size,
            actions,
            bias=True,
        )
        self.action_cost_logit_t = nn.Parameter(torch.zeros(actions))
        # Explicit recurrent workspace plus two separate learned decisions:
        # where to spend counter-Thompson exploration mass and whether replay
        # evidence justifies hill-climbing from the prior world to this one.
        self.working_memory_input = nn.Linear(
            self.cfg.hidden_size + world,
            world,
            bias=True,
        )
        self.working_memory_gate = nn.Linear(
            self.cfg.hidden_size + world,
            world,
            bias=True,
        )
        self.counter_thompson_head = nn.Linear(
            self.cfg.hidden_size + world + (2 * actions),
            2,
            bias=True,
        )
        self.hill_climb_head = nn.Linear(
            world + 4,
            2,
            bias=True,
        )
        # Exact-zero migration preserves existing routing on old checkpoints.
        # Its straight-through training path opens only from model loss.
        self.exploration_influence_scale = nn.Parameter(torch.zeros(()))

        self.token_world = nn.Linear(
            self.cfg.hidden_size,
            world,
            bias=False,
        )
        self.world_hidden = nn.Linear(
            world,
            self.cfg.hidden_size,
            bias=False,
        )
        self.proof_verifier = nn.Linear(world + 6, 1, bias=True)
        self.continuation_head = nn.Linear(world + 10, 1, bias=True)
        self.world_residual_scale = nn.Parameter(torch.zeros(()))
        self.rbo_stop_influence_scale = nn.Parameter(torch.zeros(()))

        self.register_buffer(
            "accepted_gradient_update_count_t",
            torch.zeros((), dtype=torch.long),
            persistent=True,
        )
        self.register_buffer(
            "accepted_gradient_norm_t",
            torch.zeros((), dtype=torch.float32),
            persistent=True,
        )
        self.register_buffer(
            "accepted_parameter_delta_norm_t",
            torch.zeros((), dtype=torch.float32),
            persistent=True,
        )
        self.register_buffer(
            "retained_heldout_t",
            torch.zeros((), dtype=torch.bool),
            persistent=True,
        )
        self.register_buffer(
            "falsification_verified_t",
            torch.zeros((), dtype=torch.bool),
            persistent=True,
        )
        self.register_buffer(
            "cross_domain_transfer_verified_t",
            torch.zeros((), dtype=torch.bool),
            persistent=True,
        )
        self.register_buffer(
            "cold_reload_verified_t",
            torch.zeros((), dtype=torch.bool),
            persistent=True,
        )
        # Promotion does not merely remember a host-authored boolean.  When a
        # held-out cross-domain transfer is verified, freeze the model's own
        # learned domain-transfer posterior into durable graph state.  This is
        # the abstraction that must survive checkpointing and cold reload.
        self.register_buffer(
            "verified_abstraction_transfer_t",
            torch.zeros(domains, domains, dtype=torch.float32),
            persistent=True,
        )
        self.register_buffer(
            "verified_abstraction_count_t",
            torch.zeros((), dtype=torch.long),
            persistent=True,
        )
        # Keep verified abstractions conditional on domain and intervention.
        # A single averaged transfer matrix loses which operator was reusable
        # in which scientific context; this tensor is the cold-reloadable
        # knowledge-ownership surface.
        self.register_buffer(
            "verified_operator_signature_t",
            torch.zeros(
                domains,
                actions,
                world,
                dtype=torch.float32,
            ),
            persistent=True,
        )
        self.register_buffer(
            "verified_operator_count_t",
            torch.zeros(domains, actions, dtype=torch.long),
            persistent=True,
        )
        self.register_buffer(
            "promotion_cohort_identity_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.register_buffer(
            "promotion_transaction_identity_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.register_buffer(
            "promotion_heldout_identity_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.register_buffer(
            "verified_abstraction_cohort_identity_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.register_buffer(
            "promotion_generation_t",
            torch.full((), -1, dtype=torch.long),
            persistent=True,
        )
        self.register_buffer(
            "promotion_gradient_floor_t",
            torch.zeros((), dtype=torch.long),
            persistent=True,
        )
        self.register_buffer(
            "verified_abstraction_gradient_update_count_t",
            torch.zeros((), dtype=torch.long),
            persistent=True,
        )
        # These are evaluation evidence, not routing inputs.  They are durable
        # so a cold load must reproduce the exact functional gains and
        # qualification/lineage binding that authorized promotion.
        self.register_buffer(
            "verified_coupled_gain_delta_t",
            torch.zeros(
                CAUSAL_PROMOTION_COUPLED_GAIN_COUNT,
                dtype=torch.float32,
            ),
            persistent=True,
        )
        self.register_buffer(
            "verified_native_knowledge_ownership_gain_t",
            torch.zeros((), dtype=torch.float32),
            persistent=True,
        )
        self.register_buffer(
            "verified_self_correction_gain_t",
            torch.zeros((), dtype=torch.float32),
            persistent=True,
        )
        self.register_buffer(
            "required_domain_action_coverage_t",
            torch.zeros(domains, actions, dtype=torch.bool),
            persistent=True,
        )
        self.register_buffer(
            "verified_domain_action_coverage_t",
            torch.zeros(domains, actions, dtype=torch.bool),
            persistent=True,
        )
        self.register_buffer(
            "promotion_qualification_identity_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.register_buffer(
            "promotion_candidate_lineage_t",
            torch.zeros(32, dtype=torch.uint8),
            persistent=True,
        )
        self.reset_parameters()

    def reset_parameters(self) -> None:
        for module_name, module in self.named_modules():
            if isinstance(module, nn.Linear):
                _role_seeded_xavier_uniform_(
                    module.weight,
                    f"world_graph.{module_name}.weight",
                )
                if module.bias is not None:
                    nn.init.zeros_(module.bias)
        self.compiler.reset_parameters()
        _role_seeded_xavier_uniform_(
            self.domain_basis_t,
            "world_graph.domain_basis_t",
        )
        _role_seeded_xavier_uniform_(
            self.domain_transfer_logits_t,
            "world_graph.domain_transfer_logits_t",
        )
        if self.domain_transfer_logits_t.device.type != "meta":
            with torch.no_grad():
                self.domain_transfer_logits_t.mul_(0.05).add_(
                    torch.eye(
                        self.cfg.domain_count,
                        device=self.domain_transfer_logits_t.device,
                        dtype=self.domain_transfer_logits_t.dtype,
                    )
                )
        _role_seeded_xavier_uniform_(
            self.action_read_t,
            "world_graph.action_read_t",
        )
        _role_seeded_xavier_uniform_(
            self.action_write_t,
            "world_graph.action_write_t",
        )
        _role_seeded_xavier_uniform_(
            self.action_bias_t,
            "world_graph.action_bias_t",
        )
        with torch.no_grad():
            self.domain_transfer_scale.fill_(0.25)
            self.prior_carry_scale.fill_(0.50)
            self.likelihood_scale.zero_()
            self.action_cost_logit_t.zero_()
            self.hill_climb_head.bias[1].fill_(1.0)
            self.exploration_influence_scale.zero_()
            self.world_residual_scale.zero_()
            self.rbo_stop_influence_scale.zero_()

    def promotion_authority(self) -> torch.Tensor:
        """Return durable conjunction; it has no decode or routing authority."""

        required_route_t = self.required_domain_action_coverage_t
        required_route_verified_t = (
            self.verified_domain_action_coverage_t | ~required_route_t
        ).all()
        required_operator_verified_t = (
            self.verified_operator_count_t.gt(0) | ~required_route_t
        ).all()
        return (
            self.accepted_gradient_update_count_t.gt(0)
            & self.retained_heldout_t
            & self.falsification_verified_t
            & self.cross_domain_transfer_verified_t
            & self.verified_abstraction_count_t.gt(0)
            & torch.isfinite(self.verified_coupled_gain_delta_t).all()
            & self.verified_coupled_gain_delta_t.gt(0).all()
            & torch.isfinite(
                self.verified_native_knowledge_ownership_gain_t
            )
            & self.verified_native_knowledge_ownership_gain_t.gt(0)
            & torch.isfinite(self.verified_self_correction_gain_t)
            & self.verified_self_correction_gain_t.gt(0)
            & required_route_t.any()
            & required_route_verified_t
            & required_operator_verified_t
            & self.promotion_cohort_identity_t.ne(0).any()
            & self.promotion_transaction_identity_t.ne(0).any()
            & self.promotion_heldout_identity_t.ne(0).any()
            & self.promotion_qualification_identity_t.ne(0).any()
            & self.promotion_candidate_lineage_t.ne(0).any()
            & self.verified_abstraction_cohort_identity_t.eq(
                self.promotion_cohort_identity_t
            ).all()
            & self.verified_abstraction_gradient_update_count_t.eq(
                self.accepted_gradient_update_count_t
            )
            & self.cold_reload_verified_t
        )

    def begin_promotion_cohort_boundary(
        self,
        *,
        cohort_identity_t: torch.Tensor,
        transaction_identity_t: torch.Tensor,
        heldout_identity_t: torch.Tensor,
        generation_t: torch.Tensor,
    ) -> torch.Tensor:
        """Open one generation-scoped held-out promotion cohort.

        A changed cohort resets all evaluation booleans while preserving the
        accumulated learned abstraction. It snapshots the *current accepted*
        gradient count before the cohort trains. Using the prior verified count
        here would let a failed cohort's unverified update satisfy a later
        cohort, which breaks exact transaction ownership.
        """

        identity_values = (
            cohort_identity_t,
            transaction_identity_t,
            heldout_identity_t,
        )
        if any(
            value.shape != (32,) or value.dtype != torch.uint8
            for value in identity_values
        ):
            raise ValueError(
                "causal promotion cohort identities must be uint8[32] tensors"
            )
        if generation_t.numel() != 1:
            raise ValueError("causal promotion generation must be scalar")
        active_cohort_t = cohort_identity_t.detach().to(
            device=self.promotion_cohort_identity_t.device,
            dtype=torch.uint8,
        )
        active_transaction_t = transaction_identity_t.detach().to(
            device=self.promotion_transaction_identity_t.device,
            dtype=torch.uint8,
        )
        active_heldout_t = heldout_identity_t.detach().to(
            device=self.promotion_heldout_identity_t.device,
            dtype=torch.uint8,
        )
        active_generation_t = generation_t.detach().to(
            device=self.promotion_generation_t.device,
            dtype=torch.long,
        ).reshape(())
        torch._assert_async(
            active_cohort_t.ne(0).any()
            & active_transaction_t.ne(0).any()
            & active_heldout_t.ne(0).any()
            & active_generation_t.ge(0),
            "causal promotion cohort identities must be non-zero",
        )
        same_cohort_t = (
            active_cohort_t.eq(self.promotion_cohort_identity_t).all()
            & active_transaction_t.eq(
                self.promotion_transaction_identity_t
            ).all()
            & active_heldout_t.eq(self.promotion_heldout_identity_t).all()
            & active_generation_t.eq(self.promotion_generation_t)
        )
        with torch.no_grad():
            if not bool(same_cohort_t.detach().cpu()):
                self.promotion_gradient_floor_t.copy_(
                    self.accepted_gradient_update_count_t
                )
                self.promotion_cohort_identity_t.copy_(active_cohort_t)
                self.promotion_transaction_identity_t.copy_(
                    active_transaction_t
                )
                self.promotion_heldout_identity_t.copy_(active_heldout_t)
                self.promotion_generation_t.copy_(active_generation_t)
                self.retained_heldout_t.zero_()
                self.falsification_verified_t.zero_()
                self.cross_domain_transfer_verified_t.zero_()
                self.cold_reload_verified_t.zero_()
                self.verified_coupled_gain_delta_t.zero_()
                self.verified_native_knowledge_ownership_gain_t.zero_()
                self.verified_self_correction_gain_t.zero_()
                self.required_domain_action_coverage_t.zero_()
                self.verified_domain_action_coverage_t.zero_()
                self.promotion_qualification_identity_t.zero_()
                self.promotion_candidate_lineage_t.zero_()
        return same_cohort_t

    def page_coupled_training_parameter_boundary(
        self,
        parameter: nn.Parameter,
    ) -> bool:
        """Identify one branch-local causal-head parameter.

        This is an optimizer-ownership boundary, not a routing decision.  The
        page branch may learn the complete proof-carrying causal head, including
        its compiler, recurrent workspace, experiment policy, world residual,
        continuation estimate, and RBO stop influence. Every parameter outside
        this additive head remains inherited and frozen until federation
        validates and promotes the branch delta.
        """

        return any(parameter is candidate for candidate in self.parameters())

    def activate_page_coupled_training_boundary(self) -> torch.Tensor:
        """Expose the complete additive causal head to one isolated page owner.

        The caller must already have established branch-local page ownership.
        Returning tensor counts keeps the optimizer admission proof tensor-native;
        no parameter names or host-authored routing choices cross the boundary.
        """

        for parameter in self.parameters():
            parameter.requires_grad_(True)
        trainable_parameters = tuple(
            parameter
            for parameter in self.parameters()
            if parameter.requires_grad
        )
        if not trainable_parameters or any(
            not self.page_coupled_training_parameter_boundary(parameter)
            for parameter in trainable_parameters
        ):
            raise RuntimeError(
                "page-coupled causal training escaped its isolated controls"
            )
        return self.exploration_influence_scale.detach().new_tensor(
            (
                len(trainable_parameters),
                sum(parameter.numel() for parameter in trainable_parameters),
            ),
            dtype=torch.long,
        )

    def record_gradient_update_boundary(
        self,
        *,
        gradient_norm_t: torch.Tensor,
        parameter_delta_norm_t: torch.Tensor,
    ) -> torch.Tensor:
        """Persist finite non-zero update evidence at the optimizer boundary."""

        if gradient_norm_t.numel() != 1 or parameter_delta_norm_t.numel() != 1:
            raise ValueError("causal update evidence must be scalar tensors")
        active_gradient_t = gradient_norm_t.detach().to(
            device=self.accepted_gradient_norm_t.device,
            dtype=self.accepted_gradient_norm_t.dtype,
        ).reshape(())
        active_delta_t = parameter_delta_norm_t.detach().to(
            device=self.accepted_parameter_delta_norm_t.device,
            dtype=self.accepted_parameter_delta_norm_t.dtype,
        ).reshape(())
        accepted_t = (
            torch.isfinite(active_gradient_t)
            & torch.isfinite(active_delta_t)
            & active_gradient_t.gt(0)
            & active_delta_t.gt(0)
        )
        with torch.no_grad():
            self.accepted_gradient_update_count_t.add_(
                accepted_t.to(dtype=torch.long)
            )
            self.accepted_gradient_norm_t.copy_(
                torch.where(
                    accepted_t,
                    active_gradient_t,
                    self.accepted_gradient_norm_t,
                )
            )
            self.accepted_parameter_delta_norm_t.copy_(
                torch.where(
                    accepted_t,
                    active_delta_t,
                    self.accepted_parameter_delta_norm_t,
                )
            )
        return accepted_t

    def record_promotion_evidence_boundary(
        self,
        evidence: CausalPromotionEvidence,
    ) -> torch.Tensor:
        """Monotonically persist independent held-out promotion evidence."""

        coupled_gain_t = evidence.coupled_gain_delta_t.detach().to(
            device=self.verified_coupled_gain_delta_t.device,
            dtype=self.verified_coupled_gain_delta_t.dtype,
        )
        if coupled_gain_t.shape != self.verified_coupled_gain_delta_t.shape:
            raise ValueError(
                "causal promotion coupled-gain evidence geometry differs"
            )
        if (
            evidence.native_knowledge_ownership_gain_t.numel() != 1
            or evidence.self_correction_gain_t.numel() != 1
        ):
            raise ValueError(
                "causal promotion knowledge and correction gains must be scalar"
            )
        knowledge_gain_t = (
            evidence.native_knowledge_ownership_gain_t.detach()
            .to(
                device=self.verified_native_knowledge_ownership_gain_t.device,
                dtype=self.verified_native_knowledge_ownership_gain_t.dtype,
            )
            .reshape(())
        )
        self_correction_gain_t = (
            evidence.self_correction_gain_t.detach()
            .to(
                device=self.verified_self_correction_gain_t.device,
                dtype=self.verified_self_correction_gain_t.dtype,
            )
            .reshape(())
        )
        required_route_t = evidence.required_domain_action_coverage_t.detach()
        observed_route_t = evidence.observed_domain_action_coverage_t.detach()
        if (
            required_route_t.dtype != torch.bool
            or observed_route_t.dtype != torch.bool
        ):
            raise ValueError(
                "causal promotion domain/action coverage must be boolean"
            )
        if required_route_t.numel() == 0 and observed_route_t.numel() == 0:
            required_route_t = torch.zeros_like(
                self.required_domain_action_coverage_t
            )
            observed_route_t = torch.zeros_like(
                self.verified_domain_action_coverage_t
            )
        elif (
            required_route_t.shape
            != self.required_domain_action_coverage_t.shape
            or observed_route_t.shape
            != self.verified_domain_action_coverage_t.shape
        ):
            raise ValueError(
                "causal promotion domain/action coverage geometry differs"
            )
        required_route_t = required_route_t.to(
            device=self.required_domain_action_coverage_t.device,
            dtype=torch.bool,
        )
        observed_route_t = observed_route_t.to(
            device=self.verified_domain_action_coverage_t.device,
            dtype=torch.bool,
        )
        qualification_identity_t = evidence.qualification_identity_t.detach().to(
            device=self.promotion_qualification_identity_t.device,
            dtype=torch.uint8,
        )
        candidate_lineage_t = evidence.candidate_lineage_t.detach().to(
            device=self.promotion_candidate_lineage_t.device,
            dtype=torch.uint8,
        )
        if (
            evidence.qualification_identity_t.dtype != torch.uint8
            or evidence.candidate_lineage_t.dtype != torch.uint8
            or
            qualification_identity_t.shape
            != self.promotion_qualification_identity_t.shape
            or candidate_lineage_t.shape
            != self.promotion_candidate_lineage_t.shape
        ):
            raise ValueError(
                "causal promotion qualification or lineage geometry differs"
            )
        values_t = torch.stack(
            (
                evidence.retained_heldout_t.reshape(()),
                evidence.falsification_verified_t.reshape(()),
                evidence.cross_domain_transfer_verified_t.reshape(()),
                evidence.cold_reload_verified_t.reshape(()),
            )
        ).to(
            device=self.retained_heldout_t.device,
            dtype=torch.bool,
        )
        with torch.no_grad():
            functional_gain_verified_t = (
                torch.isfinite(coupled_gain_t).all()
                & coupled_gain_t.gt(0).all()
                & torch.isfinite(knowledge_gain_t)
                & knowledge_gain_t.gt(0)
                & torch.isfinite(self_correction_gain_t)
                & self_correction_gain_t.gt(0)
                & required_route_t.any()
                & (observed_route_t | ~required_route_t).all()
                & qualification_identity_t.ne(0).any()
                & candidate_lineage_t.ne(0).any()
            )
            verified_transfer_value_t = (
                values_t[2]
                & functional_gain_verified_t
                & self.accepted_gradient_update_count_t.gt(
                    self.promotion_gradient_floor_t
                )
            )
            newly_verified_transfer_t = (
                verified_transfer_value_t
                & ~self.cross_domain_transfer_verified_t
            )
            self.retained_heldout_t.logical_or_(values_t[0])
            self.falsification_verified_t.logical_or_(values_t[1])
            self.cross_domain_transfer_verified_t.logical_or_(
                verified_transfer_value_t
            )
            self.cold_reload_verified_t.logical_or_(values_t[3])
            learned_transfer_t = torch.softmax(
                self.domain_transfer_logits_t.detach().float(),
                dim=-1,
            ).to(
                device=self.verified_abstraction_transfer_t.device,
                dtype=self.verified_abstraction_transfer_t.dtype,
            )
            next_abstraction_count_t = (
                self.verified_abstraction_count_t
                + newly_verified_transfer_t.to(dtype=torch.long)
            )
            accumulated_transfer_t = (
                self.verified_abstraction_transfer_t
                * self.verified_abstraction_count_t.to(dtype=torch.float32)
                + learned_transfer_t
            ) / next_abstraction_count_t.clamp_min(1).to(dtype=torch.float32)
            transferred_domain_t = torch.matmul(
                learned_transfer_t,
                self.domain_basis_t.detach().float(),
            )
            operator_rank_t = torch.einsum(
                "dw,awr->dar",
                transferred_domain_t,
                self.action_read_t.detach().float(),
            )
            learned_operator_signature_t = torch.tanh(
                torch.einsum(
                    "dar,arw->daw",
                    operator_rank_t,
                    self.action_write_t.detach().float(),
                )
                + self.action_bias_t.detach().float().unsqueeze(0)
            ).to(
                device=self.verified_operator_signature_t.device,
                dtype=self.verified_operator_signature_t.dtype,
            )
            verified_route_t = required_route_t & observed_route_t
            newly_verified_route_t = (
                verified_route_t & newly_verified_transfer_t
            )
            next_operator_count_t = (
                self.verified_operator_count_t
                + newly_verified_route_t.to(dtype=torch.long)
            )
            accumulated_operator_signature_t = (
                self.verified_operator_signature_t
                * self.verified_operator_count_t.to(dtype=torch.float32)
                .unsqueeze(-1)
                + learned_operator_signature_t
            ) / next_operator_count_t.clamp_min(1).to(
                dtype=torch.float32
            ).unsqueeze(-1)
            self.verified_abstraction_transfer_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    accumulated_transfer_t,
                    self.verified_abstraction_transfer_t,
                )
            )
            self.verified_abstraction_count_t.copy_(next_abstraction_count_t)
            self.verified_operator_signature_t.copy_(
                torch.where(
                    newly_verified_route_t.unsqueeze(-1),
                    accumulated_operator_signature_t,
                    self.verified_operator_signature_t,
                )
            )
            self.verified_operator_count_t.copy_(
                torch.where(
                    newly_verified_route_t,
                    next_operator_count_t,
                    self.verified_operator_count_t,
                )
            )
            self.verified_coupled_gain_delta_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    coupled_gain_t,
                    self.verified_coupled_gain_delta_t,
                )
            )
            self.verified_native_knowledge_ownership_gain_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    knowledge_gain_t,
                    self.verified_native_knowledge_ownership_gain_t,
                )
            )
            self.verified_self_correction_gain_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    self_correction_gain_t,
                    self.verified_self_correction_gain_t,
                )
            )
            self.required_domain_action_coverage_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    required_route_t,
                    self.required_domain_action_coverage_t,
                )
            )
            self.verified_domain_action_coverage_t.logical_or_(
                newly_verified_route_t
            )
            self.promotion_qualification_identity_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    qualification_identity_t,
                    self.promotion_qualification_identity_t,
                )
            )
            self.promotion_candidate_lineage_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    candidate_lineage_t,
                    self.promotion_candidate_lineage_t,
                )
            )
            self.verified_abstraction_cohort_identity_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    self.promotion_cohort_identity_t,
                    self.verified_abstraction_cohort_identity_t,
                )
            )
            self.verified_abstraction_gradient_update_count_t.copy_(
                torch.where(
                    newly_verified_transfer_t,
                    self.accepted_gradient_update_count_t,
                    self.verified_abstraction_gradient_update_count_t,
                )
            )
        return self.promotion_authority()

    def _active_verified_knowledge(
        self,
        source_domain_probability_t: torch.Tensor,
        learned_transfer_probability_t: torch.Tensor,
    ) -> CausalKnowledgeOwnershipPacket:
        """Fuse only independently verified abstractions into active reasoning.

        The zero-cohort path selects the learned tensors exactly, preserving
        legacy behavior. Each successful held-out cohort increases the
        model-owned prior monotonically while retaining the current learned
        theory as the other side of the blend. A disagreement is therefore a
        reason to continue causal traversal, not permission to overwrite the
        current theory from a host-authored record.
        """

        count_t = self.verified_abstraction_count_t.to(
            device=learned_transfer_probability_t.device,
            dtype=learned_transfer_probability_t.dtype,
        )
        ownership_strength_scalar_t = count_t / (count_t + 1.0)
        has_verified_knowledge_t = count_t.gt(0)
        verified_transfer_t = self.verified_abstraction_transfer_t.to(
            device=learned_transfer_probability_t.device,
            dtype=learned_transfer_probability_t.dtype,
        )
        blended_transfer_t = torch.lerp(
            learned_transfer_probability_t,
            verified_transfer_t,
            ownership_strength_scalar_t,
        )
        active_transfer_t = torch.where(
            has_verified_knowledge_t,
            blended_transfer_t,
            learned_transfer_probability_t,
        )
        learned_target_t = torch.matmul(
            source_domain_probability_t,
            learned_transfer_probability_t,
        )
        verified_target_t = torch.matmul(
            source_domain_probability_t,
            verified_transfer_t,
        )
        disagreement_t = torch.where(
            has_verified_knowledge_t,
            (learned_target_t - verified_target_t).abs().mean(dim=-1)
            * ownership_strength_scalar_t,
            learned_target_t.new_zeros(learned_target_t.shape[0]),
        )
        verified_operator_signature_t = torch.einsum(
            "bd,daw->baw",
            source_domain_probability_t,
            self.verified_operator_signature_t.to(
                device=source_domain_probability_t.device,
                dtype=source_domain_probability_t.dtype,
            ),
        )
        operator_count_t = self.verified_operator_count_t.to(
            device=source_domain_probability_t.device,
            dtype=source_domain_probability_t.dtype,
        )
        operator_strength_by_domain_t = operator_count_t / (
            operator_count_t + 1.0
        )
        operator_strength_t = torch.einsum(
            "bd,da->ba",
            source_domain_probability_t,
            operator_strength_by_domain_t,
        )
        return CausalKnowledgeOwnershipPacket(
            domain_transfer_probability_t=active_transfer_t,
            operator_signature_t=verified_operator_signature_t,
            operator_strength_t=operator_strength_t,
            ownership_strength_t=ownership_strength_scalar_t.expand(
                source_domain_probability_t.shape[0]
            ),
            disagreement_t=disagreement_t,
        )

    def _initial_world(
        self,
        evidence_t: torch.Tensor,
        pathway_context_t: torch.Tensor,
        source_domain_probability_t: torch.Tensor,
        target_domain_probability_t: torch.Tensor,
        prior_state: CausalWorldState | None,
    ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        batch_size = evidence_t.shape[0]
        ontology_t = torch.tanh(
            self.ontology_seed(evidence_t).reshape(
                batch_size,
                self.cfg.hypothesis_count,
                self.cfg.world_size,
            )
        )
        rule_t = torch.tanh(
            self.rule_seed(evidence_t).reshape(
                batch_size,
                self.cfg.hypothesis_count,
                self.cfg.world_size,
            )
        )
        pathway_t = self.pathway_seed(pathway_context_t).unsqueeze(1)
        domain_delta_t = torch.matmul(
            target_domain_probability_t - source_domain_probability_t,
            self.domain_basis_t.to(dtype=evidence_t.dtype),
        )
        transfer_scale_t = torch.tanh(self.domain_transfer_scale).to(
            dtype=evidence_t.dtype
        )
        ontology_t = ontology_t + pathway_t + transfer_scale_t * domain_delta_t.unsqueeze(1)
        rule_t = rule_t + transfer_scale_t * domain_delta_t.unsqueeze(1)
        prior_logits_t = self.hypothesis_prior(evidence_t)
        if prior_state is not None:
            active_prior = prior_state.validated(self.cfg, batch_size)
            carry_t = torch.sigmoid(self.prior_carry_scale).to(
                dtype=evidence_t.dtype
            )
            ontology_t = ontology_t + carry_t * active_prior.ontology_t.to(
                device=evidence_t.device,
                dtype=evidence_t.dtype,
            )
            rule_t = rule_t + carry_t * active_prior.rule_t.to(
                device=evidence_t.device,
                dtype=evidence_t.dtype,
            )
            prior_logits_t = prior_logits_t + active_prior.posterior_t.to(
                device=evidence_t.device,
                dtype=evidence_t.dtype,
            ).clamp_min(torch.finfo(evidence_t.dtype).tiny).log()
        return ontology_t, rule_t, prior_logits_t

    def _predict_interventions(
        self,
        compiled_state_t: torch.Tensor,
    ) -> torch.Tensor:
        action_read_t = torch.einsum(
            "bkw,awr->bkar",
            compiled_state_t,
            self.action_read_t.to(dtype=compiled_state_t.dtype),
        )
        action_effect_t = torch.einsum(
            "bkar,arw->bkaw",
            torch.tanh(action_read_t),
            self.action_write_t.to(dtype=compiled_state_t.dtype),
        )
        return torch.tanh(
            compiled_state_t.unsqueeze(2)
            + action_effect_t
            + self.action_bias_t.to(dtype=compiled_state_t.dtype).view(
                1,
                1,
                self.cfg.action_size,
                self.cfg.world_size,
            )
        )

    def _rbo_continuation_hold_t(
        self,
        reference_t: torch.Tensor,
        proof: CausalTheoryProofPacket,
    ) -> torch.Tensor:
        """Return learned continuation pressure with exact-zero migration."""

        if reference_t.ndim != 1:
            raise ValueError("causal RBO continuation reference expects [batch]")
        if proof.continuation_probability_t.shape != reference_t.shape:
            raise ValueError("causal continuation geometry differs")
        raw_scale_t = self.rbo_stop_influence_scale
        positive_scale_t = raw_scale_t.square()
        if self.training and torch.is_grad_enabled():
            positive_scale_t = positive_scale_t + (
                raw_scale_t - raw_scale_t.detach()
            )
        influence_t = torch.tanh(positive_scale_t).to(
            device=reference_t.device,
            dtype=reference_t.dtype,
        )
        continuation_t = proof.continuation_probability_t.to(
            device=reference_t.device,
            dtype=reference_t.dtype,
        )
        hold_t = influence_t * continuation_t
        if self.training and torch.is_grad_enabled():
            hold_t = hold_t + (1.0 - influence_t).detach() * (
                continuation_t - continuation_t.detach()
            )
        return hold_t

    def condition_rbo_stop_scores(
        self,
        stop_scores_t: torch.Tensor,
        proof: CausalTheoryProofPacket,
    ) -> torch.Tensor:
        """Keep RBO traversal open while its model-owned theory is unresolved.

        Only the two learned readiness channels are reduced. Task correctness
        confidence remains a separate conjunct. The migrated gate is exact zero,
        preserving legacy completion until training and held-out promotion open
        this path.
        """

        if stop_scores_t.ndim != 2 or stop_scores_t.shape[-1] != 3:
            raise ValueError("causal RBO stop scores expect [batch, 3]")
        hold_t = self._rbo_continuation_hold_t(
            stop_scores_t[:, 0],
            proof,
        ).unsqueeze(-1)
        readiness_t = stop_scores_t[:, :2]
        conditioned_readiness_t = readiness_t * (1.0 - hold_t)
        return torch.cat(
            (
                conditioned_readiness_t,
                stop_scores_t[:, 2:],
            ),
            dim=-1,
        )

    def condition_rbo_exit_probability(
        self,
        exit_probability_t: torch.Tensor,
        proof: CausalTheoryProofPacket,
    ) -> torch.Tensor:
        """Condition corrected-arm exit on the same learned proof obligation."""

        if exit_probability_t.ndim != 1:
            raise ValueError("causal RBO exit probability expects [batch]")
        hold_t = self._rbo_continuation_hold_t(
            exit_probability_t,
            proof,
        )
        return exit_probability_t * (1.0 - hold_t)

    def forward(
        self,
        hidden_t: torch.Tensor,
        *,
        action_context_t: torch.Tensor,
        pathway_context_t: torch.Tensor,
        prior_state: CausalWorldState | None = None,
    ) -> CausalAlgebraResult:
        if hidden_t.ndim != 3 or hidden_t.shape[-1] != self.cfg.hidden_size:
            raise ValueError(
                "causal world graph hidden expects [batch, sequence, hidden]"
            )
        batch_size = hidden_t.shape[0]
        if action_context_t.shape != (batch_size, self.cfg.action_size):
            raise ValueError("causal world graph action geometry differs")
        if pathway_context_t.shape != (batch_size, self.cfg.pathway_size):
            raise ValueError("causal world graph pathway geometry differs")
        active_prior_state = (
            None
            if prior_state is None
            else prior_state.validated(self.cfg, batch_size)
        )

        active_action_t = action_context_t.to(
            device=hidden_t.device,
            dtype=hidden_t.dtype,
        )
        active_pathway_t = pathway_context_t.to(
            device=hidden_t.device,
            dtype=hidden_t.dtype,
        )
        evidence_t = self.evidence_norm(hidden_t.mean(dim=1))
        source_domain_probability_t = _stable_probability(
            self.domain_source(evidence_t),
            dim=-1,
        )
        learned_transfer_probability_t = _stable_probability(
            self.domain_transfer_logits_t.to(dtype=evidence_t.dtype),
            dim=-1,
        )
        knowledge_ownership = self._active_verified_knowledge(
            source_domain_probability_t,
            learned_transfer_probability_t,
        )
        domain_transfer_probability_t = (
            knowledge_ownership.domain_transfer_probability_t
        )
        target_domain_probability_t = torch.matmul(
            source_domain_probability_t,
            domain_transfer_probability_t,
        )
        ontology_t, rule_t, prior_logits_t = self._initial_world(
            evidence_t,
            active_pathway_t,
            source_domain_probability_t,
            target_domain_probability_t,
            active_prior_state,
        )
        program = self.compiler(ontology_t, rule_t)
        predicted_outcome_t = self._predict_interventions(
            program.compiled_state_t
        )
        knowledge_conditioned_outcome_t = torch.tanh(
            predicted_outcome_t
            + knowledge_ownership.operator_strength_t.unsqueeze(1).unsqueeze(
                -1
            )
            * knowledge_ownership.operator_signature_t.unsqueeze(1)
        )
        predicted_outcome_t = torch.where(
            knowledge_ownership.ownership_strength_t.gt(0).view(
                batch_size,
                1,
                1,
                1,
            ),
            knowledge_conditioned_outcome_t,
            predicted_outcome_t,
        )

        observation_t = torch.tanh(
            self.observation_hidden(evidence_t)
            + self.observation_action(active_action_t)
        )
        observed_action_probability_t = _stable_probability(
            self.action_policy(active_action_t),
            dim=-1,
        )
        if active_prior_state is not None:
            action_carry_t = torch.sigmoid(self.prior_carry_scale).to(
                dtype=observed_action_probability_t.dtype
            )
            observed_action_probability_t = torch.lerp(
                observed_action_probability_t,
                active_prior_state.action_policy_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                action_carry_t,
            )
            observed_action_probability_t = (
                observed_action_probability_t
                / observed_action_probability_t.sum(
                    dim=-1,
                    keepdim=True,
                ).clamp_min(
                    torch.finfo(observed_action_probability_t.dtype).tiny
                )
            )
        predicted_observation_t = torch.einsum(
            "ba,bkaw->bkw",
            observed_action_probability_t,
            predicted_outcome_t,
        )
        observation_error_t = (
            predicted_observation_t - observation_t.unsqueeze(1)
        ).square().mean(dim=-1)
        posterior_t = _stable_probability(
            prior_logits_t
            - F.softplus(self.likelihood_scale).to(
                dtype=observation_error_t.dtype
            )
            * observation_error_t,
            dim=-1,
        )

        proof_context_t = torch.cat(
            (
                program.compiled_state_t,
                observation_t.unsqueeze(1).expand(
                    -1,
                    self.cfg.hypothesis_count,
                    -1,
                ),
                observation_error_t.unsqueeze(-1),
            ),
            dim=-1,
        )
        contradiction_t = torch.sigmoid(
            self.contradiction_head(proof_context_t).squeeze(-1)
        )
        commitment_t = torch.sigmoid(
            self.commitment_head(proof_context_t).squeeze(-1)
            - contradiction_t
        )
        reopen_probability_t = torch.sigmoid(
            self.reopen_head(proof_context_t).squeeze(-1)
            + contradiction_t
            - commitment_t
        )
        action_cost_t = F.softplus(self.action_cost_logit_t).to(
            dtype=hidden_t.dtype
        ) + torch.finfo(hidden_t.dtype).eps
        falsifying_experiment = cheapest_falsifying_experiment(
            predicted_outcome_t,
            posterior_t,
            action_cost_t,
            policy_bias_t=self.experiment_policy(evidence_t),
        )
        candidate_world_summary_t = torch.einsum(
            "bk,bkw->bw",
            posterior_t,
            program.compiled_state_t,
        )
        prior_working_memory_t = (
            candidate_world_summary_t.new_zeros(
                batch_size,
                self.cfg.world_size,
            )
            if active_prior_state is None
            else active_prior_state.working_memory_t.to(
                device=hidden_t.device,
                dtype=hidden_t.dtype,
            )
        )
        working_memory_input_t = torch.cat(
            (evidence_t, prior_working_memory_t),
            dim=-1,
        )
        working_memory_proposal_t = torch.tanh(
            self.working_memory_input(working_memory_input_t)
            + candidate_world_summary_t
        )
        working_memory_gate_t = torch.sigmoid(
            self.working_memory_gate(working_memory_input_t)
        )
        candidate_working_memory_t = torch.lerp(
            prior_working_memory_t,
            working_memory_proposal_t,
            working_memory_gate_t,
        )

        exploitation_probability_t = observed_action_probability_t
        information_probability_t = (
            falsifying_experiment.experiment_probability_t
        )
        counter_thompson_probability_t = _stable_probability(
            self.counter_thompson_head(
                torch.cat(
                    (
                        evidence_t,
                        candidate_working_memory_t,
                        exploitation_probability_t,
                        information_probability_t,
                    ),
                    dim=-1,
                )
            ),
            dim=-1,
        )[:, 1]
        raw_exploration_scale_t = self.exploration_influence_scale
        exploration_scale_t = raw_exploration_scale_t.square()
        if self.training and torch.is_grad_enabled():
            exploration_scale_t = exploration_scale_t + (
                raw_exploration_scale_t
                - raw_exploration_scale_t.detach()
            )
        exploration_influence_t = exploration_scale_t.clamp(
            min=0.0,
            max=1.0,
        ).to(
            dtype=hidden_t.dtype
        )
        exploration_counterweight_t = (
            exploration_influence_t * counter_thompson_probability_t
        )
        if self.training and torch.is_grad_enabled():
            exploration_counterweight_t = exploration_counterweight_t + (
                1.0 - exploration_influence_t
            ).detach() * (
                counter_thompson_probability_t
                - counter_thompson_probability_t.detach()
            )
        action_probability_t = (
            (1.0 - exploration_counterweight_t).unsqueeze(-1)
            * exploitation_probability_t
            + exploration_counterweight_t.unsqueeze(-1)
            * information_probability_t
        )
        action_probability_t = action_probability_t / action_probability_t.sum(
            dim=-1,
            keepdim=True,
        ).clamp_min(torch.finfo(hidden_t.dtype).tiny)
        action_index_t = action_probability_t.argmax(dim=-1)
        action_operator_t = torch.einsum(
            "awr,arv->awv",
            self.action_read_t.to(dtype=hidden_t.dtype),
            self.action_write_t.to(dtype=hidden_t.dtype),
        )
        expected_action_operator_t = torch.einsum(
            "ba,awv->bwv",
            action_probability_t,
            action_operator_t,
        )
        expected_action_bias_t = torch.einsum(
            "ba,aw->bw",
            action_probability_t,
            self.action_bias_t.to(dtype=hidden_t.dtype),
        )
        source_basis_t = F.normalize(
            torch.matmul(
                source_domain_probability_t,
                self.domain_basis_t.to(dtype=hidden_t.dtype),
            ).float(),
            dim=-1,
        ).to(dtype=hidden_t.dtype)
        target_basis_t = F.normalize(
            torch.matmul(
                target_domain_probability_t,
                self.domain_basis_t.to(dtype=hidden_t.dtype),
            ).float(),
            dim=-1,
        ).to(dtype=hidden_t.dtype)
        domain_operator_t = torch.einsum(
            "bi,bj->bij",
            target_basis_t,
            source_basis_t,
        )
        operator_commutator_error_t = (
            torch.matmul(expected_action_operator_t, domain_operator_t)
            - torch.matmul(domain_operator_t, expected_action_operator_t)
        ).float().square().mean(dim=(-1, -2)).to(dtype=hidden_t.dtype)
        transferred_basis_t = torch.einsum(
            "bij,bj->bi",
            domain_operator_t,
            source_basis_t,
        )
        reconstructed_basis_t = torch.einsum(
            "bji,bj->bi",
            domain_operator_t,
            transferred_basis_t,
        )
        domain_cycle_error_t = (
            reconstructed_basis_t - source_basis_t
        ).float().square().mean(dim=-1).to(dtype=hidden_t.dtype)
        if active_prior_state is None:
            successor_replay_error_t = candidate_world_summary_t.new_zeros(
                batch_size
            )
        else:
            prior_world_summary_t = torch.einsum(
                "bk,bkw->bw",
                active_prior_state.posterior_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                active_prior_state.ontology_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
            )
            imagined_successor_t = torch.tanh(
                torch.einsum(
                    "bwv,bv->bw",
                    expected_action_operator_t,
                    prior_world_summary_t,
                )
                + expected_action_bias_t
            )
            successor_replay_error_t = (
                imagined_successor_t
                - candidate_world_summary_t.detach()
            ).float().square().mean(dim=-1).to(dtype=hidden_t.dtype)
        replay_error_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            observation_error_t,
        )
        information_value_t = torch.einsum(
            "ba,ba->b",
            action_probability_t,
            falsifying_experiment.value_per_cost_t,
        )
        weighted_commitment_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            commitment_t,
        )
        weighted_contradiction_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            contradiction_t,
        )
        candidate_quality_t = (
            weighted_commitment_t
            - weighted_contradiction_t
            - replay_error_t
        )
        if active_prior_state is None:
            prior_quality_t = candidate_quality_t.detach()
        else:
            active_prior_posterior_t = active_prior_state.posterior_t.to(
                device=hidden_t.device,
                dtype=hidden_t.dtype,
            )
            prior_quality_t = (
                torch.einsum(
                    "bk,bk->b",
                    active_prior_posterior_t,
                    active_prior_state.commitment_t.to(
                        device=hidden_t.device,
                        dtype=hidden_t.dtype,
                    ),
                )
                - torch.einsum(
                    "bk,bk->b",
                    active_prior_posterior_t,
                    active_prior_state.contradiction_t.to(
                        device=hidden_t.device,
                        dtype=hidden_t.dtype,
                    ),
                )
                - active_prior_state.replay_error_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                )
            )
        hill_climb_gain_t = candidate_quality_t - prior_quality_t
        hill_climb_probability_t = _stable_probability(
            self.hill_climb_head(
                torch.cat(
                    (
                        candidate_working_memory_t,
                        candidate_quality_t.unsqueeze(-1),
                        prior_quality_t.unsqueeze(-1),
                        hill_climb_gain_t.unsqueeze(-1),
                        information_value_t.unsqueeze(-1),
                    ),
                    dim=-1,
                )
            ),
            dim=-1,
        )[:, 1]
        hill_climb_accept_t = (
            torch.ones_like(hill_climb_probability_t, dtype=torch.bool)
            if active_prior_state is None
            else hill_climb_probability_t.ge(0.5)
        )
        hill_climb_accept_weight_t = (
            hill_climb_accept_t.to(dtype=hidden_t.dtype)
            + hill_climb_probability_t
            - hill_climb_probability_t.detach()
        )

        if active_prior_state is None:
            accepted_ontology_t = program.compiled_state_t
            accepted_rule_t = rule_t
            accepted_posterior_t = posterior_t
            accepted_contradiction_t = contradiction_t
            accepted_commitment_t = commitment_t
            accepted_working_memory_t = candidate_working_memory_t
            accepted_action_probability_t = action_probability_t
            accepted_replay_error_t = replay_error_t
            accepted_information_value_t = information_value_t
            accepted_exploration_counterweight_t = exploration_counterweight_t
        else:
            accepted_ontology_t = _hill_climb_lerp(
                active_prior_state.ontology_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                program.compiled_state_t,
                hill_climb_accept_weight_t,
            )
            accepted_rule_t = _hill_climb_lerp(
                active_prior_state.rule_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                rule_t,
                hill_climb_accept_weight_t,
            )
            accepted_posterior_t = _hill_climb_lerp(
                active_prior_state.posterior_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                posterior_t,
                hill_climb_accept_weight_t,
            )
            accepted_contradiction_t = _hill_climb_lerp(
                active_prior_state.contradiction_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                contradiction_t,
                hill_climb_accept_weight_t,
            )
            accepted_commitment_t = _hill_climb_lerp(
                active_prior_state.commitment_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                commitment_t,
                hill_climb_accept_weight_t,
            )
            accepted_working_memory_t = _hill_climb_lerp(
                prior_working_memory_t,
                candidate_working_memory_t,
                hill_climb_accept_weight_t,
            )
            accepted_action_probability_t = _hill_climb_lerp(
                active_prior_state.action_policy_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                action_probability_t,
                hill_climb_accept_weight_t,
            )
            accepted_replay_error_t = _hill_climb_lerp(
                active_prior_state.replay_error_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                replay_error_t,
                hill_climb_accept_weight_t,
            )
            accepted_information_value_t = _hill_climb_lerp(
                active_prior_state.information_value_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                information_value_t,
                hill_climb_accept_weight_t,
            )
            accepted_exploration_counterweight_t = _hill_climb_lerp(
                active_prior_state.exploration_counterweight_t.to(
                    device=hidden_t.device,
                    dtype=hidden_t.dtype,
                ),
                exploration_counterweight_t,
                hill_climb_accept_weight_t,
            )
        world_state = CausalWorldState(
            ontology_t=accepted_ontology_t,
            rule_t=accepted_rule_t,
            posterior_t=accepted_posterior_t,
            contradiction_t=accepted_contradiction_t,
            commitment_t=accepted_commitment_t,
            working_memory_t=accepted_working_memory_t,
            action_policy_t=accepted_action_probability_t,
            replay_error_t=accepted_replay_error_t,
            information_value_t=accepted_information_value_t,
            exploration_counterweight_t=(
                accepted_exploration_counterweight_t
            ),
            hill_climb_accept_t=hill_climb_accept_t,
        )
        working_memory_delta_l2_t = (
            accepted_working_memory_t - prior_working_memory_t
        ).float().square().sum(dim=-1).sqrt().to(dtype=hidden_t.dtype)
        exploration = CounterThompsonHillClimbPacket(
            exploitation_probability_t=exploitation_probability_t,
            information_probability_t=information_probability_t,
            action_probability_t=action_probability_t,
            action_index_t=action_index_t,
            counter_thompson_probability_t=(
                counter_thompson_probability_t
            ),
            hill_climb_probability_t=hill_climb_probability_t,
            hill_climb_gain_t=hill_climb_gain_t,
            hill_climb_accept_t=hill_climb_accept_t,
            hill_climb_accept_weight_t=hill_climb_accept_weight_t,
            working_memory_t=accepted_working_memory_t,
            working_memory_delta_l2_t=working_memory_delta_l2_t,
            replay_error_t=replay_error_t,
            successor_replay_error_t=successor_replay_error_t,
            operator_commutator_error_t=operator_commutator_error_t,
            domain_cycle_error_t=domain_cycle_error_t,
            information_value_t=information_value_t,
            exploration_counterweight_t=exploration_counterweight_t,
        )
        world_summary_t = torch.einsum(
            "bk,bkw->bw",
            accepted_posterior_t,
            accepted_ontology_t,
        )
        rule_summary_t = torch.einsum(
            "bk,bkw->bw",
            accepted_posterior_t,
            accepted_rule_t,
        )
        domain_transfer_t = torch.matmul(
            target_domain_probability_t - source_domain_probability_t,
            self.domain_basis_t.to(dtype=hidden_t.dtype),
        )
        token_world_t = torch.tanh(
            self.token_world(hidden_t)
            + world_summary_t.unsqueeze(1)
            + rule_summary_t.unsqueeze(1)
            + domain_transfer_t.unsqueeze(1)
        )
        world_residual_t = self.world_hidden(token_world_t)
        residual_scale_t = torch.tanh(self.world_residual_scale).to(
            dtype=hidden_t.dtype
        )
        output_hidden_t = hidden_t + residual_scale_t * world_residual_t
        if self.training and torch.is_grad_enabled():
            output_hidden_t = output_hidden_t + (
                1.0 - residual_scale_t
            ).detach() * (world_residual_t - world_residual_t.detach())

        posterior_normalization_t = (
            posterior_t.sum(dim=-1) - 1.0
        ).abs()
        executable_consistency_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            program.executable_consistency_t,
        )
        observed_consistency_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            observation_error_t,
        )
        contradiction_commitment_t = torch.einsum(
            "bk,bk,bk->b",
            posterior_t,
            contradiction_t,
            commitment_t,
        )
        reverse_domain_probability_t = torch.matmul(
            target_domain_probability_t,
            domain_transfer_probability_t.transpose(0, 1),
        )
        reverse_domain_probability_t = (
            reverse_domain_probability_t
            / reverse_domain_probability_t.sum(
                dim=-1,
                keepdim=True,
            ).clamp_min(torch.finfo(hidden_t.dtype).tiny)
        )
        transfer_cycle_t = (
            reverse_domain_probability_t - source_domain_probability_t
        ).square().mean(dim=-1)
        normalized_hypothesis_t = F.normalize(
            program.compiled_state_t.float(),
            dim=-1,
        )
        similarity_t = torch.matmul(
            normalized_hypothesis_t,
            normalized_hypothesis_t.transpose(-1, -2),
        )
        off_diagonal_t = 1.0 - torch.eye(
            self.cfg.hypothesis_count,
            device=similarity_t.device,
            dtype=similarity_t.dtype,
        )
        hypothesis_collapse_t = (
            similarity_t.square() * off_diagonal_t
        ).sum(dim=(-1, -2)) / (
            self.cfg.hypothesis_count * (self.cfg.hypothesis_count - 1)
        )
        proof_obligation_t = torch.stack(
            (
                posterior_normalization_t,
                executable_consistency_t,
                observed_consistency_t,
                contradiction_commitment_t,
                transfer_cycle_t,
                hypothesis_collapse_t.to(dtype=hidden_t.dtype),
            ),
            dim=-1,
        )
        proof_validity_t = torch.sigmoid(
            self.proof_verifier(
                torch.cat(
                    (
                        world_summary_t,
                        proof_obligation_t.to(dtype=world_summary_t.dtype),
                    ),
                    dim=-1,
                )
            ).squeeze(-1)
            - proof_obligation_t.mean(dim=-1)
        )
        weighted_reopen_t = torch.einsum(
            "bk,bk->b",
            posterior_t,
            reopen_probability_t,
        )
        continuation_feature_t = torch.cat(
            (
                world_summary_t,
                proof_obligation_t.to(dtype=world_summary_t.dtype),
                weighted_contradiction_t.unsqueeze(-1),
                weighted_reopen_t.unsqueeze(-1),
                (1.0 - proof_validity_t).unsqueeze(-1),
                falsifying_experiment.selection_margin_t.unsqueeze(-1),
            ),
            dim=-1,
        )
        continuation_probability_t = torch.sigmoid(
            self.continuation_head(continuation_feature_t).squeeze(-1)
            + exploration_counterweight_t
            + (1.0 - hill_climb_probability_t)
            + knowledge_ownership.disagreement_t
        )
        # Both learning signals are target-free.  Information-directed loss
        # rewards interventions that distinguish still-credible worlds per
        # unit cost.  Hill-climb supervision is derived only from replayed
        # model evidence and never from target token ids or a reference answer.
        information_directed_loss_t = torch.reciprocal(
            1.0 + information_value_t.clamp_min(0.0)
        ).mean()
        hill_climb_target_t = hill_climb_gain_t.detach().ge(0).to(
            dtype=hill_climb_probability_t.dtype
        )
        hill_climb_loss_t = F.binary_cross_entropy(
            hill_climb_probability_t.clamp(
                min=torch.finfo(hill_climb_probability_t.dtype).eps,
                max=1.0
                - torch.finfo(hill_climb_probability_t.dtype).eps,
            ),
            hill_climb_target_t,
        )
        auxiliary_loss_t = (
            proof_obligation_t.mean()
            + 0.05 * information_directed_loss_t
            + 0.05 * hill_climb_loss_t
            + 0.02 * successor_replay_error_t.mean()
            + 0.02 * operator_commutator_error_t.mean()
            + 0.02 * domain_cycle_error_t.mean()
        )
        promotion_authority_t = self.promotion_authority().to(
            device=hidden_t.device
        ).expand(batch_size)
        proof = CausalTheoryProofPacket(
            world_state=world_state,
            program=program,
            predicted_outcome_t=predicted_outcome_t,
            observation_t=observation_t,
            observation_error_t=observation_error_t,
            falsifying_experiment=falsifying_experiment,
            exploration=exploration,
            source_domain_probability_t=source_domain_probability_t,
            target_domain_probability_t=target_domain_probability_t,
            domain_transfer_t=domain_transfer_t,
            knowledge_ownership_t=(
                knowledge_ownership.ownership_strength_t
            ),
            knowledge_disagreement_t=knowledge_ownership.disagreement_t,
            reopen_probability_t=reopen_probability_t,
            proof_obligation_t=proof_obligation_t,
            proof_validity_t=proof_validity_t,
            continuation_probability_t=continuation_probability_t,
            promotion_authority_t=promotion_authority_t,
        )
        return CausalAlgebraResult(
            hidden_t=output_hidden_t,
            proof=proof,
            auxiliary_loss_t=auxiliary_loss_t,
        )


def render_causal_notation_boundary(
    proof: CausalTheoryProofPacket,
    *,
    batch_index: int = 0,
    hypothesis_index: int = 0,
) -> str:
    """Render one model-owned program at an explicit diagnostic I/O boundary."""

    notation_t = proof.program.notation_probability_t
    if notation_t.ndim != 4:
        raise ValueError("causal notation tensor expects four dimensions")
    if batch_index < 0 or batch_index >= notation_t.shape[0]:
        raise IndexError("causal notation batch index is out of range")
    if hypothesis_index < 0 or hypothesis_index >= notation_t.shape[1]:
        raise IndexError("causal notation hypothesis index is out of range")
    opcode_ids = (
        notation_t[batch_index, hypothesis_index]
        .argmax(dim=-1)
        .detach()
        .cpu()
        .tolist()
    )
    opcode_names = tuple(
        CAUSAL_ALGEBRA_OPCODE_NAMES[opcode_id]
        if opcode_id < len(CAUSAL_ALGEBRA_OPCODE_NAMES)
        else f"OP_{opcode_id}"
        for opcode_id in opcode_ids
    )
    experiment_index = int(
        proof.falsifying_experiment.experiment_index_t[batch_index]
        .detach()
        .cpu()
    )
    proof_validity = float(
        proof.proof_validity_t[batch_index].detach().cpu()
    )
    posterior = float(
        proof.world_state.posterior_t[
            batch_index,
            hypothesis_index,
        ]
        .detach()
        .cpu()
    )
    return (
        f"H{hypothesis_index}[p={posterior:.6f}] := "
        f"{' |> '.join(opcode_names)} ; "
        f"TEST do(A{experiment_index}) ; PROOF {proof_validity:.6f}"
    )


def causal_algebra_state_parameter_count_boundary(
    module: CausalAlgebraWorldGraph,
) -> torch.Tensor:
    """Return parameter and persistent-state elements at a receipt boundary."""

    parameter_count = sum(
        parameter_t.numel() for parameter_t in module.parameters()
    )
    persistent_count = sum(
        buffer_t.numel()
        for buffer_name, buffer_t in module.named_buffers()
        if buffer_name not in module._non_persistent_buffers_set
    )
    reference_t = cast(torch.Tensor, module.world_residual_scale)
    return reference_t.new_tensor(
        (parameter_count, persistent_count),
        dtype=torch.long,
    )