AuthorPascal DateJuly 14, 2026 Rev1.0

ros2-memory — Implementation Spec (MCP ROS Bag Memory Server)

Repo: ros2-memory.

What it is: the robot's episodic memory. It continuously records configured ROS topics into time-segmented bags and exposes MCP tools that let the brain answer questions about the robot's recent past ("where was I 10 s ago?", "what objects did you see at the door?", "last time the path was clear?", "how far have I driven in the last minute?").

This is the history tier of perception/state memory, complementary to the two other memories:

• mcp-perception-buffer — the latest snapshot ("what do you see now"), in-RAM, tiny window.
• ros2-memory (this) — the full recent history on disk ("what did you see then").
• robot-mcp-memory — learned processes/skills (teaching mode), not sensor data.

It is read-mostly and brain-facing over MCP; it does not publish to ROS and is not in any robot↔robot data path. It posts no events (no robot-mcp-kit dependency).

---

1. System context

The brain (robot-voice-chat, ROS-free, MCP client) asks this server about the past. The server has two decoupled halves:

1. Recorder (a daemon): runs ros2 bag record as a subprocess (argv list, never shell=True) over a configured topic list, writing time-segmented MCAP bags (--max-bag-duration). Independent of the MCP server — it can be its own systemd unit.
2. MCP query server: reads the completed bag files (via the pure-Python rosbags library — no rclpy needed) and answers tool calls. Maintains an in-memory time index of bag files ([(start, end, path)]), reconciled periodically + on demand.

Decoupling means the query server can restart without losing data, and the recorder can run on a different cadence. Because reading uses rosbags (pure Python) and recording uses the ros2 CLI, the repo imports no rclpy — it only needs a sourced ROS environment for the ros2 bag record subprocess (the message definitions for custom types are registered with the rosbags typestore, §5).

Distinct from the process store: this server stores time-series sensor data, not markdown processes. Its MCP server name is episodic (tools episodic.get_topic_at_time, …) on port 9203, so it never collides with the process store (memory, 9103).

---

2. Repo layout

ros2-memory/
├── README.md  CHANGELOG.md  .pre-commit-config.yaml  .github/   # Appendix A
├── pyproject.toml                 # package: ros2_memory; deps: mcp, rosbags, mcap, pydantic,
│                                  #   hydra-core, omegaconf, numpy, structlog, pyyaml
├── uv.lock   .env.example
├── config/                        # Hydra tree
│   ├── config.yaml                # recording / data / reader knobs + defaults list
│   ├── storage/  mcap.yaml  rosbag2.yaml      # bag backend (port = adapter)
│   └── recorder/ ros2bag.yaml  none.yaml      # recorder backend (none = query-only deployments)
├── src/ros2_memory/
│   ├── server.py                  # MCP Server + tool dispatch (§6) + entrypoint
│   ├── recorder/
│   │   ├── base.py                # Recorder ABC (start/stop/is_running)
│   │   ├── ros2bag.py             # subprocess `ros2 bag record` daemon
│   │   └── index_manager.py       # scan bag dir → time index; incremental, mtime-cached
│   ├── storage/
│   │   ├── base.py                # BagReader ABC (read messages in [t0,t1] for a topic)
│   │   ├── mcap_reader.py         # MCAP via mcap / rosbags
│   │   └── rosbag2_reader.py      # rosbag2 dir fallback
│   ├── extract.py                 # SchemaExtractor: dot-path field extraction (§5)
│   ├── typestore.py               # register custom msgs (ros2_vision_interfaces) with rosbags
│   ├── queries.py                 # the query implementations behind the tools (§6)
│   └── config.py                  # Pydantic config models + Hydra load (fail-fast)
├── deploy/
│   ├── ros2-memory-recorder.service   # the recorder daemon (sources ROS, runs ros2 bag record)
│   └── ros2-memory-server.service     # the MCP query server
└── tests/{unit,integration}/

