Correct provisional action-fault attribution via supersedes_source_id

docs/api/messages.rst

@@ -179,13 +179,21 @@ Report a fault event to the FaultManager.
    uint8 severity      # Fault.SEVERITY_* constant (for FAILED events)
    string description  # Human-readable description
    string source_id    # Fully qualified node name (e.g., "/powertrain/temp_sensor")
+   string supersedes_source_id  # Optional: a prior source_id to drop before source_id is added
 
 **Response:**
 
 .. code-block:: text
 
    bool accepted       # True if event was accepted
 
+Setting ``supersedes_source_id`` drops that source from the fault's
+``reporting_sources`` before ``source_id`` is added, letting a reporter correct a
+provisional attribution (for example an action bridge that first reports under the
+action interface name, then re-reports under the resolved server node FQN once DDS
+discovery settles). Empty (default) leaves the append-only behavior unchanged. Only
+meaningful for a FAILED event on an existing fault; a self-supersede is a no-op.
+
 **Example Usage:**
 
 .. code-block:: cpp

src/ros2_medkit_action_status_bridge/CHANGELOG.rst

@@ -2,6 +2,11 @@
 Changelog for package ros2_medkit_action_status_bridge
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+Forthcoming
+-----------
+* A fault first raised before DDS discovery resolves the action server node is now corrected once the FQN becomes known, instead of staying stuck under the provisional action-interface source. The bridge records the provisional source, and on the next rescan re-reports the already-delivered fault under the resolved server FQN with ``ReportFault.supersedes_source_id`` set, so the FaultManager drops the provisional source and the fault resolves to the server node's entity under the strict-AND per-entity ``/faults`` scope filter. This refines the #466 "source fixed at first report" behavior: the source is still fixed for a fault whose server was already known, but a provisional attribution made before discovery is corrected rather than left permanently mis-scoped. A healed or not-yet-delivered fault is left untouched (`#467 <https://github.com/selfpatch/ros2_medkit/issues/467>`_)
+* Contributors: @bburda
+
 0.6.0 (2026-06-22)
 ------------------
 * Initial release: generic action-status bridge. Watches every

src/ros2_medkit_action_status_bridge/CMakeLists.txt

@@ -113,6 +113,17 @@ if(BUILD_TESTING)
   )
   medkit_add_launch_test(test_integration test/test_integration.test.py TIMEOUT 120)
 
+  # Deterministic end-to-end regression for #467 provisional re-attribution.
+  # slow_discovery_bridge is the real ActionStatusBridgeNode with only the racy
+  # DDS-FQN resolution forced to "" for the first few rescans, so the real
+  # provisional -> supersede path runs against a real FaultManager.
+  add_executable(slow_discovery_bridge test/slow_discovery_bridge_main.cpp)
+  target_link_libraries(slow_discovery_bridge action_status_bridge_lib)
+  medkit_target_dependencies(slow_discovery_bridge rclcpp)
+  install(TARGETS slow_discovery_bridge DESTINATION lib/${PROJECT_NAME})
+
+  medkit_add_launch_test(test_reattribution_e2e test/test_reattribution_e2e.test.py TIMEOUT 120)
+
   ros2_medkit_relax_vendor_warnings()
 endif()
 

src/ros2_medkit_action_status_bridge/README.md

@@ -32,7 +32,13 @@ generically, by observing the goal-status topic.
   `ACTION_NAVIGATE_TO_POSE_ABORTED`. `source_id` is the action **server's node
   FQN** (resolved from the status topic's publisher, e.g. `/bt_navigator`), so
   the fault associates with that node's SOVD entity; it falls back to the action
-  name only if no publisher is visible.
+  name only if no publisher is visible. When the FQN is not yet discovered at the
+  first report, the fault fires on time under the action-name fallback and is
+  then **re-attributed** to the FQN once discovery resolves - the bridge
+  re-reports it with `ReportFault.supersedes_source_id` set to the fallback, so
+  the FaultManager drops the provisional source and the fault lands on the server
+  node's entity (the fallback source alone would keep it out under the strict-AND
+  per-entity scope filter).
 
 ## Per-action state, not per-goal
 

src/ros2_medkit_action_status_bridge/include/ros2_medkit_action_status_bridge/action_status_bridge_node.hpp

