feat(mosaico_integration): add 3-robot fleet variant

Michał Fąferek · Michał Fąferek · commit 7ab73c45a5fb · 2026-04-15T18:29:37.000+02:00
Extends the Mosaico M0 demo with a fleet compose file so cross-robot
forensic queries have more than one sequence to filter on. The single-
robot variant proves the plumbing; the fleet variant is what makes the
.Q catalog queries interesting.

docker-compose.fleet.yml spins up three independent sensor-demos
(warehouse-A on 18081, warehouse-B on 18082, outdoor-yard on 18083)
each with its own bridge sharing a single mosaicod + postgres. Each
bridge tags ingested sequences with distinct robot_id metadata.

scripts/trigger-fleet-faults.sh injects three heterogeneous fault
signatures so the resulting Mosaico catalog has visible variance:
  robot-01 warehouse-A  LIDAR noise_stddev=0.5   (range std spikes)
  robot-02 warehouse-B  IMU failure              (different sensor)
  robot-03 outdoor-yard LIDAR drift_rate=0.5     (range mean shifts)

medkit_params.yaml: widen ring buffer to 15 s pre + 10 s post (was
10 + 2) so each snapshot carries enough pre-fault baseline for
drift-vs-noise comparison; lift max_bag_size_mb to 2000 so rosbag2
does not split the recording mid-flight (the gateway only serves the
first split).

bridge.py: honor POST_FAULT_WAIT_SEC (default 12 s) before downloading,
so the longer duration_after_sec still finishes before the bridge
grabs the bag.

README updated with fleet quick start, file inventory, and refreshed
numbers ("25 s snapshot", "~500 MB MCAP") consistent with the new
ring buffer sizes.

