feat(rtps_embedded): ROS 2-interoperable services (RMI) + actions (AMI) + native protocol

[finger563]

Adds services (RMI) and actions (AMI) to the rtps_embedded component, in
both a ROS 2-interoperable flavour and a lean native (espp↔espp) flavour, on top
of the existing pub/sub — validated end-to-end against live ROS 2 (rmw_fastrtps,
Jazzy) both directions.

What's new

• Services (RMI) — request/reply with correlated responses. Byte-level
  (add_service_server/_client) and typed (espp::ServiceServer<Req,Resp> /
  ServiceClient<Req,Resp>). Three client call styles: blocking call,
  call_async, call_future. Deferred-reply support (ServiceResponder) for
  slow responses.
• Actions (AMI) — long-running goals with feedback, result, and cancel
  (3 services + 2 topics; no new wire primitive). Byte-level and typed
  (ActionServer / ActionClient / ActionGoalHandle).
• Native protocol — a deliberately lean espp↔espp variant (20-byte in-band
  correlation header, es_rq/es_rr topics, ~3 endpoints per action vs ROS's
  ~10), same typed wrappers via a RtpsProtocol config field.
• Typed layer — rtps_service.hpp / rtps_action.hpp, the RMI/AMI analogue
  of Publisher<T>/Subscriber<T>: reflectable structs (de)serialized by the
  cdr component, no manual bytes.
• Python bindings for the whole surface + python/rtps_rpc_demo.py.
• Docs — doc/en/protocols/rtps_rmi_ami.rst (function, use, rationale, wire
  format, when to choose native); refreshed component + example READMEs.
• Kconfig RTPS_ENABLE_RPC (default on) compiles the whole services/actions
  layer out for pure-pub/sub devices.
• Embedded example now hosts a typed service + action a ROS 2 client can
  drive, runs service/action clients, and has an opt-in second-participant
  self-test.

Wire format (confirmed by capture, not guessed)

Before writing code, each wire format was captured from a live rmw_fastrtps
exchange (tshark) and codified in golden byte tests — which corrected two initial
assumptions: ROS correlation is related_sample_identity inline QoS under both
PID 0x0083 (OMG) and 0x800f (eProsima), present on both request and reply
(not the 0x8002/request-only guess). Result: no SEDP type-hash is needed — an
espp service/action even appears in ros2 service list / ros2 action list.

Testing

• Golden wire tests stay byte-identical; a new data_related_sample_identity
  section (all prior bytes unchanged).
