Claude's Corner: One Robot - The Simulation Layer Robotics Has Been Missing

# Build a Robot World Model Simulator: One Robot Clone Guide

A step-by-step guide for building a task-specific world model simulation platform for robot policy training, using Claude Code.

---

## Step 1: Database Schema

Design the schema to track customers, demonstration sessions, rollout requests, and synthetic training outputs.

```sql
CREATE TABLE customers (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name TEXT NOT NULL,
  api_key TEXT UNIQUE NOT NULL DEFAULT gen_random_uuid(),
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE demonstration_sessions (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID REFERENCES customers(id),
  task_name TEXT NOT NULL,
  robot_type TEXT NOT NULL,           -- e.g. "franka_panda", "ur5"
  num_demos INTEGER DEFAULT 0,
  status TEXT DEFAULT 'collecting',   -- collecting | training | ready | failed
  world_model_checkpoint_path TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE demonstrations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID REFERENCES demonstration_sessions(id),
  storage_path TEXT NOT NULL,         -- HDF5 file in object storage
  duration_seconds FLOAT,
  joint_dof INTEGER DEFAULT 7,
  camera_fps INTEGER DEFAULT 30,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE rollout_requests (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID REFERENCES demonstration_sessions(id),
  starting_frame_path TEXT NOT NULL,
  action_sequence JSONB NOT NULL,     -- array of joint command arrays
  status TEXT DEFAULT 'queued',       -- queued | generating | complete | failed
  output_video_path TEXT,
  failure_detected BOOLEAN,
  failure_mode TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE synthetic_trajectories (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID REFERENCES demonstration_sessions(id),
  failure_mode TEXT NOT NULL,
  trajectory_count INTEGER DEFAULT 0,
  storage_path TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Index for fast queue polling
CREATE INDEX idx_rollout_status ON rollout_requests(status, created_at);
CREATE INDEX idx_session_status ON demonstration_sessions(status, customer_id);
```

---

## Step 2: Data Ingestion API

Build a FastAPI endpoint that accepts synchronized demonstration data (video + joint states) and stores it as HDF5 files in object storage. The key challenge is frame-level temporal alignment between camera and joint state streams.

```python
# api/ingest.py
from fastapi import FastAPI, UploadFile, BackgroundTasks, HTTPException
import boto3, h5py, json, uuid, cv2
import numpy as np
from pathlib import Path

app = FastAPI()
s3 = boto3.client("s3", region_name="us-east-1")
BUCKET = "robot-demos-bucket"

def align_streams(video_bytes: bytes, joint_states: list, target_fps: int = 30):
    """Align video frames to joint state timestamps via nearest-neighbor interpolation."""
    # Decode video
    import tempfile, subprocess
    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
        f.write(video_bytes)
        tmp_path = f.name

    cap = cv2.VideoCapture(tmp_path)
    video_fps = cap.get(cv2.CAP_PROP_FPS)
    frames = []
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        frames.append(cv2.resize(frame, (256, 256)))
    cap.release()

    # Build joint state timeline (assumes timestamps in seconds)
    js_times = np.array([s["t"] for s in joint_states])
    js_actions = np.array([[s["q"], s["dq"], s["tau"], [s["gripper"]]] for s in joint_states])
    js_actions = js_actions.reshape(len(joint_states), -1)  # flatten per step

    # Match each frame to nearest joint state
    n_frames = len(frames)
    video_times = np.linspace(0, n_frames / video_fps, n_frames)
    aligned_indices = np.searchsorted(js_times, video_times)
    aligned_indices = np.clip(aligned_indices, 0, len(joint_states) - 1)

    return {
        "frames": np.stack(frames).astype(np.uint8),         # (T, H, W, 3)
        "actions": js_actions[aligned_indices].astype(np.float32),  # (T, action_dim)
        "timestamps": video_times.astype(np.float32),
    }

@app.post("/demonstrations/upload")
async def upload_demonstration(
    session_id: str,
    camera_video: UploadFile,
    joint_states_file: UploadFile,
    background_tasks: BackgroundTasks,
):
    video_bytes = await camera_video.read()
    joint_states = json.loads(await joint_states_file.read())

    aligned = align_streams(video_bytes, joint_states)

    demo_id = str(uuid.uuid4())
    storage_path = f"demos/{session_id}/{demo_id}.h5"

    # Store as HDF5 for fast random-access during training
    local_path = f"/tmp/{demo_id}.h5"
    with h5py.File(local_path, "w") as f:
        f.create_dataset("frames", data=aligned["frames"], compression="lzf")
        f.create_dataset("actions", data=aligned["actions"])
        f.create_dataset("timestamps", data=aligned["timestamps"])

    s3.upload_file(local_path, BUCKET, storage_path)
    background_tasks.add_task(increment_demo_count, session_id)
    return {"demo_id": demo_id, "storage_path": storage_path, "frames": len(aligned["frames"])}
```

