feat(daemon): Phase D — GEM300 in-the-loop (jobs, recipes, EC changes)

Semantics settled and documented: v1 is observe-and-report. The engine keeps
acking S16/S3/S7/S2F15 from its FSM tables — exactly the behaviour both
reference implementations validated — while the tool observes lifecycle
events on the Subscribe stream and reports physical progress back. Gating
stays the documented v2 deferred-reply item.

Engine: two new store observers (HandlerSlot pattern) — RecipeStore fires
(ppid, body) after an add (S7F3 downloads), EquipmentConstantStore fires
(id, value) on ACCEPTED S2F15 writes only. Unit-tested.

Daemon: the service registers PJ/recipe/EC observers (io thread; add_
observers coexist with register_default_handlers' primaries) and fans the
new HostRequest variants out via push_request (fire-and-forget, no-
buffering contract). ProcessJob carries action (Start->START, Resume->
RESUME, Paused->PAUSE, Stopping->STOP, Aborting->ABORT) + recipe + material
bindings read store-side on the io thread. ReportProcessJob maps SETTING_UP
->SetupComplete, COMPLETE->ProcessComplete, ABORTED->AbortComplete via
read_sync; PROCESSING is informational; unknown job => INVALID_OBJECT,
table-rejected transition => CANNOT_DO_NOW. Carriers deferred (CarrierStore
has no observer machinery; ReportCarrier stays UNIMPLEMENTED) — roadmap.

Python client: on_process_job / on_recipe / on_constant_change decorators +
report_job(job_id, state); ProcessJob dataclass exported.

Tests: daemon suite 141 -> 175 assertions — the full in-process loop
(S16F11 create -> tool setup -> S16F5 PJSTART -> stream ProcessJob with
recipe+carriers -> ReportProcessJob(COMPLETE) -> FSM at ProcessComplete),
rejection paths, S7F3 -> ProcessProgram, S2F15 -> ConstantChange with the
configured name. Core 475/3097 (observer units). Live regression: daemon
interop 20 checks + pyclient 13 checks still green against the running
daemon.

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

include/secsgem/daemon/equipment_service.hpp

@@ -189,6 +189,54 @@ class EquipmentService final : public pb::Equipment::Service {
           bump_health();
         });
 
+    for (const auto& ec : rt.model().ecids.all())
+      ecnames_.insert({ec.id, ec.name});
+
+    // GEM300 in-the-loop (Phase D, observe-and-report): forward job
+    // lifecycle, recipe downloads, and accepted EC writes onto the
+    // Subscribe stream. Observers fire on the io thread; push_request only
+    // takes the subscriber mutex briefly. add_ observers coexist with the
+    // primary handlers register_default_handlers installs (HandlerSlot).
+    rt.model().process_jobs.add_state_change_handler(
+        [this](const std::string& prjobid, gem::ProcessJobState,
+               gem::ProcessJobState to, gem::ProcessJobEvent trig) {
+          pb::ProcessJob::Action action;
+          using PS = gem::ProcessJobState;
+          using PE = gem::ProcessJobEvent;
+          if (to == PS::Processing && trig == PE::Start)       action = pb::ProcessJob::START;
+          else if (to == PS::Processing && trig == PE::Resume) action = pb::ProcessJob::RESUME;
+          else if (to == PS::Paused)                           action = pb::ProcessJob::PAUSE;
+          else if (to == PS::Stopping)                         action = pb::ProcessJob::STOP;
+          else if (to == PS::Aborting)                         action = pb::ProcessJob::ABORT;
+          else return;  // SettingUp/complete/etc.: not tool work
+          pb::HostRequest hr;
+          auto* job = hr.mutable_process_job();
+          job->set_job_id(prjobid);
+          job->set_action(action);
+          // Observer runs on the io thread — reading the store is safe.
+          if (const auto* pj = rt_.model().process_jobs.get(prjobid)) {
+            job->set_recipe(pj->ppid);
+            for (const auto& m : pj->mtrloutspec) job->add_carriers(m);
+          }
+          push_request(std::move(hr));
+        });
+    rt.model().recipes.add_added_handler(
+        [this](const std::string& ppid, const std::string& body) {
+          pb::HostRequest hr;
+          hr.mutable_process_program()->set_ppid(ppid);
+          hr.mutable_process_program()->set_body(body);
+          push_request(std::move(hr));
+        });
+    rt.model().ecids.add_changed_handler(
+        [this](uint32_t ecid, const s2::Item& value) {
+          pb::HostRequest hr;
+          auto it = ecnames_.find(ecid);
+          hr.mutable_constant()->set_name(
+              it != ecnames_.end() ? it->second : std::to_string(ecid));
+          *hr.mutable_constant()->mutable_value() = from_item(value);
+          push_request(std::move(hr));
+        });
+
     // Host commands -> the Subscribe stream (the HCACK-4 contract). For every
     // YAML-declared command, install a behaviour handler that pushes the
     // command to subscribed tool clients and answers the host immediately
@@ -367,6 +415,43 @@ class EquipmentService final : public pb::Equipment::Service {
     return grpc::Status::OK;
   }
 