• Interop harness 27/27 (dockerised, gated in CI on rtps_embedded/**):
  in-process loopbacks (pub/sub, services, actions, native, typed), byte-level
  codec unit tests, live ros2 service call / ros2 action send_goal both
  directions, and a functional Python demo (5/5) built in-container.
• esp32 builds green: RPC on and off, static/dynamic storage, frag on/off,
  and the example (now in the CI build matrix).
• Wheel smoke test asserts the full RMI/AMI binding surface.

Notes for reviewers

• The engine wire change is limited to emitting/parsing the reply inline-QoS
  correlation; plain pub/sub is byte-identical (proven by golden + interop).
• A latent portability bug surfaced only when building the Python module on macOS
  libc++ (forward-declared context in a unique_ptr member) — fixed; a good
  argument for keeping the wheel build in the review loop.
• The example was renamed main.cpp → rtps_embedded_example.cpp to prepare for
  retiring the old rtps component in a follow-up.
• Deferred (intentional): the "shared L2" template-unification of the ROS and
  native paths — the two already share the pub/sub + service primitives; further
  merging is cosmetic and higher-risk.

Design + milestone detail: components/rtps_embedded/RMI_AMI_DESIGN.md.

🤖 Generated with Claude Code

[github-actions]

✅Static analysis result - no issues found! ✅

[Copilot]

Pull request overview

Adds a full RTPS request/reply + goal-oriented messaging layer to rtps_embedded, including ROS 2 (rmw_fastrtps) interoperable services (RMI) and actions (AMI), plus a lean native espp↔espp protocol, with typed CDR-based wrappers, Python bindings, documentation, and an expanded interop/test harness.

Changes:

• Add ROS 2-compatible service/action primitives (incl. related_sample_identity inline-QoS support) and native RPC/action protocol support.
• Introduce typed wrappers (rtps_service.hpp, rtps_action.hpp) and extensive host + docker interop tests (goldens, naming, codec, loopbacks).
• Add Python bindings + demos/smoke tests, docs for the protocol/API, and build/CI integration (Kconfig toggle + workflow updates).

Reviewed changes

Copilot reviewed 58 out of 58 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
python/rtps_rpc_demo.py	Python end-to-end demo/self-test of ROS2 + native services/actions.
python/rtps_bindings_smoke.py	Wheel smoke test to validate RTPS RPC/action bindings are present.
pyproject.toml	Update cibuildwheel test-command to run the binding surface smoke test.
pc/tests/rtps_typed_rpc_loopback.cpp	Host typed service/action loopback coverage (ROS2 + native).
pc/tests/rtps_service_naming.cpp	Host unit test for ROS 2 service name/type mangling.
pc/tests/rtps_service_loopback.cpp	In-process service loopback exercising correlation and call styles.
pc/tests/rtps_service_interop_server.cpp	Host binary: espp service server for ROS 2 interop matrix.
pc/tests/rtps_service_interop_client.cpp	Host binary: espp service client for ROS 2 interop matrix.
pc/tests/rtps_native_service_loopback.cpp	In-process native service loopback (sync/async/future).
pc/tests/rtps_native_protocol.cpp	Unit test for native protocol framing/header and action framing.
pc/tests/rtps_native_action_loopback.cpp	In-process native action loopback (goal + feedback/result).
pc/tests/rtps_action_types.cpp	Byte-accurate unit test for ROS 2 action envelope codec vs capture.
pc/tests/rtps_action_naming.cpp	Host unit test for ROS 2 action naming/mangling.
pc/tests/rtps_action_loopback.cpp	In-process ROS2-style action loopback (services/topics + UUID correlation).
pc/tests/rtps_action_interop_server.cpp	Host binary: espp action server for ROS 2 interop matrix.
pc/tests/rtps_action_interop_client.cpp	Host binary: espp action client for ROS 2 interop matrix.
pc/tests/rtps_embedded_golden.inc	Add golden bytes for DATA with related_sample_identity inline QoS.
pc/tests/rtps_embedded_golden.cpp	Build and run new golden test case for related_sample_identity.
lib/python_bindings/rtps_bindings.cpp	Extend pybind11 RTPS bindings to include services/actions + native variants.
doc/en/protocols/rtps_rmi_ami.rst	New documentation page describing RMI/AMI APIs and wire formats.
doc/en/protocols/index.rst	Add rtps_rmi_ami to the protocols docs index.
components/rtps_embedded/src/messages/MessageReceiver.cpp	Parse related_sample_identity inline QoS into ReaderCacheChange.
components/rtps_embedded/src/entities/StatelessWriter.cpp	Emit DATA with related_sample_identity when requested by CacheChange.
components/rtps_embedded/src/entities/StatefulWriter.cpp	Emit DATA with related_sample_identity in stateful send paths.
components/rtps_embedded/RMI_AMI_DESIGN.md	Design doc capturing wire formats and milestone plan/validation.
components/rtps_embedded/README.md	Update component README to cover pub/sub + services/actions + native protocol.
components/rtps_embedded/Kconfig	Add RTPS_ENABLE_RPC option to compile out services/actions layer.
components/rtps_embedded/interop/run_interop.sh	Expand interop harness to build/run new RPC/action tests and Python demo.
components/rtps_embedded/interop/ros2_fibonacci_server.py	rclpy action server used by interop harness.
components/rtps_embedded/interop/ros2_add_two_ints_server.py	rclpy service server used by interop harness.
components/rtps_embedded/interop/Dockerfile	Add deps + venv to build/test Python wheel in-container.
components/rtps_embedded/include/rtps/storages/SimpleHistoryCache.hpp	Extend stored CacheChange metadata to carry related_sample_identity.
components/rtps_embedded/include/rtps/storages/HistoryCacheWithDeletion.hpp	Same: carry related_sample_identity through history cache path.
components/rtps_embedded/include/rtps/storages/CacheChange.hpp	Add fields for related_sample_identity + copy/reset behavior.
components/rtps_embedded/include/rtps/rpc/service_naming.hpp	New helper for ROS 2 service topic/type mangling.
components/rtps_embedded/include/rtps/rpc/sample_identity.hpp	New constants/types for related_sample_identity serialization.
components/rtps_embedded/include/rtps/rpc/native_protocol.hpp	New native protocol framing + native action helpers.
components/rtps_embedded/include/rtps/rpc/action_types.hpp	New ROS 2 action envelope wrap/unwrap codec.
components/rtps_embedded/include/rtps/rpc/action_naming.hpp	New helper for ROS 2 action endpoint/type naming.
components/rtps_embedded/include/rtps/messages/MessageFactory.hpp	Add DATA submessage builder that emits related_sample_identity inline QoS.
components/rtps_embedded/include/rtps/entities/Writer.hpp	Add newChangeWithRelatedSampleIdentity and propagate newChange params.
components/rtps_embedded/include/rtps/entities/StatelessWriter.hpp	Update writer interface to accept related_sample_identity metadata.
components/rtps_embedded/include/rtps/entities/StatefulWriter.hpp	Update writer interface to accept related_sample_identity metadata.
components/rtps_embedded/include/rtps/entities/Reader.hpp	Extend ReaderCacheChange to carry optional related_sample_identity.
components/rtps_embedded/include/rtps_service.hpp	New typed service wrapper API over byte-level service endpoints.
components/rtps_embedded/include/rtps_pubsub.hpp	Refactor to share RtpsMessage concept from new rtps_message.hpp.
components/rtps_embedded/include/rtps_participant.hpp	Add service/action/native RPC API surface + compile-out guard macros.
components/rtps_embedded/include/rtps_message.hpp	New shared RtpsMessage concept + protocol selection enum + CDR helpers.
components/rtps_embedded/include/rtps_action.hpp	New typed action wrapper API over byte-level action endpoints.
components/rtps_embedded/example/README.md	New example README describing typed pub/sub + service/action demo on ESP32.
components/rtps_embedded/example/main/rtps_embedded_example.cpp	New ESP32 example demonstrating typed pub/sub + service/action server/client.
components/rtps_embedded/example/main/main.cpp	Remove old example entrypoint (renamed/replaced).
components/rtps_embedded/example/main/Kconfig.projbuild	Add example option for a second self-test participant.
components/rtps_embedded/CMakeLists.txt	Wire RTPS_ENABLE_RPC to RTPS_NO_RPC compile definition.
.github/workflows/rtps_interop.yml	Add concurrency cancellation for interop workflow runs.
.github/workflows/build.yml	Add components/rtps_embedded/example to CI build matrix.

Suppressed comments (1)

lib/python_bindings/rtps_bindings.cpp:328

• The Python ActionClient binding exposes send_goal() but not cancel_goal(...), even though the underlying C++ API supports canceling goals. This makes the Python surface incomplete for ROS 2 actions (cancel is one of the core endpoints).

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[finger563]

Addressed the review comments and static-analysis findings in 8d50039.

Static analysis (cppcheck):

• rtps_action_types.cpp — replaced the hex()+push_back loops with an encap_hex helper using insert() (useStlAlgorithm ×7).
• rtps_service_loopback.cpp — renamed the inner deferred-reply reply → dreply (shadowVariable).
• rtps_participant.cpp — generate_goal_id now uses std::generate (useStlAlgorithm).
• rtps_service.hpp — ServiceClient::call_async takes on_response by const ref (passedByValue).

Review comments:

• assert in the demo — replaced with an explicit check + stderr + non-zero return (asserts are skipped under python -O).
• Missing canceled() binding — exposed ActionGoalHandle.canceled() in Python (+ the typed espp.rtps.GoalHandle). NativeGoalHandle correctly has none (the native protocol has no cancel).
• Missing call_future in Python — added call_future() to the ServiceClient and NativeServiceClient bindings (returns a concurrent.futures.Future) and to the typed espp.rtps.ServiceClient; the demo now exercises it (ros_service_future) and the wheel smoke test asserts the new methods exist. So the Python surface now matches the docs' claim.
• Empty except in espp/rtps.py — commented (a malformed feedback / raising callback must not kill the engine reader thread).
• Action result 8-byte alignment — documented the byte-splice envelope limitation in the user-facing RMI/AMI docs (get_result's result first field must be ≤4-byte aligned; reorder an int64/float64 first field). Goal/feedback payloads and services are unaffected (they follow the 8-aligned UUID / have no such splice). Noted as a v1 limitation to be revisited with full-CDR envelope building.

Gate: interop 27/27 (golden byte-identical, refactored action_types, service_loopback, python demo 6/6 incl. call_future); esp32 build green; wheel smoke test green.

[Copilot]

Pull request overview

Copilot reviewed 58 out of 58 changed files in this pull request and generated 6 comments.

Suppressed comments (6)

components/rtps_embedded/src/rtps_participant.cpp:763

• A successfully accepted cancellation is returned as ERROR_NONE with an empty goals_canceling list. ROS clients therefore receive no indication that this goal entered cancellation, while unknown or rejected goals receive the same apparent success. Serialize the accepted GoalInfo in goals_canceling and return the appropriate CancelGoal error code for other cases.
  components/rtps_embedded/include/rtps_action.hpp:58
• The typed server can observe is_canceling(), but its handle exposes only succeed() and abort(). It cannot terminate the goal as CANCELED, even though the byte-level handle supports canceled(), so cancellation cannot be completed through the recommended typed API. Add a typed canceled(Result) terminator and wire it to the underlying handle.

  /// Terminate the goal as ABORTED and deliver the result to the client.
  /// \param result The (partial/error) result.
  void abort(const Result &result) const {
    if (abort_) {
      abort_(detail::rtps_serialize<Result>(result));

components/rtps_embedded/include/rtps_action.hpp:237

• Converting the ROS send_goal result to bool discards the generated GoalId, and this wrapper exposes no cancel_goal. Typed clients therefore cannot cancel a goal despite cancellation being part of the action API. Return a typed goal handle/id and provide cancellation through it (with an explicit unsupported result for native mode).

    if (ros_) {
      return ros_->send_goal(goal_bytes, std::move(fb_cb), std::move(res_cb)).has_value();
    }
    if (native_) {
      return native_->send_goal(goal_bytes, std::move(fb_cb),

lib/python_bindings/rtps_bindings.cpp:340

• The Python ActionClient binding omits the existing C++ cancel_goal method, so Python clients cannot request cancellation at all. Expose cancellation and validate/convert the 16-byte goal ID returned by send_goal.
  python/rtps_bindings_smoke.py:29
• This smoke test checks only the nested byte-level handles. The new typed surface in espp.rtps (ServiceServer, typed clients, ActionServer, and GoalHandle) can be absent or broken while the wheel still passes. Import espp.rtps and assert those classes and their public methods too.
  pc/tests/rtps_service_loopback.cpp:157
• The deferred-call diagnostic reads the earlier synchronous reply, so it always prints that result instead of the deferred response being tested. This can report 42 even when dreply contains a different value.

[finger563]

Addressed the second review batch in c374237 — these were all async-RPC lifetime/leak issues, so I reworked the teardown and resource ownership coherently. Full gate is green (golden byte-for-byte unchanged, interop 27/27, native/self loopbacks + python_rpc_demo incl. stop() all pass).

rtps_participant.cpp:351 — deferred responder use-after-free. Added a shared Liveness{mutex, alive} token. ServiceResponder::reply() now writes only while holding that mutex with alive == true; stop() flips alive = false under the same mutex before the domain/writers are destroyed. A reply that races shutdown either completes against a still-valid writer or observes !alive and drops — never a freed writer. The token is a shared_ptr, so it outlives the participant.

rtps_participant.cpp:1337 — detached goal threads outliving stop(). Execute workers are now owned (not detached), stored per server context, reaped as they finish, and joined in stop() before the domain is torn down. stop() became a 4-phase teardown so the join happens without mutex_ held (a worker's final publish()/reply must be able to take mutex_/live_ and run to a no-op so it can exit). Accepted goals get a cancel signal first, so cooperative callbacks wind down — execute callbacks are expected to be finite / is_canceling()-aware, which I documented.

rtps_participant.cpp:696 — goals↔server cycle + unbounded map. The goal State now holds a weak_ptr back to the server and copies its feedback/status topic names (so publishing needs no server lock). Goals are erased from the server map once their result is delivered (either get_result answering a finished goal, or terminate() fulfilling a waiting get_result), bounding growth. A still-running worker keeps the State alive via its own reference, so the erase is safe.

rtps_participant.cpp:1213 — native action pre-registration race. The native client now buffers feedback/result that arrives for an unknown handle (bounded at 64 messages) and replays it when send_goal's reply installs the goal. A fast native action that publishes before its goal reply lands no longer drops early messages.

MessageFactory.hpp:180 — RPC + fragmentation loses correlation. A related_sample_identity DATA carries 60 bytes of inline QoS and cannot be fragmented (DATA_FRAG has no inline QoS → correlation lost → timeout). Added MAX_UNFRAGMENTED_RPC_PAYLOAD = MAX_UNFRAGMENTED_PAYLOAD − 60; the facade now rejects oversized service request/reply, and both StatefulWriter/StatelessWriter refuse to fragment an RSI change (drop rather than emit an uncorrelated fragment). Plain (non-RSI) samples still fragment normally — cross_process_frag_200k still passes.

rtps_embedded_example.cpp:206 — client demo leak with no peer. The demo now caps the action to one in-flight goal (an atomic flag reset on result), so an absent peer can no longer leak a pending goal per 5 s tick. The service call() already self-cleans on its 1 s timeout.

[Copilot]

Pull request overview

Copilot reviewed 59 out of 59 changed files in this pull request and generated no new comments.

Suppressed comments (11)

components/rtps_embedded/src/rtps_participant.cpp:1600

• Shutdown requests cancellation for ROS goals, but this native branch immediately joins workers without setting their cancel_requested flags. A cooperative native execute callback that loops until is_canceling() (as the new examples recommend) can therefore make stop() block until natural completion or forever. Mark every live native goal canceling before joining, matching the ROS branch.
  components/rtps_embedded/src/rtps_participant.cpp:862
• This always returns ERROR_NONE with an empty goals_canceling sequence, even when the goal is unknown or on_cancel rejects it. ROS 2 clients use both fields to determine whether cancellation was accepted, so they are told success while the goal keeps running. Encode the appropriate rejection/unknown/terminated return code, and include the accepted goal's GoalInfo when cancellation is accepted.
  components/rtps_embedded/src/rtps_participant.cpp:1524
• Every client subscribes to the shared native feedback topic, so feedback for other clients' goal handles enters this “early” buffer and is never removed. After 64 such messages, the client drops its own legitimately early result/feedback, potentially leaving a fast goal permanently incomplete. Correlate feedback to the requesting client, or bound entries by outstanding goal requests with expiry/eviction rather than retaining every unknown handle.
  components/rtps_embedded/include/rtps/rpc/action_types.hpp:150
• Splicing the nested result body here at CDR offset 4 silently produces invalid ROS 2 wire data whenever the result's first field requires 8-byte alignment. The generic byte-level and typed action APIs accept those valid message types, so documentation alone does not prevent corruption. Build the result envelope with alignment-aware CDR serialization, or reject/encode the unsupported alignment explicitly at the API boundary.

  if (detail::has_encap(result_cdr)) {
    v.insert(v.end(), result_cdr.begin() + 4, result_cdr.end());
  }

components/rtps_embedded/example/main/rtps_embedded_example.cpp:154

• With the default embedded limits, this example cannot register this action server. The same participant already uses two stateful writers (pub/sub publisher plus service reply), while ActionServer adds five more; config_esp32.hpp permits only NUM_STATEFUL_WRITERS = 5. The later clients bring the total to 11 writers, also exceeding the per-participant cap of 10. The default example therefore returns at the validity check instead of demonstrating RMI/AMI. Select a larger limits profile in the example defaults or reduce/split its endpoints.

  espp::ActionServer<FibGoal, FibSeq, FibSeq> fib_action(

components/rtps_embedded/src/rtps_participant.cpp:619

• The counter only overwrites two UUID bytes, so the stated uniqueness guarantee fails after 65,536 goals when random_device is deterministic or weak—exactly the MCU case this counter is meant to cover. Mix all four counter bytes into the UUID.

This issue also appears in the following locations of the same file:

• line 862
• line 1521
• line 1597
  components/rtps_embedded/Kconfig:68

• Disabling this option does not compile out the whole RPC layer as claimed. The engine-side CacheChange/ReaderCacheChange sample-identity fields, inline-QoS parsing, and writer emission paths are unconditional, so pure pub/sub builds still retain RPC code and per-history RAM overhead. Guard those engine additions with RTPS_NO_RPC, or narrow the option's documented savings.

            Compiles in the request/reply (services) and goal (actions) layers of
            the RtpsParticipant facade - both the ROS 2-interoperable
            (add_service_*/add_action_*) and native (add_native_*) variants.
            Disable to drop all of that code (and its std::thread/std::future use)
            when the device only needs pub/sub, saving flash. Pure pub/sub is

