fix(daemon): honour declared SECS-II formats + make service thread-safe

Audit fixes for two real bugs in the gRPC service:

1. Format compliance: to_item() wrote F8/I8 regardless of the variable's
   declared wire format, so values contradicted the S1F11/S1F21 namelists
   (ChamberPressure is F4, WaferCounter U4; the interop trace showed <F8 2.5>
   on the wire). Conversion now targets the declared format — verified
   end-to-end: secsgem-py now receives <F4 2.5> in S6F11.

2. Thread safety: gRPC handler threads called resolve_variable/resolve_event,
   copying live store entries (including Item values) while the io thread
   mutates them. The service now snapshots the immutable name->id/format maps
   at construction (before run_async, per the documented ordering); all writes
   already post to the io thread. Remaining known narrow race (GetControlState
   enum read) documented in DAEMON_ROADMAP.

Also: drop a stale tools/run_interop.sh reference from docker-compose.yml.
Tests: daemon in-process 16/16 (new F4/U4 format assertions), core 459/459,
secsgem-py interop green.

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

apps/equipment_service.hpp

@@ -3,13 +3,19 @@
 // The gRPC Equipment service: translates proto/secsgem/v1 RPCs onto an
 // EquipmentRuntime. Header-only so both secs_gemd and the daemon tests share
 // one definition.
+//
+// Threading: gRPC handlers run on gRPC's own threads while the engine's
+// io thread owns the data model, so handlers must not read the live stores.
+// The constructor therefore snapshots the name->id/format maps (immutable
+// after config load) — construct the service BEFORE run_async(). All writes
+// go through the runtime's posting API.
 
 #include <grpcpp/grpcpp.h>
 
 #include <cstdint>
+#include <map>
 #include <string>
 
-#include "secsgem/gem/name_index.hpp"
 #include "secsgem/gem/runtime.hpp"
 #include "secsgem/secs2/item.hpp"
 #include "secsgem/v1/equipment.grpc.pb.h"
@@ -21,15 +27,44 @@ namespace gem = secsgem::gem;
 namespace s2  = secsgem::secs2;
 namespace pb  = secsgem::v1;
 
-// proto Value -> SECS-II Item. Increment 1 uses a direct mapping; format-aware
+// proto Value -> SECS-II Item, honouring the variable's declared wire format
-// conversion (honouring the variable's declared wire format) is a refinement.
+// (from equipment.yaml) so the host sees the same format in S1F11 namelists
-inline s2::Item to_item(const pb::Value& v) {
+// and in reported values. E.g. real 2.5 -> F4 if the variable is F4.
+inline s2::Item to_item(const pb::Value& v, s2::Format want) {
+  using F = s2::Format;
   switch (v.kind_case()) {
-    case pb::Value::kText:    return s2::Item::ascii(v.text());
+    case pb::Value::kText:
-    case pb::Value::kInteger: return s2::Item::i8(static_cast<int64_t>(v.integer()));
+      return s2::Item::ascii(v.text());
-    case pb::Value::kReal:    return s2::Item::f8(v.real());
+    case pb::Value::kBoolean:
-    case pb::Value::kBoolean: return s2::Item::boolean(v.boolean());
+      return s2::Item::boolean(v.boolean());
-    default:                  return s2::Item::ascii("");
+    case pb::Value::kBinary:
+      return s2::Item::binary({v.binary().begin(), v.binary().end()});
+    case pb::Value::kReal:
+      return want == F::F4 ? s2::Item::f4(static_cast<float>(v.real()))
+                           : s2::Item::f8(v.real());
+    case pb::Value::kInteger: {
+      const int64_t n = v.integer();
+      switch (want) {
+        case F::U1: return s2::Item::u1(static_cast<uint8_t>(n));
+        case F::U2: return s2::Item::u2(static_cast<uint16_t>(n));
+        case F::U4: return s2::Item::u4(static_cast<uint32_t>(n));
+        case F::U8: return s2::Item::u8(static_cast<uint64_t>(n));
+        case F::I1: return s2::Item::i1(static_cast<int8_t>(n));
+        case F::I2: return s2::Item::i2(static_cast<int16_t>(n));
+        case F::I4: return s2::Item::i4(static_cast<int32_t>(n));
+        case F::F4: return s2::Item::f4(static_cast<float>(n));
+        case F::F8: return s2::Item::f8(static_cast<double>(n));
+        case F::Boolean: return s2::Item::boolean(n != 0);
+        default:    return s2::Item::i8(n);
+      }
+    }
+    case pb::Value::kList: {
+      s2::Item::List items;
+      for (const auto& e : v.list().items()) items.push_back(to_item(e, want));
+      return s2::Item::list(std::move(items));
+    }
+    default:
+      return s2::Item::ascii("");
   }
 }
 
@@ -46,18 +81,27 @@ inline pb::ControlState::State to_proto_state(gem::ControlState s) {
 
 class EquipmentService final : public pb::Equipment::Service {
  public:
-  explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) {}
+  // Snapshots the (immutable) name->id/format dictionaries. Construct before
+  // run_async() so the model is read while the io thread isn't running yet.
+  explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) {
+    for (const auto& sv : rt.model().svids.all())
+      vars_.insert({sv.name, {sv.id, sv.value.format()}});
+    for (const auto& dv : rt.model().dvids.all())
+      vars_.insert({dv.name, {dv.id, dv.value.format()}});
+    for (const auto& ev : rt.model().events.all_events())
+      events_.insert({ev.name, ev.id});
+  }
 
   grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req,
                             pb::Ack* resp) override {
     for (const auto& kv : req->values()) {
-      auto vid = gem::resolve_variable(rt_.model(), kv.first);
+      auto it = vars_.find(kv.first);
-      if (!vid) {
+      if (it == vars_.end()) {
         resp->set_code(pb::Ack::PARAMETER_INVALID);
         resp->set_message("no variable named '" + kv.first + "'");
         return grpc::Status::OK;
       }
-      rt_.set_variable(*vid, to_item(kv.second));
+      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
     }
     resp->set_code(pb::Ack::ACCEPT);
     return grpc::Status::OK;
@@ -67,33 +111,42 @@ class EquipmentService final : public pb::Equipment::Service {
                          pb::Ack* resp) override {
     // Optional per-fire variable values, then trigger the collection event.
     for (const auto& kv : req->data()) {
-      auto vid = gem::resolve_variable(rt_.model(), kv.first);
+      auto it = vars_.find(kv.first);
-      if (!vid) {
+      if (it == vars_.end()) {
         resp->set_code(pb::Ack::PARAMETER_INVALID);
         resp->set_message("no variable named '" + kv.first + "'");
         return grpc::Status::OK;
       }
-      rt_.set_variable(*vid, to_item(kv.second));
+      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
     }
-    auto ceid = gem::resolve_event(rt_.model(), req->name());
+    auto ev = events_.find(req->name());
-    if (!ceid) {
+    if (ev == events_.end()) {
       resp->set_code(pb::Ack::PARAMETER_INVALID);
       resp->set_message("no event named '" + req->name() + "'");
       return grpc::Status::OK;
     }
-    rt_.emit_event(*ceid);
+    rt_.emit_event(ev->second);
     resp->set_code(pb::Ack::ACCEPT);
     return grpc::Status::OK;
   }
 
   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).
     resp->set_state(to_proto_state(rt_.control_state()));
     return grpc::Status::OK;
   }
 
  private:
+  struct VarRef {
+    uint32_t vid;
+    s2::Format format;  // declared wire format from equipment.yaml
+  };
+
   gem::EquipmentRuntime& rt_;
+  std::map<std::string, VarRef> vars_;     // SVIDs + DVIDs (SVIDs win on clash)
+  std::map<std::string, uint32_t> events_; // CEID by name
 };
 
 }  // namespace secsgem::daemon

