feat(gem): multi-observer state-change handlers via HandlerSlot

The single-slot set_*_handler pattern was a structural blocker, hit twice:
the daemon could not observe control-state changes because
register_default_handlers owns the slot, forcing GetControlState to read the
FSM cross-thread (a data race), and blocking WatchHealth and the Subscribe
stream's ControlStateChange variant.

HandlerSlot<Args...> keeps a primary slot with exact legacy semantics
(set_ replaces — one existing test depends on replacement) plus an
append-only observer list (add_) that survives set_ calls. Fire sites are
textually unchanged (operator bool / operator() / assign-from-function).

Applied to ControlStateMachine + ProcessJobStore + ControlJobStore (the
roadmap-critical three; the remaining single-slot classes follow the same
3-line pattern as needed). EquipmentRuntime gains an atomic control-state
mirror registered as an observer — control_state() is now safe from any
thread, retiring the GetControlState race — plus add_control_state_observer
and add_link_observer (selected/closed fan-out), the hooks WatchHealth and
Subscribe need.

Tests: observer ordering, set-replaces-primary-but-observers-survive,
observers-without-primary, PJ-store coexistence, and the runtime scenario
that was previously impossible (mirror + observer + default-handlers set_).
Core 464/464 (2816 assertions), daemon 16/16, live GEM300 demo passes with
single-fire control-state transitions.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/equipment_service.hpp

@@ -132,8 +132,7 @@ class EquipmentService final : public pb::Equipment::Service {
 
   grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*,
                                pb::ControlState* resp) override {
-    // NOTE: reads the FSM state from a gRPC thread — a benign-width race
-    // documented in docs/DAEMON_ROADMAP.md (fix: atomic state mirror).
+    // Thread-safe: control_state() reads the runtime's atomic mirror.
     resp->set_state(to_proto_state(rt_.control_state()));
     return grpc::Status::OK;
   }

include/secsgem/gem/control_state.hpp

@@ -6,6 +6,8 @@
 #include <string>
 #include <vector>
 
+#include "secsgem/gem/handler_slot.hpp"
+
 namespace secsgem::gem {
 
 // E30 §6.2 Control State Model.
@@ -94,7 +96,10 @@ class ControlStateMachine {
   ControlState state() const { return state_; }
   bool online() const { return is_online(state_); }
 
+  // Replaces the primary handler (legacy single-slot semantics).
   void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); }
+  // Appends an observer that survives set_state_change_handler calls.
+  void add_state_change_handler(StateChangeHandler h) { on_change_.add(std::move(h)); }
 
   // Operator actions. Return true if a transition (or self-ack) was found.
   bool operator_online();
@@ -113,7 +118,7 @@ class ControlStateMachine {
 
   ControlTransitionTable table_;
   ControlState state_;
-  StateChangeHandler on_change_;
+  HandlerSlot<ControlState, ControlState, ControlEvent> on_change_;
 };
 
 }  // namespace secsgem::gem

include/secsgem/gem/handler_slot.hpp

