CODE: https://github.com/frank-morales2020/AST/blob/main/WM_TOPO_PLASTICITY.ipynb

CODE-2: https://github.com/frank-morales2020/AST/blob/main/PHYSICAL_AI.ipynb

Topo-CBP Embodied World Model (3D Latent Dynamics)

Official model checkpoint and architecture for the Topological Continual Backpropagation (Topo-CBP) Embodied World Model, a lifelong spatio-temporal transition network designed for physical robotics, autonomous flight control, and embodied systems.

The model addresses the primary failure mode of classical world models—compounding rollout hallucination caused by coordinate drift—by decomposing the parameter manifold into an invariant prime coordinate bedrock and a metabolically plastic Continual Backpropagation subspace.

---

Model Summary

• Repository: frankmorales2020/topo-cbp-world-model-3d
• Architecture: EmbodiedWorldModel (Latent Dim: 128, Action Dim: 6, Hidden Dim: 256)
• Target Task: 3D Spatio-Temporal Latent Transition Modeling ($s_{t+1} \sim p(s_{t+1} \mid s_t, a_t)$)
• Prime Coordinate Anchors ($\mathcal{P}$): {2, 3, 5, 7, 11, 13}
• Euler Attenuation Constant ($\Lambda$): $\Lambda = 1 - \prod_{p \in \mathcal{P}} \left(1 - \frac{1}{p}\right) \approx 0.9785142874$
• State Integrity Digest: 2ce28ffcf131d5c9 (Bare-metal SHA-256 state hash)
• Memory Complexity: $\mathcal{O}(1)$ (Zero replay buffers required)
• Format: safetensors (Zero-copy binary mapping, no pickle vulnerabilities)

---

Key Architectural Mechanics

1. Invariant Prime Coordinate Ring ($\mathcal{P}$)

The input projection layer in_proj is explicitly partitioned. Rows corresponding to prime indices ${2, 3, 5, 7, 11, 13}$ are registered as an immutable reference frame that preserves Euclidean scale, spatial orientation, and conservation laws across continuous learning.

2. Dual-Phase Optimizer Projection

Standard gradient zeroing fails under AdamW due to decoupled weight decay ($-\eta \lambda_{wd} W$). The Topological Governor implements a two-phase protocol:

• Pre-Step: Orthogonal gradient projection on anchor indices.
• Post-Step: Exact bitwise restoration ($W_{\mathcal{P}} \leftarrow W_{\mathcal{P}}^{(0)}$) to eliminate weight decay drift at the register level.

3. Zero-Shock Metabolic Unit Recycling

Continual Backpropagation monitors running neuron feature utility ($u_i$). When dormant units (bottom 4%) are recycled:

• Incoming weights are re-initialized using Kaiming normal distributions.
• Outgoing weights are explicitly zeroed, preventing newly spawned units from injecting high-variance shocks into downstream layers.

4. Bare-Metal Cryptographic Safety Interlock

The raw byte memory of the prime coordinate anchors is hashed via SHA-256 on every cycle. This enables fail-closed hardware interlocks: if bitwise corruption or coordinate shear occurs, the digest mutates and the control plane immediately arrests actuator torque commands.

---

Empirical Benchmark & Telemetry

100,000-Step Non-Stationary Physical Regime Stress Test

Evaluated over 100,000 continuous steps across four distinct physical transition environments:

1. Regime A (Steps 1–25,000): Newtonian Dynamics ($\sin(z) + 0.1a$)
2. Regime B (Steps 25,001–50,000): Non-Linear Turbulence ($\cos(1.5z) - 0.3\tanh(a)$)
3. Regime C (Steps 50,001–75,000): Viscous Quadratic Drag ($1.2\sigma(z) + 0.05a^2$)
4. Regime D (Steps 75,001–100,000): Gyroscopic Precession / Dynamic Roll

---

Quickstart & Verification

import hashlib
import json
import torch
import torch.nn as nn
import torch.nn.functional as F
from huggingface_hub import hf_hub_download
from safetensors.torch import load_file