Verified end-to-end on this machine: 3 fault injections produce 3
sequences in mosaicod, all listable via MosaicoClient.list_sequences()
and tagged with the expected metadata. Cold start to last verified
ingest ~75 s.
diff --git a/demos/mosaico_integration/README.md b/demos/mosaico_integration/README.md
@@ -2,7 +2,12 @@
 
 End-to-end pipeline demonstrating that **medkit fault snapshots flow into Mosaico as queryable, structured forensic data** - with no robot hardware and no recording 24/7. Designed as a homework artifact for the [Mosaico Labs](https://github.com/mosaico-labs) team.
 
-> Robot stack runs in Docker. Inject high noise on the simulated LiDAR. The medkit gateway detects it via the standard `/diagnostics` ROS topic, confirms the fault, and flushes its 10-second pre-fault + 2-second post-fault ring buffer to an `.mcap` file. A small Python bridge listens on the gateway's `/faults/stream` SSE endpoint, downloads the bag via REST, and ingests it into mosaicod via Apache Arrow Flight using Mosaico's own Python SDK. From `docker compose up` to a queryable `Sequence` in mosaicod takes about a minute.
+> Robot stack runs in Docker. Inject high noise on the simulated LiDAR. The medkit gateway detects it via the standard `/diagnostics` ROS topic, confirms the fault, and flushes its 15-second pre-fault + 10-second post-fault ring buffer to an `.mcap` file. A small Python bridge listens on the gateway's `/faults/stream` SSE endpoint, downloads the bag via REST, and ingests it into mosaicod via Apache Arrow Flight using Mosaico's own Python SDK. From `docker compose up` to a queryable `Sequence` in mosaicod takes about a minute.
+
+There are two variants of the stack:
+
+- **Single-robot** (`docker-compose.yml`): one sensor-demo + one bridge. Good for stepping through the pipeline the first time and for the notebook.
+- **Fleet** (`docker-compose.fleet.yml`): three sensor-demos (warehouse-A, warehouse-B, outdoor-yard) each with its own bridge, all ingesting into one shared mosaicod. Produces three heterogeneous fault snapshots (LiDAR noise, IMU failure, LiDAR drift) under distinct `robot_id` metadata so cross-robot queries actually have something to filter on.
 
 ## Quick start
 
@@ -29,6 +34,27 @@ jupyter notebook notebooks/mosaico_demo.ipynb
 
 The notebook connects to `localhost:16726` (mosaicod Arrow Flight) and runs four queries against your freshly-ingested fault snapshot, ending with a three-panel time-series plot showing the LiDAR noise spike alongside a stationary IMU - exactly the kind of cross-topic forensic correlation Mosaico is designed for.
 
+### Fleet variant
+
+```bash
+# Three robots + three bridges + shared mosaicod. Ports 18081/2/3 for the
+# robots, 16726 for mosaicod (same as single-robot).
+docker compose -f docker-compose.fleet.yml up -d
+
+# Wait ~25s after all containers are healthy so the ring buffer has a
+# pre-fault baseline, then inject three different fault signatures:
+#   robot-01 warehouse-A : LiDAR noise_stddev = 0.5  (range std spike)
+#   robot-02 warehouse-B : IMU failure
+#   robot-03 outdoor-yard: LiDAR drift_rate = 0.5    (range mean shift)
+./scripts/trigger-fleet-faults.sh
+
+# Each bridge independently ingests its robot's snapshot. After ~45s you
+# will have three Sequences in mosaicod with distinct robot_id metadata.
+docker compose -f docker-compose.fleet.yml logs -f bridge-01 bridge-02 bridge-03
+
+docker compose -f docker-compose.fleet.yml down -v
+```
+
 ## Architecture
 
 ```
@@ -92,19 +118,21 @@ The same draft works on the **read** side too, as long as the consumer imports `
 
 ## Smart snapshots, not 24/7 recording
 
-This is the value prop: each entry in the Mosaico catalog is **only the 12 seconds around a confirmed fault**, not hours of "nothing is happening" telemetry. With three faults injected the catalog totals ~140 MB, indexed and searchable. A naive 24/7 recording of the same four sensors at the same rates would be roughly 120 GB per robot per day. The pipeline preserves the signal and discards the noise.
+This is the value prop: each entry in the Mosaico catalog is **only the 25 seconds around a confirmed fault** (15 s pre-fault baseline + 10 s post-fault), not hours of "nothing is happening" telemetry. Three fleet snapshots together weigh roughly 1.5 GB of MCAP - a naive 24/7 recording of the same four sensors at the same rates would be closer to 120 GB per robot per day. The pipeline preserves the signal and discards the noise.
 
 ## Files in this directory
 
 | Path | What |
 |---|---|
-| `docker-compose.yml` | The four-service stack: postgres + mosaicod + sensor-demo + bridge |
-| `bridge/bridge.py` | Subscribes SSE, downloads bag, calls `RosbagInjector` |
+| `docker-compose.yml` | Single-robot stack: postgres + mosaicod + sensor-demo + bridge |
+| `docker-compose.fleet.yml` | Fleet stack: postgres + mosaicod + 3×(sensor-demo + bridge) |
+| `bridge/bridge.py` | Subscribes SSE, downloads bag, calls `RosbagInjector`. Honors `POST_FAULT_WAIT_SEC` (default 12s) before download so the post-fault ring segment is finalized |
 | `bridge/Dockerfile` | python:3.11-slim + Mosaico SDK pinned to PR #368 commit `8e090cd` |
 | `bridge/requirements.txt` | `httpx>=0.27,<0.30` |
-| `medkit_overrides/medkit_params.yaml` | Sensor-demo medkit config with `auto_cleanup: false` defensively flipped |
+| `medkit_overrides/medkit_params.yaml` | Sensor-demo medkit config: 15s pre + 10s post ring buffer, single 2 GB bag cap, `auto_cleanup: false` |
 | `notebooks/mosaico_demo.ipynb` | Connect, list, query, plot - 7 cells |
-| `scripts/trigger-fault.sh` | Calls the medkit Scripts API to inject high noise on lidar-sim |
+| `scripts/trigger-fault.sh` | Single-robot: inject high noise on lidar-sim on `localhost:18080` |
+| `scripts/trigger-fleet-faults.sh` | Fleet: inject three different fault signatures on robots 01/02/03 |
 | `docs/lidar_noise_plot.png` | The output of the notebook's money shot plot |
 | `docs/EMAIL_TO_FRANCESCO.md` | Email draft to send Mosaico Labs |
 
@@ -117,9 +145,9 @@ This is the value prop: each entry in the Mosaico catalog is **only the 12 secon
 | medkit gateway responds at `localhost:18080/api/v1/health` | ✅ |
 | `./scripts/trigger-fault.sh` injects fault, gateway returns CONFIRMED | ✅ |
 | Bridge SSE connects, picks up `fault_confirmed` event | ✅ |
-| Bridge resolves entity `apps/diagnostic-bridge` and downloads ~100 MB MCAP | ✅ |
+| Bridge resolves entity `apps/diagnostic-bridge` and downloads ~500 MB MCAP (25 s of four sensors) | ✅ |
 | `RosbagInjector` finalizes 4 TopicWriters (`/sensors/{scan,imu,fix,image_raw}`) | ✅ |
-| `MosaicoClient.list_sequences()` shows the new sequence within ~5s of fault | ✅ |
+| `MosaicoClient.list_sequences()` shows the new sequence within ~25 s of fault confirmation | ✅ |
 | Notebook reads back `LaserScan` data with `range_min`, `range_max`, `ranges`, `intensities`, `frame_id` populated | ✅ |
 | `IMU.Q.acceleration.z.between(...)` filter returns sequences | ✅ |
 
@@ -130,8 +158,9 @@ This is the value prop: each entry in the Mosaico catalog is **only the 12 secon
 3. **Faults from the legacy diagnostic path land under `apps/diagnostic-bridge`**, not `apps/lidar-sim`. The diagnostic_bridge node is what owns the snapshot bag in this demo.
 4. **Mosaico read-side registry**: even with PR #368 installed, you must `import mosaicolabs.models.futures.laser` before reading `LaserScan` data. Otherwise the topic reader raises `No ontology registered with tag 'laser_scan'`. The bridge does not need this (write side resolves adapters by ROS msg type) but the notebook does.
 5. **Not all 5 listed topics actually land in Mosaico**: `/diagnostics` drops silently because no adapter is registered. The medkit ring buffer captures it; Mosaico just does not know what to do with it. See the table above.
-6. **Initial post-fault wait**: medkit holds the rosbag2 writer open for `duration_after_sec` (2s) after `fault_confirmed`. The bridge sleeps 2.5s before downloading to make sure the file is finalized.
-7. **Gateway port conflict on dev boxes**: we publish on `18080` and `16726` instead of the conventional `8080` and `6726` because dev machines often have other demos running. Adjust if you prefer defaults.
+6. **Initial post-fault wait**: medkit holds the rosbag2 writer open for `duration_after_sec` (10s in this config) after `fault_confirmed`. The bridge waits `POST_FAULT_WAIT_SEC` seconds (default 12) before downloading so the trailing ring segment is finalized.
+7. **Gateway port conflict on dev boxes**: the single-robot stack publishes on `18080` and `16726`; the fleet stack uses `18081/18082/18083` for the three gateways with `16726` shared by mosaicod. Adjust if you prefer defaults.
+8. **`rosbag2` file splitting**: if `max_bag_size_mb` is hit mid-recording, `rosbag2` splits into `_0.mcap`, `_1.mcap`, ... and the medkit gateway serves only the first split. The 2 GB cap in `medkit_params.yaml` is there to prevent splitting for any realistic 25 s snapshot.
 
 ## Troubleshooting
 
diff --git a/demos/mosaico_integration/bridge/bridge.py b/demos/mosaico_integration/bridge/bridge.py
@@ -346,10 +346,12 @@ def handle_fault_event(fault: FaultEvent) -> None:
         LOG.warning("Confirmed fault has empty fault_code, skipping")
         return
 
-    # Tiny pause: medkit's post-fault timer keeps the bag writer open for
-    # `duration_after_sec` (default 2s in sensor_diagnostics demo). Wait
-    # so the bag is finalized and resolve_rosbag_file_path can find it.
-    time.sleep(2.5)
+    # Medkit's post-fault timer keeps the bag writer open for
+    # `duration_after_sec` (configured to 10s in fleet demo). Wait for
+    # finalization before trying to download.
+    post_fault_wait = float(os.environ.get("POST_FAULT_WAIT_SEC", "12"))
+    LOG.info("Waiting %.0fs for post-fault recording to finalize...", post_fault_wait)
+    time.sleep(post_fault_wait)
 
     bag = download_bag(MEDKIT_URL, fault)
     if bag is None:
diff --git a/demos/mosaico_integration/docker-compose.fleet.yml b/demos/mosaico_integration/docker-compose.fleet.yml
@@ -0,0 +1,160 @@
+# Mosaico M0 Fleet Demo: 3 robots, 3 bridges, 1 Mosaico archive.
+#
+# Shows cross-robot forensic queries that a folder of .mcap files cannot do.
+#
+# Robots:
+#   robot-01  warehouse aisle A   port 18081   ROS_DOMAIN_ID=41
+#   robot-02  warehouse aisle B   port 18082   ROS_DOMAIN_ID=42
+#   robot-03  outdoor yard        port 18083   ROS_DOMAIN_ID=43
+#
+# Each robot gets its own bridge that ingests fault snapshots into
+# the shared mosaicod with robot_id metadata.
+
+name: mosaico-fleet
+
+services:
+  # ── Shared infrastructure ──
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: password
+      POSTGRES_DB: mosaico
+    volumes:
+      - fleet_pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres -d mosaico"]
+      interval: 5s
+      timeout: 3s
+      retries: 20
+    networks: [fleet_net]
+
+  mosaicod:
+    image: ghcr.io/mosaico-labs/mosaicod:v0.3.0
+    environment:
+      MOSAICOD_DB_URL: postgres://postgres:password@postgres:5432/mosaico
+      RUST_LOG: info
+    command: [run, --host, 0.0.0.0, --port, "6726", --local-store, /data/store]
+    volumes:
+      - fleet_store:/data/store
+    depends_on:
+      postgres: { condition: service_healthy }
+    ports: ["16726:6726"]
+    networks: [fleet_net]
+
+  # ── Robot 01: Warehouse Aisle A ──
+  robot-01:
+    build:
+      context: ../sensor_diagnostics
+      dockerfile: Dockerfile
+    container_name: fleet_robot_01
+    environment:
+      ROS_DOMAIN_ID: "41"
+      BEACON_MODE: none
+    ports: ["18081:8080"]
+    volumes:
+      - ./medkit_overrides/medkit_params.yaml:/root/demo_ws/install/sensor_diagnostics_demo/share/sensor_diagnostics_demo/config/medkit_params.yaml:ro
+    healthcheck:
+      test: ["CMD-SHELL", "curl -f http://localhost:8080/api/v1/health || exit 1"]
+      interval: 5s
+      timeout: 3s
+      retries: 30
+      start_period: 30s
+    networks: [fleet_net]
+
+  bridge-01:
+    build: { context: ./bridge }
+    container_name: fleet_bridge_01
+    environment:
+      MEDKIT_URL: http://robot-01:8080
+      MOSAICO_HOST: mosaicod
+      MOSAICO_PORT: "6726"
+      ROBOT_ID: robot-01-warehouse-A
+      SOURCE_DEMO: fleet_demo
+      LOG_LEVEL: INFO
+      POST_FAULT_WAIT_SEC: "15"
+    depends_on:
+      robot-01: { condition: service_healthy }
+      mosaicod: { condition: service_started }
+    networks: [fleet_net]
+
+  # ── Robot 02: Warehouse Aisle B ──
+  robot-02:
+    build:
+      context: ../sensor_diagnostics
+      dockerfile: Dockerfile
+    container_name: fleet_robot_02
+    environment:
+      ROS_DOMAIN_ID: "42"
+      BEACON_MODE: none
+    ports: ["18082:8080"]
+    volumes:
+      - ./medkit_overrides/medkit_params.yaml:/root/demo_ws/install/sensor_diagnostics_demo/share/sensor_diagnostics_demo/config/medkit_params.yaml:ro
+    healthcheck:
+      test: ["CMD-SHELL", "curl -f http://localhost:8080/api/v1/health || exit 1"]
+      interval: 5s
+      timeout: 3s
+      retries: 30
+      start_period: 30s
+    networks: [fleet_net]
+
+  bridge-02:
+    build: { context: ./bridge }
+    container_name: fleet_bridge_02
+    environment:
+      MEDKIT_URL: http://robot-02:8080
+      MOSAICO_HOST: mosaicod
+      MOSAICO_PORT: "6726"
+      ROBOT_ID: robot-02-warehouse-B
+      SOURCE_DEMO: fleet_demo
+      LOG_LEVEL: INFO
+      POST_FAULT_WAIT_SEC: "15"
+    depends_on:
+      robot-02: { condition: service_healthy }
+      mosaicod: { condition: service_started }
+    networks: [fleet_net]
+
+  # ── Robot 03: Outdoor Yard ──
+  robot-03:
+    build:
+      context: ../sensor_diagnostics
+      dockerfile: Dockerfile
+    container_name: fleet_robot_03
+    environment:
+      ROS_DOMAIN_ID: "43"
+      BEACON_MODE: none
+    ports: ["18083:8080"]
+    volumes:
+      - ./medkit_overrides/medkit_params.yaml:/root/demo_ws/install/sensor_diagnostics_demo/share/sensor_diagnostics_demo/config/medkit_params.yaml:ro
+    healthcheck:
+      test: ["CMD-SHELL", "curl -f http://localhost:8080/api/v1/health || exit 1"]
+      interval: 5s
+      timeout: 3s
+      retries: 30
+      start_period: 30s
+    networks: [fleet_net]
+
+  bridge-03:
+    build: { context: ./bridge }
+    container_name: fleet_bridge_03
+    environment:
+      MEDKIT_URL: http://robot-03:8080
+      MOSAICO_HOST: mosaicod
+      MOSAICO_PORT: "6726"
+      ROBOT_ID: robot-03-outdoor-yard
+      SOURCE_DEMO: fleet_demo
+      LOG_LEVEL: INFO
+      POST_FAULT_WAIT_SEC: "15"
+    depends_on:
+      robot-03: { condition: service_healthy }
+      mosaicod: { condition: service_started }
+    networks: [fleet_net]
+
+volumes:
+  fleet_pgdata:
+  fleet_store:
+
+networks:
+  fleet_net:
+    name: mosaico_fleet_net
+    driver: bridge
diff --git a/demos/mosaico_integration/medkit_overrides/medkit_params.yaml b/demos/mosaico_integration/medkit_overrides/medkit_params.yaml
@@ -72,14 +72,14 @@ fault_manager:
       # Rosbag recording configuration
       rosbag:
         enabled: true
-        duration_sec: 10.0  # Record 10 seconds before fault confirmation
-        duration_after_sec: 2.0  # Record 2 seconds after confirmation
+        duration_sec: 15.0  # Record 15 seconds before fault confirmation
+        duration_after_sec: 10.0  # Record 10 seconds after confirmation
         lazy_start: false  # Always recording (ring buffer)
         format: "mcap"  # MCAP format (recommended for cross-platform)
         storage_path: "/var/lib/ros2_medkit/rosbags"
-        max_bag_size_mb: 100  # Max size per rosbag file
+        max_bag_size_mb: 2000  # Single-file cap; prevents rosbag2 from splitting (gateway only serves the first file)
         max_total_storage_mb: 1000  # 1GB total storage limit
-        auto_cleanup: false  # Cleanup rosbags on fault clear
+        auto_cleanup: false  # Retain rosbag on fault clear so the bridge can ingest it
 
         # Topics to record (use 'config' or 'all')
         topics: "config"  # Use include/exclude lists below
diff --git a/demos/mosaico_integration/scripts/trigger-fleet-faults.sh b/demos/mosaico_integration/scripts/trigger-fleet-faults.sh
@@ -0,0 +1,33 @@
+#!/bin/bash
+# Fleet fault injection - different fault signatures per robot:
+#   robot-01: LIDAR noise (noise_stddev=0.5) - range std spikes
+#   robot-02: IMU failure - different sensor entirely
+#   robot-03: LIDAR drift (drift_rate=0.5) - range mean shifts over time
+set -euo pipefail
+
+echo "=== Fleet Fault Injection ==="
+echo ""
+
+echo ">> robot-01 (warehouse-A): LIDAR noise_stddev → 0.5"
+curl -sf -X PUT "http://localhost:18081/api/v1/apps/lidar-sim/configurations/noise_stddev" \
+    -H "Content-Type: application/json" -d '{"value": 0.5}' > /dev/null
+
+sleep 5
+
+echo ">> robot-02 (warehouse-B): IMU failure"
+curl -sf -X POST "http://localhost:18082/api/v1/components/compute-unit/scripts/inject-failure/executions" \
+    -H "Content-Type: application/json" -d '{"execution_type":"now"}' > /dev/null
+
+sleep 5
+
+echo ">> robot-03 (outdoor-yard): LIDAR drift_rate → 0.5"
+curl -sf -X PUT "http://localhost:18083/api/v1/apps/lidar-sim/configurations/drift_rate" \
+    -H "Content-Type: application/json" -d '{"value": 0.5}' > /dev/null
+
+echo ""
+echo "=== Injected ==="
+echo "   robot-01: LIDAR noise (high variance)"
+echo "   robot-02: IMU failure"
+echo "   robot-03: LIDAR drift (shifting baseline)"
+echo ""
+echo "Wait ~60s for post-fault capture + bridge ingest."
