// Passive SECS/GEM equipment. Capabilities (SVIDs, ECIDs, CEIDs, alarms,
// recipes, host commands) come from data/equipment.yaml; the E30 control
// state machine comes from data/control_state.yaml. Dispatch is a Router
// table.  No imperative if-ladder; no in-code device data dictionary.

#include <asio.hpp>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <memory>
#include <optional>
#include <string>
#include <vector>

#include "secsgem/config/loader.hpp"
#include "secsgem/config/validate.hpp"
#include "secsgem/endpoint.hpp"
#include "secsgem/gem/control_state.hpp"
#include "secsgem/gem/data_model.hpp"
#include "secsgem/gem/e116_constants.hpp"
#include "secsgem/gem/e157_constants.hpp"
#include "secsgem/gem/e90_constants.hpp"
#include "secsgem/gem/messages.hpp"
#include "secsgem/gem/router.hpp"
#include "secsgem/secs2/message.hpp"

using namespace secsgem;
namespace s2 = secsgem::secs2;
namespace gem = secsgem::gem;

namespace {

std::string arg(int argc, char** argv, const std::string& key, const std::string& def) {
  for (int i = 1; i + 1 < argc; ++i)
    if (key == argv[i]) return argv[i + 1];
  return def;
}

bool has_flag(int argc, char** argv, const std::string& key) {
  for (int i = 1; i < argc; ++i)
    if (key == argv[i]) return true;
  return false;
}

constexpr uint32_t kSvidControlState = 1;
constexpr uint32_t kSvidClock        = 2;

void refresh(gem::EquipmentDataModel& m, const gem::ControlStateMachine& sm) {
  m.svids.set_value(kSvidControlState, s2::Item::ascii(gem::control_state_name(sm.state())));
  m.svids.set_value(kSvidClock, s2::Item::ascii(m.clock.current_time_string()));
}

}  // namespace