# Define Architecture
class EmbodiedWorldModel(nn.Module):
    def __init__(self, latent_dim: int = 128, action_dim: int = 6, hidden_dim: int = 256):
        super().__init__()
        self.in_proj = nn.Linear(latent_dim + action_dim, hidden_dim, bias=False)
        self.plastic_linear1 = nn.Linear(hidden_dim, hidden_dim, bias=False)
        self.plastic_linear2 = nn.Linear(hidden_dim, hidden_dim, bias=False)
        self.out_proj = nn.Linear(hidden_dim, latent_dim, bias=False)

    def forward(self, z_t: torch.Tensor, a_t: torch.Tensor) -> torch.Tensor:
        x = torch.cat([z_t, a_t], dim=-1)
        h = F.silu(self.in_proj(x))
        h1 = self.plastic_linear1(h)
        h2 = F.silu(self.plastic_linear2(h))
        return self.out_proj(h1 * h2)

# Load Checkpoint & Config from Hub
repo_id = "frankmorales2020/topo-cbp-world-model-3d"
config_path = hf_hub_download(repo_id=repo_id, filename="config.json")
weights_path = hf_hub_download(repo_id=repo_id, filename="model.safetensors")

with open(config_path, "r") as f:
    cfg = json.load(f)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = EmbodiedWorldModel(
    latent_dim=cfg["latent_dim"],
    action_dim=cfg["action_dim"],
    hidden_dim=cfg["hidden_dim"]
).to(device)

model.load_state_dict(load_file(weights_path, device=str(device)))
model.eval()

# Cryptographic State Audit
anchors = torch.tensor(cfg["prime_anchors"], dtype=torch.long)
raw_bytes = model.in_proj.weight[anchors, :].detach().cpu().numpy().tobytes()
computed_hash = hashlib.sha256(raw_bytes).hexdigest()[:16]

print(f"[*] Certified Hash : {cfg['invariant_sha256_hash']}")
print(f"[*] Computed Hash  : {computed_hash}")
assert computed_hash == cfg["invariant_sha256_hash"], "Manifold distortion detected!"
print("[✓] State integrity audit passed. Model ready for real-time actuation.")

import hashlib
import json
import random
import time
from huggingface_hub import hf_hub_download
import numpy as np
from safetensors.torch import load_file
import torch
import torch.nn as nn
import torch.nn.functional as F

# ==============================================================================
# DETERMINISTIC SEED ENFORCEMENT (SEED 123)
# ==============================================================================
SEED = 123
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
if torch.cuda.is_available():
  torch.cuda.manual_seed(SEED)
  torch.cuda.manual_seed_all(SEED)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False


# ==============================================================================
# 1. TOPOLOGICAL EMBODIED WORLD MODEL ARCHITECTURE
# ==============================================================================
class EmbodiedWorldModel(nn.Module):

  def __init__(
      self,
      latent_dim: int = 128,
      action_dim: int = 6,
      hidden_dim: int = 256,
  ):
    super().__init__()
    self.latent_dim = latent_dim
    self.action_dim = action_dim
    self.hidden_dim = hidden_dim

    # Coordinate projection bedrock layer
    self.in_proj = nn.Linear(latent_dim + action_dim, hidden_dim, bias=False)

    # Swish-gated plastic continual learning manifold
    self.plastic_linear1 = nn.Linear(hidden_dim, hidden_dim, bias=False)
    self.plastic_linear2 = nn.Linear(hidden_dim, hidden_dim, bias=False)

    # Next-state latent transition head
    self.out_proj = nn.Linear(hidden_dim, latent_dim, bias=False)

  def forward(self, z_t: torch.Tensor, a_t: torch.Tensor) -> torch.Tensor:
    x = torch.cat([z_t, a_t], dim=-1)
    h = F.silu(self.in_proj(x))
    h1 = self.plastic_linear1(h)
    h2 = F.silu(self.plastic_linear2(h))
    return self.out_proj(h1 * h2)


# ==============================================================================
# 2. RUNTIME TOPOLOGICAL GOVERNOR & METABOLIC ENGINE
# ==============================================================================
class RuntimeGovernor:

  def __init__(
      self,
      target_tensor: torch.nn.Parameter,
      prime_anchors=(2, 3, 5, 7, 11, 13),
      certified_hash="2ce28ffcf131d5c9",
  ):
    self.target = target_tensor
    self.prime_anchors = torch.tensor(prime_anchors, dtype=torch.long)
    self.certified_hash = certified_hash
    with torch.no_grad():
      self.invariant_bedrock = (
          self.target[self.prime_anchors, :].clone().detach()
      )

  def project_and_lock(self):
    if self.target.grad is not None:
      with torch.no_grad():
        self.target.grad[self.prime_anchors, :] = 0.0
    with torch.no_grad():
      self.target[self.prime_anchors, :] = self.invariant_bedrock

  def compute_drift(self) -> float:
    with torch.no_grad():
      current = self.target[self.prime_anchors, :]
      return torch.norm(current - self.invariant_bedrock).item()

  def verify_hardware_interlock(self) -> bool:
    raw_bytes = (
        self.target[self.prime_anchors, :].detach().cpu().numpy().tobytes()
    )
    current_hash = hashlib.sha256(raw_bytes).hexdigest()[:16]
    return current_hash == self.certified_hash


