#pragma once

// The gRPC Equipment service: translates proto/secsgem/v1 RPCs onto an
// EquipmentRuntime. Header-only so both secs_gemd and the daemon tests share
// one definition.
//
// Threading: gRPC handlers run on gRPC's own threads while the engine's
// io thread owns the data model, so handlers must not read the live stores.
// The constructor therefore snapshots the name->id/format maps (immutable
// after config load) — construct the service BEFORE run_async(). All writes
// go through the runtime's posting API.

#include <grpcpp/grpcpp.h>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "secsgem/gem/messages.hpp"
#include "secsgem/gem/runtime.hpp"
#include "secsgem/secs2/item.hpp"
#include "secsgem/v1/equipment.grpc.pb.h"
#include "secsgem/v1/equipment.pb.h"

namespace secsgem::daemon {

namespace gem = secsgem::gem;
namespace s2  = secsgem::secs2;
namespace pb  = secsgem::v1;

// proto Value -> SECS-II Item, honouring the variable's declared wire format
// (from equipment.yaml) so the host sees the same format in S1F11 namelists
// and in reported values. E.g. real 2.5 -> F4 if the variable is F4.
inline s2::Item to_item(const pb::Value& v, s2::Format want) {
  using F = s2::Format;
  switch (v.kind_case()) {
    case pb::Value::kText:
      return s2::Item::ascii(v.text());
    case pb::Value::kBoolean:
      return s2::Item::boolean(v.boolean());
    case pb::Value::kBinary:
      return s2::Item::binary({v.binary().begin(), v.binary().end()});
    case pb::Value::kReal:
      return want == F::F4 ? s2::Item::f4(static_cast<float>(v.real()))
                           : s2::Item::f8(v.real());
    case pb::Value::kInteger: {
      const int64_t n = v.integer();
      switch (want) {
        case F::U1: return s2::Item::u1(static_cast<uint8_t>(n));
        case F::U2: return s2::Item::u2(static_cast<uint16_t>(n));
        case F::U4: return s2::Item::u4(static_cast<uint32_t>(n));
        case F::U8: return s2::Item::u8(static_cast<uint64_t>(n));
        case F::I1: return s2::Item::i1(static_cast<int8_t>(n));
        case F::I2: return s2::Item::i2(static_cast<int16_t>(n));
        case F::I4: return s2::Item::i4(static_cast<int32_t>(n));
        case F::F4: return s2::Item::f4(static_cast<float>(n));
        case F::F8: return s2::Item::f8(static_cast<double>(n));
        case F::Boolean: return s2::Item::boolean(n != 0);
        default:    return s2::Item::i8(n);
      }
    }
    case pb::Value::kList: {
      // TODO(daemon): list elements inherit the variable's scalar format;
      // honouring per-element formats needs the declared Item's nested shape.
      s2::Item::List items;
      for (const auto& e : v.list().items()) items.push_back(to_item(e, want));
      return s2::Item::list(std::move(items));
    }
    default:
      // Unreachable: callers reject KIND_NOT_SET before converting (see
      // value_is_set). Kept as a safe fallback for future oneof additions.
      return s2::Item::ascii("");
  }
}

// Validate a client-supplied Value before conversion. An unset oneof would
// otherwise silently become ASCII "" — reject it at the API edge instead.
inline bool value_is_set(const pb::Value& v) {
  return v.kind_case() != pb::Value::KIND_NOT_SET;
}

// SECS-II Item -> proto Value (the read direction). Single-element numeric
// arrays — the overwhelmingly common case — become scalars; multi-element
// arrays become a List so nothing is lost.
// TODO(daemon): C2 (2-byte Unicode) currently surfaces as integer code
// points (it shares storage with U2); convert to text when a tool needs it.
// TODO(daemon): U8 values above 2^63-1 wrap through the sint64 integer
// field; switch Value to a dedicated uint64 arm if that ever bites.
inline pb::Value from_item(const s2::Item& item) {
  pb::Value out;
  switch (item.format()) {
    case s2::Format::List: {
      auto* list = out.mutable_list();
      for (const auto& child : item.as_list()) *list->add_items() = from_item(child);
      return out;
    }
    case s2::Format::ASCII:
    case s2::Format::JIS8:
      out.set_text(item.as_ascii());
      return out;
    case s2::Format::Binary: {
      const auto& b = item.as_bytes();
      out.set_binary(std::string(b.begin(), b.end()));
      return out;
    }
    case s2::Format::Boolean: {
      const auto& b = item.as_bytes();
      if (b.size() == 1) {
        out.set_boolean(b[0] != 0);
      } else {
        auto* list = out.mutable_list();
        for (auto v : b) list->add_items()->set_boolean(v != 0);
      }
      return out;
    }
    default:
      break;  // numeric formats: handled generically below
  }
  return std::visit(
      [&](const auto& vec) -> pb::Value {
        using V = std::decay_t<decltype(vec)>;
        if constexpr (std::is_same_v<V, s2::Item::List> ||
                      std::is_same_v<V, std::string>) {
          return out;  // unreachable: handled above
        } else {
          auto set_one = [](pb::Value& dst, auto v) {
            if constexpr (std::is_floating_point_v<decltype(v)>)
              dst.set_real(static_cast<double>(v));
            else
              dst.set_integer(static_cast<int64_t>(v));
          };
          if (vec.size() == 1) {
            set_one(out, vec[0]);
          } else {
            auto* list = out.mutable_list();
            for (auto v : vec) set_one(*list->add_items(), v);
          }
          return out;
        }
      },
      item.storage());
}

inline pb::ControlState::State to_proto_state(gem::ControlState s) {
  switch (s) {
    case gem::ControlState::EquipmentOffline: return pb::ControlState::EQUIPMENT_OFFLINE;
    case gem::ControlState::AttemptOnline:    return pb::ControlState::ATTEMPT_ONLINE;
    case gem::ControlState::HostOffline:      return pb::ControlState::HOST_OFFLINE;
    case gem::ControlState::OnlineLocal:      return pb::ControlState::ONLINE_LOCAL;
    case gem::ControlState::OnlineRemote:     return pb::ControlState::ONLINE_REMOTE;
  }
  return pb::ControlState::EQUIPMENT_OFFLINE;
}

class EquipmentService final : public pb::Equipment::Service {
 public:
  // Snapshots the (immutable) name->id/format dictionaries and registers the
  // health observers. Construct before run_async() so the model is read (and
  // observers land) while the io thread isn't running yet.
  explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) {
    for (const auto& sv : rt.model().svids.all())
      vars_.insert({sv.name, {sv.id, sv.value.format()}});
    for (const auto& dv : rt.model().dvids.all())
      vars_.insert({dv.name, {dv.id, dv.value.format()}});
    for (const auto& ev : rt.model().events.all_events())
      events_.insert({ev.name, ev.id});
    for (const auto& al : rt.model().alarms.all()) {
      if (!al.name.empty()) alarms_.insert({al.name, al.id});
      alarms_.insert({std::to_string(al.id), al.id});  // always addressable by id
    }
    // Health signals: bump a version + wake WatchHealth streams whenever the
    // HSMS link or the control state changes (observers fire on the io thread;
    // add_ observers survive register_default_handlers' primary set_).
    rt.add_link_observer([this](bool selected) {
      link_selected_.store(selected, std::memory_order_relaxed);
      bump_health();
    });
    rt.add_control_state_observer(
        [this](gem::ControlState, gem::ControlState, gem::ControlEvent) {
          bump_health();
        });

    for (const auto& ec : rt.model().ecids.all())
      ecnames_.insert({ec.id, ec.name});

    // GEM300 in-the-loop (Phase D, observe-and-report): forward job
    // lifecycle, recipe downloads, and accepted EC writes onto the
    // Subscribe stream. Observers fire on the io thread; push_request only
    // takes the subscriber mutex briefly. add_ observers coexist with the
    // primary handlers register_default_handlers installs (HandlerSlot).
    rt.model().process_jobs.add_state_change_handler(
        [this](const std::string& prjobid, gem::ProcessJobState,
               gem::ProcessJobState to, gem::ProcessJobEvent trig) {
          pb::ProcessJob::Action action;
          using PS = gem::ProcessJobState;
          using PE = gem::ProcessJobEvent;
          if (to == PS::Processing && trig == PE::Start)       action = pb::ProcessJob::START;
          else if (to == PS::Processing && trig == PE::Resume) action = pb::ProcessJob::RESUME;
          else if (to == PS::Paused)                           action = pb::ProcessJob::PAUSE;
          else if (to == PS::Stopping)                         action = pb::ProcessJob::STOP;
          else if (to == PS::Aborting)                         action = pb::ProcessJob::ABORT;
          else return;  // SettingUp/complete/etc.: not tool work
          pb::HostRequest hr;
          auto* job = hr.mutable_process_job();
          job->set_job_id(prjobid);
          job->set_action(action);
          // Observer runs on the io thread — reading the store is safe.
          if (const auto* pj = rt_.model().process_jobs.get(prjobid)) {
            job->set_recipe(pj->ppid);
            for (const auto& m : pj->mtrloutspec) job->add_carriers(m);
          }
          push_request(std::move(hr));
        });
    rt.model().recipes.add_added_handler(
        [this](const std::string& ppid, const std::string& body) {
          pb::HostRequest hr;
          hr.mutable_process_program()->set_ppid(ppid);
          hr.mutable_process_program()->set_body(body);
          push_request(std::move(hr));
        });
    for (const auto& rcmd : rt.model().commands.names()) commands_.push_back(rcmd);
    descr_model_ = rt.descriptor().model_name;
    descr_rev_   = rt.descriptor().software_rev;

    // E87 (D10): the host's carrier-ID decisions go to the tool. PROCEED on
    // a Confirmed transition (ProceedWithCarrier or Bind), CANCEL on
    // CancelCarrier. The tool announces carriers + drives access via
    // ReportCarrier.
    rt.model().carriers.add_id_handler(
        [this](const std::string& cid, gem::CarrierIDStatus,
               gem::CarrierIDStatus to, gem::CarrierIDEvent ev) {
          pb::CarrierAction::Action action;
          if (ev == gem::CarrierIDEvent::CancelCarrier)
            action = pb::CarrierAction::CANCEL;
          else if (to == gem::CarrierIDStatus::Confirmed)
            action = pb::CarrierAction::PROCEED;
          else
            return;
          pb::HostRequest hr;
          auto* ca = hr.mutable_carrier();
          ca->set_carrier_id(cid);
          if (const auto* c = rt_.model().carriers.get(cid))
            ca->set_port(c->port_id);
          ca->set_action(action);
          push_request(std::move(hr));
        });

    rt.model().ecids.add_changed_handler(
        [this](uint32_t ecid, const s2::Item& value) {
          pb::HostRequest hr;
          auto it = ecnames_.find(ecid);
          hr.mutable_constant()->set_name(
              it != ecnames_.end() ? it->second : std::to_string(ecid));
          *hr.mutable_constant()->mutable_value() = from_item(value);
          push_request(std::move(hr));
        });

    // 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<gem::CommandParameter>& params) {
        return forward_command(cmd, params, declarative);
      });
    }
  }

  grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req,
                            pb::Ack* resp) override {
    for (const auto& kv : req->values()) {
      auto it = vars_.find(kv.first);
      if (it == vars_.end()) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("no variable named '" + kv.first + "'");
        return grpc::Status::OK;
      }
      if (!value_is_set(kv.second)) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("value for '" + kv.first + "' has no kind set");
        return grpc::Status::OK;
      }
      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
    }
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  grpc::Status FireEvent(grpc::ServerContext*, const pb::Event* req,
                         pb::Ack* resp) override {
    // Optional per-fire variable values, then trigger the collection event.
    for (const auto& kv : req->data()) {
      auto it = vars_.find(kv.first);
      if (it == vars_.end()) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("no variable named '" + kv.first + "'");
        return grpc::Status::OK;
      }
      if (!value_is_set(kv.second)) {
        resp->set_code(pb::Ack::PARAMETER_INVALID);
        resp->set_message("value for '" + kv.first + "' has no kind set");
        return grpc::Status::OK;
      }
      rt_.set_variable(it->second.vid, to_item(kv.second, it->second.format));
    }
    auto ev = events_.find(req->name());
    if (ev == events_.end()) {
      resp->set_code(pb::Ack::PARAMETER_INVALID);
      resp->set_message("no event named '" + req->name() + "'");
      return grpc::Status::OK;
    }
    rt_.emit_event(ev->second);
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  grpc::Status SetAlarm(grpc::ServerContext*, const pb::Alarm* req,
                        pb::Ack* resp) override {
    return alarm_action(req->name(), /*set=*/true, resp);
  }

  grpc::Status ClearAlarm(grpc::ServerContext*, const pb::Alarm* req,
                          pb::Ack* resp) override {
    return alarm_action(req->name(), /*set=*/false, resp);
  }

  grpc::Status GetVariables(grpc::ServerContext*, const pb::VariableQuery* req,
                            pb::VariableSnapshot* resp) override {
    // Resolve names against the snapshot maps (empty query = everything).
    std::vector<std::pair<std::string, uint32_t>> wanted;
    if (req->names().empty()) {
      wanted.reserve(vars_.size());
      for (const auto& [name, ref] : vars_) wanted.emplace_back(name, ref.vid);
    } else {
      for (const auto& name : req->names()) {
        auto it = vars_.find(name);
        if (it == vars_.end())
          return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
                              "no variable named '" + name + "'");
        wanted.emplace_back(name, it->second.vid);
      }
    }
    // Mutable values live on the io thread — read them there (read_sync is
    // the standard pattern; see runtime.hpp).
    std::vector<uint32_t> vids;
    vids.reserve(wanted.size());
    for (const auto& w : wanted) vids.push_back(w.second);
    auto values = rt_.read_sync([this, vids]() {
      std::vector<std::optional<s2::Item>> out;
      out.reserve(vids.size());
      for (auto vid : vids) out.push_back(rt_.model().vid_value(vid));
      return out;
    });
    if (!values)
      return grpc::Status(grpc::StatusCode::UNAVAILABLE,
                          "engine io thread did not answer (not running?)");
    for (std::size_t i = 0; i < wanted.size(); ++i) {
      if (!(*values)[i]) continue;  // vid vanished — defensive, can't happen
      (*resp->mutable_values())[wanted[i].first] = from_item(*(*values)[i]);
    }
    return grpc::Status::OK;
  }

  grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*,
                               pb::ControlState* resp) override {
    // Thread-safe: control_state() reads the runtime's atomic mirror.
    resp->set_state(to_proto_state(rt_.control_state()));
    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<pb::HostRequest>* writer) override {
    auto sub = std::make_shared<Subscriber>();
    {
      std::lock_guard<std::mutex> lk(subs_mu_);
      subs_.push_back(sub);
    }
    rt_.log("tool client subscribed" +
            (req->client().empty() ? "" : " (" + req->client() + ")"));
    bool alive = true;
    // Poll at 100ms: with the sync server one thread is parked here per open
    // stream, so a cancelled client must free its worker promptly (gRPC's
    // sync API only exposes cancellation via polled IsCancelled()).
    while (alive && !ctx->IsCancelled()) {
      std::optional<pb::HostRequest> next;
      {
        std::unique_lock<std::mutex> lk(subs_mu_);
        subs_cv_.wait_for(lk, std::chrono::milliseconds(100),
                          [&] { 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<std::mutex> 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<std::mutex> 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<int>(req->ack().code())) + ")");
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  // E40 progress reports (Phase D, observe-and-report): the tool advances
  // the physical work and tells the engine; the engine drives the FSM and
  // emits the matching S16F9 alert / CEIDs to the host. PROCESSING is
  // informational (the engine entered it when the host commanded Start).
  grpc::Status ReportProcessJob(grpc::ServerContext*, const pb::ProcessJobState* req,
                                pb::Ack* resp) override {
    const std::string job_id = req->job_id();
    const auto state = req->state();
    auto outcome = rt_.read_sync([this, job_id, state]() -> std::optional<bool> {
      auto& pjs = rt_.model().process_jobs;
      if (!pjs.has(job_id)) return std::nullopt;
      switch (state) {
        case pb::ProcessJobState::SETTING_UP:
          return pjs.fire_internal(job_id, gem::ProcessJobEvent::SetupComplete);
        case pb::ProcessJobState::COMPLETE:
          return pjs.fire_internal(job_id, gem::ProcessJobEvent::ProcessComplete);
        case pb::ProcessJobState::ABORTED:
          return pjs.fire_internal(job_id, gem::ProcessJobEvent::AbortComplete);
        default:
          return true;  // PROCESSING: informational
      }
    });
    if (!outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer (not running?)");
    } else if (!*outcome) {
      resp->set_code(pb::Ack::INVALID_OBJECT);
      resp->set_message("no process job '" + job_id + "'");
    } else if (!**outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("E40 table rejects that transition from the current state");
    } else {
      resp->set_code(pb::Ack::ACCEPT);
    }
    return grpc::Status::OK;
  }

  // E87 carrier reporting (observe-and-report): WAITING announces an
  // arrived carrier (idempotent; updates the slot map), IN_ACCESS/COMPLETE
  // drive the access FSM. See the proto for the contract.
  grpc::Status ReportCarrier(grpc::ServerContext*, const pb::CarrierState* req,
                             pb::Ack* resp) override {
    const std::string cid = req->carrier_id();
    const auto port = static_cast<uint8_t>(req->port());
    const auto state = req->state();
    std::vector<bool> slots(req->slots().begin(), req->slots().end());
    auto outcome = rt_.read_sync(
        [this, cid, port, state, slots]() -> std::optional<bool> {
          auto& carriers = rt_.model().carriers;
          if (state == pb::CarrierState::WAITING) {
            carriers.create(cid, port,
                            slots.empty() ? 25 : slots.size());  // idempotent
            // E87 slot-map bytes: 0 = Empty, 1 = NotEmpty.
            for (std::size_t i = 0; i < slots.size(); ++i)
              carriers.set_slot_state(cid, i, slots[i] ? 1 : 0);
            if (!slots.empty())
              carriers.fire_slot_map_event(cid, gem::SlotMapEvent::Read);
            return true;
          }
          if (!carriers.has(cid)) return std::nullopt;
          if (state == pb::CarrierState::IN_ACCESS)
            return carriers.fire_access_event(cid, gem::CarrierAccessEvent::BeginAccess);
          if (state == pb::CarrierState::COMPLETE)
            return carriers.fire_access_event(cid, gem::CarrierAccessEvent::EndAccess);
          return true;
        });
    if (!outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer (not running?)");
    } else if (!*outcome) {
      resp->set_code(pb::Ack::INVALID_OBJECT);
      resp->set_message("no carrier '" + cid + "' (announce it with WAITING first)");
    } else if (!**outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("E87 access table rejects that transition");
    } else {
      resp->set_code(pb::Ack::ACCEPT);
    }
    return grpc::Status::OK;
  }

  grpc::Status ReportSubstrate(grpc::ServerContext*, const pb::SubstrateReport* req,
                               pb::Ack* resp) override {
    const std::string sid = req->substrate_id();
    const std::string cid = req->carrier_id();
    const auto slot = static_cast<uint8_t>(req->slot());
    const auto m = req->milestone();
    auto outcome = rt_.read_sync([this, sid, cid, slot, m]() -> std::optional<bool> {
      auto& subs = rt_.model().substrates;
      if (m == pb::SubstrateReport::ARRIVED) {
        subs.create(sid, cid, slot);  // AtSource / NeedsProcessing
        return true;
      }
      if (!subs.has(sid)) return std::nullopt;
      switch (m) {
        case pb::SubstrateReport::AT_WORK:
          return subs.fire_location_event(sid, gem::SubstrateEvent::Acquire, "");
        case pb::SubstrateReport::AT_DESTINATION:
          return subs.fire_location_event(sid, gem::SubstrateEvent::Release, "");
        case pb::SubstrateReport::PROCESSING:
          return subs.fire_processing_event(sid, gem::SubstrateProcessingEvent::StartProcessing);
        case pb::SubstrateReport::PROCESSED:
          return subs.fire_processing_event(sid, gem::SubstrateProcessingEvent::EndProcessing);
        default:
          return true;
      }
    });
    ack_from_outcome(outcome, resp, "substrate '" + sid + "'",
                     "E90 table rejects that transition");
    return grpc::Status::OK;
  }

  grpc::Status ReportModule(grpc::ServerContext*, const pb::ModuleReport* req,
                            pb::Ack* resp) override {
    const std::string mid = req->module_id();
    const auto st = req->state();
    auto outcome = rt_.read_sync([this, mid, st]() -> bool {
      auto& mods = rt_.model().modules;
      if (!mods.has(mid)) mods.create(mid);  // auto-create; starts NotExecuting
      switch (st) {
        case pb::ModuleReport::GENERAL_EXECUTING:
          return mods.fire(mid, gem::ModuleEvent::StartGeneral);
        case pb::ModuleReport::STEP_EXECUTING:
          return mods.fire(mid, gem::ModuleEvent::StartStep);
        case pb::ModuleReport::STEP_COMPLETED:
          return mods.fire(mid, gem::ModuleEvent::CompleteStep);
        case pb::ModuleReport::NOT_EXECUTING:
          return mods.fire(mid, gem::ModuleEvent::Reset);
        default:
          return true;
      }
    });
    // Module auto-creates, so "not found" can't happen; only the FSM verdict.
    if (!outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer");
    } else if (!*outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("E157 table rejects that transition from the current state");
    } else {
      resp->set_code(pb::Ack::ACCEPT);
    }
    return grpc::Status::OK;
  }

  grpc::Status Describe(grpc::ServerContext*, const pb::Empty*,
                        pb::EquipmentDescription* resp) override {
    resp->set_model_name(descr_model_);
    resp->set_software_rev(descr_rev_);
    for (const auto& [name, _] : vars_) resp->add_variables(name);
    for (const auto& [name, _] : events_) resp->add_events(name);
    for (const auto& [name, _] : alarms_) resp->add_alarms(name);
    for (const auto& rcmd : commands_) resp->add_commands(rcmd);
    for (const auto& [_, name] : ecnames_) resp->add_constants(name);
    return grpc::Status::OK;
  }

  grpc::Status FlushSpool(grpc::ServerContext*, const pb::SpoolFlushRequest* req,
                          pb::Ack* resp) override {
    const bool purge = req->purge();
    auto done = rt_.read_sync([this, purge]() {
      if (purge) rt_.model().spool.clear();
      else       rt_.model().spool.drain();  // toward the host (see S6F23)
      return true;
    });
    resp->set_code(done ? pb::Ack::ACCEPT : pb::Ack::CANNOT_DO_NOW);
    if (!done) resp->set_message("engine io thread did not answer");
    return grpc::Status::OK;
  }

  grpc::Status SendTerminalMessage(grpc::ServerContext*, const pb::TerminalMessage* req,
                                   pb::Ack* resp) override {
    const auto tid = static_cast<uint8_t>(req->tid());
    const std::string text = req->text();
    auto delivered = rt_.read_sync([this, tid, text]() {
      return rt_.deliver_or_spool(
          gem::s10f1_terminal_display_single(tid, text), "S10F1 (tool)");
    });
    if (!delivered) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer");
    } else if (!*delivered) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("no host connected and stream 10 is not spoolable");
    } else {
      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,
  // CANNOT_DO_NOW (naming the actual state) otherwise. Note the shipped table
  // has no operator path to EQUIPMENT_OFFLINE — operator_offline lands
  // HOST_OFFLINE — so honesty matters here.
  grpc::Status RequestControlState(grpc::ServerContext*,
                                   const pb::ControlStateRequest* req,
                                   pb::Ack* resp) override {
    using PS = pb::ControlState;
    const auto desired = req->desired();
    if (desired == PS::ATTEMPT_ONLINE) {
      resp->set_code(pb::Ack::PARAMETER_INVALID);
      resp->set_message("ATTEMPT_ONLINE is transient; request a settled state");
      return grpc::Status::OK;
    }
    auto final_state = rt_.read_sync([this, desired]() {
      auto& sm = rt_.control();
      switch (desired) {
        case PS::ONLINE_LOCAL:
          if (sm.state() == gem::ControlState::OnlineRemote) sm.operator_local();
          else if (!sm.online()) sm.operator_online();  // table chains to OnlineLocal
          break;
        case PS::ONLINE_REMOTE:
          if (sm.state() == gem::ControlState::OnlineLocal) sm.operator_remote();
          else if (!sm.online()) {
            sm.operator_online();
            sm.operator_remote();
          }
          break;
        case PS::HOST_OFFLINE:
        case PS::EQUIPMENT_OFFLINE:
          sm.operator_offline();
          break;
        default:
          break;
      }
      return sm.state();
    });
    if (!final_state) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer (not running?)");
      return grpc::Status::OK;
    }
    if (to_proto_state(*final_state) == desired) {
      resp->set_code(pb::Ack::ACCEPT);
    } else {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message(std::string("equipment is in ") +
                        gem::control_state_name(*final_state));
    }
    return grpc::Status::OK;
  }

  // Streams a Health snapshot immediately, then again whenever the link or
  // control state changes (and on spool-depth changes, sampled at the poll
  // interval). Ends when the client cancels or the engine stops.
  grpc::Status WatchHealth(grpc::ServerContext* ctx, const pb::Empty*,
                           grpc::ServerWriter<pb::Health>* writer) override {
    bool first = true;
    pb::Health last;
    while (!ctx->IsCancelled()) {
      const uint64_t seen = health_version_.load(std::memory_order_relaxed);
      auto snap = make_health();
      if (!snap) break;  // engine stopped — end the stream
      if (first || snap->link() != last.link() ||
          snap->control_state() != last.control_state() ||
          snap->spool_depth() != last.spool_depth()) {
        if (!writer->Write(*snap)) break;
        last = *snap;
        first = false;
      }
      std::unique_lock<std::mutex> lk(health_mu_);
      health_cv_.wait_for(lk, std::chrono::milliseconds(500), [&] {
        return health_version_.load(std::memory_order_relaxed) != seen;
      });
    }
    return grpc::Status::OK;
  }

 private:
  // Map a read_sync(std::optional<bool>) result to an Ack: outer nullopt =
  // io thread didn't answer; inner nullopt = object not found; false = FSM
  // rejected; true = accepted.
  void ack_from_outcome(const std::optional<std::optional<bool>>& outcome,
                        pb::Ack* resp, const std::string& what,
                        const std::string& reject_msg) {
    if (!outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message("engine io thread did not answer (not running?)");
    } else if (!*outcome) {
      resp->set_code(pb::Ack::INVALID_OBJECT);
      resp->set_message("no " + what);
    } else if (!**outcome) {
      resp->set_code(pb::Ack::CANNOT_DO_NOW);
      resp->set_message(reject_msg);
    } else {
      resp->set_code(pb::Ack::ACCEPT);
    }
  }

  // Fan a HostRequest out to every subscriber (fire-and-forget
  // notifications: jobs, recipes, EC changes). No subscriber = dropped,
  // matching the no-buffering contract.
  void push_request(pb::HostRequest hr) {
    std::lock_guard<std::mutex> lk(subs_mu_);
    for (auto& sub : subs_) sub->queue.push_back(hr);
    if (!subs_.empty()) subs_cv_.notify_all();
  }

  // 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<gem::CommandParameter>& params,
                                  gem::HostCmdAck declarative) {
    std::lock_guard<std::mutex> 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();
  }

  // Build a Health snapshot. Control state + link come from atomics; spool
  // depth is mutable engine state, read via read_sync. nullopt = engine down.
  std::optional<pb::Health> make_health() {
    auto depth = rt_.read_sync(
        [this]() { return static_cast<uint32_t>(rt_.model().spool.size()); });
    if (!depth) return std::nullopt;
    pb::Health h;
    // CONNECTED (TCP up, not yet SELECTED) is reserved: the runtime's link
    // observer fires on SELECTED/closed only. TODO(daemon): surface the
    // intermediate state if a tool ever needs it.
    h.set_link(link_selected_.load(std::memory_order_relaxed)
                   ? pb::Health::SELECTED
                   : pb::Health::DISCONNECTED);
    h.set_spool_depth(*depth);
    h.set_control_state(to_proto_state(rt_.control_state()));
    return h;
  }

  grpc::Status alarm_action(const std::string& name, bool set, pb::Ack* resp) {
    auto it = alarms_.find(name);
    if (it == alarms_.end()) {
      resp->set_code(pb::Ack::PARAMETER_INVALID);
      resp->set_message("no alarm named '" + name + "'");
      return grpc::Status::OK;
    }
    if (set) rt_.set_alarm(it->second);
    else     rt_.clear_alarm(it->second);
    resp->set_code(pb::Ack::ACCEPT);
    return grpc::Status::OK;
  }

  struct VarRef {
    uint32_t vid;
    s2::Format format;  // declared wire format from equipment.yaml
  };

  gem::EquipmentRuntime& rt_;
  std::map<std::string, VarRef> vars_;     // SVIDs + DVIDs (SVIDs win on clash)
  std::map<std::string, uint32_t> events_; // CEID by name
  std::map<std::string, uint32_t> alarms_; // ALID by name AND stringified id
  std::map<uint32_t, std::string> ecnames_; // ECID -> name (ConstantChange)
  std::vector<std::string> commands_;       // RCMDs (Describe)
  std::string descr_model_, descr_rev_;     // device header (Describe)

  // WatchHealth plumbing: observers (io thread) bump the version and wake the
  // per-stream wait loops (gRPC threads).
  std::atomic<bool> link_selected_{false};
  std::atomic<uint64_t> 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<pb::HostRequest> queue;
  };
  std::mutex subs_mu_;
  std::condition_variable subs_cv_;
  std::vector<std::shared_ptr<Subscriber>> subs_;
  std::map<std::string, std::string> pending_;  // command id -> rcmd (audit)
  uint64_t next_command_id_ = 1;                // guarded by subs_mu_
};

}  // namespace secsgem::daemon