feat(plugin_context): aggregate peer faults via SOVD service interface

The ROS 2 SOVD service interface (plugin context) returned only local
fault-manager faults, while the HTTP API already fans out to aggregation
peers. Wire list_entity_faults and list_all_faults to the aggregation
manager (fan_out_get) so a gateway with aggregation.enabled exposes peer
faults over the ROS 2 service path too, consistent with the HTTP
/api/v1/faults and /api/v1/<entity>/faults endpoints.

- list_entity_faults: fans out /api/v1/<type>/<id>/faults to peers and merges.
- list_all_faults: fans out /api/v1/faults and merges.
- Aggregation is gated on get_aggregation_manager() != nullptr (no behavior
  change when aggregation is disabled).

Extract entity -> source-FQN resolution and fault scoping into a shared
neutral helper (core/faults/fault_scope) used by both the HTTP fault handlers
and the plugin-context fault path. list_entity_faults now scopes its local
faults to the entity's resolved App-FQN set (APP, COMPONENT, AREA, FUNCTION)
exactly as the HTTP per-entity handler does, instead of returning a bare
fault-manager envelope.

Log partial peer fan-out and document that the ROS 2 service path queries
peers without an Authorization header. Recompute list_all_faults "count"
after merging peer faults so it does not undercount.

Add test_plugin_context_aggregation: no-aggregation baseline, live-mock-peer
fan-out, and local-fault-path coverage with a fake fault transport (scoped
local fault, component/area/function resolution, local+peer merge, and a
peer that fails the fan-out while local faults survive).

Closes #418

Author: bburda

src/ros2_medkit_gateway/CMakeLists.txt

@@ -526,6 +526,10 @@ if(BUILD_TESTING)
   target_link_libraries(test_plugin_notify_integration gateway_ros2)
   medkit_set_test_domain(test_plugin_notify_integration)
 
+  ament_add_gtest(test_plugin_context_aggregation test/test_plugin_context_aggregation.cpp)
+  target_link_libraries(test_plugin_context_aggregation gateway_ros2)
+  medkit_set_test_domain(test_plugin_context_aggregation)
+
   ament_add_gtest(test_auth_manager test/test_auth_manager.cpp)
   target_link_libraries(test_auth_manager gateway_ros2)
 
@@ -1022,6 +1026,7 @@ if(BUILD_TESTING)
       test_stream_proxy
       test_mdns_discovery
       test_network_utils
+      test_plugin_context_aggregation
     )
     foreach(_target ${_test_targets})
       target_compile_options(${_target} PRIVATE --coverage -O0 -g)

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

@@ -0,0 +1,51 @@
+// 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 <set>
+#include <string>
+
+#include "ros2_medkit_gateway/core/models/entity_types.hpp"
+
+namespace ros2_medkit_gateway {
+
+class ThreadSafeEntityCache;
+
+namespace faults {
+
+/// Resolve the set of App effective-FQNs that fall within an entity's scope by
+/// walking the entity cache:
+///   - APP: the app's own effective FQN
+///   - COMPONENT: every hosted app's FQN
+///   - AREA: every app under the area and its (recursive) subareas
+///   - FUNCTION: every app hosted directly or via a hosted component
+/// Returns an empty set for SERVER / UNKNOWN.
+///
+/// Shared by the HTTP fault handlers (`GET /{entity}/faults`) and the ROS 2
+/// plugin-context fault path so both agree on entity -> source-set resolution.
+std::set<std::string> resolve_entity_source_fqns(const ThreadSafeEntityCache & cache, SovdEntityType type,
+                                                 const std::string & entity_id);
+
+/// True when `fault` has at least one reporting source and *every* reporting
+/// source is within `source_fqns` (exact match or a path-boundary prefix).
+/// An empty scope set or an empty/absent source list returns false.
+bool fault_in_source_scope(const nlohmann::json & fault, const std::set<std::string> & source_fqns);
+
+/// Subset of `faults_array` whose faults satisfy `fault_in_source_scope`.
+nlohmann::json filter_faults_by_sources(const nlohmann::json & faults_array, const std::set<std::string> & source_fqns);
+
+}  // namespace faults
+}  // namespace ros2_medkit_gateway

src/ros2_medkit_gateway/src/core/faults/fault_scope.cpp