Ports-and-adapters + Hydra (Appendix A): Recorder and BagReader are ABCs with Hydra-selected adapters (recorder=ros2bag|none, storage=mcap|rosbag2). recorder=none ships a query-only deployment (reads bags produced elsewhere). Python ≥ 3.11, uv, ruff/mypy/pytest, CI gates.

---

3. What gets recorded (and what does NOT)

Recording raw camera images or pixel masks to bags is heavy and almost never what episodic memory needs, so the recorder captures the vision outputs and state, not pixels. For vision detections it records the mask-free meta topic /vision/{skill}/detections_meta (the vision node always publishes it alongside the masked /detections; see specs/robot-mcp-perception.md §6.1). That topic is ~100× smaller than the masked one, so history spans days rather than minutes, and it keeps exactly what gets queried (labels/boxes/scores/points over time). Old masks are useless once the scene changes, so they belong on the live data plane only, not in history. The recorded set is configurable in config/topics.yaml:

# config/topics.yaml — the topics the recorder captures
topics:
  - /odom                                  # nav_msgs/Odometry (pose + velocity)
  - /tf                                    # tf2_msgs/TFMessage
  - /tf_static
  - /cmd_vel                               # geometry_msgs/Twist
  - /vision/scene/scene_description         # ros2_vision_interfaces/StampedString
  - /vision/scene/scene_objects
  - /vision/find_object/detections_meta     # mask-FREE VisionDetectionArray (boxes/labels/points)
  # - /vision/find_object/detections        # masked variant — only for short debugging bags
  # - /camera/image_raw                     # raw frames — OFF by default (huge); debugging only

The masked /detections and raw /camera/image_raw are commented out; turn them on only for a short debugging session. The disk-budget GC (§7) bounds growth regardless.

---

4. Storage & index

• MCAP (default) via mcap / rosbags; rosbag2 directory fallback. Bags auto-split every recording.split_duration_s (default 10 s); the active bag is marked and excluded from queries until closed.
• Index: index_manager.py keeps [(start_ns, end_ns, path)] sorted; rescans bag_dir every index_refresh_s (default 5 s) and on the first query after a gap; mtime-cached so rescans are cheap. A query loads only the bag(s) overlapping the requested time window.

---

5. Message extraction (extract.py + typestore.py)

Tools return plain JSON, not raw ROS messages. SchemaExtractor pulls fields by dot-path, configured per message type so one tool works across many message types:

# config/message_schemas.yaml
schemas:
  nav_msgs/msg/Odometry:
    position:          pose.pose.position
    orientation:       pose.pose.orientation         # quaternion → auto Euler in the extractor
    linear_velocity:   twist.twist.linear
    angular_velocity:  twist.twist.angular
  sensor_msgs/msg/LaserScan:
    ranges: ranges
  ros2_vision_interfaces/msg/VisionDetectionArray:
    skill: skill
    detections: detections                            # list → array, downsampled to max_array_length
  ros2_vision_interfaces/msg/StampedString:
    value: data