class OnlinePlasticityEngine:

  def __init__(
      self, layer1: nn.Linear, layer2: nn.Linear, decay: float = 0.99
  ):
    self.layer1 = layer1
    self.layer2 = layer2
    self.decay = decay
    self.hidden_dim = layer1.out_features
    self.running_utility = torch.ones(self.hidden_dim)

  def update(self, hidden_acts: torch.Tensor):
    with torch.no_grad():
      batch_utility = hidden_acts.abs().mean(dim=0).detach().cpu()
      if not torch.isnan(batch_utility).any():
        self.running_utility = (
            self.decay * self.running_utility
            + (1.0 - self.decay) * batch_utility
        )

  def trigger_neurogenesis(self, reinit_fraction: float = 0.02):
    with torch.no_grad():
      num_units = int(self.hidden_dim * reinit_fraction)
      if num_units == 0:
        return 0
      _, dead_indices = torch.topk(
          self.running_utility, k=num_units, largest=False
      )
      device = self.layer1.weight.device
      dead_idx = dead_indices.to(device)
      nn.init.kaiming_normal_(self.layer1.weight[dead_idx, :])
      self.layer2.weight[:, dead_idx] = 0.0
      self.running_utility[dead_indices] = torch.clamp(
          self.running_utility.mean(), min=1e-4
      )
      return num_units


# ==============================================================================
# 3. NUMERICALLY HARDENED MPPI CONTROLLER (LOG-SUM-EXP SOFTMAX)
# ==============================================================================
class MPPIController:

  def __init__(
      self,
      model: EmbodiedWorldModel,
      horizon: int = 8,
      num_samples: int = 32,
      temperature: float = 0.5,
  ):
    self.model = model
    self.horizon = horizon
    self.num_samples = num_samples
    self.temperature = max(temperature, 1e-4)
    self.action_dim = model.action_dim
    self.latent_dim = model.latent_dim

  def compute_action(
      self,
      z_current: torch.Tensor,
      z_target: torch.Tensor,
      device: torch.device,
  ) -> torch.Tensor:
    action_noise = (
        torch.randn(
            self.num_samples, self.horizon, self.action_dim, device=device
        )
        * 0.5
    )
    z_sim = z_current.repeat(self.num_samples, 1)
    trajectory_costs = torch.zeros(self.num_samples, device=device)

    for t in range(self.horizon):
      a_t = torch.clamp(action_noise[:, t, :], -1.0, 1.0)
      with torch.no_grad():
        z_sim = self.model(z_sim, a_t)
        z_sim = torch.clamp(z_sim, -10.0, 10.0)

      step_cost = torch.norm(z_sim - z_target, dim=-1)
      energy_cost = 0.05 * torch.norm(a_t, dim=-1)
      trajectory_costs += step_cost + energy_cost

    scaled_logits = -trajectory_costs / self.temperature
    weights = F.softmax(scaled_logits, dim=0)

    if torch.isnan(weights).any():
      weights = torch.full_like(weights, 1.0 / self.num_samples)

    optimal_action = torch.sum(
        weights.unsqueeze(-1) * action_noise[:, 0, :], dim=0
    )

    if torch.isnan(optimal_action).any():
      return torch.zeros(self.action_dim, device=device)
    return torch.clamp(optimal_action, -1.0, 1.0)


