#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include <doctest/doctest.h>

#include <grpcpp/grpcpp.h>

#include <atomic>
#include <chrono>
#include <random>
#include <string>
#include <thread>
#include <vector>

#include "secsgem/daemon/equipment_service.hpp"
#include "secsgem/gem/default_handlers.hpp"
#include "secsgem/gem/messages.hpp"

using namespace secsgem;
namespace gem = secsgem::gem;
namespace s2  = secsgem::secs2;
namespace pb  = secsgem::v1;
namespace dmn = secsgem::daemon;

#ifndef SECSGEM_DATA_DIR
#error "SECSGEM_DATA_DIR not defined; see CMakeLists.txt"
#endif

static gem::EquipmentRuntime::Config test_config() {
  gem::EquipmentRuntime::Config c;
  c.equipment_yaml     = SECSGEM_DATA_DIR "/equipment.yaml";
  c.control_state_yaml = SECSGEM_DATA_DIR "/control_state.yaml";
  c.process_job_yaml   = SECSGEM_DATA_DIR "/process_job_state.yaml";
  c.control_job_yaml   = SECSGEM_DATA_DIR "/control_job_state.yaml";
  c.port = 0;  // ephemeral; the engine isn't run() here, only poll()ed
  return c;
}

// Exercises the real gRPC service over an in-process channel: client stub ->
// service -> EquipmentRuntime, proving the RPCs move data, not just compile.
TEST_CASE("Equipment gRPC service over an in-process channel") {
  gem::EquipmentRuntime rt(test_config());
  dmn::EquipmentService svc(rt);

  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{}));

  SUBCASE("GetControlState returns the initial control state") {
    grpc::ClientContext ctx;
    pb::Empty req;
    pb::ControlState resp;
    auto st = stub->GetControlState(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.state() == pb::ControlState::HOST_OFFLINE);
  }

  SUBCASE("SetVariables converts to the variable's declared wire format") {
    // ChamberPressure is declared F4 and WaferCounter U4 in equipment.yaml;
    // the daemon must honour those, not write F8/I8, or the host sees values
    // whose format contradicts the S1F11/S1F21 namelists.
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"].set_real(2.5);
    (*req.mutable_values())["WaferCounter"].set_integer(7);
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::ACCEPT);
    rt.poll();  // drain the posted set_variable onto this thread
    CHECK(rt.model().dvids.value(101) == s2::Item::f4(2.5f));
    CHECK(rt.model().dvids.value(100) == s2::Item::u4(uint32_t{7}));
  }

  SUBCASE("SetVariables rejects an unknown variable name") {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["definitely_not_a_var"].set_real(1.0);
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
  }

  SUBCASE("SetVariables rejects a Value with no kind set (silent-'' guard)") {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"];  // map entry, kind never set
    auto st = stub->SetVariables(&ctx, req, &resp);
    CHECK(st.ok());
    CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
    // The previous value must be untouched (no silent ASCII "" write).
    rt.poll();
    CHECK(rt.model().dvids.value(101)->format() == s2::Format::F4);
  }

  SUBCASE("property: every configured variable keeps its declared format") {
    // Iterate ALL SVIDs+DVIDs from the live config: set each via gRPC with a
    // type-appropriate plain value and assert the stored Item keeps the
    // declared wire format. Catches any future variable whose format the
    // conversion table doesn't handle — not just the two pinned above.
    auto check_all = [&](const auto& entries) {
      for (const auto& v : entries) {
        const auto declared = v.value.format();
        grpc::ClientContext ctx;
        pb::VariableUpdate req;
        pb::Ack resp;
        pb::Value val;
        switch (declared) {
          case s2::Format::ASCII: case s2::Format::JIS8:
            val.set_text("x"); break;
          case s2::Format::Boolean:
            val.set_boolean(true); break;
          case s2::Format::Binary:
            val.set_binary("\x01"); break;
          case s2::Format::F4: case s2::Format::F8:
            val.set_real(1.5); break;
          default:  // all integer widths
            val.set_integer(1); break;
        }
        (*req.mutable_values())[v.name] = val;
        REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
        REQUIRE_MESSAGE(resp.code() == pb::Ack::ACCEPT, v.name);
        rt.poll();
        auto stored = rt.model().vid_value(v.id);
        REQUIRE_MESSAGE(stored.has_value(), v.name);
        CHECK_MESSAGE(stored->format() == declared,
                      v.name, ": declared ", static_cast<int>(declared),
                      " stored ", static_cast<int>(stored->format()));
      }
    };
    check_all(rt.model().svids.all());
    check_all(rt.model().dvids.all());
  }

  SUBCASE("SetAlarm / ClearAlarm by config name, by stringified id, unknown rejected") {
    auto call = [&](auto method, const std::string& name) {
      grpc::ClientContext ctx;
      pb::Alarm req;
      pb::Ack resp;
      req.set_name(name);
      REQUIRE((stub.get()->*method)(&ctx, req, &resp).ok());
      return resp.code();
    };
    // By config name.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK(rt.model().alarms.active(1));
    CHECK(call(&pb::Equipment::Stub::ClearAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK_FALSE(rt.model().alarms.active(1));
    // By stringified ALID — always works, even for unnamed alarms.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "2") == pb::Ack::ACCEPT);
    rt.poll();
    CHECK(rt.model().alarms.active(2));
    // Unknown name.
    CHECK(call(&pb::Equipment::Stub::SetAlarm, "no_such_alarm") == pb::Ack::PARAMETER_INVALID);
  }

  SUBCASE("FireEvent accepts a known event and rejects an unknown one") {
    {
      grpc::ClientContext ctx;
      pb::Event req;
      pb::Ack resp;
      req.set_name("ProcessStarted");
      CHECK(stub->FireEvent(&ctx, req, &resp).ok());
      CHECK(resp.code() == pb::Ack::ACCEPT);
    }
    {
      grpc::ClientContext ctx;
      pb::Event req;
      pb::Ack resp;
      req.set_name("NoSuchEvent");
      CHECK(stub->FireEvent(&ctx, req, &resp).ok());
      CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
    }
  }

  server->Shutdown();
}

// GetVariables needs a live io thread (read_sync posts onto it), so this case
// runs the engine in run_async() — the daemon's PRODUCTION threading mode —
// with real concurrency between the gRPC handler thread and the io thread.
TEST_CASE("GetVariables round-trip under run_async (production threading mode)") {
  gem::EquipmentRuntime rt(test_config());
  dmn::EquipmentService svc(rt);   // snapshot BEFORE the io thread starts
  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{}));

  // Write through the API, then read back through the API: exercises BOTH
  // conversions (Value->Item with declared formats, Item->Value) end to end.
  {
    grpc::ClientContext ctx;
    pb::VariableUpdate req;
    pb::Ack resp;
    (*req.mutable_values())["ChamberPressure"].set_real(2.5);
    (*req.mutable_values())["WaferCounter"].set_integer(7);
    REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
    REQUIRE(resp.code() == pb::Ack::ACCEPT);
  }
  {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    req.add_names("ChamberPressure");
    req.add_names("WaferCounter");
    auto st = stub->GetVariables(&ctx, req, &resp);
    REQUIRE(st.ok());
    REQUIRE(resp.values().count("ChamberPressure") == 1);
    REQUIRE(resp.values().count("WaferCounter") == 1);
    CHECK(resp.values().at("ChamberPressure").real() == doctest::Approx(2.5));
    CHECK(resp.values().at("WaferCounter").integer() == 7);
  }

  SUBCASE("empty query returns every configured variable") {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    REQUIRE(stub->GetVariables(&ctx, req, &resp).ok());
    // The io thread is live here — model reads must go through read_sync
    // (the first violation of that contract was caught by the TSan lane in
    // exactly this line).
    auto expected = rt.read_sync([&rt] {
      return rt.model().svids.size() + rt.model().dvids.all().size();
    });
    REQUIRE(expected.has_value());
    CHECK(resp.values().size() == *expected);
  }

  SUBCASE("unknown name is INVALID_ARGUMENT, naming the offender") {
    grpc::ClientContext ctx;
    pb::VariableQuery req;
    pb::VariableSnapshot resp;
    req.add_names("definitely_not_a_var");
    auto st = stub->GetVariables(&ctx, req, &resp);
    CHECK(st.error_code() == grpc::StatusCode::INVALID_ARGUMENT);
    CHECK(st.error_message().find("definitely_not_a_var") != std::string::npos);
  }

  SUBCASE("RequestControlState walks the E30 table; WatchHealth pushes the change") {
    // Open the health stream first: the initial snapshot arrives immediately.
    grpc::ClientContext health_ctx;
    pb::Empty empty;
    auto reader = stub->WatchHealth(&health_ctx, empty);
    pb::Health h;
    REQUIRE(reader->Read(&h));
    CHECK(h.link() == pb::Health::DISCONNECTED);  // no HSMS host in this test
    CHECK(h.control_state() == pb::ControlState::HOST_OFFLINE);
    CHECK(h.spool_depth() == 0);

    auto request = [&](pb::ControlState::State desired) {
      grpc::ClientContext ctx;
      pb::ControlStateRequest req;
      pb::Ack resp;
      req.set_desired(desired);
      REQUIRE(stub->RequestControlState(&ctx, req, &resp).ok());
      return resp;
    };

    // HostOffline --operator_online--> (AttemptOnline) --> OnlineLocal.
    CHECK(request(pb::ControlState::ONLINE_LOCAL).code() == pb::Ack::ACCEPT);
    CHECK(rt.control_state() == gem::ControlState::OnlineLocal);

    // The stream pushes the change (well inside its 500ms poll interval).
    REQUIRE(reader->Read(&h));
    CHECK(h.control_state() == pb::ControlState::ONLINE_LOCAL);

    // OnlineLocal --operator_remote--> OnlineRemote.
    CHECK(request(pb::ControlState::ONLINE_REMOTE).code() == pb::Ack::ACCEPT);
    CHECK(rt.control_state() == gem::ControlState::OnlineRemote);

    // Transient state is not requestable.
    CHECK(request(pb::ControlState::ATTEMPT_ONLINE).code() == pb::Ack::PARAMETER_INVALID);

    // The shipped table has NO operator path to EquipmentOffline:
    // operator_offline lands HostOffline — the API must say so honestly.
    auto resp = request(pb::ControlState::EQUIPMENT_OFFLINE);
    CHECK(resp.code() == pb::Ack::CANNOT_DO_NOW);
    CHECK(resp.message().find("HostOffline") != std::string::npos);
    CHECK(rt.control_state() == gem::ControlState::HostOffline);

    // Requesting HOST_OFFLINE (the state operators actually get) succeeds —
    // idempotently, since we're already there.
    CHECK(request(pb::ControlState::HOST_OFFLINE).code() == pb::Ack::ACCEPT);

    health_ctx.TryCancel();
    pb::Health drain;
    while (reader->Read(&drain)) {}
    (void)reader->Finish();  // CANCELLED — expected
  }

  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();
}