• Quaternion→Euler conversion built in; arrays longer than data.max_array_length (default 50) are downsampled with a note.
• Custom messages: typestore.py registers ros2_vision_interfaces (and any other custom pkg) .msg definitions with the rosbags typestore so the reader can deserialize /vision/** history without a ROS runtime. Path to the interface .msg files via reader.custom_msg_paths config (or ROS_VISION_INTERFACES_PATH env). Unknown types fall back to a best-effort field dump.

---

6. MCP tools (grouped; all return JSON strings)

Times are unix seconds (float). Relative variants take seconds_ago. Docstrings are written for the planner LLM.

Metadata | Tool | Returns | |---|---| | list_recorded_topics() | [{topic, type, first_ts, last_ts, count}] | | data_time_range() | {earliest, latest} of all recorded data |

Generic value queries | Tool | Args | Returns | |---|---|---| | get_topic_at_time(topic, timestamp) | | nearest message at/just-before timestamp (within data.time_tolerance_s) as extracted JSON | | get_topic_seconds_ago(topic, seconds_ago) | | same, relative to now | | get_topic_in_range(topic, start, end, max_samples=100) | | list of samples in [start,end] (downsampled) | | get_topic_in_last(topic, seconds, max_samples=100) | | same, relative window |

Semantic extractors (use the schema map) | Tool | Returns | |---|---| | get_position_at(timestamp) / get_position_seconds_ago(seconds_ago) | {x,y,z, yaw_deg} from the pose topic | | get_velocity_at(...) / get_velocity_seconds_ago(...) | {linear, angular} |

Transforms | Tool | Returns | |---|---| | get_transform_at(parent, child, timestamp) (+ _seconds_ago) | {translation, rotation, yaw_deg} from /tf//tf_static | | list_frames() | available TF frames + parent/child pairs |

Conditional / correlation | Tool | Behavior | |---|---| | find_last_match(topic, condition_type, value) | reverse-search for the last time a condition held. condition_type ∈ contains (substring for strings/lists), equals (bool/number), near_position ({x,y} within data.position_tolerance_m). Returns {timestamp, value} or null. Lets the brain ask "when did you last see scissors?" against /vision/*/scene_objects. | | get_value_when(value_topic, condition_topic, condition_type, value) | find when condition_topic matched, return value_topic at that time (correlate two streams). |

Analysis | Tool | Returns | |---|---| | trajectory_summary(start, end) / recent_trajectory_summary(seconds) | {distance_m, avg_speed, max_speed, duration_s, waypoints[]} from the pose topic |

Each tool: clamp the window to recorded range, load only overlapping bags, extract via the schema map, downsample, return JSON. Out-of-range / no-data → {"error": "no_data", "detail": …} (not an exception). ~30 s call-timeout tolerance; tolerate list_tools() health checks (Appendix A).

---

7. Config, retention, deployment

config/config.yaml:

defaults:
  - storage: mcap
  - recorder: ros2bag
  - _self_

recording:
  bag_dir: data/bags
  split_duration_s: 10
  clear_on_startup: false        # wipe bags on recorder start (sim/testing)
  topics_file: config/topics.yaml
data:
  max_array_length: 50
  position_tolerance_m: 0.5
  time_tolerance_s: 0.1
reader:
  index_refresh_s: 5
  custom_msg_paths: []           # dirs of custom .msg (e.g. ros2_vision_interfaces); or env
retention:
  max_total_gb: 5.0              # GC oldest bags beyond this
  max_age_h: 168                 # ...and older than this (1 week)

Env	Meaning	Default
MCP_TRANSPORT / MCP_PORT	stdio | sse / port	stdio / 9203
MEMORY_BAG_DIR	override recording.bag_dir	config
ROS_VISION_INTERFACES_PATH	custom-msg defs for the reader	unset
ROS_DOMAIN_ID	for the recorder subprocess	inherit
MEMORY_CONFIG_OVERRIDES	Hydra overrides (e.g. recorder=none)	unset

Retention GC runs in the index manager: when total bag size > max_total_gb or a bag is older than max_age_h, delete oldest closed bags. Two systemd units (deploy/): the recorder (sources the catkin/colcon workspace, runs ros2 bag record) and the MCP query server; both hardened (NoNewPrivileges, PrivateTmp, ProtectSystem=strict with ReadWritePaths=<bag_dir>, dedicated user, journald). Wire into the brain's tools/mcp.yaml:

  - name: episodic
    transport: sse
    url: "http://127.0.0.1:9203/sse"

No event_secret (posts no events).

---

8. Tests