apps/secs_gemd.cpp

@@ -51,9 +51,13 @@ int main(int argc, char** argv) {
     return 1;
   }
   gem::register_default_handlers(*rt);
+
+  // Construct the service before starting the io thread: its constructor
+  // snapshots the name->id/format maps from the model (see the service's
+  // threading note).
+  secsgem::daemon::EquipmentService service(*rt);
   rt->run_async();  // engine + HSMS link on a background thread
 
-  secsgem::daemon::EquipmentService service(*rt);
   grpc::ServerBuilder builder;
   builder.AddListeningPort(grpc_addr, grpc::InsecureServerCredentials());
   builder.RegisterService(&service);

docker-compose.yml

@@ -83,8 +83,7 @@ services:
     networks: [secs]
 
   # secsgem-py running as passive equipment so the C++ active host can
-  # connect to it.  Used by tools/run_interop.sh which then launches
+  # connect to it (drive with `secs_interop_probe --host equipment_py`).
-  # `secs_interop_probe --host equipment_py`.
   equipment_py:
     build: ./interop
     volumes:

tests/test_daemon_service.cpp

@@ -46,16 +46,21 @@ TEST_CASE("Equipment gRPC service over an in-process channel") {
     CHECK(resp.state() == pb::ControlState::HOST_OFFLINE);
   }
 
-  SUBCASE("SetVariables accepts a known name and updates the model") {
+  SUBCASE("SetVariables converts to the variable's declared wire format") {
+    // ChamberPressure is declared F4 and WaferCounter U4 in equipment.yaml;
+    // the daemon must honour those, not write F8/I8, or the host sees values
+    // whose format contradicts the S1F11/S1F21 namelists.
     grpc::ClientContext ctx;
     pb::VariableUpdate req;
     pb::Ack resp;
     (*req.mutable_values())["ChamberPressure"].set_real(2.5);
+    (*req.mutable_values())["WaferCounter"].set_integer(7);
     auto st = stub->SetVariables(&ctx, req, &resp);
     CHECK(st.ok());
     CHECK(resp.code() == pb::Ack::ACCEPT);
     rt.poll();  // drain the posted set_variable onto this thread
-    CHECK(rt.model().dvids.value(101) == s2::Item::f8(2.5));
+    CHECK(rt.model().dvids.value(101) == s2::Item::f4(2.5f));
+    CHECK(rt.model().dvids.value(100) == s2::Item::u4(uint32_t{7}));
   }
 
   SUBCASE("SetVariables rejects an unknown variable name") {