@@ -0,0 +1,51 @@
+#pragma once
+
+#include <functional>
+#include <utility>
+#include <vector>
+
+namespace secsgem::gem {
+
+// Replaces the single-slot `std::function on_change_` pattern with a primary
+// slot plus an append-only observer list, so multiple parties can watch one
+// state machine without fighting over a single handler.
+//
+//   - `slot = fn;` / set_*_handler(fn)  -> sets/replaces the PRIMARY handler.
+//     Exactly the legacy single-slot semantics: callers that always owned the
+//     slot (register_default_handlers, apps, tests) keep working unchanged,
+//     including replacement.
+//   - `slot.add(fn)` / add_*_handler(fn) -> appends an OBSERVER. Observers are
+//     immune to later set_ calls — this is what lets the runtime keep an
+//     atomic control-state mirror (and later: WatchHealth, the Subscribe
+//     stream) while default_handlers still owns the primary slot.
+//
+// Invocation order: primary first, then observers in registration order.
+// Copyable (fire sites that snapshot the handler before invoking keep
+// working). Not thread-safe: register on the owning thread before the
+// io_context runs, fire on the io thread — same contract as before.
+template <typename... Args>
+class HandlerSlot {
+ public:
+  using Fn = std::function<void(Args...)>;
+
+  HandlerSlot& operator=(Fn f) {
+    primary_ = std::move(f);
+    return *this;
+  }
+
+  void add(Fn f) { observers_.push_back(std::move(f)); }
+
+  void operator()(Args... args) const {
+    if (primary_) primary_(args...);
+    for (const auto& o : observers_)
+      if (o) o(args...);
+  }
+
+  explicit operator bool() const { return static_cast<bool>(primary_) || !observers_.empty(); }
+
+ private:
+  Fn primary_;
+  std::vector<Fn> observers_;
+};
+
+}  // namespace secsgem::gem

include/secsgem/gem/runtime.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <asio.hpp>
+#include <atomic>
 #include <cstdint>
 #include <functional>
 #include <memory>
@@ -8,6 +9,7 @@
 #include <string>
 #include <thread>
 #include <utility>
+#include <vector>
 
 #include "secsgem/config/loader.hpp"
 #include "secsgem/endpoint.hpp"
@@ -76,7 +78,22 @@ class EquipmentRuntime {
   }
 
   // ---- control state -------------------------------------------------------
-  ControlState control_state() const { return sm_->state(); }
+  // Safe from any thread: reads an atomic mirror updated by a state-machine
+  // observer, so gRPC threads never touch the FSM the io thread owns.
+  ControlState control_state() const { return control_state_cache_.load(std::memory_order_relaxed); }
+
+  // Observe control-state transitions (fires on the io thread). Survives the
+  // primary set_state_change_handler that register_default_handlers installs.
+  // Register before run()/run_async().
+  void add_control_state_observer(ControlStateMachine::StateChangeHandler h) {
+    sm_->add_state_change_handler(std::move(h));
+  }
+
+  // Observe HSMS link state: fires on the io thread with true when a session
+  // reaches SELECTED, false when the connection closes. Register before
+  // run()/run_async(). Foundation for the daemon's WatchHealth stream.
+  using LinkObserver = std::function<void(bool selected)>;
+  void add_link_observer(LinkObserver h) { link_observers_.push_back(std::move(h)); }
 
   // Deliver a primary to the host, or spool it if there's no SELECTED session.
   // Call on the io thread (e.g. from a router handler or a posted emitter).
@@ -100,6 +117,8 @@ class EquipmentRuntime {
   config::EquipmentDescriptor descriptor_;
   std::unique_ptr<Server> server_;
   Router router_;
+  std::atomic<ControlState> control_state_cache_{ControlState::HostOffline};
+  std::vector<LinkObserver> link_observers_;
   std::optional<asio::executor_work_guard<asio::io_context::executor_type>> work_;
   std::thread io_thread_;
 };

include/secsgem/gem/store/control_jobs.hpp

@@ -12,6 +12,7 @@
 #include <utility>
 #include <vector>
 
+#include "secsgem/gem/handler_slot.hpp"
 #include "secsgem/gem/control_job_state.hpp"
 
 namespace secsgem::gem {
@@ -258,7 +259,8 @@ class ControlJobStore {
 
   std::map<std::string, ControlJob> jobs_;
   TransitionTableFactory factory_;
-  StateChangeHandler on_change_;
+  HandlerSlot<const std::string&, ControlJobState, ControlJobState,
+              ControlJobEvent> on_change_;
   bool persistent_ = false;
   std::filesystem::path journal_dir_;
   uint64_t next_seq_ = 0;

include/secsgem/gem/store/process_jobs.hpp

@@ -13,6 +13,7 @@
 #include <utility>
 #include <vector>
 
+#include "secsgem/gem/handler_slot.hpp"
 #include "secsgem/gem/e40_constants.hpp"
 #include "secsgem/gem/process_job_state.hpp"
 #include "secsgem/secs2/codec.hpp"
@@ -72,7 +73,10 @@ class ProcessJobStore {
   ProcessJobStore& operator=(ProcessJobStore&&) = delete;
 
   void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); }
+  // set_ replaces the primary handler (legacy semantics); add_ appends an
+  // observer that survives set_ calls (see handler_slot.hpp).
   void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); }