Unit (no ROS runtime — fixture bags + the rosbags reader): 1. index: bag dir with several split bags → correct sorted time index; active bag excluded; incremental rescan picks up a new bag; GC by size + age. 2. extract: dot-path extraction for Odometry/Twist/StampedString/VisionDetectionArray; quaternion → yaw; array downsampling; custom-msg typestore registration deserializes a VisionDetectionArray. 3. queries: get_topic_at_time nearest-within-tolerance; range downsample; find_last_match for contains/equals/near_position; get_value_when correlation; trajectory_summary math (distance/speed) against a known synthetic path; out-of-range → no_data. 4. tools (MCP in-memory client): every tool returns valid JSON; bad args → clean error. 5. recorder (recorder=ros2bag): builds the correct ros2 bag record argv (no shell=True); recorder=none no-ops. (The actual subprocess is integration-only.)

Integration (not CI): record a short live session (or replay a fixture bag), then get_position_seconds_ago(10) and find_last_match("/vision/scene/scene_objects","contains","scissors") return sane answers.

9. Acceptance criteria

• All gates pass (Appendix A: pytest, ruff, ruff-format, mypy, pip-audit).
• Recorder writes split MCAP bags for the configured topics; query server indexes them.
• data_time_range / list_recorded_topics reflect live data; get_position_seconds_ago(N) returns the pose from N s ago.
• find_last_match answers "when did you last see X" against recorded /vision/** history.
• Retention GC keeps the bag dir under the disk budget.
• recorder=none deployment serves queries over pre-existing bags with no recording.

---

Appendix A — Repository & tooling conventions

Standard "how we build repos here". This is an MCP server (rclpy-free: rosbags for reading, ros2 bag record subprocess for recording) — all sections apply, including MCP servers specifically.

Structure — src/<package>/ layout (never flat); config/ Hydra tree; tests/{unit,integration}/; committed README.md (quickstart + how-to table + troubleshooting table), CHANGELOG.md (Keep-a-Changelog + SemVer), .pre-commit-config.yaml, .github/ (CI + dependabot.yml), optional multi-stage non-root Dockerfile, .env.example committed / .env gitignored.

Tooling — uv for env + deps (uv sync, lockfile committed), Python ≥ 3.11, optional extras for heavy/optional deps (the --extra observability pattern). ruff (lint + format), mypy --strict, pytest + pytest-asyncio, coverage gate ≥ 80%. Pre-commit: trailing-whitespace, end-of-file-fixer, check-yaml/toml/merge-conflict/large-files, ruff (--fix), ruff-format, and a local mypy hook run via uv run (so it picks up the pydantic.mypy plugin). CI on every PR: pytest, ruff check, ruff format --check, mypy, pip-audit. Dependabot weekly for the uv ecosystem.

Architecture — ports-and-adapters: every capability with >1 implementation or that touches hardware/network gets an abc.ABC port in base.py; concrete adapters live beside it; application code imports only ports. Hydra + Pydantic config: one config-group dir per port, one YAML per implementation carrying _target_; swap implementations by changing one line in the root config.yaml defaults list — no if provider == … ladders. Validate the composed config through Pydantic dataclasses at startup (fail fast). Deployment presets under config/deployment/<name>.yaml selected via an env var.

Secrets — never in YAML or git. .env + OmegaConf ${oc.env:KEY} resolvers, typed SecretStr; startup refuses to boot if a required key is empty.

Logging — structlog JSON by default, pretty renderer toggle for dev; bind correlation IDs via contextvars; log duration_ms per stage. Subprocess — always argv lists, never shell=True. Testing — fakes at the port boundary (don't patch internals); respx for HTTP-level mocking; integration via httpx.AsyncClient + ASGITransport.

MCP servers specifically — one repo per capability from a copy-me template; transport stdio if the consumer supervises it (1–2 co-located), else sse/streamable_http with its own systemd unit/container/logs/restart policy and the consumer just holds a URL. Tools namespaced server.tool; mark slow tools "blocking", else fire-and-forget; expect a ~30 s call timeout; tolerate clients health-checking via list_tools(). Resilience: the consumer reconnects on an interval — design for clean restart, never assume connection order at boot. Deploy: systemd unit with hardening (NoNewPrivileges, PrivateTmp, ProtectSystem=strict), bind localhost, dedicated user, journald.