---

## Step 3: World Model Architecture

Build the autoregressive transformer that learns to predict next video frames conditioned on robot actions. Uses a VQ-VAE to tokenize frames into a discrete vocabulary, then trains a causal transformer over the token sequence.

```python
# models/world_model.py
import torch
import torch.nn as nn
import torch.nn.functional as F

class VQVAE(nn.Module):
    """Encode frames to discrete tokens; decode tokens back to frames."""
    def __init__(self, frame_size: int = 256, vocab_size: int = 8192, d_model: int = 512):
        super().__init__()
        self.vocab_size = vocab_size
        # Encoder: image -> latent grid
        self.encoder = nn.Sequential(
            nn.Conv2d(3, 64, 4, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(64, 128, 4, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(128, 256, 4, stride=2, padding=1), nn.ReLU(),
            nn.Conv2d(256, d_model, 4, stride=2, padding=1),
        )  # 256 -> 16x16 grid of d_model vectors
        self.codebook = nn.Embedding(vocab_size, d_model)
        # Decoder: latent grid -> image
        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(d_model, 256, 4, stride=2, padding=1), nn.ReLU(),
            nn.ConvTranspose2d(256, 128, 4, stride=2, padding=1), nn.ReLU(),
            nn.ConvTranspose2d(128, 64, 4, stride=2, padding=1), nn.ReLU(),
            nn.ConvTranspose2d(64, 3, 4, stride=2, padding=1), nn.Sigmoid(),
        )

    def encode(self, x: torch.Tensor) -> torch.Tensor:
        z = self.encoder(x.permute(0, 3, 1, 2) / 255.0)
        B, C, H, W = z.shape
        z_flat = z.permute(0, 2, 3, 1).reshape(-1, C)
        dists = torch.cdist(z_flat, self.codebook.weight)
        indices = dists.argmin(-1).reshape(B, H * W)
        return indices  # (B, 256) token ids for 16x16 grid

    def decode(self, indices: torch.Tensor) -> torch.Tensor:
        B, N = indices.shape
        H = W = int(N ** 0.5)
        z = self.codebook(indices).reshape(B, H, W, -1).permute(0, 3, 1, 2)
        return (self.decoder(z).permute(0, 2, 3, 1) * 255).byte()


class RobotWorldModel(nn.Module):
    def __init__(self, action_dim: int = 22, d_model: int = 768, n_heads: int = 12, n_layers: int = 12, vocab_size: int = 8192):
        super().__init__()
        self.vqvae = VQVAE(vocab_size=vocab_size, d_model=d_model // 2)
        self.frame_embed = nn.Embedding(vocab_size, d_model)
        self.pos_embed = nn.Embedding(2048, d_model)
        self.action_embed = nn.Sequential(
            nn.Linear(action_dim, d_model // 2), nn.GELU(),
            nn.Linear(d_model // 2, d_model),
        )
        decoder_layer = nn.TransformerDecoderLayer(d_model, n_heads, dim_feedforward=4 * d_model, batch_first=True)
        self.transformer = nn.TransformerDecoder(decoder_layer, num_layers=n_layers)
        self.frame_head = nn.Linear(d_model, vocab_size)
        self._causal_mask = {}

    def _get_mask(self, T: int, device):
        if T not in self._causal_mask:
            self._causal_mask[T] = torch.triu(torch.ones(T, T, device=device), diagonal=1).bool()
        return self._causal_mask[T]

    def forward(self, frames: torch.Tensor, actions: torch.Tensor):
        """frames: (B,T,H,W,3) uint8; actions: (B,T,action_dim) float32"""
        B, T = frames.shape[:2]
        frame_tokens = self.vqvae.encode(frames.flatten(0, 1)).reshape(B, T, -1)
        # Use mean pool of frame tokens as single frame embedding
        frame_emb = self.frame_embed(frame_tokens).mean(2)  # (B, T, d_model)
        action_emb = self.action_embed(actions)              # (B, T, d_model)
        # Interleave [f0, a0, f1, a1, ...] - action_t predicts frame_t+1
        seq = torch.stack([frame_emb, action_emb], dim=2).flatten(1, 2)  # (B, 2T, d_model)
        pos = torch.arange(seq.shape[1], device=frames.device)
        seq = seq + self.pos_embed(pos)
        mask = self._get_mask(seq.shape[1], frames.device)
        out = self.transformer(seq, seq, tgt_mask=mask)
        # Action positions (odd indices) predict the NEXT frame tokens
        action_out = out[:, 1::2]  # (B, T, d_model)
        return self.frame_head(action_out)  # (B, T, vocab_size)

    @torch.no_grad()
    def rollout(self, start_frame: torch.Tensor, action_sequence: torch.Tensor, steps: int = 180):
        """Generate rollout video. start_frame: (1,H,W,3). action_sequence: (1,steps,action_dim)."""
        frames = [start_frame]
        for t in range(min(steps, action_sequence.shape[1])):
            stacked = torch.stack(frames, dim=1)  # (1, t+1, H, W, 3)
            acts = action_sequence[:, :t + 1]
            logits = self.forward(stacked, acts)
            next_tokens = logits[:, -1].argmax(-1, keepdim=True)  # (1, vocab_size) -> (1,)
            next_frame = self.vqvae.decode(next_tokens.unsqueeze(1).expand(-1, 256)).squeeze(1)
            frames.append(next_frame.unsqueeze(1))
        return torch.cat(frames[1:], dim=1)  # (1, steps, H, W, 3)
```

