feat: demo-reset + collision-storm fixes

Makefile

@@ -422,6 +422,30 @@ sim-drive-stop: ## Stop the rover
 	@$(CONTAINER_ENGINE) exec $(SIM_CONTAINER) bash -c \
 	  'ign topic -t $(GZ_DRIVE_TOPIC) -m ignition.msgs.Twist -p "linear: {x: 0}, angular: {z: 0}"'
 
+# Teleport the rover back to the origin. Useful when an OTA controller
+# runs the rover off the world or into the boulder. Override
+# TELEPORT_X / TELEPORT_Y / TELEPORT_Z to land somewhere else.
+SIM_GAZEBO_CONTAINER ?= temporal-hack-lab-gazebo-1
+SIM_WORLD_NAME ?= moon
+ROBOT_MODEL ?= perseverance
+TELEPORT_X ?= 0
+TELEPORT_Y ?= 0
+TELEPORT_Z ?= 0.30
+
+.PHONY: sim-teleport
+sim-teleport: ## Teleport the rover back to the origin (override TELEPORT_X/Y/Z)
+	@echo "[sim-teleport] $(ROBOT_MODEL) -> ($(TELEPORT_X), $(TELEPORT_Y), $(TELEPORT_Z))"
+	@# Stop the rover first so its old velocity doesn't get carried over.
+	@$(CONTAINER_ENGINE) exec $(SIM_GAZEBO_CONTAINER) bash -c \
+	  'ign topic -t $(GZ_DRIVE_TOPIC) -m ignition.msgs.Twist -p "linear: {x: 0}, angular: {z: 0}"' >/dev/null 2>&1 || true
+	@# set_pose service from inside the gazebo container. Orientation
+	@# is identity quaternion (w=1) — rover faces +x.
+	@$(CONTAINER_ENGINE) exec $(SIM_GAZEBO_CONTAINER) bash -c \
+	  'ign service -s /world/$(SIM_WORLD_NAME)/set_pose \
+	     --reqtype ignition.msgs.Pose --reptype ignition.msgs.Boolean \
+	     --timeout 2000 \
+	     --req "name: \"$(ROBOT_MODEL)\", position: {x: $(TELEPORT_X), y: $(TELEPORT_Y), z: $(TELEPORT_Z)}, orientation: {w: 1.0}"'
+
 # =============================================================================
 # CI cluster (smoke / pre-push parity) — alternate ports so it can run
 # alongside `make lab-up` on the same host. Used by .git-hooks/installer-smoke.sh
@@ -437,6 +461,44 @@ sim-drive-stop: ## Stop the rover
 CI_PROJECT := temporal-hack-ci
 CI_FILES   := -f docker-compose.yml -f docker-compose.ci.yml
 
+# =============================================================================
+# Demo reset — wipes all transient demo state and (optionally) brings
+# the stack back up clean.
+#
+#   make demo-reset           stops everything, wipes volumes + .run/,
+#                             and BRINGS THE STACK BACK UP fresh
+#   make demo-reset NOUP=1    same, but stops short of starting again
+# =============================================================================
+
+.PHONY: demo-reset
+demo-reset: container-check ## Stop everything, wipe demo state, and start fresh (NOUP=1 to skip the bring-up)
+	@echo "[demo-reset] stopping host-side processes"
+	-@$(MAKE) -s controlplane-down
+	-@$(MAKE) -s workers-down
+	-@$(MAKE) -s agent-down
+	@echo "[demo-reset] killing OTA-spawned robot-app containers"
+	-@$(CONTAINER_ENGINE) rm -f robot-app robot-app-new >/dev/null 2>&1 || true
+	@echo "[demo-reset] tearing down sim + lab compose, wiping volumes"
+	-@cd installer/docker-compose && $(COMPOSE) -p $(LAB_PROJECT) \
+	  -f docker-compose.yml -f docker-compose.sim.yml down -v >/dev/null 2>&1 || true
+	@echo "[demo-reset] clearing .run/ pid files and logs"
+	@rm -rf .run/
+	@if [ "$${NOUP:-0}" = "1" ]; then \
+	  echo "[demo-reset] NOUP=1 — stopping after teardown"; \
+	  exit 0; \
+	fi
+	@echo "[demo-reset] bringing the stack back up"
+	@$(MAKE) -s sim-up
+	@$(MAKE) -s agent-up
+	@$(MAKE) -s workers-up
+	@$(MAKE) -s controlplane-up
+	@echo
+	@echo "  demo reset complete. Re-run a demo:"
+	@echo "    make ota-circle"
+	@echo "    make collide"
+	@echo "  GUI: http://localhost:14680/vnc.html?autoconnect=1&resize=scale"
+	@echo "  Temporal UI: http://localhost:14080"
+
 .PHONY: ci-up
 ci-up: container-check ## Bring up an isolated CI/smoke cluster on alternate ports
 	@echo "[$(CONTAINER_ENGINE)] bringing up CI stack on alt ports"