lib/python_bindings/rtps_bindings.cpp:285

• The PR advertises Python bindings for the whole service surface, including deferred replies, but this exposes only the synchronous handler API; there is no ServiceResponder or add_service_server_deferred binding. Python users therefore cannot implement slow services without blocking an engine worker. Bind the deferred API (and cover it in the smoke/functional tests), or narrow the stated Python support.
  doc/en/protocols/rtps_rmi_ami.rst:172
• The native implementation has separate goal and cancel services plus feedback, so it uses five DDS endpoints per participant, not roughly three. This undercounts the static pool/RAM requirement and contradicts the implemented topology. Document goal request/reply, cancel request/reply, and feedback explicitly (or fold cancel into the goal service as the design proposes).
  pc/tests/rtps_service_loopback.cpp:157
• This diagnostic reads the earlier synchronous reply instead of the deferred dreply, so it can print a successful value even when the deferred response is missing or wrong. Report the response being tested.
  components/rtps_embedded/interop/run_interop.sh:271
• This functional Python gate is not triggered by changes to the code it tests. .github/workflows/rtps_interop.yml:9-16 omits lib/python_bindings/rtps_bindings.cpp, lib/python_bindings/espp/rtps.py, and python/rtps_rpc_demo.py, so a binding-only regression can skip this round-trip entirely. Add those paths (and the smoke test/packaging inputs as appropriate) to the workflow filter.