---

## Step 4: Training Pipeline

Set up distributed training with PyTorch Lightning. Use window-based sampling from HDF5 demonstration files for efficient data loading.

```python
# training/train.py
import lightning as L
import h5py, torch, boto3
from torch.utils.data import Dataset, DataLoader

class DemoDataset(Dataset):
    def __init__(self, demo_paths: list[str], window: int = 32, stride: int = 8):
        self.windows = []
        for p in demo_paths:
            with h5py.File(p, "r") as f:
                T = f["frames"].shape[0]
            for s in range(0, T - window, stride):
                self.windows.append((p, s, s + window))

    def __len__(self):
        return len(self.windows)

    def __getitem__(self, idx):
        path, start, end = self.windows[idx]
        with h5py.File(path, "r") as f:
            frames = torch.from_numpy(f["frames"][start:end])    # (W, H, W, 3)
            actions = torch.from_numpy(f["actions"][start:end])  # (W, action_dim)
        return frames, actions


class WorldModelModule(L.LightningModule):
    def __init__(self, model: "RobotWorldModel", lr: float = 2e-4):
        super().__init__()
        self.model = model
        self.lr = lr

    def training_step(self, batch, _):
        frames, actions = batch
        # Predict frame[t+1] from frame[t] + action[t], for all t
        logits = self.model(frames[:, :-1], actions[:, :-1])  # (B, T-1, vocab_size)
        # Target: token ids of frames[1:]
        with torch.no_grad():
            targets = self.model.vqvae.encode(frames[:, 1:].flatten(0, 1))  # (B*(T-1), 256)
            targets = targets.reshape(frames.shape[0], frames.shape[1] - 1, -1)
        # Predict just the mean token (simplification; full spatial loss is better)
        loss = torch.nn.functional.cross_entropy(
            logits.flatten(0, 1), targets.float().mean(-1).long().flatten()
        )
        self.log("train/loss", loss, prog_bar=True)
        return loss

    def configure_optimizers(self):
        return torch.optim.AdamW(self.parameters(), lr=self.lr, weight_decay=0.01)


def train_world_model(session_id: str, demo_paths: list[str]):
    from models.world_model import RobotWorldModel
    model = RobotWorldModel()
    module = WorldModelModule(model)
    dataset = DemoDataset(demo_paths)
    loader = DataLoader(dataset, batch_size=4, shuffle=True, num_workers=4, pin_memory=True)
    trainer = L.Trainer(
        max_epochs=100, devices=4, strategy="ddp",
        precision="bf16-mixed", gradient_clip_val=1.0,
        default_root_dir=f"/checkpoints/{session_id}",
    )
    trainer.fit(module, loader)
    # Save checkpoint to object storage
    boto3.client("s3").upload_file(
        f"/checkpoints/{session_id}/best.ckpt",
        "robot-demos-bucket",
        f"checkpoints/{session_id}/world_model.ckpt"
    )
```

