feat(daemon): Subscribe command stream + CompleteCommand — the vendor loop closes

The HCACK-4 contract, implemented end to end. For every YAML-declared
command the service registers a forwarding handler (new HostCommandRegistry
names()/spec() accessors): with a subscribed tool client the command is
queued onto the Subscribe stream (id + name + params via from_item) and the
host is answered S2F42 HCACK=4 immediately — never blocking the io thread or
the T3 window; with NO subscriber the command takes its declarative YAML ack
(the honest pre-daemon behaviour). Settled + documented in the proto: v1 is
a firehose with no buffering/replay. CompleteCommand correlates the pending
id (audit; unknown id => PARAMETER_INVALID). Side effects stay suppressed on
HCACK-4 (router applies them only on Accept), so the completion event the
TOOL fires is the host's real signal — exactly E30's intent.

Tests (daemon suite 101 -> 124 assertions): a real S2F41 dispatched through
the full default-handler router ON the io thread under run_async — HCACK 4
with subscriber + params on the stream, declarative Accept without,
CompleteCommand known/unknown, fallback restored after unsubscribe.

Interop (now 20 checks, all green): the complete conformant loop against
the secsgem-py reference host — S2F41 START -> S2F42 HCACK=4 -> tool
receives Command(name=START, id=1) -> CompleteCommand -> FireEvent -> host
receives S6F11.

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

apps/equipment_service.hpp

@@ -12,11 +12,14 @@
 
 #include <grpcpp/grpcpp.h>
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <cstdint>
+#include <deque>
 #include <map>
+#include <memory>
 #include <mutex>
 #include <optional>
 #include <string>
@@ -185,6 +188,23 @@ class EquipmentService final : public pb::Equipment::Service {
         [this](gem::ControlState, gem::ControlState, gem::ControlEvent) {
           bump_health();
         });
+
+    // 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
+    // with AcceptedWillFinishLater; with NO subscriber it returns the
+    // command's declarative ack — i.e. exactly the pre-daemon behaviour, and
+    // honest (never "will finish later" for work nobody will do). Runs on
+    // the io thread; only briefly takes the subscriber mutex.
+    for (const auto& rcmd : rt.model().commands.names()) {
+      const auto declarative =
+          rt.model().commands.spec(rcmd)->ack;  // names() came from the map
+      rt.on_command(rcmd, [this, declarative](
+                              const std::string& cmd,
+                              const std::vector<gem::CommandParameter>& params) {
+        return forward_command(cmd, params, declarative);
+      });
+    }
   }
 
   grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req,
@@ -289,6 +309,64 @@ class EquipmentService final : public pb::Equipment::Service {
     return grpc::Status::OK;
   }
 
