tests: S9F7 wire emission for malformed primaries

S9F3/F5 are covered by test_s9_fallback (router path); S9F9/F11 by
test_hsms_timers (timer/over-length). This commit adds S9F7 wire-level
tests for the third path — a primary whose body fails secs2::decode.

Three new cases:
  - hand-built primary with truncated <B> body provokes S9F7
    carrying the original 10-byte MHEAD (sys + stream + function)
  - emission is non-fatal: the next well-formed primary still routes
    to the registered handler
  - data-while-NOT-SELECTED still echoes Reject(EntityNotSelected)
    (sanity copy of the test_hsms_connection case so the "what does
    the equipment say when a peer sends garbage" family lives together)

Closes #6 in the test-gap backlog.

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

CMakeLists.txt

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

tests/test_hsms_s9.cpp

@@ -0,0 +1,211 @@
+// Real-socket tests for S9F7 (illegal data) emission.
+//
+// S9F3 / F5 are covered by test_s9_fallback.cpp; S9F9 / F11 by
+// test_hsms_timers.cpp.  S9F7 sits in between: it fires when the peer
+// sends a primary (or a reply) whose body fails secs2::decode.
+
+#include <doctest/doctest.h>
+
+#include <asio.hpp>
+#include <array>
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <thread>
+#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 {
+
+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 ea, eb;
+    bool da = false, db = false;
+    acc.async_accept(a, [&](std::error_code ec) { ea = ec; da = true; });
+    b.async_connect({asio::ip::address_v4::loopback(), port},
+                    [&](std::error_code ec) { eb = ec; db = true; });
+    while (!(da && db)) {
+      if (io.stopped()) io.restart();
+      if (io.poll() == 0) std::this_thread::sleep_for(1ms);
+    }
+    REQUIRE_FALSE(ea);
+    REQUIRE_FALSE(eb);
+  }
+};
+
+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>>();
+  bool len_done = false;
+  std::error_code rec_ec;
+  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());
+}
+
+Timers permissive_timers() {
+  Timers t;
+  t.t3 = 5s; t.t6 = 5s; t.t7 = 5s; t.t8 = 5s; t.linktest = 0ms;
+  return t;
+}
+
+// Build a SELECTED passive connection, ready to be poked.
+std::pair<std::shared_ptr<Connection>, std::unique_ptr<SocketPair>>
+make_selected_passive() {
+  auto sp = std::make_unique<SocketPair>();
+  auto conn = std::make_shared<Connection>(std::move(sp->a),
+                                            Connection::Mode::Passive,
+                                            /*device_id=*/0, permissive_timers());
+  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; });
+  return {conn, std::move(sp)};
+}
+
+}  // namespace
+
+TEST_CASE("S9F7: primary with malformed body emits S9F7 echoing original MHEAD") {
+  auto [conn, sp] = make_selected_passive();
+
+  // Build a primary S1F3 frame with a deliberately broken body — the
+  // formatcode prefix 0x21 says "binary, 1 length byte" but we don't
+  // supply the length byte.  Body of just {0x21}.
+  Frame primary(Header::data_message(/*session=*/0, /*stream=*/1,
+                                       /*function=*/3, /*reply_expected=*/true,
+                                       /*sys=*/777),
+                std::vector<uint8_t>{0x21});
+  send_bytes(*sp, primary.encode());
+
+  auto s9 = try_recv_frame(*sp);
+  REQUIRE(s9.has_value());
+  CHECK(s9->header.stream() == 9);
+  CHECK(s9->header.function() == 7);
+  // S9F7 body: <B 10> MHEAD echo.
+  REQUIRE(s9->body.size() >= 12);
+  CHECK(s9->body[0] == 0x21);
+  CHECK(s9->body[1] == 10);
+  // MHEAD bytes 6..9 are the original sys=777 = 0x00000309.
+  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 == 777);
+  // MHEAD byte2's lower 7 bits should be stream 1; bit7 carries the W-bit.
+  CHECK((s9->body[2 + 2] & 0x7F) == 1);
+  CHECK(s9->body[2 + 3] == 3);  // function
+
+  conn->close("test done");
+}
+
+TEST_CASE("S9F7: emission keeps the connection up (next frame still routes)") {
+  auto [conn, sp] = make_selected_passive();
+
+  bool got_good = false;
+  conn->set_message_handler([&](const secsgem::secs2::Message& m)
+                              -> std::optional<secsgem::secs2::Message> {
+    if (m.stream == 1 && m.function == 13) {
+      got_good = true;
+      return secsgem::secs2::Message(1, 14, false);
+    }
+    return std::nullopt;
+  });
+
+  // First send a broken frame to provoke S9F7.
+  Frame broken(Header::data_message(0, 1, 3, true, 100), {0x21});
+  send_bytes(*sp, broken.encode());
+  auto s9 = try_recv_frame(*sp);
+  REQUIRE(s9.has_value());
+  CHECK(s9->header.function() == 7);
+
+  // Now send a well-formed S1F13 — it should still route to the handler.
+  Frame good(Header::data_message(0, 1, 13, true, 101), {});
+  send_bytes(*sp, good.encode());
+
+  auto reply = try_recv_frame(*sp);
+  REQUIRE(reply.has_value());
+  CHECK(reply->header.stream() == 1);
+  CHECK(reply->header.function() == 14);
+  CHECK(got_good);
+
+  conn->close("test done");
+}
+
+TEST_CASE("Data frame while NOT_SELECTED still echoes Reject (sanity vs S9 paths)") {
+  // This case lives in test_hsms_connection.cpp already; including it
+  // here too because the family of "what does the equipment say when
+  // peer sends garbage" tests reads better as one place.
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a),
+                                            Connection::Mode::Passive,
+                                            /*device_id=*/0, permissive_timers());
+  conn->start();
+
+  Frame data(Header::data_message(0, 1, 1, true, 42));
+  send_bytes(sp, data.encode());
+
+  auto rej = try_recv_frame(sp);
+  REQUIRE(rej.has_value());
+  CHECK(rej->header.stype == SType::RejectReq);
+  CHECK(rej->header.byte3 == static_cast<uint8_t>(RejectReason::EntityNotSelected));
+
+  conn->close("test done");
+}