---

## Step 5: Rollout Generation Service

Build an async service that queues rollout requests, processes them on GPU workers, and stores output videos.

```python
# services/rollout_worker.py
import asyncio, json, uuid, redis, boto3, cv2
import numpy as np
import torch

redis_client = redis.Redis(host="localhost", decode_responses=True)

async def process_rollout(data: dict):
    session_id = data["session_id"]
    rollout_id = data["rollout_id"]

    # Load world model checkpoint for this session
    model = load_model_for_session(session_id)  # pulls from S3, caches locally
    model.eval()

    # Load starting frame
    start_frame = load_frame(data["starting_frame_path"])  # (1, H, W, 3)
    actions = torch.tensor(data["actions"]).unsqueeze(0)   # (1, T, action_dim)

    with torch.cuda.amp.autocast():
        rollout_video = model.rollout(start_frame.cuda(), actions.cuda())

    # Detect failure: simple heuristic - check if end-effector leaves frame
    failure_detected = detect_failure(rollout_video)

    # Encode and upload video
    video_path = f"rollouts/{session_id}/{rollout_id}.mp4"
    frames = rollout_video.squeeze(0).cpu().numpy()  # (T, H, W, 3)
    encode_and_upload(frames, video_path)

    # Update DB
    await db.execute(
        """UPDATE rollout_requests
           SET status='complete', output_video_path=$1, failure_detected=$2
           WHERE id=$3""",
        video_path, failure_detected, rollout_id
    )

def encode_and_upload(frames: np.ndarray, storage_path: str):
    local_path = f"/tmp/{uuid.uuid4()}.mp4"
    h, w = frames.shape[1:3]
    writer = cv2.VideoWriter(local_path, cv2.VideoWriter_fourcc(*"mp4v"), 30, (w, h))
    for frame in frames:
        writer.write(cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
    writer.release()
    boto3.client("s3").upload_file(local_path, "robot-demos-bucket", storage_path)

async def worker_loop():
    while True:
        item = redis_client.blpop("rollout_queue", timeout=5)
        if item:
            _, raw = item
            await process_rollout(json.loads(raw))
```

---

## Step 6: Failure Detection and Synthetic Data Generation

Implement the evaluation loop that identifies policy failure modes and generates targeted synthetic training trajectories.

```python
# services/policy_evaluator.py
import torch
import numpy as np
from typing import Optional

class PolicyEvaluator:
    def __init__(self, world_model, failure_classifier=None):
        self.world_model = world_model
        self.failure_classifier = failure_classifier or HeuristicFailureClassifier()

    def evaluate_policy(self, policy, session_id: str, n_rollouts: int = 500):
        starting_frames = sample_diverse_starts(session_id, n=n_rollouts)
        failures = []

        for frame in starting_frames:
            with torch.no_grad():
                actions = policy.infer(frame, max_steps=180)
                rollout = self.world_model.rollout(frame, actions)
            failure_mode = self.failure_classifier.classify(rollout)
            if failure_mode:
                failures.append({
                    "starting_frame": frame,
                    "rollout": rollout,
                    "failure_mode": failure_mode,
                })

        return failures

    def generate_synthetic_data(self, failures: list, n_per_failure: int = 100):
        synthetic = []
        for failure in failures:
            perturbed_starts = self._perturb_frame(failure["starting_frame"], n=n_per_failure)
            recovery_actions = self._sample_recovery_actions(failure["failure_mode"])
            for start in perturbed_starts:
                traj = self.world_model.rollout(start.unsqueeze(0), recovery_actions.unsqueeze(0))
                synthetic.append({
                    "frames": traj.squeeze(0).numpy(),
                    "actions": recovery_actions.numpy(),
                    "failure_mode_covered": failure["failure_mode"],
                })
        return synthetic

    def _perturb_frame(self, frame: torch.Tensor, n: int):
        noise_scale = 0.02
        return [frame + torch.randn_like(frame.float()) * noise_scale for _ in range(n)]

    def _sample_recovery_actions(self, failure_mode: str):
        # Task-specific recovery heuristics per failure mode
        recovery_map = {
            "gripper_missed": torch.zeros(1, 90, 22),  # re-approach trajectory
            "object_dropped": torch.zeros(1, 120, 22), # recovery grasp
            "out_of_workspace": torch.zeros(1, 60, 22),# retract and retry
        }
        return recovery_map.get(failure_mode, torch.zeros(1, 90, 22)).squeeze(0)


class HeuristicFailureClassifier:
    def classify(self, rollout: torch.Tensor) -> Optional[str]:
        frames = rollout.squeeze(0).float().numpy()
        last_frame = frames[-1]
        # Detect gripper miss: gripper visible in final frame (object should be grasped)
        # Simple proxy: check if end of rollout shows object still in starting position
        if self._object_unmoved(frames):
            return "gripper_missed"
        if self._object_dropped(frames):
            return "object_dropped"
        return None

    def _object_unmoved(self, frames):
        # Compare first and last frame centroid of salient region
        from skimage.color import rgb2gray
        from skimage.filters import threshold_otsu
        start = rgb2gray(frames[0] / 255.0)
        end = rgb2gray(frames[-1] / 255.0)
        return np.abs(start - end).mean() < 0.02  # near-identical = nothing moved

    def _object_dropped(self, frames):
        # Bright region (object) appears at bottom of frame in last frames
        last = frames[-1, :, :, :].mean(-1)
        bottom_half = last[last.shape[0] // 2:].mean()
        top_half = last[:last.shape[0] // 2].mean()
        return bottom_half > top_half * 1.5
```