@@ -139,8 +139,17 @@ class ActionStatusBridgeNode : public rclcpp::Node {
   /// Resolve the action server's node FQN from its status-topic publisher, for
   /// use as the fault source_id so faults associate with the gateway's SOVD
   /// entity. Returns "" if no publisher with a discovery-resolved node name is
-  /// visible yet (the caller falls back and re-resolves later).
-  std::string server_fqn_for_action(const std::string & action_name);
+  /// visible yet (the caller falls back and re-resolves later). virtual so a unit
+  /// test can drive the re-attribution path without a live action graph.
+  virtual std::string server_fqn_for_action(const std::string & action_name);
+
+  /// Re-attribute faults first reported under a provisional (action-name) source
+  /// to the resolved server FQN. For each provisional action still reported failed,
+  /// once the FQN resolves it re-reports the fault under the FQN with
+  /// supersedes_source_id set to the provisional source, so the FaultManager swaps
+  /// the source and the fault resolves to the server node's SOVD entity. A no-op
+  /// once every provisional attribution has been corrected. Runs on the rescan tick.
+  void reattribute_provisional();
 
   /// Returns true if this (goal, status) pair was not logged before, marking it
   /// logged. Bounded to avoid unbounded growth. Suppresses duplicate LOG lines
@@ -172,6 +181,17 @@ class ActionStatusBridgeNode : public rclcpp::Node {
   std::map<std::string, std::unique_ptr<ros2_medkit_fault_reporter::FaultReporter>> reporters_;
   std::mutex reporters_mutex_;
 
+  // Actions whose fault was first reported under a provisional (action-name) source
+  // because the server node FQN was not yet discovered. reattribute_provisional()
+  // corrects each to the FQN once discovery resolves. Guarded by reporters_mutex_.
+  struct Provisional {
+    std::string old_source;  // the provisional source_id to supersede
+    // The FQN-sourced reporter, created once the FQN resolves and reused across
+    // rescan ticks until its service client is ready (avoids re-discovery churn).
+    std::unique_ptr<ros2_medkit_fault_reporter::FaultReporter> fqn_reporter;
+  };
+  std::map<std::string, Provisional> provisional_;
+
   // Desired (latest observed) action-level state derived from status messages:
   // the source of truth the bridge tries to make the FaultManager reflect.
   // `canceled` records whether a failure was a CANCELED (vs ABORTED), to pick
@@ -187,6 +207,10 @@ class ActionStatusBridgeNode : public rclcpp::Node {
   // so reconcile_pending() retries it on the next rescan instead of dropping it.
   std::map<std::string, DesiredState> desired_state_;
   std::map<std::string, ActionState> last_reported_state_;
+  // The `canceled` flag as of the report that put each action into kFailed, so a
+  // deferred re-attribution supersedes the fault code that was actually reported
+  // (not a value that flipped afterwards). Guarded by state_mutex_.
+  std::map<std::string, bool> reported_canceled_;
   std::mutex state_mutex_;
 
   // Bounded dedup of logged (goal_id:status) keys (LOG suppression only).
@@ -251,6 +275,22 @@ class ActionStatusBridgeTestAccess {
   /// test assert the reporter is created once and never swapped out.
   const void * reporter_identity(const std::string & action_name);
 
+  /// Run the re-attribution pass (what the rescan tick fires) so a test can drive
+  /// the provisional -> FQN correction without a live rescan timer.
+  void run_reattribute_provisional();
+
+  /// True if the action currently has a provisional (action-name) attribution
+  /// pending correction to the server FQN.
+  bool is_provisional(const std::string & action_name) const;
+
+  /// True if re-attribution has resolved the FQN and created (but not yet
+  /// delivered, e.g. service not ready) the FQN-sourced reporter for this action.
+  bool provisional_has_fqn_reporter(const std::string & action_name) const;
+
+  /// The `canceled` flag captured for the action's currently-reported fault
+  /// (what a deferred re-attribution will supersede under).
+  bool reported_canceled(const std::string & action_name) const;
+
  private:
   ActionStatusBridgeNode * node_;
 };

src/ros2_medkit_action_status_bridge/src/action_status_bridge_node.cpp

@@ -214,6 +214,10 @@ void ActionStatusBridgeNode::rescan_actions() {
   // arming can never strand a pending report. The fast timer is what keeps the
   // FaultManager freeze-frame contemporaneous; this is just belt-and-suspenders.
   reconcile_pending();
+
+  // Correct any faults reported under a provisional (action-name) source now that
+  // discovery has had another rescan period to resolve the server FQN.
+  reattribute_provisional();
 }
 
 void ActionStatusBridgeNode::prune_vanished(const std::map<std::string, std::string> & present_topics) {
@@ -257,11 +261,13 @@ void ActionStatusBridgeNode::prune_vanished(const std::map<std::string, std::str
         }
       }
       reporters_.erase(action_name);
+      provisional_.erase(action_name);
     }
     {
       std::lock_guard<std::mutex> lock(state_mutex_);
       last_reported_state_.erase(action_name);
       desired_state_.erase(action_name);
+      reported_canceled_.erase(action_name);
     }
     RCLCPP_INFO(get_logger(), "Action '%s' vanished; dropped (topic %s)", action_name.c_str(), topic.c_str());
     it = subs_.erase(it);
@@ -358,6 +364,9 @@ ActionStatusBridgeNode::reconcile(const std::string & action_name,
       RCLCPP_INFO(get_logger(), "Action %s -> fault %s", action_name.c_str(), code.c_str());
     }
     last_reported_state_[action_name] = ActionState::kFailed;
+    // Remember which code (aborted vs canceled) was actually raised, so a later
+    // re-attribution supersedes that exact fault, not one derived from a flipped flag.
+    reported_canceled_[action_name] = desired.canceled;
     return ActionState::kFailed;
   }
 
@@ -462,27 +471,117 @@ ros2_medkit_fault_reporter::FaultReporter * ActionStatusBridgeNode::reporter_for
   std::lock_guard<std::mutex> lock(reporters_mutex_);
   auto it = reporters_.find(action_name);
   if (it != reporters_.end()) {
-    return it->second.get();  // created on the first report; source is fixed thereafter
+    return it->second.get();  // created on the first report; source fixed until re-attributed
   }
   // First report for this action: attribute it to the action SERVER's node FQN so
   // the gateway can resolve the fault to its SOVD entity. During DDS discovery the
   // FQN may be unresolved (rcl reports placeholders); fall back to the action name
   // so the fault still fires on time. Reporting the fault on time takes priority
   // over entity attribution when discovery is slow.
   //
-  // The source is NOT re-attributed later: reporting_sources is append-only on the
-  // manager side and the per-entity /faults scope filter is strict-AND, so a
-  // provisional action-name source cannot be swapped for the FQN afterwards.
-  // Correct attribution for the slow-discovery case is a separate concern (it
-  // needs a way to supersede a provisional source).
+  // A fallback (action-name) attribution is recorded as provisional and later
+  // corrected to the FQN by reattribute_provisional(): reporting_sources is
+  // append-only and the per-entity /faults scope filter is strict-AND, so the
+  // provisional source cannot merely be added-to - it is swapped for the FQN via
+  // ReportFault.supersedes_source_id once discovery resolves.
   const std::string fqn = server_fqn_for_action(action_name);
   const std::string & source_id = fqn.empty() ? action_name : fqn;
+  if (fqn.empty()) {
+    provisional_[action_name] = Provisional{action_name, nullptr};
+  }
   auto reporter = std::make_unique<ros2_medkit_fault_reporter::FaultReporter>(this->shared_from_this(), source_id);
   auto * raw = reporter.get();
   reporters_[action_name] = std::move(reporter);
   return raw;
 }
 
