B4: Host-mode integration tests

Five end-to-end tests wire a real HostHandler against a real passive
HSMS Connection over a TCP loopback pair and assert wire-level
behaviour matches expectations:

  - establish_communication + go_remote sequence S1F13 then S1F17
  - send_remote_command produces a wire-correct S2F41 the equipment
    can re-parse with parse_s2f41 and recover CPNAME/CPVAL
  - send_terminal_display round-trips through S10F1/F2
  - E40/E94 create+command sequence (S16F11, S14F9, S16F5)
  - Inbound S5F1 alarm fires the host's alarm observer + auto-acks

Each test uses the existing pump_until / SocketPair harness pattern
from test_hsms_connection.cpp.  The recorder pattern keeps the
equipment-side dispatch table small — every test installs the same
canned reply handler.

This closes Tranche B (host mode).  HostHandler now has the inbound
+ outbound surface secsgem-py's GemHostHandler exposes.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

CMakeLists.txt

@@ -88,6 +88,7 @@ add_executable(secsgem_tests
   tests/test_hsms_connection.cpp
   tests/test_secsi.cpp
   tests/test_secsi_tcp.cpp
+  tests/test_host_handler.cpp
   tests/test_control_state.cpp
   tests/test_communication_state.cpp
   tests/test_data_model.cpp

tests/test_host_handler.cpp

@@ -0,0 +1,286 @@
+// Integration tests for the GEM HostHandler.  Wires a real Active
+// HostHandler vs a real Passive Connection over a TCP loopback pair,
+// drives the host through its workflow + sender APIs, and asserts the
+// equipment side observed the right primaries and the host received the
+// right replies.
+
+#include <doctest/doctest.h>
+
+#include <asio.hpp>
+#include <chrono>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "secsgem/gem/host_handler.hpp"
+#include "secsgem/gem/messages.hpp"
+#include "secsgem/hsms/connection.hpp"
+#include "secsgem/hsms/header.hpp"
+
+using namespace secsgem;
+namespace gem  = secsgem::gem;
+namespace s2   = secsgem::secs2;
+
+namespace {
+
+struct Wired {
+  asio::io_context io;
+  std::shared_ptr<hsms::Connection> host;       // Active
+  std::shared_ptr<hsms::Connection> equipment;  // Passive
+  std::shared_ptr<gem::HostHandler> handler;
+
+  Wired() {
+    asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
+                                        asio::ip::address_v4::loopback(), 0));
+    const auto port = acc.local_endpoint().port();
+
+    asio::ip::tcp::socket eq_sock(io);
+    asio::ip::tcp::socket host_sock(io);
+    bool a_done = false, c_done = false;
+    std::error_code ec_a, ec_c;
+    acc.async_accept(eq_sock,    [&](std::error_code e) { ec_a = e; a_done = true; });
+    host_sock.async_connect({asio::ip::address_v4::loopback(), port},
+                            [&](std::error_code e) { ec_c = e; c_done = true; });
+    while (!(a_done && c_done)) {
+      if (io.stopped()) io.restart();
+      if (io.poll() == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    REQUIRE_FALSE(ec_a);
+    REQUIRE_FALSE(ec_c);
+
+    hsms::Timers timers;
+    timers.linktest = std::chrono::milliseconds(0);
+    timers.t3 = std::chrono::milliseconds(1000);
+
+    host = std::make_shared<hsms::Connection>(
+        std::move(host_sock), hsms::Connection::Mode::Active, 0, timers);
+    equipment = std::make_shared<hsms::Connection>(
+        std::move(eq_sock), hsms::Connection::Mode::Passive, 0, timers);
+    handler = std::make_shared<gem::HostHandler>(host);
+  }
+};
+
+template <typename Pred>
+void pump_until(asio::io_context& io, Pred pred,
+                std::chrono::milliseconds budget = std::chrono::seconds(3)) {
+  const auto deadline = std::chrono::steady_clock::now() + budget;
+  while (!pred()) {
+    if (std::chrono::steady_clock::now() > deadline) FAIL("pump_until budget exceeded");
+    if (io.stopped()) io.restart();
+    if (io.poll() == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+}
+
+// Tracks primaries seen on the equipment side and the canned reply each
+// gets.  Used by every test to capture wire-level behaviour without
+// having to repeat the dispatch boilerplate.
+struct Recorder {
+  std::vector<std::pair<uint8_t, uint8_t>> primaries;  // (stream, function)
+  std::vector<s2::Message> last;
+
+  hsms::Connection::MessageHandler handler() {
+    return [this](const s2::Message& msg) -> std::optional<s2::Message> {
+      primaries.emplace_back(msg.stream, msg.function);
+      last.push_back(msg);
+      // Return a generic-ish reply for each known primary.
+      if (msg.stream == 1 && msg.function == 13)
+        return gem::s1f14_establish_comms_ack(gem::CommAck::Accept, {"EQUIP", "1.0"});
+      if (msg.stream == 1 && msg.function == 17)
+        return gem::s1f18_online_ack(gem::OnlineAck::Accept);
+      if (msg.stream == 1 && msg.function == 15)
+        return gem::s1f16_offline_ack(gem::OfflineAck::Accept);
+      if (msg.stream == 2 && msg.function == 41)
+        return gem::s2f42_host_command_ack(gem::HostCmdAck::Accept, {});
+      if (msg.stream == 10 && msg.function == 1)
+        return gem::s10f2_terminal_display_ack(gem::TerminalAck::Accepted);
+      if (msg.stream == 16 && msg.function == 11)
+        return gem::s16f12_pr_job_create_ack(gem::HostCmdAck::Accept);
+      if (msg.stream == 16 && msg.function == 5)
+        return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::Accept);
+      if (msg.stream == 14 && msg.function == 9)
+        return gem::s14f10_create_control_job_ack("CJ-1", gem::ObjectAck::Success);
+      // Unhandled — leave the host to time out on T3.
+      return std::nullopt;
+    };
+  }
+};
+
+}  // namespace
+
+TEST_CASE("HostHandler: establish_communication + go_remote workflow") {
+  Wired w;
+  Recorder rec;
+  w.equipment->set_message_handler(rec.handler());
+
+  bool selected = false;
+  w.host->set_selected_handler([&] { selected = true; });
+  w.handler->install();
+  w.host->start();
+  w.equipment->start();
+  pump_until(w.io, [&] { return selected; });
+
+  bool s1f14_seen = false, s1f18_seen = false;
+  w.handler->establish_communication([&](std::error_code ec, const s2::Message& m) {
+    REQUIRE_FALSE(ec);
+    CHECK(m.stream == 1);
+    CHECK(m.function == 14);
+    s1f14_seen = true;
+  });
+  pump_until(w.io, [&] { return s1f14_seen; });
+
+  w.handler->go_remote([&](std::error_code ec, const s2::Message& m) {
+    REQUIRE_FALSE(ec);
+    CHECK(m.stream == 1);
+    CHECK(m.function == 18);
+    s1f18_seen = true;
+  });
+  pump_until(w.io, [&] { return s1f18_seen; });
+
+  // Equipment saw both primaries in order.
+  REQUIRE(rec.primaries.size() >= 2);
+  CHECK(rec.primaries[0] == std::make_pair<uint8_t, uint8_t>(1, 13));
+  CHECK(rec.primaries[1] == std::make_pair<uint8_t, uint8_t>(1, 17));
+}
+
+TEST_CASE("HostHandler: send_remote_command routes through S2F41") {
+  Wired w;
+  Recorder rec;
+  w.equipment->set_message_handler(rec.handler());
+  w.handler->install();
+  w.host->start();
+  w.equipment->start();
+  bool selected = false;
+  w.host->set_selected_handler([&] { selected = true; });
+  pump_until(w.io, [&] { return selected; });
+
+  bool acked = false;
+  std::vector<gem::CommandParameter> params{
+      {"LOTID", s2::Item::ascii("LOT-1")},
+      {"PPID",  s2::Item::ascii("RECIPE-A")},
+  };
+  w.handler->send_remote_command("START", params,
+                                 [&](std::error_code ec, const s2::Message& m) {
+                                   REQUIRE_FALSE(ec);
+                                   CHECK(m.stream == 2);
+                                   CHECK(m.function == 42);
+                                   acked = true;
+                                 });
+  pump_until(w.io, [&] { return acked; });
+
+  // Equipment received the right primary with the right parameters.
+  REQUIRE_FALSE(rec.last.empty());
+  const auto& s2f41 = rec.last.back();
+  auto parsed = gem::parse_s2f41(s2f41);
+  REQUIRE(parsed.has_value());
+  CHECK(parsed->rcmd == "START");
+  REQUIRE(parsed->params.size() == 2);
+  CHECK(parsed->params[0].name == "LOTID");
+}
+
+TEST_CASE("HostHandler: terminal display + S10F1 ack round-trip") {
+  Wired w;
+  Recorder rec;
+  w.equipment->set_message_handler(rec.handler());
+  w.handler->install();
+  w.host->start();
+  w.equipment->start();
+  bool selected = false;
+  w.host->set_selected_handler([&] { selected = true; });
+  pump_until(w.io, [&] { return selected; });
+
+  bool done = false;
+  w.handler->send_terminal_display(7, "hello operator",
+                                   [&](std::error_code ec, const s2::Message& m) {
+                                     REQUIRE_FALSE(ec);
+                                     CHECK(m.stream == 10);
+                                     CHECK(m.function == 2);
+                                     done = true;
+                                   });
+  pump_until(w.io, [&] { return done; });
+
+  REQUIRE_FALSE(rec.last.empty());
+  auto td = gem::parse_s10f1(rec.last.back());
+  REQUIRE(td.has_value());
+  CHECK(td->tid == 7);
+  CHECK(td->text == "hello operator");
+}
+
+TEST_CASE("HostHandler: E40/E94 job creation senders produce correct wire") {
+  Wired w;
+  Recorder rec;
+  w.equipment->set_message_handler(rec.handler());
+  w.handler->install();
+  w.host->start();
+  w.equipment->start();
+  bool selected = false;
+  w.host->set_selected_handler([&] { selected = true; });
+  pump_until(w.io, [&] { return selected; });
+
+  // Create a PJ.
+  bool pj_ack = false;
+  w.handler->send_create_process_job(
+      "PJ-1", "RECIPE-A", {"W1", "W2"},
+      [&](std::error_code ec, const s2::Message& m) {
+        REQUIRE_FALSE(ec);
+        CHECK(m.stream == 16);
+        CHECK(m.function == 12);
+        pj_ack = true;
+      });
+  pump_until(w.io, [&] { return pj_ack; });
+
+  // Create a CJ that references it.
+  bool cj_ack = false;
+  w.handler->send_create_control_job(
+      "CJ-1", {"PJ-1"},
+      [&](std::error_code ec, const s2::Message& m) {
+        REQUIRE_FALSE(ec);
+        CHECK(m.stream == 14);
+        CHECK(m.function == 10);
+        cj_ack = true;
+      });
+  pump_until(w.io, [&] { return cj_ack; });
+
+  // Command the PJ.
+  bool cmd_ack = false;
+  w.handler->send_pr_job_command(
+      "PJ-1", "PJSTART",
+      [&](std::error_code ec, const s2::Message& m) {
+        REQUIRE_FALSE(ec);
+        CHECK(m.stream == 16);
+        CHECK(m.function == 6);
+        cmd_ack = true;
+      });
+  pump_until(w.io, [&] { return cmd_ack; });
+}
+
+TEST_CASE("HostHandler: inbound S5F1 alarm auto-acks with S5F2") {
+  Wired w;
+  // Equipment side stays silent on inbound; we only need the host to
+  // process and auto-ack an alarm primary the equipment pushes at it.
+  w.equipment->set_message_handler(
+      [](const s2::Message&) { return std::optional<s2::Message>{}; });
+
+  // The host handler's alarm observer gets called.
+  bool alarm_seen = false;
+  uint32_t got_alid = 0;
+  w.handler->set_alarm_handler([&](uint32_t alid, uint8_t, const std::string&) {
+    got_alid = alid;
+    alarm_seen = true;
+  });
+  w.handler->install();
+  w.host->start();
+  w.equipment->start();
+  bool selected = false;
+  w.host->set_selected_handler([&] { selected = true; });
+  pump_until(w.io, [&] { return selected; });
+
+  // Equipment side pushes S5F1 at the host.  We don't care about the
+  // ack body — just that the handler observes the alarm.
+  w.equipment->send_request(
+      gem::s5f1_alarm_report(0x80, 42, "vacuum lost"),
+      [](std::error_code, const s2::Message&) {});
+
+  pump_until(w.io, [&] { return alarm_seen; });
+  CHECK(got_alid == 42);
+}