fix(opcua,#386): per-MI active flag for event MI removal (Copilot review)

Two correctness fixes from Copilot's review of PR #387 in the same
concurrency cluster:

[Blocker] ``remove_event_monitored_item`` was bumping the global
``generation`` counter to invalidate in-flight callbacks for the MI
being removed. That works for the removed MI but ALSO trips the
trampoline staleness check for every peer monitored item registered
against the same subscription, silently dropping their callbacks even
though their server-side MIs are still live. Replace with a per-context
``std::atomic<bool> active`` flag flipped to false before the unique_ptr
is moved out; the trampoline reads it (acquire) ahead of dispatch and
bails for that MI only. Global ``generation`` stays reserved for full
disconnect / ``remove_subscriptions``.

[Defensive] ``setup_event_subscriptions`` was capturing
``const auto & cfg`` from a range-for by reference. Per current C++
semantics the captured reference resolves to the underlying vector
element (not the local reference variable), so the closure stays valid
- the integration test fires three distinct alarms and gets three
distinct fault_codes, confirming each callback uses its own cfg. But
value capture is unambiguous and matches the review feedback.
``AlarmEventConfig`` is a small struct of strings, so the copy is
cheap.

New test ``RemoveEventMonitoredItemUnknownIdDoesNotBumpGeneration``
locks the contract: peer MI staleness must not be triggered by an
unknown-ID single removal.

Local verify: 27/27 test_opcua_client + 27/27 test_alarm_state_machine.

Author: mfaferek93

src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_client.cpp

@@ -38,11 +38,21 @@ static void on_event_trampoline_c(UA_Client * client, UA_UInt32 sub_id, void * s
 /// Heap-owned context passed to the open62541 C event callback. Lifetime is
 /// owned by ``Impl::event_callbacks`` (unique_ptr); the raw pointer handed to C
 /// is valid until ``remove_event_monitored_item`` or ``remove_subscriptions``
-/// erases the entry. The ``generation_snapshot`` lets the trampoline drop
-/// callbacks that fire from a defunct subscription after a reconnect.
+/// erases the entry.
+///
+/// Two staleness guards layered together:
+/// - ``generation_snapshot`` filters callbacks fired from a defunct
+///   subscription after the whole client reconnected (bumped on disconnect
+///   and on ``remove_subscriptions``).
+/// - ``active`` filters callbacks fired from a single monitored item that has
+///   been individually removed via ``remove_event_monitored_item`` while
+///   other items in the same subscription are still live. Per-MI flag,
+///   not the global generation, so a single removal does not invalidate
+///   peer callbacks.
 struct EventCallbackContext {
   OpcuaClient * owner{nullptr};
   uint64_t generation_snapshot{0};
+  std::atomic<bool> active{true};
   uint32_t subscription_id{0};
   uint32_t monitored_item_id{0};  // populated after createEvent returns
   OpcuaClient::EventCallback callback;
@@ -602,6 +612,13 @@ static void on_event_trampoline_c(UA_Client * /*client*/, UA_UInt32 sub_id, void
   if (ctx->generation_snapshot != ctx->owner->current_generation()) {
     return;
   }
+  // Single MI removed via remove_event_monitored_item while peers remain
+  // live - the global generation has not changed, so the peer trampolines
+  // (in the same subscription) must keep firing. Drop only this MI's late
+  // notifications via the per-context flag.
+  if (!ctx->active.load(std::memory_order_acquire)) {
+    return;
+  }
 
   // Copy the UA_Variant fields into open62541pp wrappers. UA_Variant_copy
   // duplicates the underlying buffer, which the opcua::Variant destructor
@@ -743,14 +760,26 @@ bool OpcuaClient::remove_event_monitored_item(uint32_t subscription_id, uint32_t
     if (it == impl_->event_callbacks.end() || it->second->subscription_id != subscription_id) {
       return false;
     }
+    // Flip the per-MI active flag BEFORE moving the unique_ptr so any
+    // in-flight trampoline call observes the cleared flag and bails out
+    // before touching the about-to-be-freed callback. Order:
+    //   1. set active=false (release): trampoline reads it (acquire) and
+    //      returns without invoking ``callback``.
+    //   2. move unique_ptr out of map and erase: ctx still alive in
+    //      ``retired`` until end of function.
+    //   3. delete server-side MI synchronously (mutex serializes against
+    //      the client's notification dispatch).
+    //   4. ``retired`` falls out of scope -> ctx freed.
+    // We deliberately do NOT bump the global ``generation`` counter here:
+    // it would invalidate every peer monitored item registered against the
+    // same subscription, silently dropping their callbacks even though
+    // their MIs are still live on the server. Generation is reserved for
+    // full disconnect / remove_subscriptions.
+    it->second->active.store(false, std::memory_order_release);
     retired = std::move(it->second);
     impl_->event_callbacks.erase(it);
   }
 
-  // Bump generation BEFORE freeing the ctx so any in-flight trampoline call
-  // captured the old generation and will drop its work.
-  impl_->generation.fetch_add(1, std::memory_order_release);
-
   if (impl_->connected) {
     UA_StatusCode status = UA_Client_MonitoredItems_deleteSingle(impl_->client.handle(), subscription_id, mi_id);
     (void)status;  // best effort; on failure the server may already have dropped it

src/ros2_medkit_plugins/ros2_medkit_opcua/src/opcua_poller.cpp

@@ -232,8 +232,16 @@ void OpcuaPoller::setup_event_subscriptions() {
   event_monitored_item_ids_.clear();
 
   for (const auto & cfg : node_map_.event_alarms()) {
-    auto callback = [this, &cfg](const std::vector<opcua::Variant> & values, const opcua::NodeId & source_node,
-                                 const opcua::NodeId & event_type, const opcua::NodeId & condition_id) {
+    // Capture cfg BY VALUE: even though range-for ``const auto & cfg`` binds
+    // to a vector element that outlives the loop, an `&cfg`-by-reference
+    // capture would chain through a local reference variable whose name
+    // goes out of scope after each iteration. Defensible per current C++
+    // semantics (the captured reference resolves to the underlying vector
+    // element), but value capture is unambiguous and matches Copilot's
+    // review feedback. AlarmEventConfig is a small struct of strings, so
+    // copying is cheap.
+    auto callback = [this, cfg](const std::vector<opcua::Variant> & values, const opcua::NodeId & source_node,
+                                const opcua::NodeId & event_type, const opcua::NodeId & condition_id) {
       on_event(cfg, values, source_node, event_type, condition_id);
     };
     uint32_t mi_id =

src/ros2_medkit_plugins/ros2_medkit_opcua/test/test_opcua_client.cpp

@@ -131,6 +131,19 @@ TEST(OpcuaClientTest, RemoveEventMonitoredItemUnknownIdReturnsFalse) {
   EXPECT_FALSE(client.remove_event_monitored_item(/*sub_id=*/1, /*mi_id=*/9999));
 }
 
+TEST(OpcuaClientTest, RemoveEventMonitoredItemUnknownIdDoesNotBumpGeneration) {
+  // remove_event_monitored_item is per-MI cleanup. It must NOT touch the
+  // global generation counter - that would silently invalidate every peer
+  // monitored item's trampoline check and drop their callbacks even though
+  // the underlying server-side MIs are still live. Per-MI staleness is
+  // tracked by EventCallbackContext::active instead. (Copilot review on
+  // PR #387.)
+  OpcuaClient client;
+  uint64_t before = client.current_generation();
+  client.remove_event_monitored_item(/*sub_id=*/1, /*mi_id=*/9999);
+  EXPECT_EQ(client.current_generation(), before);
+}
+
 TEST(OpcuaClientTest, CallMethodWhenDisconnected) {
   OpcuaClient client;
   auto result = client.call_method(opcua::NodeId(0, UA_NS0ID_SERVER), opcua::NodeId(0, 11489), {});