Digital Twins Design

User Plane (ros.dev.gpu)
  Ray worker writes live sensor data → lakehouse (data plane) during job
  Ray worker writes live pose snapshots → lakehouse (data plane) during job
  Cosmos-Reason2: reads twin state snapshots as visual context for feasibility evaluation

User Plane (torch.dev.gpu)
  Cosmos-Predict2: reads observed twin state → generates predicted future state video
  Cosmos-Transfer2.5: translates predicted synthetic video → photorealistic; stored as predicted snapshots

Control Plane
  TwinAgent subprocess (claude -p) reconciles and validates at job end
  TwinAgent exposes create / sync / query / retire / predict operations
  FastAPI /api/v1/twins router accepts HTTP calls from UI and other agents

Data Plane (Lakehouse)
  Seven Parquet tables under twins/ prefix
  state_snapshots.source distinguishes observed vs predicted vs sim2real snapshots
  DuckDB queries served by TwinAgent and LakehouseAgent

# control-plane/backend/agents/twin_agent.py
def create_twin(asset_id: str, asset_type: str, urdf_path: str | None = None) -> dict: ...
def sync_twin(twin_id: str, job_id: str) -> dict: ...
def query_twin(twin_id: str, query: str) -> dict: ...
def get_twin_state(twin_id: str, at: str | None = None) -> dict: ...
def predict_twin(twin_id: str, action: dict, horizon_s: float) -> dict: ...
def annotate_twin(twin_id: str, annotation: dict) -> dict: ...
def retire_twin(twin_id: str) -> dict: ...

Gazebo + Nav2
  → DDS topics
  → Zenoh bridge
  → Ray worker (Python)
      → pyarrow: append rows to sensor_readings Parquet partition
      → pyarrow: append rows to state_snapshots Parquet partition (reconciled=False)

GET    /api/v1/twins                      List all registered twins
POST   /api/v1/twins                      Register a new twin (calls TwinAgent.create_twin)
GET    /api/v1/twins/{id}                 Twin details (asset record + latest state)
GET    /api/v1/twins/{id}/state           Latest state snapshot (or ?at=ISO8601 for point-in-time)
POST   /api/v1/twins/{id}/sync            Trigger post-job reconciliation
POST   /api/v1/twins/{id}/predict         Generate predicted future snapshots via Cosmos-Predict2 + Transfer2.5
GET    /api/v1/twins/{id}/events          Event log (paginated)
GET    /api/v1/twins/{id}/annotations     Annotations
POST   /api/v1/twins/{id}/annotations     Add annotation

1. TwinAgent.create_twin("turtlebot-01", "robot", urdf_path="user-plane/turtlebot/urdf/turtlebot3.urdf")
2. Control plane submits turtlebot-maze RayJob to ros.dev.gpu
3. Ray worker connects to Gazebo via Zenoh; writes pose + IMU to the lakehouse in-job
4. Job completes → POST /api/v1/twins/turtlebot-01/sync?job_id=<id>
5. TwinAgent reconciles; twin.synced event logged
6. bd query_twin("turtlebot-01", "SELECT timestamp, position_x, position_y FROM state_snapshots ORDER BY timestamp")
   → returns full trajectory from the job

# v1.5 extension — Cosmos-driven predicted twin state:
7. POST /api/v1/twins/turtlebot-01/predict  \{action: \{cmd_vel: \{linear: 0.3, angular: 0.1\}\}, horizon_s: 5.0\}
   → TwinAgent calls Cosmos-Predict2 (torch.dev.gpu) with latest observed snapshot as seed frame
   → Cosmos-Transfer2.5 translates synthetic prediction → photorealistic
   → Predicted snapshots written to state_snapshots (source=predicted, cosmos_model=transfer2.5)
8. Cosmos-Reason2 (ros.dev.gpu) reads predicted snapshots as visual context
   → evaluates feasibility → go / no-go before Nav2 goal dispatch

v1   — Persistent world model: lakehouse-backed twins for TurtleBot; in-job writes + post-job reconciliation
v1.5 — Real-time sync: Redis hot-cache for live pose (sub-second); Zenoh → twin writer as a persistent Ray actor
       Cosmos-Predict2 + Cosmos-Transfer2.5: predicted twin state snapshots (source=predicted/sim2real)
       predict_twin() operation + POST /api/v1/twins/{id}/predict endpoint
       Cosmos-Reason2: reads predicted snapshots as visual context for pre-execution feasibility evaluation
       Predict → Transfer → Reason → Execute loop integrated with twin state
v2   — Agent memory substrate: TwinAgent is the canonical world-model interface for all control-plane agents
       Cosmos models post-trained on turtlebot-maze ROS bags; domain-specific predicted snapshots
       Visualisation: Rerun viewer embedded in Next.js dashboard (observed + predicted state overlay)
       counter-uas (aegean-ai/counter-uas): UE5 + VisDrone perception/tracking twin; third reference asset
       Management plane: per-twin access control, retention policies

control-plane/backend/
  agents/twin_agent.py          TwinAgent wrapper (claude -p subprocess)
  api/twins.py                  FastAPI router
  models/twin.py                Pydantic models

data-plane/
  schema/twins/                 Schema definitions and migration scripts

.claude/agents/
  twin-agent.md                 Agent definition (YAML frontmatter + system prompt)