tests: HSMS T3/T6/T7/T8 wire-level enforcement

Real-socket tests for the timer family in E37 §10 — these replace
the "the timer fires somewhere" implicit assumption with
end-to-end observations on a loopback pair:

  T3: send_request that gets no reply emits S9F9 with the original
      MHEAD echoed in the body and surfaces Timeout to the caller.
  T6: active mode whose Select.req goes unanswered self-closes
      with a "T6 timeout on Select" reason.
  T7: passive mode that never receives Select.req self-closes
      with a "T7 not-selected timeout" reason.
  T8: peer sends only the 4-byte length prefix; T8 expires mid-read
      and closes with "T8 intercharacter timeout".

Plus S9F11 emission for an over-length frame (length prefix of
1 GiB+1) — body's <B 10> echoes the offending bytes verbatim.

Per-test timer profiles (only the timer under test is short, the
rest are 5s) so the FSM isn't racing against unrelated timers.

Closes #5 in the test-gap backlog.

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

CMakeLists.txt

@@ -98,6 +98,7 @@ add_executable(secsgem_tests
   tests/test_secs2.cpp
   tests/test_hsms.cpp
   tests/test_hsms_connection.cpp
+  tests/test_hsms_timers.cpp
   tests/test_secsi.cpp
   tests/test_secsi_tcp.cpp
   tests/test_host_handler.cpp

tests/test_hsms_timers.cpp

@@ -0,0 +1,298 @@
+// Real-socket tests for the HSMS T-timer family (E37 §10).
+//
+// Drives a real `hsms::Connection` over a loopback socket pair with
+// deliberately tiny timer budgets and asserts:
+//
+//   T3 — reply timeout fires S9F9 with the original MHEAD and surfaces
+//        Timeout to the caller.
+//   T6 — control-transaction timeout (Select.rsp dropped) closes the
+//        connection.
+//   T7 — not-selected timeout (passive waiting for Select.req) closes
+//        the connection.
+//   T8 — intercharacter timeout (peer sends length prefix then stalls)
+//        closes the connection.
+//
+// Plus S9F11 emission for an over-length frame.
+
+#include <doctest/doctest.h>
+
+#include <asio.hpp>
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "secsgem/hsms/connection.hpp"
+#include "secsgem/hsms/header.hpp"
+#include "secsgem/secs2/message.hpp"
+
+using namespace secsgem::hsms;
+using namespace std::chrono_literals;
+
+namespace {
+
+// Identical pattern to test_hsms_connection.cpp's SocketPair, modulo the
+// shared `io` pump.  Kept local so the two test files stay independent.
+struct SocketPair {
+  asio::io_context io;
+  asio::ip::tcp::socket a{io};
+  asio::ip::tcp::socket b{io};
+
+  SocketPair() {
+    asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
+                                        asio::ip::address_v4::loopback(), 0));
+    const auto port = acc.local_endpoint().port();
+    std::error_code ec_accept, ec_connect;
+    bool accepted = false, connected = false;
+    acc.async_accept(a, [&](std::error_code ec) { ec_accept = ec; accepted = true; });
+    b.async_connect(asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), port),
+                    [&](std::error_code ec) { ec_connect = ec; connected = true; });
+    while (!(accepted && connected)) {
+      if (io.stopped()) io.restart();
+      if (io.poll() == 0) std::this_thread::sleep_for(1ms);
+    }
+    REQUIRE_FALSE(ec_accept);
+    REQUIRE_FALSE(ec_connect);
+  }
+};
+
+template <typename Pred>
+void pump_until(asio::io_context& io, Pred pred,
+                std::chrono::milliseconds budget = 3s) {
+  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(1ms);
+  }
+}
+
+void send_bytes(SocketPair& sp, std::vector<uint8_t> bytes) {
+  auto buf = std::make_shared<std::vector<uint8_t>>(std::move(bytes));
+  bool done = false;
+  asio::async_write(sp.b, asio::buffer(*buf),
+                    [buf, &done](std::error_code ec, std::size_t) {
+                      REQUIRE_FALSE(ec);
+                      done = true;
+                    });
+  pump_until(sp.io, [&] { return done; });
+}
+
+std::optional<Frame> try_recv_frame(SocketPair& sp,
+                                     std::chrono::milliseconds budget = 2s) {
+  auto lenbuf = std::make_shared<std::array<uint8_t, 4>>();
+  std::error_code rec_ec;
+  bool len_done = false;
+  asio::async_read(sp.b, asio::buffer(*lenbuf),
+                   [lenbuf, &len_done, &rec_ec](std::error_code ec, std::size_t) {
+                     rec_ec = ec;
+                     len_done = true;
+                   });
+  const auto deadline = std::chrono::steady_clock::now() + budget;
+  while (!len_done) {
+    if (std::chrono::steady_clock::now() > deadline) return std::nullopt;
+    if (sp.io.stopped()) sp.io.restart();
+    if (sp.io.poll() == 0) std::this_thread::sleep_for(1ms);
+  }
+  if (rec_ec) return std::nullopt;
+  const uint32_t len = (uint32_t((*lenbuf)[0]) << 24) | (uint32_t((*lenbuf)[1]) << 16) |
+                       (uint32_t((*lenbuf)[2]) << 8) | uint32_t((*lenbuf)[3]);
+  auto payload = std::make_shared<std::vector<uint8_t>>(len);
+  bool payload_done = false;
+  asio::async_read(sp.b, asio::buffer(*payload),
+                   [payload, &payload_done](std::error_code, std::size_t) {
+                     payload_done = true;
+                   });
+  pump_until(sp.io, [&] { return payload_done; });
+  return Frame::decode(payload->data(), payload->size());
+}
+
+// Drive the passive Connection into SELECTED so we can exercise data-path
+// timers (T3) cleanly.  Tiny T3, T6, T7, T8 so tests stay fast.
+struct Selected {
+  SocketPair sp;
+  std::shared_ptr<Connection> conn;
+
+  explicit Selected(Timers t) {
+    conn = std::make_shared<Connection>(std::move(sp.a),
+                                         Connection::Mode::Passive,
+                                         /*device_id=*/0, t);
+    bool selected = false;
+    conn->set_selected_handler([&] { selected = true; });
+    conn->start();
+    send_bytes(sp, Frame(Header::control(SType::SelectReq, 1)).encode());
+    auto rsp = try_recv_frame(sp);
+    REQUIRE(rsp.has_value());
+    REQUIRE(rsp->header.stype == SType::SelectRsp);
+    pump_until(sp.io, [&] { return selected; });
+  }
+};
+
+// Timer profiles tuned per test so the timer of interest is the FIRST to
+// fire.  Without this, e.g. T7 (armed on passive start) wins over T8
+// because they're both 150ms and T7 was armed first.
+Timers profile_for_t3() {
+  Timers t;
+  t.t3 = 150ms; t.t6 = 5s; t.t7 = 5s; t.t8 = 5s; t.linktest = 0ms;
+  return t;
+}
+Timers profile_for_t6() {
+  Timers t;
+  t.t3 = 5s; t.t6 = 150ms; t.t7 = 5s; t.t8 = 5s; t.linktest = 0ms;
+  return t;
+}
+Timers profile_for_t7() {
+  Timers t;
+  t.t3 = 5s; t.t6 = 5s; t.t7 = 150ms; t.t8 = 5s; t.linktest = 0ms;
+  return t;
+}
+Timers profile_for_t8() {
+  Timers t;
+  // T7 must outlast T8 here, otherwise T7 wins and closes first.
+  t.t3 = 5s; t.t6 = 5s; t.t7 = 5s; t.t8 = 150ms; t.linktest = 0ms;
+  return t;
+}
+
+}  // namespace
+
+TEST_CASE("T3: reply timeout emits S9F9 and surfaces Timeout to caller") {
+  Selected s(profile_for_t3());
+
+  // Driver sends a primary message expecting a reply.  The peer doesn't
+  // reply at all — only consumes the frame.
+  std::optional<std::error_code> got_ec;
+  s.conn->send_request(secsgem::secs2::Message(1, 1, true),
+                       [&](std::error_code ec, const secsgem::secs2::Message&) {
+                         got_ec = ec;
+                       });
+
+  // First frame on the peer is the primary S1F1 itself.
+  auto primary = try_recv_frame(s.sp);
+  REQUIRE(primary.has_value());
+  CHECK(primary->header.stream() == 1);
+  CHECK(primary->header.function() == 1);
+  const uint32_t primary_sys = primary->header.system_bytes;
+
+  // Pump past T3.  The reply callback should fire with Timeout AND an
+  // S9F9 frame should land on the peer carrying the original MHEAD.
+  pump_until(s.sp.io, [&] { return got_ec.has_value(); }, 2s);
+  REQUIRE(got_ec.has_value());
+  CHECK(got_ec->value() == static_cast<int>(Error::Timeout));
+
+  auto s9 = try_recv_frame(s.sp);
+  REQUIRE(s9.has_value());
+  CHECK(s9->header.stream() == 9);
+  CHECK(s9->header.function() == 9);
+  // S9F9 body is the failed primary's MHEAD as <B 10>; the formatcode
+  // byte for binary is 0x21 followed by length-byte then 10 payload bytes.
+  REQUIRE(s9->body.size() >= 12);
+  CHECK(s9->body[0] == 0x21);
+  CHECK(s9->body[1] == 10);
+  // Bytes 6..9 of the MHEAD are the system_bytes (big-endian).
+  const uint32_t echoed_sys =
+      (uint32_t(s9->body[2 + 6]) << 24) | (uint32_t(s9->body[2 + 7]) << 16) |
+      (uint32_t(s9->body[2 + 8]) << 8) | uint32_t(s9->body[2 + 9]);
+  CHECK(echoed_sys == primary_sys);
+
+  s.conn->close("test done");
+}
+
+TEST_CASE("T6: dropped Select.rsp closes the active connection") {
+  SocketPair sp;
+  bool closed = false;
+  std::string close_reason;
+  auto conn = std::make_shared<Connection>(std::move(sp.a),
+                                            Connection::Mode::Active,
+                                            /*device_id=*/0, profile_for_t6());
+  conn->set_closed_handler([&](const std::string& r) {
+    closed = true;
+    close_reason = r;
+  });
+  conn->start();
+
+  // Active mode emits Select.req on start.  We deliberately do NOT reply.
+  auto sel = try_recv_frame(sp);
+  REQUIRE(sel.has_value());
+  CHECK(sel->header.stype == SType::SelectReq);
+
+  // Drain the io_context past T6; the connection should self-close.
+  pump_until(sp.io, [&] { return closed; }, 2s);
+  CHECK(closed);
+  CHECK(close_reason.find("T6") != std::string::npos);
+}
+
+TEST_CASE("T7: passive connection with no Select.req self-closes") {
+  SocketPair sp;
+  bool closed = false;
+  std::string close_reason;
+  auto conn = std::make_shared<Connection>(std::move(sp.a),
+                                            Connection::Mode::Passive,
+                                            /*device_id=*/0, profile_for_t7());
+  conn->set_closed_handler([&](const std::string& r) {
+    closed = true;
+    close_reason = r;
+  });
+  conn->start();
+  // Peer sends nothing — T7 should fire.
+  pump_until(sp.io, [&] { return closed; }, 2s);
+  CHECK(closed);
+  CHECK(close_reason.find("T7") != std::string::npos);
+}
+
+TEST_CASE("T8: length-prefix-then-stall closes the connection") {
+  SocketPair sp;
+  bool closed = false;
+  std::string close_reason;
+  auto conn = std::make_shared<Connection>(std::move(sp.a),
+                                            Connection::Mode::Passive,
+                                            /*device_id=*/0, profile_for_t8());
+  conn->set_closed_handler([&](const std::string& r) {
+    closed = true;
+    close_reason = r;
+  });
+  conn->start();
+
+  // Send only the 4-byte length prefix.  Connection will allocate the
+  // body buffer and arm T8 — when no bytes follow, T8 fires.
+  send_bytes(sp, {0x00, 0x00, 0x00, 0x0A});
+
+  pump_until(sp.io, [&] { return closed; }, 2s);
+  CHECK(closed);
+  CHECK(close_reason.find("T8") != std::string::npos);
+}
+
+TEST_CASE("S9F11: over-length frame triggers emission + close") {
+  SocketPair sp;
+  bool closed = false;
+  // T7 long so the close reason isn't ambiguously T7.
+  auto conn = std::make_shared<Connection>(std::move(sp.a),
+                                            Connection::Mode::Passive,
+                                            /*device_id=*/0, profile_for_t8());
+  conn->set_closed_handler([&](const std::string&) { closed = true; });
+  conn->start();
+
+  // Length prefix of 0x40000001 = 1 GiB + 1, well above kMaxFrameLength.
+  send_bytes(sp, {0x40, 0x00, 0x00, 0x01});
+
+  auto s9 = try_recv_frame(sp);
+  REQUIRE(s9.has_value());
+  CHECK(s9->header.stream() == 9);
+  CHECK(s9->header.function() == 11);
+  // Body's <B 10> starts with the offending length prefix bytes.
+  REQUIRE(s9->body.size() >= 12);
+  CHECK(s9->body[0] == 0x21);
+  CHECK(s9->body[1] == 10);
+  CHECK(s9->body[2] == 0x40);
+  CHECK(s9->body[3] == 0x00);
+  CHECK(s9->body[4] == 0x00);
+  CHECK(s9->body[5] == 0x01);
+
+  pump_until(sp.io, [&] { return closed; }, 2s);
+  CHECK(closed);
+}