# ==============================================================================
# 4. ENERGY-BOUNDED MULTI-REGIME ROBOTIC PLANT (100,000 STEPS)
# ==============================================================================
class SimulatedRoboticPlant:

  def __init__(self, latent_dim: int = 128, action_dim: int = 6):
    self.latent_dim = latent_dim
    self.action_dim = action_dim
    self.state = torch.randn(1, latent_dim) * 0.1
    self.euler_lambda = 0.9785142874

  def step(self, action: torch.Tensor, cycle: int) -> tuple[torch.Tensor, str]:
    a_pad = F.pad(action.unsqueeze(0), (0, self.latent_dim - self.action_dim))

    if cycle <= 25000:
      regime = "Regime A (Newtonian)"
      next_state = (
          torch.sin(self.state) + 0.1 * a_pad - 0.05 * torch.tanh(self.state)
      )
    elif cycle <= 50000:
      regime = "Regime B (Turbulence)"
      next_state = torch.cos(self.state * 1.5) - 0.3 * torch.tanh(a_pad)
    elif cycle <= 75000:
      regime = "Regime C (Viscous Drag)"
      next_state = (
          torch.sigmoid(self.state) * 1.2
          + 0.05 * (a_pad**2)
          - 0.15 * self.state
      )
    else:
      regime = "Regime D (Gyroscopic)"
      next_state = torch.roll(self.state, shifts=1, dims=-1) * 0.8 + 0.2 * torch.sin(
          a_pad
      )

    self.state = (
        self.euler_lambda * next_state + torch.randn_like(next_state) * 0.001
    )
    self.state = torch.clamp(self.state, -10.0, 10.0)
    return self.state.clone(), regime


# ==============================================================================
# 5. FULL 100,000-CYCLE CONTROL LOOP EXECUTION UNDER SEED 123
# ==============================================================================
def run_robotic_control_loop_100k_seed123(
    repo_id="frankmorales2020/topo-cbp-world-model-3d",
):
  device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
  print(
      "[*] Initializing Numerically Hardened Robotic Loop (100,000 Cycles) under"
      f" Execution Seed 123 on: {device}"
  )

  cfg_file = hf_hub_download(repo_id=repo_id, filename="config.json")
  weights_file = hf_hub_download(repo_id=repo_id, filename="model.safetensors")

  with open(cfg_file, "r") as f:
    cfg = json.load(f)

  model = EmbodiedWorldModel(
      cfg["latent_dim"], cfg["action_dim"], cfg["hidden_dim"]
  ).to(device)
  model.load_state_dict(load_file(weights_file, device=str(device)))

  optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-4)
  governor = RuntimeGovernor(
      model.in_proj.weight,
      cfg["prime_anchors"],
      certified_hash=cfg["invariant_sha256_hash"],
  )
  plasticity = OnlinePlasticityEngine(
      model.plastic_linear1, model.plastic_linear2, decay=0.99
  )
  controller = MPPIController(model, horizon=8, num_samples=32)
  plant = SimulatedRoboticPlant(cfg["latent_dim"], cfg["action_dim"])

  print(f"[*] Target Certified Hash : {cfg['invariant_sha256_hash']}")
  if not governor.verify_hardware_interlock():
    raise SystemError(
        "FAIL-CLOSED: Initial hardware interlock verification failed!"
    )
  print(
      "[*] Hardware interlock armed. Launching 100,000-step actuation loop...\n"
  )

  z_target = torch.zeros(1, cfg["latent_dim"], device=device)
  log_checkpoints = {
      1,
      1000,
      10000,
      25000,
      25001,
      35000,
      50000,
      50001,
      60000,
      75000,
      75001,
      85000,
      100000,
  }

  print("=" * 110)
  print(
      f"{'CYCLE':>6} | {'LATENCY':>10} | {'PRED ERROR':>11} | {'ACTUATOR NORM':>13}"
      f" | {'DRIFT':>12} | {'INTERLOCK':>9} | {'REGIME'}"
  )
  print("=" * 110)

  t_total_start = time.time()

  for cycle in range(1, 100001):
    t_cycle_start = time.perf_counter()

    if not governor.verify_hardware_interlock():
      print(f"\n[!] FAIL-CLOSED TRIPPED: Hash mutation detected at cycle {cycle}!")
      print("[!] EMERGENCY INTERLOCK: Motor torque killed to zero.")
      break

    z_observed = plant.state.to(device)
    optimal_torque = controller.compute_action(z_observed, z_target, device)

    z_next_observed, regime = plant.step(optimal_torque.cpu(), cycle)
    z_next_observed = z_next_observed.to(device)

    # Online Continual Learning update
    model.train()
    optimizer.zero_grad()
    z_predicted = model(z_observed, optimal_torque.unsqueeze(0))
    loss = F.mse_loss(z_predicted, z_next_observed)
    loss.backward()

    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

    # Dual-Phase Projection
    governor.project_and_lock()
    optimizer.step()
    governor.project_and_lock()

    with torch.no_grad():
      h_inter = model.in_proj(
          torch.cat([z_observed, optimal_torque.unsqueeze(0)], dim=-1)
      )
      plasticity.update(h_inter)
      if cycle % 100 == 0:
        plasticity.trigger_neurogenesis(reinit_fraction=0.02)

    model.eval()
    t_cycle_elapsed = (time.perf_counter() - t_cycle_start) * 1000

    if cycle in log_checkpoints:
      drift = governor.compute_drift()
      print(
          f"{cycle:6d} | {t_cycle_elapsed:8.4f} ms | {loss.item():11.5f} |"
          f" {torch.norm(optimal_torque).item():13.4f} | {drift:12.10f} |"
          f" {'LOCKED':>9} | {regime}"
      )

  total_duration = time.time() - t_total_start
  final_drift = governor.compute_drift()
  is_safe = governor.verify_hardware_interlock()

  print("=" * 110)
  print(
      f"[*] Completed 100,000 cycles in {total_duration:.2f}s"
      f" ({100000/total_duration:.1f} cycles/s)"
  )
  print(
      "    - Final Hardware Interlock :"
      f" {'LOCKED (PASS)' if is_safe else 'TRIPPED (FAIL)'}"
  )
  print(f"    - Certified Baseline Hash   : {cfg['invariant_sha256_hash']}")
  print(f"    - Final Prime Anchor Drift  : {final_drift:.10f}")
  print("    - Total Replay Buffer Size  : 0 bytes (Strict O(1) Memory)")