+  // Everything the host asks of the tool, as a stream. v1 contract:
+  //   - firehose: every subscriber receives every host request;
+  //   - NO buffering: commands arriving with no subscriber take the
+  //     declarative ack instead (see the constructor's forwarding handler);
+  //   - the S2F42 already went out as HCACK-4 when a command appears here —
+  //     report the real outcome via FireEvent/SetAlarm; CompleteCommand
+  //     correlates/audits.
+  grpc::Status Subscribe(grpc::ServerContext* ctx, const pb::SubscribeRequest* req,
+                         grpc::ServerWriter<pb::HostRequest>* writer) override {
+    auto sub = std::make_shared<Subscriber>();
+    {
+      std::lock_guard<std::mutex> lk(subs_mu_);
+      subs_.push_back(sub);
+    }
+    rt_.log("tool client subscribed" +
+            (req->client().empty() ? "" : " (" + req->client() + ")"));
+    bool alive = true;
+    while (alive && !ctx->IsCancelled()) {
+      std::optional<pb::HostRequest> next;
+      {
+        std::unique_lock<std::mutex> lk(subs_mu_);
+        subs_cv_.wait_for(lk, std::chrono::milliseconds(500),
+                          [&] { return !sub->queue.empty(); });
+        if (!sub->queue.empty()) {
+          next = std::move(sub->queue.front());
+          sub->queue.pop_front();
+        }
+      }
+      if (next) alive = writer->Write(*next);  // write OUTSIDE the lock
+    }
+    {
+      std::lock_guard<std::mutex> lk(subs_mu_);
+      subs_.erase(std::remove(subs_.begin(), subs_.end(), sub), subs_.end());
+    }
+    rt_.log("tool client unsubscribed");
+    return grpc::Status::OK;
+  }
+
+  grpc::Status CompleteCommand(grpc::ServerContext*, const pb::CommandResult* req,
+                               pb::Ack* resp) override {
+    std::string rcmd;
+    {
+      std::lock_guard<std::mutex> lk(subs_mu_);
+      auto it = pending_.find(req->id());
+      if (it == pending_.end()) {
+        resp->set_code(pb::Ack::PARAMETER_INVALID);
+        resp->set_message("unknown command id '" + req->id() + "'");
+        return grpc::Status::OK;
+      }
+      rcmd = it->second;
+      pending_.erase(it);
+    }
+    rt_.log("command '" + rcmd + "' id=" + req->id() + " completed by tool (ack=" +
+            std::to_string(static_cast<int>(req->ack().code())) + ")");
+    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,
@@ -370,6 +448,27 @@ class EquipmentService final : public pb::Equipment::Service {
   }
 
  private:
+  // 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.
+  gem::HostCmdAck forward_command(const std::string& rcmd,
+                                  const std::vector<gem::CommandParameter>& params,
+                                  gem::HostCmdAck declarative) {
+    std::lock_guard<std::mutex> lk(subs_mu_);
+    if (subs_.empty()) return declarative;
+    const std::string id = std::to_string(next_command_id_++);
+    pb::HostRequest hr;
+    auto* cmd = hr.mutable_command();
+    cmd->set_id(id);
+    cmd->set_name(rcmd);
+    for (const auto& p : params) (*cmd->mutable_params())[p.name] = from_item(p.value);
+    pending_.emplace(id, rcmd);
+    for (auto& sub : subs_) sub->queue.push_back(hr);
+    subs_cv_.notify_all();
+    rt_.log("command '" + rcmd + "' id=" + id + " -> tool stream (HCACK=4)");
+    return gem::HostCmdAck::AcceptedWillFinishLater;
+  }
+
   void bump_health() {
     health_version_.fetch_add(1, std::memory_order_relaxed);
     health_cv_.notify_all();
@@ -422,6 +521,17 @@ class EquipmentService final : public pb::Equipment::Service {
   std::atomic<uint64_t> health_version_{0};
   std::mutex health_mu_;
   std::condition_variable health_cv_;
+
+  // Subscribe plumbing: the io-thread forwarding handler queues HostRequests;
+  // each Subscribe call drains its own queue on a gRPC thread.
+  struct Subscriber {
+    std::deque<pb::HostRequest> queue;
+  };
+  std::mutex subs_mu_;
+  std::condition_variable subs_cv_;
+  std::vector<std::shared_ptr<Subscriber>> subs_;
+  std::map<std::string, std::string> pending_;  // command id -> rcmd (audit)
+  uint64_t next_command_id_ = 1;                // guarded by subs_mu_
 };
 
 }  // namespace secsgem::daemon

docs/DAEMON_ROADMAP.md

@@ -28,7 +28,7 @@ host and stays conformant while the tool software restarts/upgrades/crashes.
 | `secs_gemd`: `SetVariables` / `FireEvent` / `GetControlState` / `GetVariables` | ✅ | **format-aware** both directions (declared SECS-II formats on write, `from_item` on read) and thread-safe (snapshot maps + posted writes + `read_sync` reads). In-process gRPC tests incl. run_async production mode (`test_daemon_service.cpp`, 61 assertions) |
 | Daemon interop vs **secsgem-py** reference host | ✅ | `interop/daemon_interop.py` (via `gemd` compose service): gRPC `SetVariables(ChamberPressure=2.5)` + `FireEvent` → host receives `S6F11 CEID 300` carrying `<F4 2.5>` — value *and declared format* flow gRPC→engine→HSMS→host |
 | Daemon interop vs **secs4j** (Java) | ⬜ | mirror the secsgem-py harness against `interop/secs4j` |
-| `Subscribe` host→tool command stream | ⬜ | design settled (HCACK-4, see below); not implemented |
+| `Subscribe` host→tool command stream + `CompleteCommand` | ✅ | HCACK-4 contract implemented + tested in-process AND live vs secsgem-py (full loop: S2F41 → stream → complete → S6F11) |
 | Universal RPC surface complete (vars/events/alarms/control-state/health) | ✅ | Phase A done; daemon tests 101 assertions, interop 15 checks |
 | Python client package (the "beautiful API") | ⬜ | thin wrapper over generated stubs |
 
@@ -72,11 +72,11 @@ your command" from "the work finished". The conformant, non-blocking flow:
    goes out) requires a deferred-reply mechanism in the engine — explicitly a
    v2 refinement, not needed for conformance.
 
