diff --git a/apps/equipment_service.hpp b/apps/equipment_service.hpp index 738c113..2fe31a2 100644 --- a/apps/equipment_service.hpp +++ b/apps/equipment_service.hpp @@ -12,11 +12,14 @@ #include +#include #include #include #include #include +#include #include +#include #include #include #include @@ -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& 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* writer) override { + auto sub = std::make_shared(); + { + std::lock_guard 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 next; + { + std::unique_lock 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 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 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(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& params, + gem::HostCmdAck declarative) { + std::lock_guard 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 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 queue; + }; + std::mutex subs_mu_; + std::condition_variable subs_cv_; + std::vector> subs_; + std::map pending_; // command id -> rcmd (audit) + uint64_t next_command_id_ = 1; // guarded by subs_mu_ }; } // namespace secsgem::daemon diff --git a/docs/DAEMON_ROADMAP.md b/docs/DAEMON_ROADMAP.md index 5bd1d3b..5ecfa2a 100644 --- a/docs/DAEMON_ROADMAP.md +++ b/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 `` — 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 diff --git a/include/secsgem/gem/store/host_commands.hpp b/include/secsgem/gem/store/host_commands.hpp index 4d1867d..4f848f8 100644 --- a/include/secsgem/gem/store/host_commands.hpp +++ b/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 names() const { + std::vector out; + out.reserve(by_rcmd_.size()); + for (const auto& [rcmd, _] : by_rcmd_) out.push_back(rcmd); + return out; + } + std::optional 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& params) const { auto it = by_rcmd_.find(rcmd); diff --git a/interop/daemon_interop.py b/interop/daemon_interop.py index eba230e..7ecf79b 100644 --- a/interop/daemon_interop.py +++ b/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)) diff --git a/proto/secsgem/v1/equipment.proto b/proto/secsgem/v1/equipment.proto index 4363d1e..d1256bc 100644 --- a/proto/secsgem/v1/equipment.proto +++ b/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`. diff --git a/tests/test_daemon_service.cpp b/tests/test_daemon_service.cpp index 2679cb3..85db663 100644 --- a/tests/test_daemon_service.cpp +++ b/tests/test_daemon_service.cpp @@ -3,7 +3,12 @@ #include +#include +#include + #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 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 params) { + auto reply = rt.read_sync([&rt, &rcmd, ¶ms]() { + 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(); +}