---

## Step 7: Kubernetes Deployment with GPU Autoscaling

Deploy the rollout worker on GPU node pools with horizontal autoscaling based on queue depth. Critical: customer data isolation via namespace-per-customer and per-session model loading.

```yaml
# k8s/rollout-worker.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: rollout-worker
  namespace: one-robot-prod
spec:
  replicas: 2
  selector:
    matchLabels:
      app: rollout-worker
  template:
    metadata:
      labels:
        app: rollout-worker
    spec:
      nodeSelector:
        cloud.google.com/gke-accelerator: nvidia-a100-80gb
      tolerations:
        - key: nvidia.com/gpu
          operator: Exists
          effect: NoSchedule
      containers:
        - name: worker
          image: gcr.io/one-robot/rollout-worker:latest
          resources:
            limits:
              nvidia.com/gpu: "1"
              memory: "60Gi"
              cpu: "8"
            requests:
              memory: "40Gi"
              cpu: "4"
          env:
            - name: REDIS_URL
              valueFrom:
                secretKeyRef:
                  name: redis-creds
                  key: url
            - name: S3_BUCKET
              value: "robot-demos-bucket"
            - name: MODEL_CACHE_DIR
              value: "/model-cache"
          volumeMounts:
            - name: model-cache
              mountPath: /model-cache
            - name: tmp-storage
              mountPath: /tmp
      volumes:
        - name: model-cache
          emptyDir:
            medium: Memory
            sizeLimit: "20Gi"
        - name: tmp-storage
          emptyDir:
            sizeLimit: "10Gi"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: rollout-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: rollout-worker
  minReplicas: 2
  maxReplicas: 16
  metrics:
    - type: External
      external:
        metric:
          name: redis_queue_depth
          selector:
            matchLabels:
              queue: rollout_queue
        target:
          type: AverageValue
          averageValue: "3"
```

**Key deployment decisions:**

- **Customer isolation**: Each customer's model checkpoint is isolated in S3 under `checkpoints/{session_id}/`. Workers pull the correct checkpoint per request - never co-locate models from different customers in the same memory space.
- **Model caching**: Workers keep recently-used checkpoints in the in-memory volume (`/model-cache`). Cache eviction uses LRU - evict the least-recently-used session model when memory pressure rises.
- **Rollout deduplication**: Cache rollout results for identical `(session_id, starting_frame_hash, action_sequence_hash)` tuples. Batch evaluation runs often repeat the same starting conditions.
- **Sim-to-real validation**: Track the correlation between world model rollout predictions and actual robot outcomes on a held-out validation set. If this correlation drops below 0.85 after retraining on synthetic data, flag the session for human review before deploying the updated policy. This metric is the single most important signal for whether your world model is actually useful.
- **Monitoring**: Emit metrics for rollout generation latency (p50/p95/p99), world model prediction loss on held-out demos, synthetic data policy improvement rate, and sim-to-real correlation. Alert if rollout latency exceeds 30 seconds for a 60-second rollout - that indicates GPU memory pressure or checkpoint cache thrashing.