refactor(gateway): SSE fault stream reads a neutral event source

scripts/check_no_naked_subscriptions.sh

@@ -45,7 +45,7 @@ ALLOWED_DIRS_PATTERN="(${GATEWAY_ROOT}/src/ros2_common/|${GATEWAY_ROOT}/include/
 # Explicit files that still use the naked APIs pending provider migration.
 # Keep this list SHORT and add a TODO in the file referencing the follow-up.
 ALLOWED_LEGACY_FILES=(
-  "${GATEWAY_ROOT}/src/http/handlers/sse_fault_handler.cpp"        # faults provider follow-up
+  "${GATEWAY_ROOT}/src/ros2/ros2_fault_event_source.cpp"           # faults provider follow-up
   "${GATEWAY_ROOT}/src/trigger_fault_subscriber.cpp"               # faults provider follow-up
   "${GATEWAY_ROOT}/src/ros2/transports/ros2_action_transport.cpp"  # operations provider follow-up
   "${GATEWAY_ROOT}/src/ros2/transports/ros2_log_source.cpp"        # /rosout adapter (LogSource port)

src/ros2_medkit_gateway/CMakeLists.txt

@@ -36,6 +36,7 @@ find_package(ament_index_cpp REQUIRED)
 find_package(action_msgs REQUIRED)
 find_package(lifecycle_msgs REQUIRED)
 find_package(ros2_medkit_msgs REQUIRED)
+find_package(ros2_medkit_fault_detection REQUIRED)
 find_package(ros2_medkit_serialization REQUIRED)
 find_package(SQLite3 REQUIRED)
 
@@ -116,6 +117,7 @@ target_link_libraries(gateway_core
     OpenSSL::SSL
     OpenSSL::Crypto
     tl::expected
+    ros2_medkit_fault_detection::ros2_medkit_fault_detection
     jwt-cpp::jwt-cpp
     SQLite::SQLite3
     ${CMAKE_DL_LIBS}
@@ -166,6 +168,7 @@ add_library(gateway_ros2 STATIC
   src/ros2/transports/ros2_topic_subscription_transport.cpp
   src/ros2/transports/ros2_topic_transport.cpp
   src/ros2/status/ros2_lifecycle_state_reader.cpp
+  src/ros2/ros2_fault_event_source.cpp
   src/ros2/ros2_log_sink.cpp
   src/ros2/ros2_type_schema_provider.cpp
   src/ros2/trigger_topic_subscriber.cpp

src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp

@@ -29,7 +29,6 @@ struct RestConfig {
   bool docs_enabled{true};
   std::string server_host{"localhost"};
   int64_t server_port{8080};
-  std::string fault_events_topic;
   /// Logger name the REST server reports its warnings under.
   std::string logger_name{"rest_server"};
 };

src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/faults/fault_event_source.hpp

@@ -0,0 +1,44 @@
+// Copyright 2026 bburda
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <functional>
+
+#include "ros2_medkit_fault_detection/fault_types.hpp"
+
+namespace ros2_medkit_gateway {
+
+/// Push side of the fault stream: delivers each fault event the fault manager
+/// publishes. Implementations must be thread-safe.
+class FaultEventSource {
+ public:
+  using Callback = std::function<void(const ros2_medkit::fault_detection::FaultEventRecord &)>;
+
+  FaultEventSource() = default;
+  virtual ~FaultEventSource() = default;
+  FaultEventSource(const FaultEventSource &) = delete;
+  FaultEventSource & operator=(const FaultEventSource &) = delete;
+  FaultEventSource(FaultEventSource &&) = delete;
+  FaultEventSource & operator=(FaultEventSource &&) = delete;
+
+  /// Start delivering events to `callback`; replaces any earlier callback.
+  /// The callback may run on any thread.
+  virtual void subscribe(Callback callback) = 0;
+
+  /// Stop delivery. Returns only when no callback is running any more.
+  virtual void unsubscribe() = 0;
+};
+
+}  // namespace ros2_medkit_gateway

src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/faults/fault_json.hpp

@@ -0,0 +1,27 @@
+// Copyright 2026 bburda
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <nlohmann/json.hpp>
+
+#include "ros2_medkit_fault_detection/fault_types.hpp"
+
+namespace ros2_medkit_gateway {
+
+/// Fault JSON object served by /faults and the fault stream. Times are seconds
+/// as doubles; ``last_passed`` is present only when set.
+nlohmann::json fault_record_to_json(const ros2_medkit::fault_detection::FaultRecord & fault);
+
+}  // namespace ros2_medkit_gateway

src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/gateway_services.hpp

@@ -31,6 +31,7 @@ class UpdateManager;
 class LogManager;
 class BulkDataStore;
 class FaultTriggerEngine;
+class FaultEventSource;
 class FreezeFrameSource;
 class TopicDataProvider;
 class TriggerManager;
@@ -79,6 +80,8 @@ class GatewayServices {
   virtual BulkDataStore * bulk_data_store() = 0;
   virtual FaultTriggerEngine * fault_trigger_engine() = 0;
   virtual FreezeFrameSource * freeze_frames() = 0;
+  // May be null: a host without a fault manager has no fault stream.
+  virtual FaultEventSource * fault_events() = 0;
   virtual TopicDataProvider * topic_data_provider() = 0;
   virtual TriggerManager * trigger_manager() = 0;
 };

src/ros2_medkit_gateway/include/ros2_medkit_gateway/gateway_node.hpp

@@ -57,6 +57,7 @@
 #include "ros2_medkit_gateway/core/trigger_store.hpp"
 #include "ros2_medkit_gateway/discovery/discovery_manager.hpp"
 #include "ros2_medkit_gateway/entity_freeze_frame_capture.hpp"
+#include "ros2_medkit_gateway/ros2/ros2_fault_event_source.hpp"
 #include "ros2_medkit_gateway/ros2/ros2_type_schema_provider.hpp"
 #include "ros2_medkit_gateway/ros2/transports/ros2_action_transport.hpp"
 #include "ros2_medkit_gateway/ros2/transports/ros2_fault_service_transport.hpp"
@@ -134,6 +135,9 @@ class GatewayNode : public rclcpp::Node, public GatewayServices {
   FreezeFrameSource * freeze_frames() override {
     return get_entity_freeze_frame_capture();
   }
+  FaultEventSource * fault_events() override {
+    return fault_event_source_.get();
+  }
   TopicDataProvider * topic_data_provider() override {
     return get_topic_data_provider();
   }
@@ -495,6 +499,8 @@ class GatewayNode : public rclcpp::Node, public GatewayServices {
   std::unique_ptr<TriggerFaultSubscriber> trigger_fault_subscriber_;
   // Zero-config freeze-frames for plugin-backed entities (nullptr when disabled)
   std::unique_ptr<EntityFreezeFrameCapture> entity_freeze_frame_capture_;
+  // Destroyed after rest_server_, whose SSE fault handler holds a pointer to it.
+  std::unique_ptr<Ros2FaultEventSource> fault_event_source_;
   // Config-less threshold-rule engine + its dedicated evaluation loop (issue
   // #235). The thread is joined in ~GatewayNode BEFORE plugin/fault shutdown.
   std::unique_ptr<FaultTriggerEngine> fault_trigger_engine_;

src/ros2_medkit_gateway/include/ros2_medkit_gateway/http/handlers/sse_fault_handler.hpp

@@ -28,13 +28,13 @@
 
 #include <nlohmann/json.hpp>
 
-#include "rclcpp/rclcpp.hpp"
+#include "ros2_medkit_fault_detection/fault_types.hpp"
 #include "ros2_medkit_gateway/core/aggregation/peer_fault_relay.hpp"
+#include "ros2_medkit_gateway/core/faults/fault_event_source.hpp"
 #include "ros2_medkit_gateway/core/http/sse_client_tracker.hpp"
 #include "ros2_medkit_gateway/http/handlers/handler_context.hpp"
 #include "ros2_medkit_gateway/http/response_types.hpp"
 #include "ros2_medkit_gateway/http/typed_router.hpp"
-#include "ros2_medkit_msgs/msg/fault_event.hpp"
 
 namespace ros2_medkit_gateway {
 namespace handlers {
@@ -81,19 +81,19 @@ class SSEFaultHandler {
    * @brief Construct SSE fault handler with shared context.
    * @param ctx The shared handler context
    * @param client_tracker Shared SSE client counter (across all SSE handlers)
-   * @param fault_events_topic Fully qualified topic the FaultManager publishes fault events on
+   *
+   * Fault events come from ``ctx.services().fault_events()``; with no source
+   * the stream stays open and silent.
    */
-  SSEFaultHandler(HandlerContext & ctx, std::shared_ptr<SSEClientTracker> client_tracker,
-                  const std::string & fault_events_topic);
+  SSEFaultHandler(HandlerContext & ctx, std::shared_ptr<SSEClientTracker> client_tracker);
   /**
    * @brief Test-only constructor that overrides the keepalive interval.
    * @param ctx The shared handler context
    * @param client_tracker Shared SSE client counter (across all SSE handlers)
    * @param keepalive_interval Must be positive; non-positive values fall back to the 30s default.
-   * @param fault_events_topic Fully qualified topic the FaultManager publishes fault events on
    */
   SSEFaultHandler(HandlerContext & ctx, std::shared_ptr<SSEClientTracker> client_tracker,
-                  std::chrono::milliseconds keepalive_interval, const std::string & fault_events_topic);
+                  std::chrono::milliseconds keepalive_interval);
 
   /// Destructor. Signals shutdown, then blocks until every stream closure has
   /// been destroyed: the per-stream unregister deleter locks queue_mutex_ and
@@ -205,8 +205,8 @@ class SSEFaultHandler {
   /// without it a chain would relay the same event back round the loop.
   std::function<bool(httplib::DataSink &)> make_stream_loop(uint64_t initial_last_event_id, bool relay_peers);
 
-  /// Callback for fault events from ROS 2 topic
-  void on_fault_event(const ros2_medkit_msgs::msg::FaultEvent::ConstSharedPtr & msg);
+  /// Callback for events from the fault event source
+  void on_fault_event(const ros2_medkit::fault_detection::FaultEventRecord & event);
 
   /// Callback for an event relayed from a peer's stream. Runs on that peer
   /// proxy's reader thread.
@@ -224,7 +224,7 @@ class SSEFaultHandler {
   /// flip the entity reported to consumers.
   struct QueuedEvent {
     uint64_t id;
-    ros2_medkit_msgs::msg::FaultEvent event;
+    ros2_medkit::fault_detection::FaultEventRecord event;
     std::optional<EntityContext> entity;
     /// Name of the peer this event was relayed from; empty when this gateway
     /// saw the fault on its own graph. Part of the supersede key, because the
@@ -291,13 +291,13 @@ class SSEFaultHandler {
   /// ros2_runtime_introspection.cpp's renaming rule. Only emits a value when
   /// an App with the resolved id actually exists in the cache - returns
   /// ``std::nullopt`` otherwise so the consumer falls back to discovery.
-  std::optional<EntityContext> resolve_entity_context(const ros2_medkit_msgs::msg::Fault & fault) const;
+  std::optional<EntityContext> resolve_entity_context(const ros2_medkit::fault_detection::FaultRecord & fault) const;
 
   HandlerContext & ctx_;
   std::shared_ptr<SSEClientTracker> client_tracker_;
 
-  /// Subscription to fault events topic
-  rclcpp::Subscription<ros2_medkit_msgs::msg::FaultEvent>::SharedPtr subscription_;
+  /// Fault event source; null on a host without a fault manager.
+  FaultEventSource * fault_source_{nullptr};
 
   /// Relays the peers' fault streams into this one on an aggregating gateway.
   /// Always constructed; it opens nothing when this gateway has no peers.

src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/conversions/fault_msg_conversions.hpp

@@ -16,23 +16,18 @@
 
 #include <nlohmann/json.hpp>
 
+#include "ros2_medkit_fault_detection/fault_types.hpp"
 #include "ros2_medkit_msgs/msg/environment_data.hpp"
 #include "ros2_medkit_msgs/msg/fault.hpp"
+#include "ros2_medkit_msgs/msg/fault_event.hpp"
 
 namespace ros2_medkit_gateway::ros2::conversions {
 
-/// Convert a ros2_medkit_msgs::msg::Fault to JSON.
-///
-/// Produces the flat per-fault representation consumed by `list_faults` items,
-/// the SSE fault-event payload, and the trigger subsystem's notifier change
-/// values. Timestamps become seconds-as-double; severity gains a label.
-///
-/// Lives at the ROS-coupled boundary because three independent call sites
-/// each translate `ros2_medkit_msgs::msg::Fault` directly into JSON: the
-/// fault-service transport adapter, the SSE fault handler (subscribes to the
-/// FaultEvent topic directly), and the trigger fault subscriber. Keeping the
-/// helper free standing avoids code duplication while preserving the neutral
-/// FaultManager contract.
+/// Message to neutral record; times become nanoseconds since the epoch.
+ros2_medkit::fault_detection::FaultRecord from_msg(const ros2_medkit_msgs::msg::Fault & msg);
+ros2_medkit::fault_detection::FaultEventRecord from_msg(const ros2_medkit_msgs::msg::FaultEvent & msg);
+
+/// Fault message to JSON through fault_record_to_json.
 nlohmann::json fault_to_json(const ros2_medkit_msgs::msg::Fault & fault);
 
 /// Convert a ros2_medkit_msgs::msg::EnvironmentData to JSON.

src/ros2_medkit_gateway/include/ros2_medkit_gateway/ros2/ros2_fault_event_source.hpp

@@ -0,0 +1,53 @@
+// Copyright 2026 bburda
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <memory>
+#include <mutex>
+#include <string>
+
+#include "rclcpp/rclcpp.hpp"
+#include "ros2_medkit_gateway/core/faults/fault_event_source.hpp"
+#include "ros2_medkit_msgs/msg/fault_event.hpp"
+
+namespace ros2_medkit_gateway {
+
+/// Fault events from the fault manager's events topic.
+class Ros2FaultEventSource final : public FaultEventSource {
+ public:
+  /// @param node Node the subscription lives on; must outlive this object.
+  /// @param topic Fully qualified events topic.
+  Ros2FaultEventSource(rclcpp::Node & node, std::string topic);
+  ~Ros2FaultEventSource() override;
+  Ros2FaultEventSource(const Ros2FaultEventSource &) = delete;
+  Ros2FaultEventSource & operator=(const Ros2FaultEventSource &) = delete;
+  Ros2FaultEventSource(Ros2FaultEventSource &&) = delete;
+  Ros2FaultEventSource & operator=(Ros2FaultEventSource &&) = delete;
+
+  void subscribe(Callback callback) override;
+  void unsubscribe() override;
+
+ private:
+  void on_message(const ros2_medkit_msgs::msg::FaultEvent & msg);
+
+  rclcpp::Node & node_;
+  std::string topic_;
+  /// Held while a callback runs, so unsubscribe() waits for it.
+  std::mutex callback_mutex_;
+  Callback callback_;
+  rclcpp::Subscription<ros2_medkit_msgs::msg::FaultEvent>::SharedPtr subscription_;
+};
+
+}  // namespace ros2_medkit_gateway

src/ros2_medkit_gateway/package.xml

@@ -28,6 +28,7 @@
   <depend>ros2_medkit_msgs</depend>
   <depend>ros2_medkit_serialization</depend>
   <depend>libsqlite3-dev</depend>
+  <depend>ros2_medkit_fault_detection</depend>
   <depend>rosidl_typesupport_cpp</depend>
   <depend>rosidl_typesupport_introspection_cpp</depend>
   <exec_depend>rosidl_runtime_py</exec_depend>

src/ros2_medkit_gateway/scripts/check_core_purity.sh

@@ -53,6 +53,9 @@ ROS_FREE_SOURCES=(
   "include/ros2_medkit_gateway/core/gateway_services.hpp"
   "include/ros2_medkit_gateway/core/discovery/entity_discovery.hpp"
   "include/ros2_medkit_gateway/core/faults/freeze_frame_source.hpp"
+  "include/ros2_medkit_gateway/core/faults/fault_event_source.hpp"
+  "include/ros2_medkit_gateway/core/faults/fault_json.hpp"
+  "src/core/faults/fault_json.cpp"
   "include/ros2_medkit_gateway/core/discovery/linking_result.hpp"
   "include/ros2_medkit_gateway/core/types/type_schema_provider.hpp"
   "include/ros2_medkit_gateway/http/handlers/handler_context.hpp"
@@ -65,7 +68,7 @@ for src in "${ROS_FREE_SOURCES[@]}"; do
     continue
   fi
   if ! out=$(g++ -std=c++17 -fsyntax-only -Wno-pragma-once-outside-header \
-    -Iinclude -Isrc -Isrc/vendored/tl_expected/include -Isrc/vendored/cpp_httplib "$src" 2>&1); then
+    -Iinclude -I../ros2_medkit_fault_detection/include -Isrc -Isrc/vendored/tl_expected/include -Isrc/vendored/cpp_httplib "$src" 2>&1); then
     echo "DOES NOT COMPILE WITHOUT ROS: $src"
     echo "$out" | grep -E 'error' | head -5
     violations=$((violations + 1))