README.md

@@ -29,39 +29,187 @@ ops/          runbooks
 
 ## Service shape
 
+The runtime splits across **three containers** + **four host-side
+processes**. The agent owns the OTA path: it shells out to the host
+docker/podman CLI to pull, run, swap, and roll back the
+**robot-app** container — which runs alongside the others on the lab
+network and joins the same ROS DDS domain.
+
+```mermaid
+flowchart TB
+    classDef host fill:#e8f0ff,stroke:#5277b8,color:#000
+    classDef cont fill:#f4ecd8,stroke:#a98246,color:#000
+    classDef ota  fill:#ffe6cc,stroke:#d79b00,color:#000
+    classDef lab  fill:#dae8fc,stroke:#6c8ebf,color:#000
+    classDef ext  fill:#f5f5f5,stroke:#666,color:#000
+
+    User([browser /<br/>operator]):::ext
+
+    subgraph LabCluster["lab cluster (compose)"]
+        direction TB
+        Gz["gazebo container<br/>ign gazebo<br/>ros_gz_bridge<br/>Xvfb + x11vnc + noVNC :14680"]:::cont
+        Robot["robot container<br/>bridge_node :50051<br/>sim_battery<br/>collision_publisher<br/>twist_subscriber"]:::cont
+        RobotApp["robot-app container<br/>drive-circle | drive-figure-eight<br/>(OTA-swappable)"]:::ota
+        MQTT[("EMQX MQTT<br/>:14883")]:::lab
+        Tmp[("Temporal :14733<br/>Postgres :14432<br/>UI :14080")]:::lab
+        Reg[("Registry :14050")]:::lab
+    end
+
+    subgraph Host["host-side Go binaries (make-managed)"]
+        direction TB
+        Agent["agent<br/>ROS gRPC client<br/>MQTT pub/sub<br/>OTA executor"]:::host
+        Otaw["ota-worker<br/>Temporal worker +<br/>MQTT bridge"]:::host
+        Colw["collision-worker<br/>Temporal worker +<br/>MQTT bridge"]:::host
+        Cp["controlplane<br/>HTTP :8081"]:::host
+        Eng["docker / podman CLI<br/>host engine"]:::host
+    end
+
+    User -->|noVNC :14680<br/>UI :14080| LabCluster
+    User -->|POST /v1/ota/rollouts| Cp
+
+    Gz <-->|ROS DDS<br/>domain 42| Robot
+    Gz <-->|ROS DDS<br/>/cmd_vel| RobotApp
+    Robot <-->|gRPC| Agent
+    Agent <-->|MQTT| MQTT
+    Otaw <-->|MQTT| MQTT
+    Colw <-->|MQTT| MQTT
+    Otaw <-->|gRPC| Tmp
+    Colw <-->|gRPC| Tmp
+    Cp -->|StartWorkflow| Tmp
+
+    Agent ==>|shells| Eng
+    Eng ==>|pull / run / rename| RobotApp
+    Reg --o RobotApp
+
+    linkStyle 11 stroke:#d79b00,stroke-width:2px
+    linkStyle 12 stroke:#d79b00,stroke-width:2px
 ```