-Open sub-decisions to settle while implementing:
-- Per-command routing (subscribe to specific RCMDs?) or one firehose? (v1: firehose.)
-- Reconnect semantics: buffer commands while no subscriber (bounded queue +
-  declarative fallback after timeout) or reject with HCACK 2? Must be decided
-  and TESTED before calling the stream production-ready.
+Sub-decisions (settled 2026-06-10, implemented + tested):
+- v1 is a firehose: every subscriber receives every host request.
+- NO buffering: with no subscriber a command takes its declarative YAML ack
+  and is not replayed on reconnect — never "will finish later" for work no
+  tool will do. Documented in the proto's Subscribe contract.
 
 ## Plan — ordered next steps
 
@@ -147,13 +147,19 @@ debts tax every later phase, and the most valuable tests aren't automated.
 4. ✅ Done per-item above (daemon suite at 101 assertions; interop at 15 checks).
 
 ### Phase B — the command stream (the big one)
-5. ⬜ Implement `Subscribe`/`CompleteCommand` per the design above, including
-   the no-subscriber fallback and bounded buffering. In-process gRPC tests:
-   command arrives on stream; HCACK 4 on the wire; declarative fallback when
-   unsubscribed.
-6. ⬜ Extend `daemon_interop.py`: secsgem-py host sends `S2F41 START` → gRPC
-   tool receives it on the stream → tool fires completion event → host sees
-   `S6F11`. (The full conformant loop against the reference implementation.)
+5. ✅ `Subscribe`/`CompleteCommand` implemented per the HCACK-4 design.
+   Reconnect decision settled and documented in the proto: **no buffering** —
+   a command with no subscriber takes its declarative YAML ack (the honest
+   pre-daemon behaviour) and is not replayed. Firehose fan-out; per-command
+   forwarding handlers registered from the registry (new `names()`/`spec()`
+   accessors); pending-id audit map. In-process tests drive a REAL S2F41
+   through the default-handler router on the io thread: HCACK 4 with a
+   subscriber (params arrive on the stream), declarative Accept without,
+   CompleteCommand known/unknown ids, fallback restored after unsubscribe.
+6. ✅ The full conformant loop runs against secsgem-py live: host `S2F41
+   START` → `S2F42 HCACK=4` → tool receives Command(name=START, id) on the
+   stream → `CompleteCommand` → tool fires the event → host receives `S6F11`.
+   (interop now 20 checks.)
 7. ⬜ Java interop: `secs4j` host variant of the same scenario.
 
 ### Phase C — the beautiful Python client

include/secsgem/gem/store/host_commands.hpp

@@ -71,6 +71,21 @@ class HostCommandRegistry {
   }
   bool has(const std::string& rcmd) const { return by_rcmd_.count(rcmd) > 0; }
   bool has_handler(const std::string& rcmd) const { return handlers_.count(rcmd) > 0; }