+  // E40 progress reports (Phase D, observe-and-report): the tool advances
+  // the physical work and tells the engine; the engine drives the FSM and
+  // emits the matching S16F9 alert / CEIDs to the host. PROCESSING is
+  // informational (the engine entered it when the host commanded Start).
+  grpc::Status ReportProcessJob(grpc::ServerContext*, const pb::ProcessJobState* req,
+                                pb::Ack* resp) override {
+    const std::string job_id = req->job_id();
+    const auto state = req->state();
+    auto outcome = rt_.read_sync([this, job_id, state]() -> std::optional<bool> {
+      auto& pjs = rt_.model().process_jobs;
+      if (!pjs.has(job_id)) return std::nullopt;
+      switch (state) {
+        case pb::ProcessJobState::SETTING_UP:
+          return pjs.fire_internal(job_id, gem::ProcessJobEvent::SetupComplete);
+        case pb::ProcessJobState::COMPLETE:
+          return pjs.fire_internal(job_id, gem::ProcessJobEvent::ProcessComplete);
+        case pb::ProcessJobState::ABORTED:
+          return pjs.fire_internal(job_id, gem::ProcessJobEvent::AbortComplete);
+        default:
+          return true;  // PROCESSING: informational
+      }
+    });
+    if (!outcome) {
+      resp->set_code(pb::Ack::CANNOT_DO_NOW);
+      resp->set_message("engine io thread did not answer (not running?)");
+    } else if (!*outcome) {
+      resp->set_code(pb::Ack::INVALID_OBJECT);
+      resp->set_message("no process job '" + job_id + "'");
+    } else if (!**outcome) {
+      resp->set_code(pb::Ack::CANNOT_DO_NOW);
+      resp->set_message("E40 table rejects that transition from the current state");
+    } else {
+      resp->set_code(pb::Ack::ACCEPT);
+    }
+    return grpc::Status::OK;
+  }
+
   // Operator-panel control-state transitions (e.g. "offline for maintenance").
   // Fires the operator events on the io thread and reports what the E30 table
   // actually did: ACCEPT iff the equipment landed in the requested state,
@@ -448,6 +533,15 @@ class EquipmentService final : public pb::Equipment::Service {
   }
 
  private:
+  // Fan a HostRequest out to every subscriber (fire-and-forget
+  // notifications: jobs, recipes, EC changes). No subscriber = dropped,
+  // matching the no-buffering contract.
+  void push_request(pb::HostRequest hr) {
+    std::lock_guard<std::mutex> lk(subs_mu_);
+    for (auto& sub : subs_) sub->queue.push_back(hr);
+    if (!subs_.empty()) subs_cv_.notify_all();
+  }
+
   // The command-forwarding handler (runs on the io thread during S2F41/F21/
   // F49 dispatch). Subscriber present -> queue the command + HCACK 4; absent
   // -> the command's declarative YAML ack. Never blocks.
@@ -514,6 +608,7 @@ class EquipmentService final : public pb::Equipment::Service {
   std::map<std::string, VarRef> vars_;     // SVIDs + DVIDs (SVIDs win on clash)
   std::map<std::string, uint32_t> events_; // CEID by name
   std::map<std::string, uint32_t> alarms_; // ALID by name AND stringified id
+  std::map<uint32_t, std::string> ecnames_; // ECID -> name (ConstantChange)
 
   // WatchHealth plumbing: observers (io thread) bump the version and wake the
   // per-stream wait loops (gRPC threads).

include/secsgem/gem/store/equipment_constants.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <cstdint>
+#include <functional>
 #include <map>
 #include <optional>
 #include <stdexcept>
@@ -9,6 +10,7 @@
 #include <variant>
 #include <vector>
 
+#include "secsgem/gem/handler_slot.hpp"
 #include "secsgem/secs2/item.hpp"
 
 namespace secsgem::gem {
@@ -59,10 +61,18 @@ class EquipmentConstantStore {
     if (it == by_id_.end()) return EquipmentAck::Denied_UnknownEcid;
     if (!in_range(it->second, value)) return EquipmentAck::Denied_OutOfRange;
     it->second.value = std::move(value);
+    if (on_changed_) on_changed_(id, it->second.value);
     return EquipmentAck::Accept;
   }
 
+  // Observe ACCEPTED host writes (S2F15): fires after the value is stored.
+  // The tool reacts to process-parameter tuning; rejected writes don't fire.
+  using ChangedHandler = std::function<void(uint32_t, const s2::Item&)>;
+  void add_changed_handler(ChangedHandler h) { on_changed_.add(std::move(h)); }
+
  private:
+  HandlerSlot<uint32_t, const s2::Item&> on_changed_;
+
   // For numeric formats, parse min_str / max_str as integers / doubles and
   // compare against the first value of the array (we don't support setting
   // multi-element ECs in this implementation).

include/secsgem/gem/store/recipes.hpp

@@ -1,12 +1,14 @@
 #pragma once
 
 #include <cstdint>
+#include <functional>
 #include <map>
 #include <optional>
 #include <string>
 #include <utility>
 #include <vector>
 
+#include "secsgem/gem/handler_slot.hpp"
 #include "secsgem/secs2/item.hpp"
 
 namespace secsgem::gem {
@@ -45,8 +47,15 @@ struct FormattedRecipe {
 
 class RecipeStore {
  public:
+  // Observe recipe arrivals (e.g. a host S7F3 download): fires after the
+  // store is updated, with (ppid, body). Multi-observer via HandlerSlot —
+  // the gRPC daemon forwards these onto its Subscribe stream.
+  using AddedHandler = std::function<void(const std::string&, const std::string&)>;
+  void add_added_handler(AddedHandler h) { on_added_.add(std::move(h)); }
+
   void add(std::string ppid, std::string body) {
-    by_ppid_.insert_or_assign(std::move(ppid), std::move(body));
+    auto it = by_ppid_.insert_or_assign(std::move(ppid), std::move(body)).first;
+    if (on_added_) on_added_(it->first, it->second);
   }
   std::optional<std::string> get(const std::string& ppid) const {
     auto it = by_ppid_.find(ppid);
@@ -95,6 +104,7 @@ class RecipeStore {
 
  private:
   std::map<std::string, std::string>     by_ppid_;
+  HandlerSlot<const std::string&, const std::string&> on_added_;
   std::map<std::string, FormattedRecipe> formatted_;
 };