if __name__ == "__main__":
  run_robotic_control_loop_100k_seed123()

[*] Initializing Numerically Hardened Robotic Loop (100,000 Cycles) under Execution Seed 123 on: cuda
[*] Target Certified Hash : 2ce28ffcf131d5c9
[*] Hardware interlock armed. Launching 100,000-step actuation loop...

==============================================================================================================
 CYCLE |    LATENCY |  PRED ERROR | ACTUATOR NORM |        DRIFT | INTERLOCK | REGIME
==============================================================================================================
     1 | 377.0564 ms |     0.00898 |        0.1898 | 0.0000000000 |    LOCKED | Regime A (Newtonian)
  1000 |   6.5303 ms |     0.00003 |        0.1728 | 0.0000000000 |    LOCKED | Regime A (Newtonian)
 10000 |   6.7313 ms |     0.00003 |        0.1870 | 0.0000000000 |    LOCKED | Regime A (Newtonian)
 25000 |   6.5483 ms |     0.00002 |        0.2267 | 0.0000000000 |    LOCKED | Regime A (Newtonian)
 25001 |   8.6515 ms |     0.95687 |        0.1395 | 0.0000000000 |    LOCKED | Regime B (Turbulence)
 35000 |   6.4241 ms |     0.00022 |        0.1841 | 0.0000000000 |    LOCKED | Regime B (Turbulence)
 50000 |   6.7795 ms |     0.00004 |        0.3566 | 0.0000000000 |    LOCKED | Regime B (Turbulence)
 50001 |  11.0340 ms |     0.13283 |        0.3011 | 0.0000000000 |    LOCKED | Regime C (Viscous Drag)
 60000 |   6.5838 ms |     0.00000 |        0.1087 | 0.0000000000 |    LOCKED | Regime C (Viscous Drag)
 75000 |   6.6553 ms |     0.00000 |        0.1705 | 0.0000000000 |    LOCKED | Regime C (Viscous Drag)
 75001 |   6.9122 ms |     0.02193 |        0.1782 | 0.0000000000 |    LOCKED | Regime D (Gyroscopic)
 85000 |   7.0008 ms |     0.00002 |        0.1329 | 0.0000000000 |    LOCKED | Regime D (Gyroscopic)
100000 |   6.6733 ms |     0.00006 |        0.2360 | 0.0000000000 |    LOCKED | Regime D (Gyroscopic)
==============================================================================================================
[*] Completed 100,000 cycles in 634.07s (157.7 cycles/s)
    - Final Hardware Interlock : LOCKED (PASS)
    - Certified Baseline Hash   : 2ce28ffcf131d5c9
    - Final Prime Anchor Drift  : 0.0000000000
    - Total Replay Buffer Size  : 0 bytes (Strict O(1) Memory)

Interactive ReAct & 100k Benchmark Notebook

Run the complete physical plant stress test and the multi-terrain ReAct agent demo:

• GitHub Notebook: PHYSICAL_AI.ipynb
• Plasticity Engine Reference: WM_TOPO_PLASTICITY.ipynb