+  void add_state_change_handler(StateChangeHandler h) { on_change_.add(std::move(h)); }
 
   enum class CreateResult {
     Created,
@@ -512,7 +516,8 @@ class ProcessJobStore {
   std::map<std::string, ProcessJob> jobs_;
   std::vector<std::string> order_;        // queue position (E40 HOQ-aware)
   TransitionTableFactory factory_;
-  StateChangeHandler on_change_;
+  HandlerSlot<const std::string&, ProcessJobState, ProcessJobState,
+              ProcessJobEvent> on_change_;
   bool persistent_ = false;
   std::filesystem::path journal_dir_;
   uint64_t next_seq_ = 0;

src/gem/runtime.cpp

@@ -17,6 +17,14 @@ EquipmentRuntime::EquipmentRuntime(const Config& cfg)
   descriptor_ = config::load_equipment(cfg.equipment_yaml, *model_);
   auto sm_cfg = config::load_control_state(cfg.control_state_yaml);
   sm_ = std::make_shared<ControlStateMachine>(sm_cfg.table, sm_cfg.initial);
+  // Keep an atomic mirror of the control state so control_state() is safe
+  // from any thread. Registered as an observer (add_), so it survives the
+  // primary handler register_default_handlers installs later.
+  control_state_cache_.store(sm_->state(), std::memory_order_relaxed);
+  sm_->add_state_change_handler(
+      [this](ControlState, ControlState to, ControlEvent) {
+        control_state_cache_.store(to, std::memory_order_relaxed);
+      });
 
   auto pj = config::load_process_job_state(cfg.process_job_yaml);
   auto cj = config::load_control_job_state(cfg.control_job_yaml);
@@ -107,10 +115,14 @@ void EquipmentRuntime::clear_alarm(uint32_t alid) {
 void EquipmentRuntime::wire_connection() {
   server_->on_connection([this](std::shared_ptr<Connection> conn) {
     *active_conn_ = conn;
-    conn->set_closed_handler([this](const std::string&) { active_conn_->reset(); });
+    conn->set_closed_handler([this](const std::string&) {
+      active_conn_->reset();
+      for (const auto& o : link_observers_) o(false);
+    });
 
     conn->set_selected_handler([this]() {
       log(std::string("host is online; control=") + control_state_name(sm_->state()));
+      for (const auto& o : link_observers_) o(true);
       if (model_->spool.size() == 0) return;
       asio::post(io_, [this]() {
         auto c = active_conn_->lock();

tests/test_control_state.cpp

@@ -151,3 +151,56 @@ TEST_CASE("custom table: a row that only sets ack, no transition") {
   CHECK(sm.on_host_request_online() == OnlineAck::NotAccept);
   CHECK(sm.state() == ControlState::HostOffline);
 }
+
+// ---- multi-observer (HandlerSlot) ------------------------------------------
+
+TEST_CASE("add_state_change_handler observers fire alongside the primary, in order") {
+  ControlStateMachine sm;
+  std::vector<int> order;
+  Recorder primary, obs1, obs2;
+
+  sm.set_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) {
+    order.push_back(0);
+    primary.handler()(f, t, e);
+  });
+  sm.add_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) {
+    order.push_back(1);
+    obs1.handler()(f, t, e);
+  });
+  sm.add_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) {
+    order.push_back(2);
+    obs2.handler()(f, t, e);
+  });
+
+  CHECK(sm.on_host_request_online() == OnlineAck::Accept);  // 2 transitions
+
+  // All three saw both transitions, primary before observers each time.
+  CHECK(primary.changes.size() == 2);
+  CHECK(obs1.changes.size() == 2);
+  CHECK(obs2.changes.size() == 2);
+  REQUIRE(order.size() == 6);
+  CHECK(order == std::vector<int>{0, 1, 2, 0, 1, 2});
+}
+
+TEST_CASE("set_state_change_handler replaces the primary but observers survive") {
+  ControlStateMachine sm;
+  Recorder old_primary, new_primary, observer;
+
+  sm.set_state_change_handler(old_primary.handler());
+  sm.add_state_change_handler(observer.handler());
+  sm.set_state_change_handler(new_primary.handler());  // replaces old_primary ONLY
+
+  sm.on_host_request_online();
+
+  CHECK(old_primary.changes.empty());     // replaced
+  CHECK(new_primary.changes.size() == 2); // active primary
+  CHECK(observer.changes.size() == 2);    // survived the re-set
+}
+
+TEST_CASE("observers work with no primary handler installed") {
+  ControlStateMachine sm;
+  Recorder observer;
+  sm.add_state_change_handler(observer.handler());
+  sm.on_host_request_online();
+  CHECK(observer.changes.size() == 2);
+}

tests/test_process_jobs.cpp

@@ -221,3 +221,19 @@ TEST_CASE("Store: set_alert toggles per-PJ alert flag") {
   CHECK(store.get("PJ-1")->alert_enabled == false);
   CHECK_FALSE(store.set_alert("ghost", false));
 }
+
+TEST_CASE("ProcessJobStore: add_state_change_handler observer coexists with primary") {
+  ProcessJobStore store;
+  int primary_calls = 0, observer_calls = 0;
+  store.set_state_change_handler(
+      [&](const std::string&, ProcessJobState, ProcessJobState,
+          ProcessJobEvent) { ++primary_calls; });
+  store.add_state_change_handler(
+      [&](const std::string&, ProcessJobState, ProcessJobState,
+          ProcessJobEvent) { ++observer_calls; });
+
+  CHECK(store.create("PJ-OBS", "RECIPE-A", {"lot1"}) ==
+        ProcessJobStore::CreateResult::Created);  // NoState -> Queued fires both
+  CHECK(primary_calls == 1);
+  CHECK(observer_calls == 1);
+}

tests/test_runtime.cpp

@@ -64,3 +64,26 @@ TEST_CASE("EquipmentRuntime.on_command registers the behaviour hook on the model
   CHECK(ran);
   CHECK(res.ack == gem::HostCmdAck::Accept);
 }
+
+TEST_CASE("EquipmentRuntime control-state mirror tracks transitions; observers coexist with default handlers") {
+  gem::EquipmentRuntime rt(test_config());
+
+  int observed = 0;
+  rt.add_control_state_observer(
+      [&](gem::ControlState, gem::ControlState, gem::ControlEvent) { ++observed; });
+
+  // Simulate what register_default_handlers does: claim the PRIMARY slot
+  // after the runtime (mirror) and the observer are already registered.
+  rt.control().set_state_change_handler(
+      [](gem::ControlState, gem::ControlState, gem::ControlEvent) {});
+
+  CHECK(rt.control_state() == gem::ControlState::HostOffline);  // mirror initial
+
+  rt.control().on_host_request_online();  // HostOffline -> ... -> OnlineRemote
+
+  // The atomic mirror followed the FSM, and the added observer survived the
+  // primary set_ call — the exact scenario that used to be impossible.
+  CHECK(rt.control_state() == gem::ControlState::OnlineRemote);
+  CHECK(rt.control_state() == rt.control().state());
+  CHECK(observed == 2);  // two chained transitions
+}