"""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, )