@@ -0,0 +1,159 @@
+// 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.
+
+#include "ros2_medkit_gateway/core/faults/fault_scope.hpp"
+
+#include <utility>
+#include <vector>
+
+#include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp"
+
+namespace ros2_medkit_gateway {
+namespace faults {
+
+namespace {
+
+void collect_app_fqn(const ThreadSafeEntityCache & cache, const std::string & app_id, std::set<std::string> & out) {
+  auto app = cache.get_app(app_id);
+  if (!app) {
+    return;
+  }
+  auto fqn = app->effective_fqn();
+  if (fqn.empty()) {
+    return;
+  }
+  out.insert(std::move(fqn));
+}
+
+void collect_component_app_fqns(const ThreadSafeEntityCache & cache, const std::string & comp_id,
+                                std::set<std::string> & out) {
+  for (const auto & app_id : cache.get_apps_for_component(comp_id)) {
+    collect_app_fqn(cache, app_id, out);
+  }
+}
+
+void collect_area_app_fqns(const ThreadSafeEntityCache & cache, const std::string & area_id,
+                           std::set<std::string> & out) {
+  // BFS over (area, subareas...) so a top-level area whose components live in
+  // nested subareas still resolves to the union of every descendant's apps.
+  // Without the recursion, e.g. `/areas/powertrain/...` returns an empty set
+  // when components are attached to `engine` (subarea of `powertrain`).
+  std::vector<std::string> pending = {area_id};
+  std::set<std::string> visited;
+  while (!pending.empty()) {
+    auto current = std::move(pending.back());
+    pending.pop_back();
+    if (!visited.insert(current).second) {
+      continue;
+    }
+    for (const auto & comp_id : cache.get_components_for_area(current)) {
+      collect_component_app_fqns(cache, comp_id, out);
+    }
+    for (const auto & sub_id : cache.get_subareas(current)) {
+      pending.push_back(sub_id);
+    }
+  }
+}
+
+void collect_function_app_fqns(const ThreadSafeEntityCache & cache, const std::string & function_id,
+                               std::set<std::string> & out) {
+  // Function.hosts can contain either App IDs or Component IDs; the indexed
+  // lookups in the cache only resolve the App-host case (function_to_apps_),
+  // so we walk the raw `hosts` list and dispatch per host kind ourselves.
+  auto func = cache.get_function(function_id);
+  if (!func) {
+    return;
+  }
+  for (const auto & host_id : func->hosts) {
+    if (cache.get_app(host_id)) {
+      collect_app_fqn(cache, host_id, out);
+    } else if (cache.get_component(host_id)) {
+      collect_component_app_fqns(cache, host_id, out);
+    }
+    // Unknown host - silently skip; it would have been flagged by manifest validation.
+  }
+}
+
+bool source_matches_scope(const std::string & src, const std::set<std::string> & scope_fqns) {
+  for (const auto & fqn : scope_fqns) {
+    if (src == fqn) {
+      return true;
+    }
+    if (src.size() > fqn.size() && src.compare(0, fqn.size(), fqn) == 0 && src[fqn.size()] == '/') {
+      return true;
+    }
+  }
+  return false;
+}
+
+}  // namespace
+
+std::set<std::string> resolve_entity_source_fqns(const ThreadSafeEntityCache & cache, SovdEntityType type,
+                                                 const std::string & entity_id) {
+  std::set<std::string> fqns;
+  switch (type) {
+    case SovdEntityType::APP:
+      collect_app_fqn(cache, entity_id, fqns);
+      break;
+    case SovdEntityType::COMPONENT:
+      collect_component_app_fqns(cache, entity_id, fqns);
+      break;
+    case SovdEntityType::AREA:
+      collect_area_app_fqns(cache, entity_id, fqns);
+      break;
+    case SovdEntityType::FUNCTION:
+      collect_function_app_fqns(cache, entity_id, fqns);
+      break;
+    case SovdEntityType::SERVER:
+    case SovdEntityType::UNKNOWN:
+      break;
+  }
+  return fqns;
+}
+
+bool fault_in_source_scope(const nlohmann::json & fault, const std::set<std::string> & source_fqns) {
+  if (source_fqns.empty()) {
+    return false;
+  }
+  if (!fault.contains("reporting_sources") || !fault["reporting_sources"].is_array()) {
+    return false;
+  }
+  const auto & sources = fault["reporting_sources"];
+  if (sources.empty()) {
+    return false;
+  }
+  for (const auto & src : sources) {
+    if (!src.is_string()) {
+      return false;
+    }
+    if (!source_matches_scope(src.get<std::string>(), source_fqns)) {
+      return false;
+    }
+  }
+  return true;
+}
+
+nlohmann::json filter_faults_by_sources(const nlohmann::json & faults_array,
+                                        const std::set<std::string> & source_fqns) {
+  nlohmann::json filtered = nlohmann::json::array();
+  for (const auto & fault : faults_array) {
+    if (fault_in_source_scope(fault, source_fqns)) {
+      filtered.push_back(fault);
+    }
+  }
+  return filtered;
+}
+
+}  // namespace faults
+}  // namespace ros2_medkit_gateway

src/ros2_medkit_gateway/src/http/handlers/fault_handlers.cpp

@@ -27,6 +27,7 @@
 #include <vector>
 
 #include "ros2_medkit_gateway/aggregation/aggregation_manager.hpp"
+#include "ros2_medkit_gateway/core/faults/fault_scope.hpp"
 #include "ros2_medkit_gateway/core/http/entity_path_utils.hpp"
 #include "ros2_medkit_gateway/core/http/error_codes.hpp"
 #include "ros2_medkit_gateway/core/http/fan_out_helpers.hpp"