// Phase D — GEM300 in-the-loop (observe-and-report): job lifecycle, recipe
// downloads, and EC writes flow host -> engine -> Subscribe stream; the tool
// reports physical progress back and the engine drives the E40 FSM.
TEST_CASE("Phase D: jobs, recipes, and EC changes on the stream; ReportProcessJob") {
  gem::EquipmentRuntime rt(test_config());
  gem::register_default_handlers(rt);
  dmn::EquipmentService svc(rt);
  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{}));

  grpc::ClientContext sub_ctx;
  pb::SubscribeRequest sreq;
  auto reader = stub->Subscribe(&sub_ctx, sreq);
  std::this_thread::sleep_for(std::chrono::milliseconds(100));  // registration

  auto dispatch = [&](s2::Message msg) {
    auto reply = rt.read_sync([&rt, &msg]() { return rt.router().dispatch(msg); });
    REQUIRE(reply.has_value());
    REQUIRE(reply->has_value());
  };

  // Host creates the PJ (S16F11), the tool sets up, the host starts it.
  gem::PRJobCreateRequest pj{"PJ-D-1", gem::MaterialFlag::Substrate,
                             gem::ProcessRecipeMethod::RecipeOnly,
                             gem::RecipeSpec{"RECIPE-A", {}}, {"WFR-D-1"}, {}};
  dispatch(gem::s16f11_pr_job_create(pj));
  auto setup = rt.read_sync([&rt]() {
    rt.model().process_jobs.fire_internal("PJ-D-1", gem::ProcessJobEvent::Select);
    return rt.model().process_jobs.fire_internal("PJ-D-1",
                                                 gem::ProcessJobEvent::SetupComplete);
  });
  REQUIRE(setup.has_value());
  REQUIRE(*setup);
  dispatch(gem::s16f5_pr_job_command("PJ-D-1", "PJSTART"));

  // The tool's stream receives the job with recipe + material bindings.
  pb::HostRequest hr;
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_process_job());
  CHECK(hr.process_job().job_id() == "PJ-D-1");
  CHECK(hr.process_job().action() == pb::ProcessJob::START);
  CHECK(hr.process_job().recipe() == "RECIPE-A");
  REQUIRE(hr.process_job().carriers_size() == 1);
  CHECK(hr.process_job().carriers(0) == "WFR-D-1");

  // The tool finishes the physical work and reports; the engine's FSM follows.
  {
    grpc::ClientContext ctx;
    pb::ProcessJobState rep;
    pb::Ack ack;
    rep.set_job_id("PJ-D-1");
    rep.set_state(pb::ProcessJobState::COMPLETE);
    REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
    CHECK(ack.code() == pb::Ack::ACCEPT);
  }
  auto state = rt.read_sync(
      [&rt]() { return rt.model().process_jobs.state("PJ-D-1"); });
  REQUIRE(state.has_value());
  CHECK(*state == gem::ProcessJobState::ProcessComplete);

  // Reporting against an unknown job / an illegal transition is rejected.
  {
    grpc::ClientContext ctx;
    pb::ProcessJobState rep;
    pb::Ack ack;
    rep.set_job_id("PJ-GHOST");
    rep.set_state(pb::ProcessJobState::COMPLETE);
    REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
    CHECK(ack.code() == pb::Ack::INVALID_OBJECT);
  }
  {
    grpc::ClientContext ctx;
    pb::ProcessJobState rep;
    pb::Ack ack;
    rep.set_job_id("PJ-D-1");  // already ProcessComplete
    rep.set_state(pb::ProcessJobState::COMPLETE);
    REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
    CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW);
  }

  // Host downloads a recipe (S7F3) -> ProcessProgram on the stream.
  dispatch(gem::s7f3_process_program_send("RECIPE-NEW", "step1\nstep2\n"));
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_process_program());
  CHECK(hr.process_program().ppid() == "RECIPE-NEW");
  CHECK(hr.process_program().body() == "step1\nstep2\n");

  // Host writes an EC (S2F15) -> ConstantChange with the configured name.
  dispatch(gem::s2f15_ec_send({{10, s2::Item::u4(uint32_t{1})}}));
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_constant());
  CHECK(hr.constant().name() == "TimeFormat");
  CHECK(hr.constant().value().integer() == 1);

  sub_ctx.TryCancel();
  pb::HostRequest drain;
  while (reader->Read(&drain)) {}
  (void)reader->Finish();
  server->Shutdown();
  rt.stop();
}

