Add rosbag_blackbox: ROS2 ring buffer recorder + AI-accessible snapshot server

[Copilot]

Implements a new ament_python ROS2 Humble package with two nodes: a memory-efficient ring buffer recorder and a FastAPI+MCP snapshot introspection server.

Ring Buffer Node (ringbuffer_node)

• Dynamically discovers topic types at runtime (mirrors ros2 bag record behaviour)
• Topic selection via zenoh-style glob patterns in YAML config — * matches one path segment, ** matches any namespace depth
• Per-topic configurable QoS (reliability, durability, depth); default is best_effort/volatile to minimise system impact
• durability: transient_local for latched topics
• Optional max_frequency per topic — messages arriving faster than the configured Hz are silently dropped
• Buffer retains the last N seconds (default 10 s) of serialised CDR bytes in a thread-safe deque; old entries evicted on every add()
• std_srvs/srv/Trigger service (~/save_snapshot) dumps buffer → SQLite3 rosbag; filename pattern supports {datetime}, {date}, {time}, {timestamp} placeholders

# config/example_config.yaml
buffer:
  duration_seconds: 10
topics:
  - pattern: "/camera/**"
    max_frequency: 10.0
    qos: {reliability: best_effort, durability: volatile, depth: 1}
  - pattern: "/map"
    qos: {reliability: reliable, durability: transient_local, depth: 1}  # latched
  - pattern: "/**"       # catch-all
    max_frequency: 5.0

Snapshot Server (snapshot_server)

• FastAPI REST endpoints: list snapshots, get metadata, extract messages (topic/time filter, keys_only, truncate_arrays), stream image as PNG
• FastMCP (MCP protocol) at /mcp (streamable-HTTP) — AI agent tools mirror the REST surface: list_snapshots, select_snapshot, get_snapshot_metadata, list_topics, search_topics, get_messages, get_image
• sensor_msgs/Image + CompressedImage → PNG via numpy/Pillow; all other types → JSON via rosidl_runtime_py
• All non-ROS pip deps (fastapi, uvicorn, fastmcp, numpy, Pillow) are optional with graceful fallback — the recorder node builds and runs without them

Package structure

Module	Notes
ring_buffer.py	No ROS2 imports — independently unit-testable
topic_filter.py	Pure-Python zenoh glob engine
ringbuffer_node.py	ROS2 node wiring
snapshot_server.py	FastAPI + FastMCP app factory
launch/rosbag_blackbox.launch.py	Launches both nodes with configurable args
config/example_config.yaml	Annotated reference config

Original prompt

Create a new new ros2 package named rosbag_blackbox in python

It should utilise the rosbag python API to implement an efficient ringbuffer for Ros topics configured via a yaml config file. The config file describes the topics to be recorded, optionally the max frequency they should be recorded at (with messages discarded if they arrive at higher frequency), and their QoS. By default, messages on topics have a low QoS to keep the impact of the ring buffer on system performance low. But support for "latched" topics must be easily provided and documented. The configuration files must support glob topic matching,. adopting a similar syntax as zenoh query selectors (* matching on hierarchy level name, and ** matching any depth of namespace hierarchy).

The topic types should be dynamically discovered just like the normal ros2 bag cli tools.

The ringbuffer is designed to store a fixed duration of messages (e.g. always the last N seconds with N being 10 seconds by default).

The implementation must be memory efficient as it may not be possible to retain all messages in memory.

The ringbuffer is dumped into a snapshot (rosbag file) when a dedicated service provided by the new node is called. The node can be configured with the naming pattern the snapshot files should be created with. By default they are using a filename representing the snapshots date and time.

A separate node to be developed implements a FASTAPI and fastmcp2 server that can provide an Ai agent with tools.to access and introspect snapshot in a given directory. The server exposes both a fast API endpoint and a streamable http endpoint (at /MCP). By default, the server introspects the latest snapshot, but tools are available to list and choose snapshot, display their meta data, and have a comprehensive tools set with error handling to extract specific messages. Support images explicitly to be returned as PNG images, but support all mesaage types, returned as JSON. tools should have parameters to requested truncated messages, e.g. not return arrays or only return the keys of a message before requesting specific values. Filtering by time and topic name must be supported through the tools.

Implement a launch file for both nodes and document the new package and it's usage well.

---

💬 We'd love your input! Share your thoughts on Copilot coding agent in our 2 minute survey.