-                 ┌────────────────── browser ──────────────────┐
-                 │  http://localhost:14680  Gazebo GUI (noVNC) │
-                 └──────────────────┬──────────────────────────┘
-                                    │
-            ┌───────────────────────▼───────────────────────┐
-            │ gazebo container                              │
-            │   • ign gazebo + ros_gz_bridge                │
-            │   • Xvfb + x11vnc + noVNC                     │
-            └────┬─────────────────────┬────────────────────┘
-                 │ ROS DDS (domain 42) │
-       ┌─────────▼──────────┐  ┌───────▼────────────────────┐
-       │ robot container    │  │ robot-app container        │
-       │ • bridge_node      │  │ (drive-circle | -fig-eight) │
-       │ • sim_battery      │  │ — OTA-swappable             │
-       │ • collision_pub    │  └────────────────────────────┘
-       │ • twist_subscriber │
-       └─────────┬──────────┘
-                 │ gRPC (TCP)
-       ┌─────────▼──────────────────────────┐
-       │ agent (Go, native macOS binary)    │
-       │ • MQTT pub/sub on lab broker       │
-       │ • OTA executor (docker/podman CLI) │
-       └─────────┬──────────────────────────┘
-                 │
-       ┌─────────▼──────────┐    ┌───────────────────────┐
-       │ MQTT (lab :14883)  │◀──▶│ ota-worker            │
-       │                    │    │ collision-worker      │
-       │                    │    │ Temporal :14733       │
-       └────────────────────┘    └───────────────────────┘
+
+Bold orange edges are the OTA path: the agent shells out to the host
+engine, which pulls from the lab registry and swaps the robot-app
+container in place.
+
+### OTA flow
+
+```mermaid
+sequenceDiagram
+    autonumber
+    actor Op as Operator
+    participant Cp as controlplane
+    participant Ow as ota-worker
+    participant Tmp as Temporal
+    participant MQ as MQTT
+    participant Ag as agent
+    participant Eng as host engine
+    participant App as robot-app
+
+    Op->>Cp: POST /v1/ota/rollouts
+    Cp->>Tmp: StartWorkflow OTARollout
+    Tmp-->>Ow: dispatch task
+    Ow->>MQ: publish cmd/{robot_id}/ota
+    MQ-->>Ag: deliver
+    Ag->>Eng: pull image_ref
+    Ag->>MQ: ack PHASE_PULLED
+    Ag->>Eng: run new under temp name
+    Ag->>Eng: rm old then rename new
+    Ag->>MQ: ack PHASE_SWAPPED
+    Eng->>App: container starts
+    Ag->>Eng: exec smoke_command
+    Ag->>MQ: ack PHASE_HEALTHY
+    MQ-->>Ow: signal workflow per phase
+    Ow->>Tmp: RecordRolloutEnded
+    Cp-->>Op: GET /v1/ota/rollouts shows completed
+```
+
+### Collision flow
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant Gz as gazebo
+    participant RGB as ros_gz_bridge
+    participant Pub as collision_publisher
+    participant MQ as MQTT
+    participant Cw as collision-worker
+    participant Tmp as Temporal
+    participant Sub as twist_subscriber
+    participant Rover as gz DiffDrive
+
+    Gz->>RGB: ignition.msgs.Contacts
+    RGB->>Pub: ROS /contacts
+    Pub->>MQ: events/{id}/collision after 2s debounce
+    MQ-->>Cw: deliver
+    Cw->>Tmp: StartWorkflow CollisionResponse
+    Tmp-->>Cw: dispatch SendTwist back 3s
+    loop each phase back, settle, turn, forward, stop
+        Cw->>MQ: cmd/{id}/twist at 10 Hz QoS 0
+        MQ-->>Sub: deliver
+        Sub->>Rover: ROS /cmd_vel forwards to gz cmd_vel
+    end
+    Cw->>MQ: cmd/{id}/twist 0,0 QoS 1 final stop
+    Cw->>Tmp: workflow Completed
 ```
 
-## Lab quickstart
+## Make-target interaction map
+
+The four bring-up targets in the **baseline** lane are the ones you
+run; everything else either depends on those or operates on them.
+Demo triggers (right column) need the baseline lane up to function.
+
+```mermaid
+flowchart LR
+    classDef baseline fill:#dae8fc,stroke:#6c8ebf,color:#000
+    classDef demo fill:#ffe6cc,stroke:#d79b00,color:#000
+    classDef status fill:#d5e8d4,stroke:#82b366,color:#000
+    classDef cleanup fill:#f8cecc,stroke:#b85450,color:#000
+
+    subgraph Baseline["BASELINE — bring up in any order"]
+        direction TB
+        SimUp["make sim-up<br/>compose up: gazebo + robot + lab cluster<br/>publishes :14050 :14080 :14432 :14680<br/>:14733 :14883 :14900 :50051"]:::baseline
+        AgentUp["make agent-up<br/>./bin/agent (.run/agent.pid)<br/>BROKER_URL=tcp://localhost:14883<br/>BRIDGE_ADDR=localhost:50051"]:::baseline
+        WorkersUp["make workers-up<br/>./bin/ota-worker + ./bin/collision-worker<br/>(.run/*.pid)<br/>TEMPORAL_ADDR=localhost:14733"]:::baseline
+        CpUp["make controlplane-up<br/>./bin/controlplane :8081<br/>(.run/controlplane.pid)"]:::baseline
+    end
+
+    subgraph Demos["DEMO TRIGGERS"]
+        direction TB
+        OtaC["make ota-circle<br/>build + push +<br/>POST /v1/ota/rollouts"]:::demo
+        OtaF["make ota-figure-eight<br/>(same shape)"]:::demo
+        Coll["make collide<br/>publish events/{id}/collision"]:::demo
+        OtaS["make ota-status<br/>GET /v1/ota/rollouts"]:::demo
+        Drive["make sim-drive-fwd LX= /<br/>-back / -left / -right / -stop<br/>(no Temporal in the loop)"]:::demo
+    end
+
+    subgraph Lifecycle["LIFECYCLE"]
+        direction TB
+        Down["make sim-down / agent-down /<br/>workers-down / controlplane-down"]:::cleanup
+        Reset["make demo-reset<br/>(or demo-reset NOUP=1)"]:::cleanup
+    end
+
+    subgraph Obs["STATUS / OBSERVABILITY"]
+        direction TB
+        Stat["make agent-status / workers-status<br/>controlplane-status / lab-status"]:::status
+        Gui["make sim-gui<br/>open noVNC URL"]:::status
+        Logs["make sim-logs<br/>tail sim+robot+agent"]:::status
+    end
+
+    SimUp -- gazebo, robot, lab containers --> AgentUp
+    AgentUp -- shells host docker/podman --> SimUp
+    WorkersUp -- gRPC --> SimUp
+    CpUp -- gRPC --> SimUp
+
+    OtaC -- HTTP --> CpUp
+    OtaC -- via MQTT --> AgentUp
+    OtaF -- HTTP --> CpUp
+    OtaF -- via MQTT --> AgentUp
+    Coll -- MQTT publish --> WorkersUp
+    OtaS -- HTTP GET --> CpUp
+    Drive -- ign topic exec --> SimUp
+
+    Reset -. wipes + restarts .- Baseline
+```
+
+### Quickstart
 
 Requires Go 1.22+, Python 3.10+, and either Docker or Podman with
 compose. The Makefile auto-detects the container engine.
@@ -78,6 +226,8 @@ That's the whole baseline. Tear down:
 
 ```bash
 make controlplane-down && make workers-down && make agent-down && make sim-down