// D10 + E16: carriers on the stream, Describe / FlushSpool / terminal.
TEST_CASE("carriers (D10) and the operations RPCs (E16)") {
  gem::EquipmentRuntime rt(test_config());
  gem::register_default_handlers(rt);
  dmn::EquipmentService svc(rt);
  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{}));

  grpc::ClientContext sub_ctx;
  pb::SubscribeRequest sreq;
  auto reader = stub->Subscribe(&sub_ctx, sreq);
  std::this_thread::sleep_for(std::chrono::milliseconds(100));

  auto report_carrier = [&](const std::string& cid, pb::CarrierState::State st,
                            std::vector<bool> slots = {}) {
    grpc::ClientContext ctx;
    pb::CarrierState req;
    pb::Ack ack;
    req.set_carrier_id(cid);
    req.set_port(2);
    req.set_state(st);
    for (bool b : slots) req.add_slots(b);
    REQUIRE(stub->ReportCarrier(&ctx, req, &ack).ok());
    return ack.code();
  };

  // Tool announces an arrived FOUP with a slot map.
  CHECK(report_carrier("CAR-D-1", pb::CarrierState::WAITING,
                       {true, true, false}) == pb::Ack::ACCEPT);
  auto created = rt.read_sync([&rt]() {
    const auto* c = rt.model().carriers.get("CAR-D-1");
    return c && c->slots.size() == 3 && c->slots[0].state == 1 &&
           c->slots[2].state == 0 && c->port_id == 2;
  });
  REQUIRE(created.has_value());
  CHECK(*created);

  // Host says ProceedWithCarrier (S3F17) -> tool stream gets PROCEED.
  auto reply = rt.read_sync([&rt]() {
    return rt.router().dispatch(
        gem::s3f17_carrier_action(0u, "ProceedWithCarrier", "CAR-D-1", {}));
  });
  REQUIRE(reply.has_value());
  REQUIRE(reply->has_value());
  pb::HostRequest hr;
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_carrier());
  CHECK(hr.carrier().carrier_id() == "CAR-D-1");
  CHECK(hr.carrier().port() == 2);
  CHECK(hr.carrier().action() == pb::CarrierAction::PROCEED);

  // Tool drives access: begin + end; reporting against an unknown id fails.
  CHECK(report_carrier("CAR-D-1", pb::CarrierState::IN_ACCESS) == pb::Ack::ACCEPT);
  CHECK(report_carrier("CAR-D-1", pb::CarrierState::COMPLETE) == pb::Ack::ACCEPT);
  CHECK(report_carrier("CAR-GHOST", pb::CarrierState::IN_ACCESS) ==
        pb::Ack::INVALID_OBJECT);

  // Host cancels CAR-D-1 (Confirmed -> NotConfirmed is a real transition, so
  // the observer fires) -> CANCEL on the stream. NOTE: a CancelCarrier on a
  // still-NotConfirmed carrier is a self-transition the FSM doesn't signal,
  // so the tool isn't told — a known E87 edge (see DAEMON_ROADMAP).
  (void)rt.read_sync([&rt]() {
    return rt.router().dispatch(
        gem::s3f17_carrier_action(0u, "CancelCarrier", "CAR-D-1", {}));
  });
  REQUIRE(reader->Read(&hr));
  REQUIRE(hr.has_carrier());
  CHECK(hr.carrier().action() == pb::CarrierAction::CANCEL);

  // ---- E16: Describe / FlushSpool / SendTerminalMessage --------------------
  {
    grpc::ClientContext ctx;
    pb::Empty req;
    pb::EquipmentDescription d;
    REQUIRE(stub->Describe(&ctx, req, &d).ok());
    CHECK(d.model_name() == "SECSGEM-SIM");
    auto has = [](const auto& list, const std::string& want) {
      for (const auto& e : list)
        if (e == want) return true;
      return false;
    };
    CHECK(has(d.variables(), "ChamberPressure"));
    CHECK(has(d.events(), "ProcessStarted"));
    CHECK(has(d.alarms(), "chiller_temp_high"));
    CHECK(has(d.commands(), "START"));
    CHECK(has(d.constants(), "TimeFormat"));
  }
  {
    // No host: a fired event spools (stream 6 is spoolable); purge empties it.
    grpc::ClientContext fctx;
    pb::Event ev;
    pb::Ack ack;
    ev.set_name("ProcessStarted");
    // Enable the event first so emit isn't suppressed.
    (void)rt.read_sync([&rt]() {
      return rt.model().enable_events(true, {300});
    });
    REQUIRE(stub->FireEvent(&fctx, ev, &ack).ok());
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    auto depth = rt.read_sync([&rt]() { return rt.model().spool.size(); });
    REQUIRE(depth.has_value());
    CHECK(*depth == 1);

    grpc::ClientContext pctx;
    pb::SpoolFlushRequest freq;
    pb::Ack fack;
    freq.set_purge(true);
    REQUIRE(stub->FlushSpool(&pctx, freq, &fack).ok());
    CHECK(fack.code() == pb::Ack::ACCEPT);
    depth = rt.read_sync([&rt]() { return rt.model().spool.size(); });
    CHECK(*depth == 0);
  }
  {
    // Stream 10 is not spoolable and no host is connected -> honest refusal.
    grpc::ClientContext ctx;
    pb::TerminalMessage req;
    pb::Ack ack;
    req.set_tid(0);
    req.set_text("hello fab");
    REQUIRE(stub->SendTerminalMessage(&ctx, req, &ack).ok());
    CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW);
  }

  sub_ctx.TryCancel();
  pb::HostRequest drain;
  while (reader->Read(&drain)) {}
  (void)reader->Finish();
  server->Shutdown();
  rt.stop();
}