+
+  // Enumeration + spec lookup, so a front-end (e.g. the gRPC daemon) can
+  // attach behaviour to every declared command and fall back to the
+  // declarative ack when its tool client isn't connected.
+  std::vector<std::string> names() const {
+    std::vector<std::string> out;
+    out.reserve(by_rcmd_.size());
+    for (const auto& [rcmd, _] : by_rcmd_) out.push_back(rcmd);
+    return out;
+  }
+  std::optional<Spec> spec(const std::string& rcmd) const {
+    auto it = by_rcmd_.find(rcmd);
+    if (it == by_rcmd_.end()) return std::nullopt;
+    return it->second;
+  }
   Result dispatch(const std::string& rcmd,
                   const std::vector<CommandParameter>& params) const {
     auto it = by_rcmd_.find(rcmd);

interop/daemon_interop.py

@@ -201,6 +201,51 @@ def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int,
                   f"body={last_s5f1}")
         stub.ClearAlarm(pb.Alarm(name="chiller_temp_high"))
 
+        # --- the full conformant command loop (HCACK-4 contract) ---
+        # Tool subscribes; host sends S2F41 START; daemon answers HCACK=4 and
+        # forwards the command to the tool; tool completes it and fires the
+        # event; host receives S6F11 — command -> tool -> outcome, end to end.
+        received_cmd = {}
+        cmd_seen = threading.Event()
+
+        def consume_stream():
+            try:
+                for hr in stub.Subscribe(pb.SubscribeRequest(client="interop-tool")):
+                    if hr.HasField("command"):
+                        received_cmd["cmd"] = hr.command
+                        cmd_seen.set()
+                        return
+            except grpc.RpcError:
+                pass  # stream cancelled at teardown
+
+        tool_thread = threading.Thread(target=consume_stream, daemon=True)
+        tool_thread.start()
+        time.sleep(0.5)  # let the subscription register
+
+        ceid300.clear()
+        rsp = client.send_and_waitfor_response(
+            F.SecsS02F41({"RCMD": "START", "PARAMS": []}))
+        body = client.settings.streams_functions.decode(rsp).get()
+        hcack = body.get("HCACK") if isinstance(body, dict) else None
+        check("host got S2F42 HCACK=4 (accepted, will finish later)",
+              int(hcack or -1) == 4, f"HCACK={hcack!r}")
+
+        got_cmd = cmd_seen.wait(timeout=10)
+        check("tool received START on the Subscribe stream", got_cmd)
+        if got_cmd:
+            cmd = received_cmd["cmd"]
+            check("streamed command carries name + id",
+                  cmd.name == "START" and bool(cmd.id),
+                  f"name={cmd.name} id={cmd.id}")
+            ack = stub.CompleteCommand(pb.CommandResult(
+                id=cmd.id, ack=pb.Ack(code=pb.Ack.ACCEPT)))
+            check("CompleteCommand(id) -> ACCEPT",
+                  ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
+            # The tool reports the real outcome as an event; the host sees it.
+            stub.FireEvent(pb.Event(name="ProcessStarted"))
+            check("host received S6F11 after tool completed the command",
+                  ceid300.wait(timeout=10))
+
         # --- operator-panel control state: take the tool offline via gRPC ---
         ack = stub.RequestControlState(pb.ControlStateRequest(
             desired=pb.ControlState.HOST_OFFLINE))

proto/secsgem/v1/equipment.proto

@@ -57,6 +57,15 @@ service Equipment {
 
   // Subscribe to everything the host asks of this equipment. The daemon streams
   // HostRequest messages for as long as the call stays open.
+  //
+  // Delivery contract (v1):
+  //   - firehose: every subscriber receives every host request;
+  //   - NO buffering: a command arriving while no client is subscribed is
+  //     answered with its declarative ack from the equipment config (the
+  //     pre-daemon behaviour) and is NOT replayed on reconnect — the daemon
+  //     never tells the host "will finish later" for work no tool will do;
+  //   - when a Command does arrive here, the host has ALREADY been answered
+  //     with S2F42 HCACK=4; report the real outcome via FireEvent/SetAlarm.
   rpc Subscribe(SubscribeRequest) returns (stream HostRequest);
 
   // Report the outcome of a Command delivered on the stream, quoting its `id`.

tests/test_daemon_service.cpp

@@ -3,7 +3,12 @@
 
 #include <grpcpp/grpcpp.h>
 
+#include <chrono>
+#include <thread>
+
 #include "equipment_service.hpp"
+#include "secsgem/gem/default_handlers.hpp"
+#include "secsgem/gem/messages.hpp"
 
 using namespace secsgem;
 namespace gem = secsgem::gem;
@@ -284,3 +289,88 @@ TEST_CASE("GetVariables round-trip under run_async (production threading mode)")
   server->Shutdown();
   rt.stop();
 }
+
+// The Subscribe command stream, per the HCACK-4 contract: a real S2F41 is
+// dispatched through the full default-handler router ON the io thread, while
+// the gRPC tool client consumes the stream — the daemon's production shape.
+TEST_CASE("Subscribe: S2F41 -> stream -> HCACK 4; declarative fallback without subscriber") {
+  gem::EquipmentRuntime rt(test_config());
+  gem::register_default_handlers(rt);   // the real S2F41/F21/F49 router path
+  dmn::EquipmentService svc(rt);        // registers the forwarding handlers
+  rt.run_async();
+
+  grpc::ServerBuilder builder;
+  builder.RegisterService(&svc);
+  std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
+  REQUIRE(server);
+  auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
+
+  // Dispatch a wire-shaped S2F41 on the io thread (the model's owner) and
+  // hand back the parsed S2F42 ack.
+  auto send_s2f41 = [&](const std::string& rcmd,
+                        std::vector<gem::CommandParameter> params) {
+    auto reply = rt.read_sync([&rt, &rcmd, &params]() {
+      return rt.router().dispatch(gem::s2f41_host_command(rcmd, params));
+    });
+    REQUIRE(reply.has_value());          // read_sync answered
+    REQUIRE(reply->has_value());         // router produced an S2F42
+    auto parsed = gem::parse_s2f42(**reply);
+    REQUIRE(parsed.has_value());
+    return parsed->hcack;
+  };
+
+  // --- no subscriber: declarative ack (START is Accept in equipment.yaml) ---
+  CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);
+
+  // --- subscriber connected: HCACK 4 + the command arrives on the stream ----
+  grpc::ClientContext sub_ctx;
+  pb::SubscribeRequest sreq;
+  sreq.set_client("test-tool");
+  auto reader = stub->Subscribe(&sub_ctx, sreq);
+
+  // Subscription registration races the dispatch below only in this test
+  // (the registry insert happens on the gRPC server thread); give it a beat.
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+  CHECK(send_s2f41("START", {{"PPID", s2::Item::ascii("RECIPE-A")}}) ==
+        gem::HostCmdAck::AcceptedWillFinishLater);
+
+  pb::HostRequest hr;
+  REQUIRE(reader->Read(&hr));
+  REQUIRE(hr.has_command());
+  CHECK(hr.command().name() == "START");
+  CHECK_FALSE(hr.command().id().empty());
+  REQUIRE(hr.command().params().count("PPID") == 1);
+  CHECK(hr.command().params().at("PPID").text() == "RECIPE-A");
+
+  // --- CompleteCommand: known id accepted, unknown rejected ------------------
+  {
+    grpc::ClientContext ctx;
+    pb::CommandResult res;
+    pb::Ack ack;
+    res.set_id(hr.command().id());
+    res.mutable_ack()->set_code(pb::Ack::ACCEPT);
+    REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
+    CHECK(ack.code() == pb::Ack::ACCEPT);
+  }
+  {
+    grpc::ClientContext ctx;
+    pb::CommandResult res;
+    pb::Ack ack;
+    res.set_id("no-such-id");
+    REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
+    CHECK(ack.code() == pb::Ack::PARAMETER_INVALID);
+  }
+
+  // --- unsubscribe: the fallback returns ------------------------------------
+  sub_ctx.TryCancel();
+  pb::HostRequest drain;
+  while (reader->Read(&drain)) {}
+  (void)reader->Finish();  // CANCELLED — expected
+  // The server-side Subscribe loop notices the cancel within its 500ms poll.
+  std::this_thread::sleep_for(std::chrono::milliseconds(700));
+  CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);
+
+  server->Shutdown();
+  rt.stop();
+}