+void ActionStatusBridgeNode::reattribute_provisional() {
+  // Snapshot the provisional action names under reporters_mutex_, then release it
+  // so the per-action pass does not iterate provisional_ while it mutates the map
+  // (it erases corrected entries below). Each action's delivered state is read
+  // under state_mutex_; the graph query and re-report then run under reporters_mutex_
+  // - held while we look up, cache the FQN reporter in, and erase from provisional_,
+  // the same lock reporter_for() takes, so the correction stays atomic against
+  // concurrent status callbacks. Neither server_fqn_for_action() nor
+  // FaultReporter::report() re-acquires reporters_mutex_, so there is no deadlock.
+  std::vector<std::string> actions;
+  {
+    std::lock_guard<std::mutex> lock(reporters_mutex_);
+    actions.reserve(provisional_.size());
+    for (const auto & [action_name, prov] : provisional_) {
+      actions.push_back(action_name);
+    }
+  }
+  if (actions.empty()) {
+    return;
+  }
+
+  for (const auto & action_name : actions) {
+    // Re-attribution applies only to a fault that is actually LIVE in the
+    // FaultManager under the provisional source. Read the delivered state:
+    //   - not delivered yet (no last_reported entry): the fault does not exist yet
+    //     (its FAILED is still deferred on service discovery). Keep the entry and
+    //     wait - erasing here would strand the report so it later lands under the
+    //     provisional source with nothing left to correct it.
+    //   - healed (kHealthy): nothing active to correct now, but a re-failure would
+    //     reactivate under the still-provisional source (reporter_for reuses the
+    //     frozen reporter), so keep the entry to correct it then.
+    //   - failed (kFailed): correct it.
+    // reported_canceled_ carries the canceled flag AS OF the report that raised the
+    // fault, so the supersede targets the code actually reported, not a later flip.
+    bool has_reported = false;
+    ActionState reported = ActionState::kUnknown;
+    bool reported_canceled = false;
+    {
+      std::lock_guard<std::mutex> lock(state_mutex_);
+      auto sit = last_reported_state_.find(action_name);
+      has_reported = sit != last_reported_state_.end();
+      if (has_reported) {
+        reported = sit->second;
+      }
+      auto cit = reported_canceled_.find(action_name);
+      reported_canceled = cit != reported_canceled_.end() && cit->second;
+    }
+
+    std::lock_guard<std::mutex> lock(reporters_mutex_);
+    auto pit = provisional_.find(action_name);
+    if (pit == provisional_.end()) {
+      continue;  // raced away (e.g. pruned)
+    }
+    if (!has_reported || reported != ActionState::kFailed) {
+      continue;  // not delivered yet, or healed: keep the entry, nothing to correct now
+    }
+
+    // Resolve the server FQN once, then keep the FQN reporter across ticks until
+    // its service client is ready (recreating it every tick would reset discovery).
+    if (!pit->second.fqn_reporter) {
+      const std::string fqn = server_fqn_for_action(action_name);
+      if (fqn.empty() || fqn == pit->second.old_source) {
+        continue;  // still unresolved; retry on the next rescan
+      }
+      pit->second.fqn_reporter =
+          std::make_unique<ros2_medkit_fault_reporter::FaultReporter>(this->shared_from_this(), fqn);
+    }
+    if (!pit->second.fqn_reporter->is_service_ready()) {
+      continue;  // FaultManager service not discovered by the new client yet; retry
+    }
+
+    // Re-report the same fault (the code actually reported, per reported_canceled_)
+    // under the FQN, superseding the provisional source, so the FaultManager ends
+    // with reporting_sources = {FQN} and the fault resolves to the server node's
+    // SOVD entity under the strict-AND scope filter. The fault is still active, so this
+    // second FAILED report does not inflate occurrence_count (the store counts occurrence
+    // edges, not reports) - it only swaps the source. Happens at most once per fault.
+    const std::string code = fault_code_for(action_name, reported_canceled);
+    const std::string desc = std::string("Action ") + action_name + (reported_canceled ? " canceled" : " aborted");
+    pit->second.fqn_reporter->report(code, aborted_severity_, desc, pit->second.old_source);
+    RCLCPP_INFO(get_logger(), "Re-attributed action '%s' fault '%s' from provisional '%s' to server FQN",
+                action_name.c_str(), code.c_str(), pit->second.old_source.c_str());
+    reporters_[action_name] = std::move(pit->second.fqn_reporter);
+    provisional_.erase(pit);
+  }
+}
+
 std::string ActionStatusBridgeNode::server_fqn_from_endpoint(const std::string & node_name,
                                                              const std::string & node_namespace) {
   // During DDS discovery the participant is known before its node name/namespace
@@ -670,4 +769,25 @@ const void * ActionStatusBridgeTestAccess::reporter_identity(const std::string &
   return node_->reporter_for(action_name);
 }
 
+void ActionStatusBridgeTestAccess::run_reattribute_provisional() {
+  node_->reattribute_provisional();
+}
+
+bool ActionStatusBridgeTestAccess::is_provisional(const std::string & action_name) const {
+  std::lock_guard<std::mutex> lock(node_->reporters_mutex_);
+  return node_->provisional_.count(action_name) > 0;
+}
+
+bool ActionStatusBridgeTestAccess::provisional_has_fqn_reporter(const std::string & action_name) const {
+  std::lock_guard<std::mutex> lock(node_->reporters_mutex_);
+  auto it = node_->provisional_.find(action_name);
+  return it != node_->provisional_.end() && it->second.fqn_reporter != nullptr;
+}
+
+bool ActionStatusBridgeTestAccess::reported_canceled(const std::string & action_name) const {
+  std::lock_guard<std::mutex> lock(node_->state_mutex_);
+  auto it = node_->reported_canceled_.find(action_name);
+  return it != node_->reported_canceled_.end() && it->second;
+}
+
 }  // namespace ros2_medkit_action_status_bridge