@@ -106,39 +107,6 @@ tl::expected<FaultStatusFilter, ErrorInfo> read_fault_status_filter(const http::
 // SOVD-compliant response helpers (legacy free functions kept verbatim)
 // =============================================================================
 
-/// Check if a ROS node FQN falls within the entity's source FQN set.
-///
-/// A node is in scope iff it equals one of the entity's owned FQNs, OR is a
-/// strict path-child of one (i.e., `<owned-fqn>/<...>`). We deliberately do
-/// NOT use raw `rfind(prefix, 0)` because that would let `/ns/node_extra`
-/// pass for an entity owning `/ns/node`. Path boundary is enforced by
-/// requiring the byte after the prefix to be `/`.
-bool source_matches_scope(const std::string & src, const std::set<std::string> & scope_fqns) {
-  for (const auto & fqn : scope_fqns) {
-    if (src == fqn) {
-      return true;
-    }
-    if (src.size() > fqn.size() && src.compare(0, fqn.size(), fqn) == 0 && src[fqn.size()] == '/') {
-      return true;
-    }
-  }
-  return false;
-}
-
-/// Filter a faults JSON array down to faults whose every reporting source is
-/// in the entity's scope. Shares the same all-sources / path-boundary
-/// semantics as `FaultHandlers::fault_in_source_scope` so per-entity
-/// collection routes and per-fault routes agree on what counts as "in scope".
-json filter_faults_by_sources(const json & faults_array, const std::set<std::string> & source_fqns) {
-  json filtered = json::array();
-  for (const auto & fault : faults_array) {
-    if (FaultHandlers::fault_in_source_scope(fault, source_fqns)) {
-      filtered.push_back(fault);
-    }
-  }
-  return filtered;
-}
-
 /// Build SOVD status object from fault status string
 /// Maps ROS 2 medkit status (PREFAILED, PREPASSED, CONFIRMED, HEALED, CLEARED)
 /// to SOVD aggregated status (active, passive, cleared)
@@ -249,25 +217,9 @@ dto::FaultDetailResult wrap_detail_result(json payload) {
 }  // namespace
 
 bool FaultHandlers::fault_in_source_scope(const json & fault, const std::set<std::string> & source_fqns) {
-  if (source_fqns.empty()) {
-    return false;
-  }
-  if (!fault.contains("reporting_sources") || !fault["reporting_sources"].is_array()) {
-    return false;
-  }
-  const auto & sources = fault["reporting_sources"];
-  if (sources.empty()) {
-    return false;
-  }
-  for (const auto & src : sources) {
-    if (!src.is_string()) {
-      return false;
-    }
-    if (!source_matches_scope(src.get<std::string>(), source_fqns)) {
-      return false;
-    }
-  }
-  return true;
+  // Thin wrapper preserving the public static API; the scope logic now lives in
+  // the neutral core helper shared with the ROS 2 plugin-context fault path.
+  return faults::fault_in_source_scope(fault, source_fqns);
 }
 
 // Static method: Build SOVD-compliant fault response from transport-supplied JSON.
@@ -553,7 +505,7 @@ http::Result<dto::FaultListResult> FaultHandlers::list_faults(const http::TypedR
       const auto & cache = ctx_.node()->get_thread_safe_cache();
       auto source_fqns = HandlerContext::resolve_entity_source_fqns(cache, entity_info);
 
-      json filtered_faults = filter_faults_by_sources(result.data["faults"], source_fqns);
+      json filtered_faults = faults::filter_faults_by_sources(result.data["faults"], source_fqns);
       json response = {{"items", filtered_faults}};
 
       // x-medkit extension (typed DTO)
@@ -600,7 +552,7 @@ http::Result<dto::FaultListResult> FaultHandlers::list_faults(const http::TypedR
       const auto & cache = ctx_.node()->get_thread_safe_cache();
       auto app_fqns = HandlerContext::resolve_entity_source_fqns(cache, entity_info);
 
-      json filtered_faults = filter_faults_by_sources(result.data["faults"], app_fqns);
+      json filtered_faults = faults::filter_faults_by_sources(result.data["faults"], app_fqns);
       json response = {{"items", filtered_faults}};
 
       dto::FaultListAggXMedkit xm;
@@ -639,7 +591,7 @@ http::Result<dto::FaultListResult> FaultHandlers::list_faults(const http::TypedR
                      json{{"details", result.error_message}, {entity_info.id_field, entity_id}}));
     }
 
-    json filtered_faults = filter_faults_by_sources(result.data["faults"], app_fqns);
+    json filtered_faults = faults::filter_faults_by_sources(result.data["faults"], app_fqns);
     json response = {{"items", filtered_faults}};
 
     // x-medkit extension for ros2_medkit-specific fields (typed DTO)
@@ -981,7 +933,7 @@ http::Result<http::NoContent> FaultHandlers::clear_all_faults(const http::TypedR
     }
     const auto & cache = ctx_.node()->get_thread_safe_cache();
     auto entity_fqns = HandlerContext::resolve_entity_source_fqns(cache, entity_info);
-    json faults_to_clear = filter_faults_by_sources(result.data["faults"], entity_fqns);
+    json faults_to_clear = faults::filter_faults_by_sources(result.data["faults"], entity_fqns);
 
     // Clear each matching fault. Use `skip_correlation_auto_clear=true` for
     // the same reason as the single-fault DELETE: keep this entity's clear