// Randomized concurrent stress: several client threads fire a seeded random
// mix of RPCs against the live service while the io thread runs — the
// strongest TSan target we have, and a probe for ordering/lifetime bugs the
// scenario tests can't reach. Failures print the seed for reproduction.
TEST_CASE("randomized concurrent RPC stress (seeded)") {
  const unsigned seed = static_cast<unsigned>(
      std::random_device{}());  // logged below; rerun by hardcoding it
  INFO("stress seed = " << seed);

  gem::EquipmentRuntime rt(test_config());
  gem::register_default_handlers(rt);
  dmn::EquipmentService svc(rt);
  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{}));

  constexpr int kThreads = 4;
  constexpr int kOpsPerThread = 250;
  std::atomic<int> failures{0};

  auto worker = [&](unsigned tseed) {
    std::mt19937 rng(tseed);
    std::uniform_int_distribution<int> op(0, 7);
    std::uniform_real_distribution<double> val(0.0, 10.0);
    for (int i = 0; i < kOpsPerThread; ++i) {
      grpc::ClientContext ctx;
      pb::Ack ack;
      grpc::Status st = grpc::Status::OK;
      switch (op(rng)) {
        case 0: {
          pb::VariableUpdate r;
          (*r.mutable_values())["ChamberPressure"].set_real(val(rng));
          st = stub->SetVariables(&ctx, r, &ack);
          break;
        }
        case 1: {
          pb::VariableQuery r;
          pb::VariableSnapshot snap;
          st = stub->GetVariables(&ctx, r, &snap);
          if (st.ok() && snap.values().empty()) ++failures;  // config never empty
          break;
        }
        case 2: {
          pb::Event r;
          r.set_name("ProcessStarted");
          st = stub->FireEvent(&ctx, r, &ack);
          break;
        }
        case 3: {
          pb::Alarm r;
          r.set_name("chiller_temp_high");
          st = stub->SetAlarm(&ctx, r, &ack);
          break;
        }
        case 4: {
          pb::Alarm r;
          r.set_name("chiller_temp_high");
          st = stub->ClearAlarm(&ctx, r, &ack);
          break;
        }
        case 5: {
          pb::Empty r;
          pb::ControlState cs;
          st = stub->GetControlState(&ctx, r, &cs);
          break;
        }
        case 6: {
          pb::Empty r;
          pb::EquipmentDescription d;
          st = stub->Describe(&ctx, r, &d);
          if (st.ok() && d.variables_size() == 0) ++failures;
          break;
        }
        case 7: {
          // Subscribe/cancel churn: exercises subscriber add/remove under load.
          grpc::ClientContext sctx;
          pb::SubscribeRequest sr;
          auto rd = stub->Subscribe(&sctx, sr);
          sctx.TryCancel();
          pb::HostRequest hr;
          while (rd->Read(&hr)) {}
          (void)rd->Finish();
          break;
        }
      }
      if (!st.ok()) ++failures;
    }
  };

  std::vector<std::thread> threads;
  for (int t = 0; t < kThreads; ++t) threads.emplace_back(worker, seed + t);
  for (auto& th : threads) th.join();

  CHECK(failures.load() == 0);
  // The engine must still be fully responsive afterwards.
  grpc::ClientContext ctx;
  pb::Empty req;
  pb::ControlState cs;
  CHECK(stub->GetControlState(&ctx, req, &cs).ok());

  server->Shutdown();
  rt.stop();
}