+# OR, full wipe + restart in one shot:
+make demo-reset
 ```
 
 ## Drive demo (no Temporal in the loop)
@@ -106,6 +256,11 @@ What you'll see: a `rollout-…` workflow appears at
 1–2 seconds, and the `robot-app` container under `podman ps` flips to
 the new image. The rover's behaviour changes immediately.
 
+The full data path for a rollout — see the **OTA flow** sequence
+diagram in *Service shape* above. The agent is the only process that
+runs `podman pull / run / rename`; workers never touch the host
+engine, they orchestrate via MQTT.
+
 ## Collision demo (Temporal drives the rover out of an obstacle)
 
 The moon world spawns the rover with a 0.9 m boulder at `x = 8` —

bridge/bridge_node/collision_publisher.py

@@ -26,6 +26,12 @@
 
 DEBOUNCE_SEC = 2.0
 
+# Collision partners we IGNORE — these are continuous "of course
+# you're touching the ground" contacts that shouldn't trigger an
+# avoidance workflow. Substring match on the contact's collision2
+# name (e.g. "lunar_ground::link::collision").
+IGNORE_PARTNERS = ("lunar_ground", "ground_plane")
+
 
 class CollisionPublisher(Node):
     def __init__(self) -> None:
@@ -53,12 +59,23 @@ def _on_contacts(self, msg: Contacts) -> None:
         # means "no contacts this step"; only act on non-empty.
         if not msg.contacts:
             return
+        # Pick a contact whose partner is *not* the ground — the rover
+        # stands on its wheels (or rests on its deck) so the ground is
+        # always touching something. We only care about novel obstacles.
+        partner = None
+        for c in msg.contacts:
+            name = c.collision2.name or ""
+            if any(ig in name for ig in IGNORE_PARTNERS):
+                continue
+            partner = name
+            break
+        if partner is None:
+            return
         now = time.monotonic()
         if now - self._last_emit < DEBOUNCE_SEC:
             return
         self._last_emit = now
         self._counter += 1
-        partner = msg.contacts[0].collision2.name if msg.contacts else "unknown"
         body = json.dumps({
             "robot_id": self.robot_id,
             "at": time.time(),

cloud/cmd/collision-worker/main.go

@@ -71,7 +71,12 @@ func connectMQTT(url string) (mqtt.Client, error) {
 	opts := mqtt.NewClientOptions().
 		AddBroker(url).
 		SetClientID("collision-worker").
-		SetCleanSession(false).
+		// CleanSession=true on purpose: a missed collision event is
+		// fine (the next contact will fire another). With clean=false
+		// EMQX queues every event the bridge missed; after a storm or
+		// restart the worker reconnects to a flood of replays that
+		// blocks the publisher with backpressure.
+		SetCleanSession(true).
 		SetAutoReconnect(true).
 		SetMaxReconnectInterval(60 * time.Second).
 		SetOrderMatters(false)

cloud/internal/collision/activities.go

@@ -34,11 +34,11 @@ func (a *Activities) SendTwist(ctx context.Context, args SendTwistArgs) error {
 	tick := time.NewTicker(100 * time.Millisecond)
 	defer tick.Stop()
 	for {
-		// paho handles its own reconnects; don't second-guess via
-		// IsConnectionOpen(), which has tight semantics around
-		// reconnect windows. The token's WaitTimeout + Error are
-		// the authoritative result for a single publish.
-		tok := a.MQTT.Publish(topic, 1, false, body)
+		// QoS 0: twist messages are republished at 10 Hz so a dropped
+		// frame doesn't matter, and the QoS-1 PUBACK round-trip
+		// causes inflight-queue saturation against EMQX. Fire-and-
+		// forget is the right choice for high-rate control commands.
+		tok := a.MQTT.Publish(topic, 0, false, body)
 		if !tok.WaitTimeout(2 * time.Second) {
 			return fmt.Errorf("mqtt publish timeout for %s", topic)
 		}
@@ -55,7 +55,8 @@ func (a *Activities) SendTwist(ctx context.Context, args SendTwistArgs) error {
 		}
 	}
 
-	// Final explicit stop frame.
+	// Final explicit stop frame (QoS 1 here so the rover always sees the
+	// stop even if the last QoS 0 packet got dropped).
 	tok := a.MQTT.Publish(topic, 1, false, stop)
 	tok.WaitTimeout(2 * time.Second)
 	return tok.Error()