int main(int argc, char** argv) {
  const auto port = static_cast<uint16_t>(std::stoi(arg(argc, argv, "--port", "5000")));
  // Defaults are relative to the current working directory so the
  // binary works both inside the docker image (WORKDIR=/app) and on a
  // host build run from the repo root.  Pass explicit --config etc.
  // when running from anywhere else.
  const auto equipment_yaml = arg(argc, argv, "--config", "data/equipment.yaml");
  const auto state_yaml     = arg(argc, argv, "--state-table", "data/control_state.yaml");
  const auto pj_state_yaml  = arg(argc, argv, "--pj-state-table",
                                   "data/process_job_state.yaml");
  const auto cj_state_yaml  = arg(argc, argv, "--cj-state-table",
                                   "data/control_job_state.yaml");
  const auto spool_dir      = arg(argc, argv, "--spool-dir", "");
  const bool validate_only  = has_flag(argc, argv, "--validate-config");

  auto logfn = [](const std::string& m) { std::cout << "[equip] " << m << std::endl; };

  // --validate-config: read every YAML, accumulate every issue we can
  // find, print to stderr, and exit 0/1.  Does NOT bind the port — this
  // is the day-1 friction killer the README points customers at.
  if (validate_only) {
    config::ConfigValidator v;
    v.validate_equipment(equipment_yaml);
    v.validate_control_state(state_yaml);
    v.validate_process_job_state(pj_state_yaml);
    v.validate_control_job_state(cj_state_yaml);
    config::format_issues_to(std::cerr, v.issues());
    std::cerr << v.error_count() << " error(s), "
              << v.warning_count() << " warning(s) across 4 files\n";
    return v.has_errors() ? 1 : 0;
  }

  auto model = std::make_shared<gem::EquipmentDataModel>();
  if (!spool_dir.empty()) {
    model->spool.enable_persistence(spool_dir);
    logfn("spool: persisting to " + spool_dir +
          " (replayed " + std::to_string(model->spool.size()) + " messages)");
  }
  config::EquipmentDescriptor desc;
  config::ControlStateConfig sm_cfg;
  try {
    desc = config::load_equipment(equipment_yaml, *model);
    sm_cfg = config::load_control_state(state_yaml);
  } catch (const std::exception& e) {
    std::cerr << "[equip] config error: " << e.what() << std::endl;
    return 1;
  }
  logfn("loaded " + std::to_string(model->svids.size()) + " SVIDs, " +
        std::to_string(model->ecids.all().size()) + " ECIDs, " +
        std::to_string(model->events.all_events().size()) + " CEIDs, " +
        std::to_string(model->alarms.all().size()) + " alarms");

  auto sm = std::make_shared<gem::ControlStateMachine>(sm_cfg.table, sm_cfg.initial);

  asio::io_context io;
  Server::Config server_cfg{port, desc.device_id, {}};
  Server server(io, server_cfg);
  server.on_log(logfn);

  auto active_conn = std::make_shared<std::weak_ptr<Connection>>();

  // Deliver a primary message to the host, or spool it if we can't (either
  // there is no SELECTED session or the spool's force-flag is on).  Returns
  // true if the message was delivered or queued; false if dropped because
  // the stream isn't spoolable and there is no live session.
  auto deliver_or_spool = [active_conn, model, logfn](s2::Message msg,
                                                      const std::string& what) -> bool {
    auto conn = active_conn->lock();
    const bool spooling = model->spool.force_spool() || !conn;
    if (spooling) {
      auto r = model->spool.enqueue(msg);
      if (r == gem::SpoolStore::EnqueueResult::Queued) {
        logfn("spool: " + what + " queued (depth=" +
              std::to_string(model->spool.size()) + ")");
        return true;
      }
      logfn("spool: " + what + " dropped (stream not spoolable, no host)");
      return false;
    }
    // W=1 primaries use send_request (transaction tracking); W=0 primaries
    // (e.g. S16F9 PRJobAlert) go via send_data so we don't register a
    // never-arriving "reply" and time out on T3.
    if (msg.reply_expected) {
      conn->send_request(std::move(msg), [](std::error_code, const s2::Message&) {});
    } else {
      conn->send_data(std::move(msg));
    }
    return true;
  };

  auto emit_event = [&io, model, logfn, deliver_or_spool](uint32_t ceid) {
    asio::post(io, [model, logfn, deliver_or_spool, ceid]() {
      if (!model->is_event_enabled(ceid)) {
        logfn("CEID " + std::to_string(ceid) + " not enabled; suppressed");
        return;
      }
      auto reports = model->compose_reports_for(ceid);
      logfn("emit S6F11 CEID=" + std::to_string(ceid) + " (" +
            std::to_string(reports.size()) + " reports)");
      deliver_or_spool(gem::s6f11_event_report(0, ceid, reports),
                       "S6F11 CEID=" + std::to_string(ceid));
    });
  };

  auto emit_alarm_set = [&io, model, logfn, emit_event, deliver_or_spool](uint32_t alid) {
    asio::post(io, [model, logfn, emit_event, deliver_or_spool, alid]() {
      auto alarm = model->alarms.get(alid);
      if (!alarm) return;
      auto alcd = model->alarms.set_active(alid);
      if (!alcd) return;
      if (model->alarms.enabled(alid)) {
        logfn("emit S5F1 alarm set ALID=" + std::to_string(alid));
        deliver_or_spool(gem::s5f1_alarm_report(*alcd, alid, alarm->text),
                         "S5F1 ALID=" + std::to_string(alid));
      } else {
        logfn("alarm " + std::to_string(alid) + " not enabled; suppressed");
      }
      // E30: an AlarmSetEvent CEID also fires (if linked + enabled).
    });
  };

  sm->set_state_change_handler(
      [logfn, emit_event, desc](gem::ControlState from, gem::ControlState to, gem::ControlEvent ev) {
        logfn(std::string("control: ") + gem::control_state_name(from) + " -> " +
              gem::control_state_name(to) + "  (" + gem::control_event_name(ev) + ")");
        if (desc.emit_on_control_change) emit_event(*desc.emit_on_control_change);
      });

  // ---- E40/E94: load the PJ/CJ transition tables and wire emitters -----
  // (pj_state_yaml / cj_state_yaml already parsed at the top so
  // --validate-config sees them.)
  config::ProcessJobStateConfig pj_cfg;
  config::ControlJobStateConfig cj_cfg;
  try {
    pj_cfg = config::load_process_job_state(pj_state_yaml);
    cj_cfg = config::load_control_job_state(cj_state_yaml);
  } catch (const std::exception& e) {
    std::cerr << "[equip] E40/E94 config error: " << e.what() << std::endl;
    return 1;
  }
  // Each new PJ/CJ gets a fresh copy of the loaded transition table.
  model->process_jobs.set_table_factory([t = pj_cfg.table]() { return t; });
  model->control_jobs.set_table_factory([t = cj_cfg.table]() { return t; });

  // Emit S16F9 PRJobAlert (E40-0705 §10.3). Equipment-initiated; the spec
  // is silent on reply expectation, so we send it as a primary one-way.
  auto emit_pj_alert = [&io, model, logfn, deliver_or_spool](
                          const std::string& prjobid,
                          gem::ProcessJobState state) {
    asio::post(io, [model, logfn, deliver_or_spool, prjobid, state]() {
      const auto* pj = model->process_jobs.get(prjobid);
      if (pj && !pj->alert_enabled) return;
      logfn("emit S16F9 PJ=" + prjobid + " state=" +
            gem::process_job_state_name(state));
      deliver_or_spool(gem::s16f9_pr_job_alert(prjobid, state),
                       "S16F9 PJ=" + prjobid);
    });
  };

  // PJ state-change handler: log + emit S16F9 (skip the synthetic
  // NoState->Queued so we don't alert on a freshly-created PJ that's
  // still being acked).
  model->process_jobs.set_state_change_handler(
      [logfn, emit_pj_alert](const std::string& prjobid,
                             gem::ProcessJobState from,
                             gem::ProcessJobState to,
                             gem::ProcessJobEvent trig) {
        logfn(std::string("PJ ") + prjobid + ": " +
              gem::process_job_state_name(from) + " -> " +
              gem::process_job_state_name(to) + "  (" +
              gem::process_job_event_name(trig) + ")");
        if (from != gem::ProcessJobState::NoState) emit_pj_alert(prjobid, to);
      });

  // CEIDs the equipment.yaml is expected to register for CJ state
  // changes (best-effort: if missing they're silently no-ops via the
  // existing CEID-not-enabled guard in emit_event).
  constexpr uint32_t kCeidCJExecuting = 400;
  constexpr uint32_t kCeidCJCompleted = 401;

  model->control_jobs.set_state_change_handler(
      [logfn, emit_event](const std::string& ctljobid,
                          gem::ControlJobState from,
                          gem::ControlJobState to,
                          gem::ControlJobEvent trig) {
        logfn(std::string("CJ ") + ctljobid + ": " +
              gem::control_job_state_name(from) + " -> " +
              gem::control_job_state_name(to) + "  (" +
              gem::control_job_event_name(trig) + ")");
        if (to == gem::ControlJobState::Executing) emit_event(kCeidCJExecuting);
        if (to == gem::ControlJobState::Completed) emit_event(kCeidCJCompleted);
      });

  // Drive the contained PJs through the demo lifecycle when the host
  // tells the CJ to start. Real equipment would step PJs one at a time
  // as material arrives; our simulator cascades through every legal
  // state so the wire trace exercises the whole FSM.
  auto run_cj_lifecycle = [&io, model, logfn](const std::string& ctljobid) {
    asio::post(io, [model, logfn, ctljobid]() {
      auto* cj = model->control_jobs.get(ctljobid);
      if (!cj) return;
      // CJ promotion path: Queued -> Selected -> WaitingForStart -> Executing.
      model->control_jobs.fire_internal(ctljobid, gem::ControlJobEvent::Select);
      model->control_jobs.fire_internal(ctljobid, gem::ControlJobEvent::SetupComplete);
      model->control_jobs.fire_internal(ctljobid, gem::ControlJobEvent::Start);
      // Cascade every contained PJ through to ProcessComplete.
      for (const auto& pjid : cj->prjobids) {
        model->process_jobs.fire_internal(pjid, gem::ProcessJobEvent::Select);
        model->process_jobs.fire_internal(pjid, gem::ProcessJobEvent::SetupComplete);
        model->process_jobs.fire_internal(pjid, gem::ProcessJobEvent::Start);
        model->process_jobs.fire_internal(pjid, gem::ProcessJobEvent::ProcessComplete);
      }
      // All PJs done -> CJ Completed.
      model->control_jobs.fire_internal(ctljobid, gem::ControlJobEvent::AllJobsComplete);
      logfn("CJ " + ctljobid + " lifecycle complete");
    });
  };

  // ---- E5 exception state-change emitters -------------------------------
  // Translate ExceptionStore state changes into S5F9/F11/F15 on the wire,
  // mirroring how PJ state changes become S16F9.  The synthetic
  // NoState->Posted event is the trigger for emitting S5F9 (we look up
  // the metadata from the store so we can build the full body).
  model->exceptions.set_state_change_handler(
      [model, logfn, deliver_or_spool](uint32_t exid,
                                       gem::ExceptionState from,
                                       gem::ExceptionState to,
                                       gem::ExceptionEvent trig) {
        logfn(std::string("EX ") + std::to_string(exid) + ": " +
              gem::exception_state_name(from) + " -> " +
              gem::exception_state_name(to) + "  (" +
              gem::exception_event_name(trig) + ")");

        if (from == gem::ExceptionState::NoState &&
            to   == gem::ExceptionState::Posted) {
          const auto* ex = model->exceptions.get(exid);
          if (!ex) return;
          deliver_or_spool(
              gem::s5f9_exception_post_notify(exid, ex->extype, ex->exmessage,
                                              ex->exrecvra),
              "S5F9 EXID=" + std::to_string(exid));
          return;
        }

        if (to == gem::ExceptionState::Cleared) {
          if (from == gem::ExceptionState::Recovering) {
            // RecoveryComplete -> emit S5F15 with success EXRESULT.
            deliver_or_spool(
                gem::s5f15_exception_recover_complete_notify(exid, "OK"),
                "S5F15 EXID=" + std::to_string(exid));
          }
          // S5F11 for any Cleared transition that didn't go through
          // Recovering (autonomous clear or post-failure clear).
          if (from == gem::ExceptionState::Posted ||
              from == gem::ExceptionState::RecoverFailed) {
            deliver_or_spool(
                gem::s5f11_exception_clear_notify(exid, "", ""),
                "S5F11 EXID=" + std::to_string(exid));
          }
          return;
        }

        if (from == gem::ExceptionState::Recovering &&
            to   == gem::ExceptionState::RecoverFailed) {
          deliver_or_spool(
              gem::s5f15_exception_recover_complete_notify(exid, "FAILED"),
              "S5F15 EXID=" + std::to_string(exid));
        }
      });

  // ---- E90 substrate state-change emitters -----------------------------
  // Map SubstrateState / SubstrateProcessingState transitions to the
  // standard E90 CEIDs.  Only the post-transition state matters; the
  // event-report machinery already gates on enabled CEIDs so emitting
  // for every transition is safe and gives the host a complete trace.
  model->substrates.set_location_handler(
      [logfn, emit_event](const std::string& substid,
                          gem::SubstrateState from,
                          gem::SubstrateState to,
                          gem::SubstrateEvent ev) {
        logfn(std::string("SUB ") + substid + ": " +
              gem::substrate_state_name(from) + " -> " +
              gem::substrate_state_name(to) + "  (" +
              gem::substrate_event_name(ev) + ")");
        switch (to) {
          case gem::SubstrateState::AtSource:
            emit_event(gem::e90::kCeidSubstrateAtSource); break;
          case gem::SubstrateState::AtWork:
            emit_event(gem::e90::kCeidSubstrateAtWork); break;
          case gem::SubstrateState::AtDestination:
            emit_event(gem::e90::kCeidSubstrateAtDestination); break;
          case gem::SubstrateState::NoState: break;
        }
      });

  model->substrates.set_processing_handler(
      [logfn, emit_event](const std::string& substid,
                          gem::SubstrateProcessingState from,
                          gem::SubstrateProcessingState to,
                          gem::SubstrateProcessingEvent ev) {
        logfn(std::string("SUB ") + substid + " proc: " +
              gem::substrate_processing_state_name(from) + " -> " +
              gem::substrate_processing_state_name(to) + "  (" +
              gem::substrate_processing_event_name(ev) + ")");
        switch (to) {
          case gem::SubstrateProcessingState::NeedsProcessing:
            emit_event(gem::e90::kCeidSubstrateNeedsProcessing); break;
          case gem::SubstrateProcessingState::InProcess:
            emit_event(gem::e90::kCeidSubstrateInProcess); break;
          case gem::SubstrateProcessingState::Processed:
            emit_event(gem::e90::kCeidSubstrateProcessed); break;
          case gem::SubstrateProcessingState::Aborted:
            emit_event(gem::e90::kCeidSubstrateAborted); break;
          case gem::SubstrateProcessingState::Stopped:
            emit_event(gem::e90::kCeidSubstrateStopped); break;
          case gem::SubstrateProcessingState::Rejected:
            emit_event(gem::e90::kCeidSubstrateRejected); break;
          case gem::SubstrateProcessingState::Lost:
            emit_event(gem::e90::kCeidSubstrateLost); break;
          case gem::SubstrateProcessingState::Skipped:
            emit_event(gem::e90::kCeidSubstrateSkipped); break;
          case gem::SubstrateProcessingState::NoState: break;
        }
      });

  // ---- E116 EPT state-change emitter -----------------------------------
  model->ept.set_state_change_handler(
      [logfn, emit_event](gem::EptState from, gem::EptState to,
                          gem::EptEvent ev, std::chrono::milliseconds dwell) {
        logfn(std::string("EPT: ") + gem::ept_state_name(from) + " -> " +
              gem::ept_state_name(to) + "  (" + gem::ept_event_name(ev) +
              ", dwelt " + std::to_string(dwell.count()) + "ms)");
        switch (to) {
          case gem::EptState::NonScheduledTime:
            emit_event(gem::e116::kCeidNonScheduledTime); break;
          case gem::EptState::ScheduledDowntime:
            emit_event(gem::e116::kCeidScheduledDowntime); break;
          case gem::EptState::UnscheduledDowntime:
            emit_event(gem::e116::kCeidUnscheduledDowntime); break;
          case gem::EptState::Engineering:
            emit_event(gem::e116::kCeidEngineering); break;
          case gem::EptState::Standby:
            emit_event(gem::e116::kCeidStandby); break;
          case gem::EptState::Productive:
            emit_event(gem::e116::kCeidProductive); break;
          case gem::EptState::NoState: break;
        }
      });

  // ---- E157 module state-change emitter --------------------------------
  // Every module transition fires the generic ModuleProcessStateChange
  // CEID plus the state-specific one, mirroring secsgem-py's per-module
  // CEID granularity.  Hosts that don't subscribe to the specific CEID
  // can still listen on the generic one to drive an internal FSM.
  model->modules.set_state_change_handler(
      [logfn, emit_event](const std::string& mod, gem::ModuleState from,
                          gem::ModuleState to, gem::ModuleEvent ev) {
        logfn(std::string("MOD ") + mod + ": " +
              gem::module_state_name(from) + " -> " +
              gem::module_state_name(to) + "  (" +
              gem::module_event_name(ev) + ")");
        emit_event(gem::e157::kCeidModuleProcessStateChange);
        switch (to) {
          case gem::ModuleState::NotExecuting:
            emit_event(gem::e157::kCeidModuleNotExecuting); break;
          case gem::ModuleState::GeneralExecuting:
            emit_event(gem::e157::kCeidModuleGeneralExecuting); break;
          case gem::ModuleState::StepExecuting:
            emit_event(gem::e157::kCeidModuleStepExecuting); break;
          case gem::ModuleState::StepCompleted:
            emit_event(gem::e157::kCeidModuleStepCompleted); break;
          case gem::ModuleState::NoState: break;
        }
      });

  // ---- Build the SECS dispatch table once -------------------------------
  gem::Router router;

  router.on(1, 1, [desc, logfn](const s2::Message&) {
    logfn("S1F1 -> S1F2");
    return gem::s1f2_on_line_data(desc.model_name, desc.software_rev);
  });
  router.on(1, 3, [model, sm, logfn](const s2::Message& msg) -> std::optional<s2::Message> {
    refresh(*model, *sm);
    auto svids = gem::parse_s1f3(msg);
    if (!svids) return s2::Message(1, 0, false);
    std::vector<std::optional<s2::Item>> values;
    if (svids->empty()) {
      for (const auto& sv : model->svids.all()) values.push_back(sv.value);
    } else {
      for (auto id : *svids) {
        auto sv = model->svids.get(id);
        values.push_back(sv ? std::optional<s2::Item>(sv->value) : std::nullopt);
      }
    }
    logfn("S1F3 -> S1F4 (" + std::to_string(values.size()) + " values)");
    return gem::s1f4_selected_status_data(values);
  });
  router.on(1, 11, [model, logfn](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& sv : model->svids.all())
      rows.push_back({sv.id, sv.name, sv.units});
    logfn("S1F11 -> S1F12 (namelist, " + std::to_string(rows.size()) + ")");
    return gem::s1f12_status_namelist_data(rows);
  });
  router.on(1, 13, [desc, logfn](const s2::Message&) {
    logfn("S1F13 -> S1F14");
    return gem::s1f14_establish_comms_ack(gem::CommAck::Accept,
                                          {desc.model_name, desc.software_rev});
  });
  router.on(1, 15, [sm, logfn](const s2::Message&) {
    auto ack = sm->on_host_request_offline();
    logfn("S1F15 -> S1F16 OFLACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s1f16_offline_ack(ack);
  });
  router.on(1, 17, [sm, logfn](const s2::Message&) {
    auto ack = sm->on_host_request_online();
    logfn("S1F17 -> S1F18 ONLACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s1f18_online_ack(ack);
  });
  router.on(1, 19, [desc, logfn](const s2::Message&) {
    std::vector<gem::CapabilityEntry> caps;
    for (const auto& c : desc.capabilities) caps.push_back({c.first, c.second});
    logfn("S1F19 -> S1F20 (" + std::to_string(caps.size()) + " capabilities)");
    return gem::s1f20_get_gem_compliance_data(desc.software_rev, desc.equipment_type, caps);
  });
  router.on(1, 21, [model, logfn](const s2::Message&) {
    std::vector<gem::StatusName> rows;
    for (const auto& dv : model->dvids.all())
      rows.push_back({dv.id, dv.name, dv.units});
    logfn("S1F21 -> S1F22 (" + std::to_string(rows.size()) + " DVIDs)");
    return gem::s1f22_data_variable_namelist_data(rows);
  });
  // S1F23 — Collection Event Namelist Request.  Empty CEID list means
  // "every CEID in the catalog"; otherwise we filter to the requested
  // set and silently skip unknown CEIDs (per SEMI E5: "all unidentified
  // are returned in S1F24 with an empty VID list").
  router.on(1, 23, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s1f23(msg);
    std::vector<gem::CollectionEventName> rows;
    if (req && req->empty()) {
      for (const auto& e : model->events.all_events())
        rows.push_back({e.id, e.name, model->events.vids_for(e.id)});
    } else if (req) {
      for (auto id : *req) {
        if (auto info = model->events.event_info(id)) {
          rows.push_back({id, info->name, model->events.vids_for(id)});
        } else {
          rows.push_back({id, "", {}});
        }
      }
    }
    logfn("S1F23 -> S1F24 (" + std::to_string(rows.size()) + " CEIDs)");
    return gem::s1f24_collection_event_namelist_data(rows);
  });

  router.on(2, 13, [model, logfn](const s2::Message& msg) -> std::optional<s2::Message> {
    auto ids = gem::parse_u4_list_body(msg);
    if (!ids) return s2::Message(2, 0, false);
    std::vector<s2::Item> values;
    for (auto id : *ids) {
      auto ec = model->ecids.get(id);
      values.push_back(ec ? ec->value : s2::Item::list({}));
    }
    logfn("S2F13 -> S2F14 (" + std::to_string(values.size()) + " values)");
    return gem::s2f14_ec_data(values);
  });
  router.on(2, 15, [model, logfn](const s2::Message& msg) {
    auto sets = gem::parse_s2f15(msg);
    auto eac = gem::EquipmentAck::Accept;
    if (!sets) eac = gem::EquipmentAck::Denied_OutOfRange;
    else
      for (const auto& s : *sets) {
        auto r = model->ecids.set_value(s.ecid, s.value);
        if (r != gem::EquipmentAck::Accept) eac = r;
      }
    logfn("S2F15 -> S2F16 EAC=" + std::to_string(static_cast<int>(eac)));
    return gem::s2f16_ec_ack(eac);
  });
  router.on(2, 17, [model, logfn](const s2::Message&) {
    logfn("S2F17 -> S2F18 (clock)");
    return gem::s2f18_date_time_data(model->clock.current_time_string());
  });
  router.on(2, 29, [model, logfn](const s2::Message& msg) {
    auto ids = gem::parse_u4_list_body(msg);
    std::vector<gem::EquipmentConstant> ecs;
    if (ids && ids->empty()) ecs = model->ecids.all();
    else if (ids)
      for (auto id : *ids) {
        auto ec = model->ecids.get(id);
        if (ec) ecs.push_back(*ec);
      }
    std::vector<gem::EcNameRow> rows;
    for (const auto& ec : ecs)
      rows.push_back({ec.id, ec.name, ec.min_str, ec.max_str, "", ec.units});
    logfn("S2F29 -> S2F30 (" + std::to_string(rows.size()) + " ECs)");
    return gem::s2f30_ec_namelist_data(rows);
  });
  router.on(2, 31, [model, logfn](const s2::Message& msg) {
    auto t = gem::parse_s2f31(msg);
    auto ack = t ? model->clock.set_time_string(*t) : gem::TimeAck::Error;
    logfn("S2F31 -> S2F32 TIACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f32_date_time_ack(ack);
  });
  router.on(2, 33, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f33(msg);
    auto ack = gem::DefineReportAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      rows.reserve(req->reports.size());
      for (const auto& r : req->reports) rows.emplace_back(r.rptid, r.vids);
      ack = model->define_reports(rows);
    }
    logfn("S2F33 -> S2F34 DRACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f34_define_report_ack(ack);
  });
  router.on(2, 35, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f35(msg);
    auto ack = gem::LinkEventAck::InvalidFormat;
    if (req) {
      std::vector<std::pair<uint32_t, std::vector<uint32_t>>> rows;
      rows.reserve(req->links.size());
      for (const auto& l : req->links) rows.emplace_back(l.ceid, l.rptids);
      ack = model->link_event_reports(rows);
    }
    logfn("S2F35 -> S2F36 LRACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f36_link_event_report_ack(ack);
  });
  router.on(2, 37, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f37(msg);
    auto ack = req ? model->enable_events(req->enable, req->ceids)
                   : gem::EnableEventAck::UnknownCeid;
    logfn(std::string("S2F37 ") + (req && req->enable ? "enable" : "disable") +
          " -> S2F38 ERACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f38_enable_event_ack(ack);
  });
  router.on(2, 41, [model, logfn, emit_event, emit_alarm_set](const s2::Message& msg) {
    auto cmd = gem::parse_s2f41(msg);
    if (!cmd) return gem::s2f42_host_command_ack(gem::HostCmdAck::ParameterInvalid, {});
    auto result = model->commands.dispatch(cmd->rcmd, cmd->params);
    logfn("S2F41 RCMD=" + cmd->rcmd + " -> S2F42 HCACK=" +
          std::to_string(static_cast<int>(result.ack)));
    if (result.ack == gem::HostCmdAck::Accept) {
      if (result.emit_ceid) emit_event(*result.emit_ceid);
      if (result.set_alarm) emit_alarm_set(*result.set_alarm);
      if (result.force_spool) {
        model->spool.set_force_spool(*result.force_spool);
        logfn(std::string("spool: force_spool=") + (*result.force_spool ? "true" : "false") +
              " (depth=" + std::to_string(model->spool.size()) + ")");
      }
    }
    return gem::s2f42_host_command_ack(result.ack, {});
  });

  // S2F21 — legacy Remote Command (no parameter list).  Delegated to
  // the same HostCommandRegistry as S2F41 so a single YAML row defines
  // both behaviours; we just don't pass any parameters along.
  router.on(2, 21, [model, logfn, emit_event, emit_alarm_set](const s2::Message& msg) {
    auto rcmd = gem::parse_s2f21(msg);
    if (!rcmd) return gem::s2f22_remote_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto result = model->commands.dispatch(*rcmd, {});
    logfn("S2F21 RCMD=" + *rcmd + " -> S2F22 CMDA=" +
          std::to_string(static_cast<int>(result.ack)));
    if (result.ack == gem::HostCmdAck::Accept) {
      if (result.emit_ceid) emit_event(*result.emit_ceid);
      if (result.set_alarm) emit_alarm_set(*result.set_alarm);
      if (result.force_spool) model->spool.set_force_spool(*result.force_spool);
    }
    return gem::s2f22_remote_command_ack(result.ack);
  });

  // S2F49 — Enhanced Remote Command.  OBJSPEC scopes the command at a
  // specific object instance (e.g. a CJ or PJ id); for now we delegate
  // to the same command registry as S2F41 and surface OBJSPEC in the
  // log so downstream tooling can audit it.  The richer CPACK/CEPACK
  // shape lets us return per-parameter outcomes; until a command in
  // the registry produces per-CP failures we just reply with an empty
  // cpacks list, matching the spec's "all OK" interpretation.
  router.on(2, 49, [model, logfn, emit_event, emit_alarm_set](const s2::Message& msg) {
    auto cmd = gem::parse_s2f49(msg);
    if (!cmd) return gem::s2f50_enhanced_host_command_ack(gem::HostCmdAck::ParameterInvalid, {});
    auto result = model->commands.dispatch(cmd->rcmd, cmd->params);
    logfn("S2F49 DATAID=" + std::to_string(cmd->dataid) +
          " OBJSPEC=" + cmd->objspec + " RCMD=" + cmd->rcmd +
          " -> S2F50 HCACK=" + std::to_string(static_cast<int>(result.ack)));
    if (result.ack == gem::HostCmdAck::Accept) {
      if (result.emit_ceid) emit_event(*result.emit_ceid);
      if (result.set_alarm) emit_alarm_set(*result.set_alarm);
      if (result.force_spool) {
        model->spool.set_force_spool(*result.force_spool);
      }
    }
    return gem::s2f50_enhanced_host_command_ack(result.ack, {});
  });

  router.on(2, 23, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f23(msg);
    auto ack = gem::TraceAck::Accept;
    if (!req) {
      ack = gem::TraceAck::InvalidPeriod;
    } else {
      for (auto v : req->svids) {
        if (!model->vid_exists(v)) { ack = gem::TraceAck::UnknownVid; break; }
      }
      if (ack == gem::TraceAck::Accept) {
        model->traces.add({req->trid, req->dsper, req->totsmp, req->repgsz, req->svids});
      }
    }
    logfn("S2F23 -> S2F24 TIAACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f24_trace_initialize_ack(ack);
  });

  router.on(2, 45, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s2f45(msg);
    auto ack = gem::LimitMonitorAck::Accept;
    if (!req) {
      ack = gem::LimitMonitorAck::LimitValueError;
    } else {
      for (const auto& entry : req->entries) {
        if (!model->vid_exists(entry.vid)) { ack = gem::LimitMonitorAck::VidNotExist; break; }
      }
      if (ack == gem::LimitMonitorAck::Accept) {
        for (const auto& entry : req->entries)
          model->limits.set_for_vid(entry.vid, entry.limits);
      }
    }
    logfn("S2F45 -> S2F46 VLAACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s2f46_define_variable_limits_ack(ack);
  });
  router.on(2, 47, [model, logfn](const s2::Message& msg) {
    auto vids = gem::parse_s2f47(msg);
    std::vector<gem::VidLimitsEntry> rows;
    if (vids) {
      const auto target = vids->empty() ? model->limits.all_vids() : *vids;
      for (auto v : target) rows.push_back({v, model->limits.get_for_vid(v)});
    }
    logfn("S2F47 -> S2F48 (" + std::to_string(rows.size()) + " entries)");
    return gem::s2f48_variable_limit_attribute_data(rows);
  });

  router.on(2, 43, [model, logfn](const s2::Message& msg) {
    auto streams = gem::parse_s2f43(msg);
    auto ack = gem::ResetSpoolAck::Accept;
    std::vector<gem::ResetSpoolStreamAck> per;
    if (!streams) {
      ack = gem::ResetSpoolAck::Denied_NotAllowed;
    } else {
      model->spool.set_spoolable_streams(*streams);
      logfn("S2F43 spoolable=" + std::to_string(streams->size()) + " streams");
    }
    return gem::s2f44_reset_spooling_ack(ack, per);
  });

  // S6F15 — Event Report Request.  Host pulls the current payload for
  // a CEID without waiting for the equipment to emit it.  Reply mirrors
  // S6F11 (DATAID=0, the same CEID, and the latest report rows).
  router.on(6, 15, [model, logfn](const s2::Message& msg) {
    auto ceid = gem::parse_s6f15(msg);
    if (!ceid)
      return gem::s6f16_event_report_data({0, 0, {}});
    auto reports = model->compose_reports_for(*ceid);
    logfn("S6F15 CEID=" + std::to_string(*ceid) + " -> S6F16 (" +
          std::to_string(reports.size()) + " reports)");
    return gem::s6f16_event_report_data({0, *ceid, reports});
  });

  // S6F19 — Individual Report Request.  Host pulls a specific RPTID;
  // we return just that report's VID values (no annotation).
  router.on(6, 19, [model, logfn](const s2::Message& msg) {
    auto rptid = gem::parse_s6f19(msg);
    std::vector<s2::Item> values;
    if (rptid) {
      // Resolve each VID in the report against the current values.
      for (const auto& r : model->events.all_reports()) {
        if (r.id != *rptid) continue;
        for (auto vid : r.vids) {
          auto v = model->vid_value(vid);
          values.push_back(v ? *v : s2::Item::list({}));
        }
        break;
      }
    }
    logfn("S6F19 RPTID=" + std::to_string(rptid.value_or(0)) +
          " -> S6F20 (" + std::to_string(values.size()) + " values)");
    return gem::s6f20_individual_report_data(values);
  });

  // S6F21 — Annotated Individual Report Request.  Same lookup as F19
  // but the reply carries (VID, value) pairs so the host doesn't need
  // to remember the report definition.
  router.on(6, 21, [model, logfn](const s2::Message& msg) {
    auto rptid = gem::parse_s6f21(msg);
    std::vector<gem::AnnotatedValue> rows;
    if (rptid) {
      for (const auto& r : model->events.all_reports()) {
        if (r.id != *rptid) continue;
        for (auto vid : r.vids) {
          auto v = model->vid_value(vid);
          rows.push_back({vid, v ? *v : s2::Item::list({})});
        }
        break;
      }
    }
    logfn("S6F21 RPTID=" + std::to_string(rptid.value_or(0)) +
          " -> S6F22 (" + std::to_string(rows.size()) + " annotated values)");
    return gem::s6f22_annotated_report_data(rows);
  });

  // S6F5 — Multi-block Data Send Inquire.  When the host plays this
  // role we grant unconditionally (HSMS doesn't have the SECS-I
  // 244-byte block ceiling that motivates the handshake).  Real hosts
  // would gate on storage or busy state.
  router.on(6, 5, [logfn](const s2::Message& msg) {
    auto req = gem::parse_s6f5(msg);
    logfn("S6F5 DATAID=" + std::to_string(req ? req->dataid : 0) +
          " LEN=" + std::to_string(req ? req->datalength : 0) +
          " -> S6F6 GRANT6=0");
    return gem::s6f6_multi_block_grant(gem::MultiBlockGrant::Ok);
  });

  router.on(6, 23, [&io, active_conn, model, logfn](const s2::Message& msg) {
    auto rsdc = gem::parse_s6f23(msg);
    if (!rsdc) return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Denied);
    if (*rsdc == gem::SpoolRequestCode::Purge) {
      const auto n = model->spool.size();
      model->spool.clear();
      logfn("S6F23 purge: dropped " + std::to_string(n) + " messages");
      return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Accept);
    }
    // Transmit: drain the queue, fire each as a fresh primary.  Defer to
    // the executor so the S6F24 ack flushes before the drained primaries
    // go out — the host should see ACK first, then the spooled traffic.
    auto drained = model->spool.drain();
    logfn("S6F23 transmit: draining " + std::to_string(drained.size()) +
          " messages");
    asio::post(io, [active_conn, drained = std::move(drained), logfn]() mutable {
      auto conn = active_conn->lock();
      if (!conn) return;
      for (auto& m : drained) {
        const bool w = m.reply_expected;
        if (w)
          conn->send_request(std::move(m), [](std::error_code, const s2::Message&) {});
        else
          conn->send_data(std::move(m));
      }
    });
    return gem::s6f24_request_spool_data_ack(gem::SpoolRequestAck::Accept);
  });

  router.on(5, 3, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s5f3(msg);
    auto ack = req ? model->alarms.set_enabled(req->alid, (req->aled & 0x80) != 0)
                   : gem::AlarmAck::Error;
    logfn(std::string("S5F3 -> S5F4 ACKC5=") + std::to_string(static_cast<int>(ack)));
    return gem::s5f4_enable_alarm_ack(ack);
  });
  router.on(5, 7, [model, logfn](const s2::Message&) {
    std::vector<gem::AlarmListing> rows;
    for (const auto& a : model->alarms.all()) {
      if (!model->alarms.enabled(a.id)) continue;
      const uint8_t alcd = (a.severity_category & 0x7F) |
                           static_cast<uint8_t>(model->alarms.active(a.id) ? 0x80 : 0x00);
      rows.push_back({alcd, a.id, a.text});
    }
    logfn("S5F7 -> S5F8 (" + std::to_string(rows.size()) + " enabled)");
    return gem::s5f8_list_enabled_alarms_data(rows);
  });

  router.on(5, 5, [model, logfn](const s2::Message& msg) {
    auto ids = gem::parse_u4_list_body(msg);
    std::vector<gem::Alarm> alarms;
    if (ids && ids->empty()) alarms = model->alarms.all();
    else if (ids)
      for (auto id : *ids) {
        auto a = model->alarms.get(id);
        if (a) alarms.push_back(*a);
      }
    logfn("S5F5 -> S5F6 (" + std::to_string(alarms.size()) + " alarms)");
    return gem::s5f6_list_alarms_data(
        alarms, [model](uint32_t id) { return model->alarms.active(id); });
  });

  // S5F13/F14 — Exception Recover Request.  Validates EXRECVRA against
  // the candidates the matching S5F9 advertised; on Accept the FSM
  // transitions Posted/RecoverFailed -> Recovering.  Equipment-side
  // recovery progress is signalled by the application calling
  // model->exceptions.fire_internal(exid, RecoveryComplete/Failed).
  router.on(5, 13, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s5f13(msg);
    auto ack = req ? model->exceptions.on_recover(req->exid, req->exrecvra)
                   : gem::AlarmAck::Error;
    logfn("S5F13 EXID=" + std::to_string(req ? req->exid : 0) +
          " action=" + (req ? req->exrecvra : std::string{"?"}) +
          " -> S5F14 ACKC5=" + std::to_string(static_cast<int>(ack)));
    return gem::s5f14_exception_recover_ack(ack);
  });

  router.on(5, 17, [model, logfn](const s2::Message& msg) {
    auto exid = gem::parse_s5f17(msg);
    auto ack = exid ? model->exceptions.on_recover_abort(*exid)
                    : gem::AlarmAck::Error;
    logfn("S5F17 EXID=" + std::to_string(exid.value_or(0)) +
          " -> S5F18 ACKC5=" + std::to_string(static_cast<int>(ack)));
    return gem::s5f18_exception_recover_abort_ack(ack);
  });

  // ---- E87 Carrier Management dispatch ---------------------------------
  // S3F17 maps the textual CARRIERACTION string onto a CarrierIDEvent
  // and fires it against the matching carrier.  Unknown actions return
  // CarrierActionInvalid; unknown carriers return CarrierIDUnknown.
  router.on(3, 17, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s3f17(msg);
    if (!req) return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::ParameterInvalid);
    if (!model->carriers.has(req->carrierid))
      return gem::s3f18_carrier_action_ack(gem::CarrierActionAck::CarrierIDUnknown);

    auto ack = gem::CarrierActionAck::Accept;
    if (req->carrieraction == "ProceedWithCarrier") {
      model->carriers.fire_id_event(req->carrierid,
                                    gem::CarrierIDEvent::ProceedWithCarrier);
    } else if (req->carrieraction == "CancelCarrier") {
      model->carriers.fire_id_event(req->carrierid,
                                    gem::CarrierIDEvent::CancelCarrier);
    } else if (req->carrieraction == "BindCarrierID") {
      model->carriers.fire_id_event(req->carrierid, gem::CarrierIDEvent::Bind);
    } else {
      ack = gem::CarrierActionAck::CarrierActionInvalid;
    }
    logfn("S3F17 CARRIER=" + req->carrierid + " action=" + req->carrieraction +
          " -> S3F18 CAACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s3f18_carrier_action_ack(ack);
  });

  // S3F25 — host instructs equipment to move a carrier between ports.
  // We record the new port binding on the Carrier and fire the source
  // port's StartUnloading + target port's StartLoading transfer events;
  // application code is responsible for completing them.
  router.on(3, 25, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s3f25(msg);
    if (!req)
      return gem::s3f26_carrier_transfer_ack(gem::CarrierActionAck::ParameterInvalid);
    auto* c = model->carriers.get(req->carrierid);
    if (!c)
      return gem::s3f26_carrier_transfer_ack(gem::CarrierActionAck::CarrierIDUnknown);
    model->load_ports.fire_transfer_event(req->source_portid,
                                          gem::LoadPortTransferEvent::StartUnloading);
    model->load_ports.fire_transfer_event(req->target_portid,
                                          gem::LoadPortTransferEvent::StartLoading);
    c->port_id = req->target_portid;
    logfn("S3F25 CARRIER=" + req->carrierid +
          " " + std::to_string(req->source_portid) + "->" +
          std::to_string(req->target_portid));
    return gem::s3f26_carrier_transfer_ack(gem::CarrierActionAck::Accept);
  });

  // S3F19 — Slot Map Verify.  Host sends its expected slot map for
  // CARRIERID; equipment compares against locally-stored slots and
  // drives CSMS (NotRead -> Read on Accept, -> Mismatched on Mismatch).
  router.on(3, 19, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s3f19(msg);
    if (!req)
      return gem::s3f20_slot_map_verify_ack(gem::SlotMapVerifyAck::Error);
    auto* c = model->carriers.get(req->carrierid);
    if (!c)
      return gem::s3f20_slot_map_verify_ack(gem::SlotMapVerifyAck::CarrierUnknown);
    bool match = c->slots.size() == req->slots.size();
    if (match) {
      for (std::size_t i = 0; i < req->slots.size(); ++i) {
        if (static_cast<uint8_t>(c->slots[i].state) !=
            static_cast<uint8_t>(req->slots[i])) { match = false; break; }
      }
    }
    if (match) {
      model->carriers.fire_slot_map_event(req->carrierid, gem::SlotMapEvent::Read);
      logfn("S3F19 CARRIER=" + req->carrierid + " -> S3F20 Accept");
      return gem::s3f20_slot_map_verify_ack(gem::SlotMapVerifyAck::Accept);
    }
    model->carriers.fire_slot_map_event(req->carrierid, gem::SlotMapEvent::Mismatch);
    logfn("S3F19 CARRIER=" + req->carrierid + " -> S3F20 Mismatch");
    return gem::s3f20_slot_map_verify_ack(gem::SlotMapVerifyAck::Mismatch);
  });

  // S3F27 — Cancel Carrier (single-EXID form).
  router.on(3, 27, [model, logfn](const s2::Message& msg) {
    auto cid = gem::parse_s3f27(msg);
    if (!cid || !model->carriers.has(*cid))
      return gem::s3f28_cancel_carrier_ack(gem::CarrierActionAck::CarrierIDUnknown);
    model->carriers.fire_id_event(*cid, gem::CarrierIDEvent::CancelCarrier);
    model->carriers.fire_access_event(*cid, gem::CarrierAccessEvent::Cancel);
    logfn("S3F27 CARRIER=" + *cid + " cancelled");
    return gem::s3f28_cancel_carrier_ack(gem::CarrierActionAck::Accept);
  });

  // S7F1 — Process Program Load Inquire.  Host asks permission to send
  // LENGTH bytes for PPID; equipment responds with PPGNT.  Policy here:
  // accept any reasonable size (< 16 MiB which is also our HSMS frame
  // cap) and reject empty PPIDs.  Real equipment would gate on
  // available recipe storage.
  router.on(7, 1, [logfn](const s2::Message& msg) {
    auto req = gem::parse_s7f1(msg);
    auto ack = gem::ProcessProgramAck::Accept;
    if (!req || req->ppid.empty()) ack = gem::ProcessProgramAck::PpidNotFound;
    else if (req->length > 16u * 1024u * 1024u) ack = gem::ProcessProgramAck::MatrixOverflow;
    logfn("S7F1 PPID=" + (req ? req->ppid : std::string{"?"}) +
          " LEN=" + std::to_string(req ? req->length : 0) +
          " -> S7F2 PPGNT=" + std::to_string(static_cast<int>(ack)));
    return gem::s7f2_pp_load_grant(ack);
  });
  router.on(7, 3, [model, logfn](const s2::Message& msg) {
    auto pp = gem::parse_s7f3(msg);
    if (!pp) return gem::s7f4_process_program_ack(gem::ProcessProgramAck::LengthError);
    model->recipes.add(pp->ppid, pp->ppbody);
    logfn("S7F3 PPID=" + pp->ppid + " -> S7F4 (Accept)");
    return gem::s7f4_process_program_ack(gem::ProcessProgramAck::Accept);
  });
  // S7F17 — Delete Process Program.  Empty PPID list deletes all;
  // otherwise we remove each PPID and aggregate the worst ack.
  router.on(7, 17, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s7f17(msg);
    if (!req) return gem::s7f18_delete_pp_ack(gem::ProcessProgramAck::LengthError);
    auto ack = gem::ProcessProgramAck::Accept;
    if (req->empty()) {
      const auto all = model->recipes.list();
      for (const auto& id : all) model->recipes.remove(id);
      logfn("S7F17 delete-all (" + std::to_string(all.size()) + ") -> S7F18 Accept");
    } else {
      for (const auto& id : *req) {
        auto r = model->recipes.remove(id);
        if (r != gem::ProcessProgramAck::Accept) ack = r;
      }
      logfn("S7F17 delete " + std::to_string(req->size()) +
            " PPIDs -> S7F18 ACKC7=" + std::to_string(static_cast<int>(ack)));
    }
    return gem::s7f18_delete_pp_ack(ack);
  });
  router.on(7, 5, [model, logfn](const s2::Message& msg) {
    auto ppid = gem::parse_s7f5(msg);
    if (!ppid) return gem::s7f6_process_program_data("", "");
    auto body = model->recipes.get(*ppid);
    logfn("S7F5 PPID=" + *ppid + " -> S7F6");
    return gem::s7f6_process_program_data(*ppid, body ? *body : "");
  });
  router.on(7, 19, [model, logfn](const s2::Message&) {
    auto list = model->recipes.list();
    logfn("S7F19 -> S7F20 (" + std::to_string(list.size()) + " PPIDs)");
    return gem::s7f20_current_eppd_data(list);
  });

  // ---- E39 generic ObjectService ----------------------------------------
  // S14F1 GetAttr / S14F3 SetAttr against the CemObjectStore.  OBJTYPE
  // is validated against the stored object's type name (case-sensitive).
  router.on(14, 1, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s14f1(msg);
    if (!req) return gem::s14f2_get_attr_data({}, gem::ObjectAck::Error);
    auto* obj = model->cem.get(req->objspec);
    if (!obj)
      return gem::s14f2_get_attr_data({}, gem::ObjectAck::Denied_UnknownObject);
    if (gem::cem_object_type_name(obj->objtype) != req->objtype)
      return gem::s14f2_get_attr_data({}, gem::ObjectAck::Denied_InvalidAttribute);
    std::vector<gem::AttrValue> attrs;
    attrs.reserve(req->attrids.size());
    for (const auto& id : req->attrids) {
      auto v = model->cem.get_attr(req->objspec, id);
      attrs.push_back({id, v.value_or(s2::Item::ascii(""))});
    }
    logfn("S14F1 " + req->objspec + " (" +
          std::to_string(req->attrids.size()) + " attrs) -> S14F2");
    return gem::s14f2_get_attr_data(attrs, gem::ObjectAck::Success);
  });
  router.on(14, 3, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s14f3(msg);
    if (!req)
      return gem::s14f4_set_attr_ack({}, gem::ObjectAck::Error);
    auto* obj = model->cem.get(req->objspec);
    if (!obj)
      return gem::s14f4_set_attr_ack({}, gem::ObjectAck::Denied_UnknownObject);
    if (gem::cem_object_type_name(obj->objtype) != req->objtype)
      return gem::s14f4_set_attr_ack({}, gem::ObjectAck::Denied_InvalidAttribute);
    for (const auto& a : req->attrs) {
      model->cem.set_attr(req->objspec, a.attrid, a.value);
    }
    // Reply echoes back the now-stored values.
    std::vector<gem::AttrValue> reply;
    reply.reserve(req->attrs.size());
    for (const auto& a : req->attrs) {
      auto v = model->cem.get_attr(req->objspec, a.attrid);
      reply.push_back({a.attrid, v.value_or(s2::Item::ascii(""))});
    }
    logfn("S14F3 " + req->objspec + " (" +
          std::to_string(req->attrs.size()) + " attrs) -> S14F4");
    return gem::s14f4_set_attr_ack(reply, gem::ObjectAck::Success);
  });

  // ---- E40 / E94 -------------------------------------------------------
  router.on(14, 9, [model, logfn, run_cj_lifecycle](const s2::Message& msg) {
    (void)run_cj_lifecycle;
    auto req = gem::parse_s14f9(msg);
    if (!req) {
      logfn("S14F9 -> S14F10 Error (malformed body)");
      return gem::s14f10_create_control_job_ack("", gem::ObjectAck::Error);
    }
    auto r = model->control_jobs.create(
        req->ctljobid, req->prjobids,
        [model](const std::string& id) { return model->process_jobs.has(id); });
    gem::ObjectAck ack = gem::ObjectAck::Success;
    switch (r) {
      case gem::ControlJobStore::CreateResult::Created: ack = gem::ObjectAck::Success; break;
      case gem::ControlJobStore::CreateResult::Denied_AlreadyExists:
        ack = gem::ObjectAck::Denied_AlreadyExists; break;
      case gem::ControlJobStore::CreateResult::Denied_UnknownPRJob:
        ack = gem::ObjectAck::Denied_UnknownObject; break;
      case gem::ControlJobStore::CreateResult::Denied_Empty:
        ack = gem::ObjectAck::Denied_InvalidAttribute; break;
    }
    logfn("S14F9 CJ=" + req->ctljobid + " -> S14F10 OBJACK=" +
          std::to_string(static_cast<int>(ack)));
    return gem::s14f10_create_control_job_ack(req->ctljobid, ack);
  });
  router.on(14, 11, [model, logfn](const s2::Message& msg) {
    auto id = gem::parse_s14f11(msg);
    if (!id) return gem::s14f12_delete_control_job_ack(gem::ObjectAck::Error);
    const auto removed = model->control_jobs.remove(*id);
    logfn("S14F11 delete CJ=" + *id + " -> S14F12 " +
          (removed ? "Success" : "UnknownObject"));
    return gem::s14f12_delete_control_job_ack(
        removed ? gem::ObjectAck::Success : gem::ObjectAck::Denied_UnknownObject);
  });

  router.on(16, 11, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s16f11(msg);
    if (!req) return gem::s16f12_pr_job_create_ack(gem::HostCmdAck::ParameterInvalid);
    auto r = model->process_jobs.create(
        req->prjobid, req->rcpspec.ppid, req->mtrloutspec,
        [model](const std::string& ppid) {
          return model->recipes.get(ppid).has_value();
        });
    gem::HostCmdAck ack = gem::HostCmdAck::Accept;
    switch (r) {
      case gem::ProcessJobStore::CreateResult::Created: ack = gem::HostCmdAck::Accept; break;
      case gem::ProcessJobStore::CreateResult::Denied_AlreadyExists:
        ack = gem::HostCmdAck::Rejected; break;
      case gem::ProcessJobStore::CreateResult::Denied_InvalidPpid:
        ack = gem::HostCmdAck::ParameterInvalid; break;
    }
    if (ack == gem::HostCmdAck::Accept) {
      // Persist the optional E40-0705 trailers (MF / recipe-method /
      // recipe variables / process parameters) on the freshly created PJ.
      std::vector<gem::RcpVar> rcpvars;
      rcpvars.reserve(req->rcpspec.rcpvars.size());
      for (auto& v : req->rcpspec.rcpvars) rcpvars.push_back({v.name, v.value});
      std::vector<gem::ProcessParam> params;
      params.reserve(req->prprocessparams.size());
      for (auto& p : req->prprocessparams) params.push_back({p.name, p.value});
      model->process_jobs.set_e40_extras(req->prjobid, req->mf,
                                          req->prrecipemethod,
                                          std::move(rcpvars),
                                          std::move(params));
    }
    logfn("S16F11 PJ=" + req->prjobid + " PPID=" + req->rcpspec.ppid +
          " MF=" + std::to_string(static_cast<int>(req->mf)) +
          " RM=" + std::to_string(static_cast<int>(req->prrecipemethod)) +
          " rcpvars=" + std::to_string(req->rcpspec.rcpvars.size()) +
          " params=" + std::to_string(req->prprocessparams.size()) +
          " -> S16F12 HCACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s16f12_pr_job_create_ack(ack);
  });
  router.on(16, 13, [model, logfn](const s2::Message& msg) {
    auto id = gem::parse_s16f13(msg);
    auto ack = id ? model->process_jobs.dequeue(*id) : gem::HostCmdAck::ParameterInvalid;
    logfn("S16F13 PJ=" + (id ? *id : std::string{"?"}) +
          " -> S16F14 HCACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s16f14_pr_job_dequeue_ack(ack);
  });
  router.on(16, 7, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s16f7(msg);
    if (!req) return gem::s16f8_pr_job_monitor_ack(gem::HostCmdAck::ParameterInvalid);
    bool any_bad = false;
    for (const auto& e : req->entries) {
      const bool enable = (e.pralert & 0x80) != 0;
      if (!model->process_jobs.set_alert(e.prjobid, enable)) any_bad = true;
    }
    const auto ack = any_bad ? gem::HostCmdAck::InvalidObject : gem::HostCmdAck::Accept;
    logfn("S16F7 monitor " + std::to_string(req->entries.size()) +
          " jobs -> S16F8 HCACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s16f8_pr_job_monitor_ack(ack);
  });
  router.on(16, 15, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s16f15(msg);
    if (!req)
      return gem::s16f16_pr_job_create_multi_ack(std::vector<gem::PRJobCreateMultiResult>{});
    std::vector<gem::PRJobCreateMultiResult> results;
    results.reserve(req->jobs.size());
    for (const auto& job : req->jobs) {
      auto r = model->process_jobs.create(
          job.prjobid, job.ppid, job.mtrloutspec,
          [model](const std::string& ppid) {
            return model->recipes.get(ppid).has_value();
          });
      gem::HostCmdAck ack = gem::HostCmdAck::Accept;
      switch (r) {
        case gem::ProcessJobStore::CreateResult::Created: break;
        case gem::ProcessJobStore::CreateResult::Denied_AlreadyExists:
          ack = gem::HostCmdAck::Rejected; break;
        case gem::ProcessJobStore::CreateResult::Denied_InvalidPpid:
          ack = gem::HostCmdAck::ParameterInvalid; break;
      }
      results.push_back({job.prjobid, ack});
    }
    logfn("S16F15 multi-create " + std::to_string(req->jobs.size()) +
          " jobs -> S16F16");
    return gem::s16f16_pr_job_create_multi_ack(results);
  });
  router.on(16, 5, [model, logfn](const s2::Message& msg) {
    auto req = gem::parse_s16f5(msg);
    if (!req) return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto ev = gem::pr_cmd_to_event(req->prcmd);
    if (!ev) return gem::s16f6_pr_job_command_ack(gem::HostCmdAck::InvalidCommand);
    auto ack = model->process_jobs.on_host_command(req->prjobid, *ev);
    logfn("S16F5 PJ=" + req->prjobid + " " + req->prcmd +
          " -> S16F6 HCACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s16f6_pr_job_command_ack(ack);
  });
  router.on(16, 27, [model, logfn, run_cj_lifecycle](const s2::Message& msg) {
    auto req = gem::parse_s16f27(msg);
    if (!req) return gem::s16f28_cj_command_ack(gem::HostCmdAck::ParameterInvalid);
    auto ev = gem::ctl_cmd_to_event(req->ctljobcmd);
    if (!ev) return gem::s16f28_cj_command_ack(gem::HostCmdAck::InvalidCommand);
    // CJSTART semantics: implicit Select -> SetupComplete -> Start
    // cascade so a Queued CJ can be started in one host action. The
    // cascade is the equipment policy; the FSM rules still gate every
    // step.
    auto ack = gem::HostCmdAck::Accept;
    if (*ev == gem::ControlJobEvent::Start) {
      run_cj_lifecycle(req->ctljobid);
    } else {
      ack = model->control_jobs.on_host_command(req->ctljobid, *ev);
    }
    logfn("S16F27 CJ=" + req->ctljobid + " " + req->ctljobcmd +
          " -> S16F28 HCACK=" + std::to_string(static_cast<int>(ack)));
    return gem::s16f28_cj_command_ack(ack);
  });

  router.on(10, 1, [logfn](const s2::Message& msg) {
    auto td = gem::parse_s10f1(msg);
    if (td) logfn("TERMINAL[" + std::to_string(td->tid) + "] " + td->text);
    return gem::s10f2_terminal_display_ack(gem::TerminalAck::Accepted);
  });
  // S10F3 is the canonical SEMI E5 host→equipment "Terminal Display Single"
  // (S10F1 is documented in the spec as equipment→host); secsgem-py and
  // other reference libraries use F3.  We accept both for compatibility.
  router.on(10, 3, [logfn](const s2::Message& msg) {
    auto td = gem::parse_s10f3(msg);
    if (td) logfn("TERMINAL[" + std::to_string(td->tid) + "] " + td->text);
    return gem::s10f4_terminal_display_ack(gem::TerminalAck::Accepted);
  });
  router.on(10, 5, [logfn](const s2::Message& msg) {
    auto td = gem::parse_s10f5(msg);
    if (td) {
      logfn("TERMINAL[" + std::to_string(td->tid) + "] (" +
            std::to_string(td->lines.size()) + " lines)");
      for (const auto& l : td->lines) logfn("  | " + l);
    }
    return gem::s10f6_terminal_display_multi_ack(gem::TerminalAck::Accepted);
  });

  logfn("registered " + std::to_string(router.size()) + " (stream,function) handlers");

  // ---- Wire the router into accepted connections -----------------------
  server.on_connection([&io, sm, model, logfn, active_conn, &router, desc](
                           std::shared_ptr<Connection> conn) {
    *active_conn = conn;
    conn->set_closed_handler([active_conn](const std::string&) { active_conn->reset(); });

    // E30 §6.22: on entering SELECTED, if there's spooled data, notify
    // the host via S6F25 so it can decide (S6F23 Transmit vs Purge).
    // Our happy-path demo never drops the link so this branch doesn't
    // fire there — but the wiring is the canonical re-SELECT trigger.
    conn->set_selected_handler([logfn, sm, model, &io, active_conn]() {
      logfn(std::string("host is online; control=") + gem::control_state_name(sm->state()));
      if (model->spool.size() == 0) return;
      asio::post(io, [active_conn, model, logfn]() {
        auto c = active_conn->lock();
        if (!c) return;
        const uint32_t n = static_cast<uint32_t>(model->spool.size());
        logfn("spool: notifying host of " + std::to_string(n) + " queued messages");
        c->send_request(gem::s6f25_spool_data_ready(n),
                        [](std::error_code, const s2::Message&) {});
      });
    });

    // Wrap router.dispatch so we can emit S9F3 / S9F5 when an inbound
    // primary has no registered handler.  We use Connection's
    // current_header() accessor to capture the offending MHEAD; without
    // it we'd be left fabricating a synthetic header.
    std::weak_ptr<Connection> wconn = conn;
    conn->set_message_handler([&router, wconn, logfn](const s2::Message& msg)
                                  -> std::optional<s2::Message> {
      if (!router.has_handler(msg.stream, msg.function)) {
        auto c = wconn.lock();
        if (c && c->current_header()) {
          const uint8_t s9_function =
              router.has_handler_for_stream(msg.stream) ? 5 : 3;
          logfn("unhandled S" + std::to_string(msg.stream) + "F" +
                std::to_string(msg.function) + "; emitting S9F" +
                std::to_string(s9_function));
          c->emit_s9(s9_function, c->current_header()->encode());
        }
      }
      return router.dispatch(msg);
    });
  });

  server.start();
  io.run();
  return 0;
}