gea_node.cpp

// SPDX-License-Identifier: Apache-2.0
// node-compat native runtime.
//
// Emitted as its own generated C++ source by the node-compat geatsc plugin.
// This is the thin native layer the geatsc-compiled Node builtins
// (runtime/node/*.ts) sit on: a small reactor event loop and TCP/HTTP over
// sockets, written as proper namespaced OOP C++ (RAII, a polymorphic watcher
// interface — not C soup).
//
// Compiled under -std=gnu++20 -fno-exceptions -fno-rtti -Os.
#ifndef GEA_NODE_RUNTIME_INCLUDED
#define GEA_NODE_RUNTIME_INCLUDED

// The native carriers and byte adapters arrive from `gea_node.hpp`.

#include <algorithm>
#include <atomic>
#include <charconv>
#if defined(__GLIBCXX__) && __has_include(<bits/chrono.h>)
// Clocks and durations only; libstdc++'s <chrono> also drags in <format> (see
// gea_runtime.h).
#include <bits/chrono.h>
#else
#include <chrono>
#endif
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <deque>
#include <functional>
#include <memory>
#include <string>
#include <string_view>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include <cctype>
#include <cerrno>
#include <climits>
#include <cmath>
#include <csignal>

// Every system call below goes through this one interface (see the header).
#include "gea_node_platform.hpp"

#if defined(__APPLE__)
#include <sys/event.h>
#include <malloc/malloc.h>
#endif
#ifdef __linux__
#include <sys/epoll.h>
#endif

// ---------------------------------------------------------------------------
// The microtask queue.
//
// The loop below drains microtasks at six points, and the request path enqueues
// them. The compiler's runtime owns no queue of its own -- deliberately, a
// microtask queue being a host's concern rather than the language's -- so the
// queue is defined here, beside the loop that drains it, and
// `plugin/index.mjs` points the compiled program's `queueMicrotask` at the same
// symbol.
//
// It is reached through ONE spelling; a call site that named a queue directly
// would be a second authority on which one this reactor uses.
// ---------------------------------------------------------------------------
namespace gea::node {
// Promise jobs, the bulk of this queue, arrive as `gea::detail::PromiseJob`:
// moving one through here relocates it instead of cloning a std::function.
inline std::deque<gea::detail::PromiseJob> &microtasks() {
  static std::deque<gea::detail::PromiseJob> queue;
  return queue;
}
inline void queue_microtask(gea::detail::PromiseJob callback) {
  if (callback) microtasks().push_back(std::move(callback));
}
inline std::deque<std::function<void()>> &next_ticks();
inline void drain_microtasks() {
  // A drain entered from inside a microtask finds the running job still at the
  // head of the queue (see below). Nothing in this runtime does that -- a
  // blocking wait inside a job aborts, see `hostCallbackDepth` -- and the
  // outer drain reaches whatever is queued behind the running job anyway, in
  // the same order, so it does nothing.
  static bool draining = false;
  if (draining) return;
  // Every microtask is a host callback: a blocking `.awaited()` reached from
  // one is a nested pump, which the runtime refuses (`hostCallbackDepth`).
  gea::detail::HostCallbackScope scope;
  draining = true;
  struct Reset {
    ~Reset() { draining = false; }
  } reset;
  while (!microtasks().empty()) {
    // Node's next-tick queue outranks both Promise reactions and host
    // `queueMicrotask` callbacks. A callback may enqueue more ticks, so the
    // priority boundary is every individual microtask, not just the start of
    // one reactor turn. Checked before the call: nearly every drain finds the
    // queue empty, and entering `drain_next_ticks` costs a scope and a loop.
    if (!next_ticks().empty()) drain_next_ticks();
    // Run where it was queued. Moving the job out first relocated it through
    // an indirect call per microtask -- every await and every generator step
    // -- for a closure that is destroyed right after. A deque keeps
    // references to its other elements across `push_back`, so a job that
    // enqueues more work is safe; the entry is destroyed (by `invokeAndReset`)
    // BEFORE it is popped, because destroying a job can release objects that
    // enqueue, and a `push_back` from inside `pop_front` would modify the
    // container while it is being modified.
    struct Retire {
      std::deque<gea::detail::PromiseJob> &jobs;
      ~Retire() { jobs.pop_front(); }
    } retire{microtasks()};
    microtasks().front().invokeAndReset();
  }
  if (!next_ticks().empty()) drain_next_ticks();
}
inline std::deque<std::function<void()>> &next_ticks() {
  static std::deque<std::function<void()>> queue;
  return queue;
}
inline void queue_next_tick(std::function<void()> callback) { next_ticks().push_back(std::move(callback)); }
inline void drain_next_ticks() {
  gea::detail::HostCallbackScope scope;
  while (!next_ticks().empty()) {
    std::function<void()> task = std::move(next_ticks().front());
    next_ticks().pop_front();
    task();
  }
}
}  // namespace gea::node

namespace gea::node {

// A pooled request/response exchange over TCP for wire protocols whose
// messages start with their own total length as a 4-byte little-endian
// integer: a request is written whole, and the reply is read as that header
// and then the rest of the frame it announces.
namespace {

[[noreturn]] inline void framed_socket_exchange_fail(const char *operation) {
  std::fprintf(stderr, "gea: framed socket exchange %s failed: %s\n", operation, platform::errorText(errno));
  std::abort();
}

inline platform::Fd framed_socket_connect(const std::string &host, unsigned int port) {
  platform::ensureSockets();
  const std::string service = std::to_string(port);
  addrinfo query{};
  query.ai_family = AF_UNSPEC;
  query.ai_socktype = SOCK_STREAM;
  addrinfo *resolved = nullptr;
  if (::getaddrinfo(host.c_str(), service.c_str(), &query, &resolved) != 0) framed_socket_exchange_fail("resolve");

  platform::Fd fd = platform::kInvalidFd;
  for (addrinfo *candidate = resolved; candidate != nullptr; candidate = candidate->ai_next) {
    fd = platform::openSocket(candidate->ai_family, candidate->ai_socktype, candidate->ai_protocol);
    if (!platform::valid(fd)) continue;
    platform::setIntOption(fd, SOL_SOCKET, SO_KEEPALIVE, 1);
    platform::setIntOption(fd, IPPROTO_TCP, TCP_NODELAY, 1);
    if (platform::connectSocket(fd, candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen)) == 0) break;
    platform::closeSocket(fd);
    fd = platform::kInvalidFd;
  }
  ::freeaddrinfo(resolved);
  if (!platform::valid(fd)) framed_socket_exchange_fail("connect");
  return fd;
}

class FramedSocketPool {
 public:
  FramedSocketPool(std::string host, unsigned int port, std::size_t max_size)
      : host_(std::move(host)), port_(port), max_size_(max_size) {
    idle_.push_back(framed_socket_connect(host_, port_));
  }

  ~FramedSocketPool() {
    for (const platform::Fd fd : idle_) platform::closeSocket(fd);
  }

  platform::Fd lease() {
    if (!idle_.empty()) {
      const platform::Fd fd = idle_.back();
      idle_.pop_back();
      ++leased_;
      return fd;
    }
    if (leased_ >= max_size_) {
      errno = EBUSY;
      framed_socket_exchange_fail("pool lease");
    }
    ++leased_;
    return framed_socket_connect(host_, port_);
  }

  void release(platform::Fd fd) {
    if (leased_ == 0) {
      errno = EINVAL;
      framed_socket_exchange_fail("pool release");
    }
    --leased_;
    idle_.push_back(fd);
  }

 private:
  std::string host_;
  unsigned int port_;
  std::size_t max_size_;
  std::size_t leased_ = 0;
  std::vector<platform::Fd> idle_;
};

inline std::unordered_map<std::uint64_t, std::unique_ptr<FramedSocketPool>> &framed_socket_pools() {
  static std::unordered_map<std::uint64_t, std::unique_ptr<FramedSocketPool>> pools;
  return pools;
}

inline std::uint64_t &framed_socket_next_pool_id() {
  static std::uint64_t id = 1;
  return id;
}

inline FramedSocketPool &framed_socket_pool_for(double pool_id) {
  const auto found = framed_socket_pools().find(static_cast<std::uint64_t>(pool_id));
  if (found == framed_socket_pools().end()) {
    errno = EINVAL;
    framed_socket_exchange_fail("pool lookup");
  }
  return *found->second;
}

inline void framed_socket_send_all(platform::Fd fd, const std::uint8_t *bytes, std::size_t length) {
  std::size_t sent = 0;
  while (sent < length) {
    const platform::ssize count = platform::sendBytes(fd, bytes + sent, length - sent);
    if (count < 0 && errno == EINTR) continue;
    if (count <= 0) framed_socket_exchange_fail("write");
    sent += static_cast<std::size_t>(count);
  }
}

inline void framed_socket_read_all(platform::Fd fd, std::uint8_t *bytes, std::size_t length) {
  std::size_t received = 0;
  while (received < length) {
    const platform::ssize count = platform::recvBytes(fd, bytes + received, length - received);
    if (count < 0 && errno == EINTR) continue;
    if (count <= 0) framed_socket_exchange_fail("read");
    received += static_cast<std::size_t>(count);
  }
}

}  // namespace

inline double framed_socket_pool_open(std::string host, double port, double max_size) {
  const std::uint64_t id = framed_socket_next_pool_id()++;
  const std::size_t bounded_size =
      std::max<std::size_t>(1, std::min<std::size_t>(64, static_cast<std::size_t>(max_size)));
  framed_socket_pools().emplace(
      id, std::make_unique<FramedSocketPool>(std::move(host), static_cast<unsigned int>(port), bounded_size));
  return static_cast<double>(id);
}

inline std::vector<std::uint8_t> framed_socket_pool_exchange(double pool_id, const gea::TypedArray<std::uint8_t> &request) {
  FramedSocketPool &pool = framed_socket_pool_for(pool_id);
  const platform::Fd fd = pool.lease();
  framed_socket_send_all(fd, request.data(), request.size());
  std::uint8_t header[4]{};
  framed_socket_read_all(fd, header, sizeof(header));
  const std::uint32_t length = static_cast<std::uint32_t>(header[0]) |
                               (static_cast<std::uint32_t>(header[1]) << 8) |
                               (static_cast<std::uint32_t>(header[2]) << 16) |
                               (static_cast<std::uint32_t>(header[3]) << 24);
  if (length < 21 || length > 64u * 1024u * 1024u) {
    std::fprintf(stderr, "gea: framed socket exchange returned invalid message length %u\n", length);
    std::abort();
  }
  std::vector<std::uint8_t> response(length);
  std::copy(std::begin(header), std::end(header), response.begin());
  framed_socket_read_all(fd, response.data() + sizeof(header), response.size() - sizeof(header));
  pool.release(fd);
  return response;
}

inline void framed_socket_pool_close(double pool_id) {
  framed_socket_pools().erase(static_cast<std::uint64_t>(pool_id));
}

}  // namespace gea::node


// ---------------------------------------------------------------------------
// Reactor-integrated timers. setTimeout/setInterval in compiled TS lower to
// the __gea_node_set_timeout/... intrinsics (plugin-mapped), which register
// here; the EventLoop polls with nextDelayMs() and fires due timers each pass.
// ---------------------------------------------------------------------------
namespace gea::node::timers {

struct Timer {
  double id;
  double dueMs;
  double intervalMs;
  bool repeats;
  bool referenced;
  std::function<void()> callback;
};

inline std::vector<Timer> &registry() {
  static std::vector<Timer> timers;
  return timers;
}

inline double nowMs() {
  using namespace std::chrono;
  return duration_cast<duration<double, std::milli>>(steady_clock::now().time_since_epoch()).count();
}

inline double nextTimerId() {
  static double id = 0;
  return ++id;
}

inline bool hasPending() { return !registry().empty(); }
inline bool hasReferenced() {
  for (const Timer &timer : registry())
    if (timer.referenced) return true;
  return false;
}

// Poll timeout in ms until the earliest timer is due (0 if already due,
// capped so a far-future timer cannot starve the loop of periodic passes).
inline int nextDelayMs() {
  const auto &timers = registry();
  if (timers.empty()) return -1;
  double earliest = timers[0].dueMs;
  for (const Timer &timer : timers) earliest = std::min(earliest, timer.dueMs);
  const double delay = earliest - nowMs();
  if (delay <= 0) return 0;
  if (delay > 60000) return 60000;
  return static_cast<int>(delay) + 1;
}

inline double add(std::function<void()> callback, double delayMs, bool repeats) {
  const double id = nextTimerId();
  if (delayMs < 0) delayMs = 0;
  registry().push_back(Timer{id, nowMs() + delayMs, delayMs, repeats, true, std::move(callback)});
  return id;
}

// The due timers of the pass `fireDue` is running, collected before any of
// them runs; clearing one of them takes it out of this set as well.
inline std::unordered_set<double> &firingDue() {
  static std::unordered_set<double> ids;
  return ids;
}

inline void remove(double id) {
  firingDue().erase(id);
  auto &timers = registry();
  timers.erase(std::remove_if(timers.begin(), timers.end(), [&](const Timer &timer) { return timer.id == id; }),
               timers.end());
}

inline void unref(double id) {
  for (Timer &timer : registry()) {
    if (timer.id == id) {
      timer.referenced = false;
      return;
    }
  }
}

inline void fireDue() {
  const double now = nowMs();
  // Collect due timers first: callbacks may add/clear timers while running.
  std::vector<Timer> due;
  auto &timers = registry();
  for (std::size_t i = 0; i < timers.size();) {
    if (timers[i].dueMs <= now) {
      due.push_back(timers[i]);
      if (timers[i].repeats) {
        timers[i].dueMs = now + std::max(timers[i].intervalMs, 1.0);
        ++i;
      } else {
        timers.erase(timers.begin() + static_cast<std::ptrdiff_t>(i));
      }
    } else {
      ++i;
    }
  }
  std::sort(due.begin(), due.end(), [](const Timer &a, const Timer &b) {
    return a.dueMs != b.dueMs ? a.dueMs < b.dueMs : a.id < b.id;
  });
  gea::detail::HostCallbackScope scope;
  firingDue().clear();
  for (const Timer &timer : due) firingDue().insert(timer.id);
  for (Timer &timer : due) {
    // A callback that ran earlier in this pass may have cleared this one.
    if (firingDue().erase(timer.id) == 0) continue;
    if (timer.callback) timer.callback();
    // Node runs the microtask checkpoint after EACH timer callback, so a
    // promise job queued by the first of two due timers runs before the second.
    gea::node::drain_microtasks();
  }
  firingDue().clear();
}

}  // namespace gea::node::timers

namespace gea::node::immediates {

struct Immediate {
  double id;
  bool referenced;
  std::function<void()> callback;
};

inline std::vector<Immediate> &queue() {
  static std::vector<Immediate> immediates;
  return immediates;
}

inline double nextImmediateId() {
  static double id = 0;
  return ++id;
}

inline bool hasPending() { return !queue().empty(); }
inline bool hasReferenced() {
  for (const Immediate &immediate : queue())
    if (immediate.referenced) return true;
  return false;
}

inline double add(std::function<void()> callback) {
  const double id = nextImmediateId();
  queue().push_back(Immediate{id, true, std::move(callback)});
  return id;
}

// Each pass of runAll owns the ids of the batch it is running, and
// clearImmediate drops an id from every pass in flight. A callback can pump the
// loop again (an await on host work does), so a nested pass runs only what was
// queued since and leaves its caller's batch alone.
inline std::vector<std::unordered_set<double> *> &runningBatches() {
  static std::vector<std::unordered_set<double> *> batches;
  return batches;
}

inline void remove(double id) {
  for (std::unordered_set<double> *batch : runningBatches()) batch->erase(id);
  auto &immediates = queue();
  immediates.erase(
      std::remove_if(immediates.begin(), immediates.end(), [&](const Immediate &immediate) { return immediate.id == id; }),
      immediates.end());
}

inline void setReferenced(double id, bool referenced) {
  for (Immediate &immediate : queue()) {
    if (immediate.id == id) {
      immediate.referenced = referenced;
      return;
    }
  }
}

// Whether the check phase runs this turn. An unreferenced immediate does not
// keep the loop alive, but it still runs while other referenced work does.
inline bool due(bool loopAlive) { return !queue().empty() && (loopAlive || hasReferenced()); }

inline void runAll(bool loopAlive) {
  if (!due(loopAlive)) return;
  std::vector<Immediate> batch;
  batch.swap(queue());
  gea::detail::HostCallbackScope scope;
  std::unordered_set<double> pending;
  for (const Immediate &immediate : batch) pending.insert(immediate.id);
  runningBatches().push_back(&pending);
  struct Pop {
    ~Pop() { runningBatches().pop_back(); }
  } pop;
  for (Immediate &immediate : batch) {
    if (pending.erase(immediate.id) == 0) continue;
    if (immediate.callback) immediate.callback();
    gea::node::drain_microtasks();
  }
}

}

// ---------------------------------------------------------------------------
// Size-class free-list allocator (global operator new/delete override).
//
// A server's steady-state request path allocates the same handful of block
// sizes over and over (req/res control blocks, header vectors, body/response
// strings, map entries). Routing those through the system malloc dominated
// the non-syscall profile (~650 leaf samples of malloc/free vs ~4200 syscall).
// Recycling freed blocks per size class turns each of those into a pointer
// pop/push.
//
// Safety model: pooled blocks are carved from 1 MB-aligned arena chunks whose
// bases are registered in a global table. `release` masks the pointer to its
// chunk base and looks it up — a hit recycles onto the size-class free list, a
// miss means the block came from some other allocator (malloc passthrough for
// large sizes, or a system `operator new` in a library path that did not
// resolve to this override) and is handed to `free`, which is correct for the
// malloc-backed default `operator new` on both macOS and Linux. This makes
// cross-pairing (their new → our delete) safe by construction. The reverse
// direction (our new → their delete) is prevented by exporting the override
// with default visibility so every image in the process resolves to it.
//
// Free lists are thread_local: the node runtime is a single-threaded reactor,
// and any stray secondary thread gets its own lists (a cross-thread free just
// migrates the block). The chunk table is global and append-only (chunks are
// never returned to the OS), with atomic entries so any thread can look up
// membership. Set GEA_NO_POOL_ALLOC=1 to route every allocation to malloc.
// ---------------------------------------------------------------------------
namespace gea::node::alloc {

inline constexpr std::size_t kGranule = 16;                    // size-class step
inline constexpr std::size_t kMaxPooled = 4096;                // larger blocks pass through
inline constexpr std::size_t kClassCount = kMaxPooled / kGranule;
inline constexpr std::size_t kChunkSize = std::size_t(1) << 20;  // 1 MB, alignment == size
inline constexpr std::size_t kChunkTableSize = 8192;             // power of two; 8 GB of chunks
inline constexpr std::size_t kMaxProbe = 64;

// Each entry packs (chunkBase >> 20) << 12 | sizeClass. 0 = empty. Entries are
// written once and never removed, so a relaxed-read hit is always valid.
inline std::atomic<std::uint64_t> g_chunk_table[kChunkTableSize];

// POD, zero-initialized — safe to touch during static initialization.
//
// Plain statics under GEA_RUNTIME_SINGLE_THREADED (the reactor build): every
// `thread_local` read on macOS is a `_tlv_get_addr` call, and `release` made
// one per free even with the pool off, just to read the one-entry cache.
#if defined(GEA_RUNTIME_SINGLE_THREADED) && GEA_RUNTIME_SINGLE_THREADED
#define GEA_NODE_ALLOC_THREAD_LOCAL
#else
#define GEA_NODE_ALLOC_THREAD_LOCAL thread_local
#endif
inline GEA_NODE_ALLOC_THREAD_LOCAL void *g_free[kClassCount];
inline GEA_NODE_ALLOC_THREAD_LOCAL char *g_bump[kClassCount];
inline GEA_NODE_ALLOC_THREAD_LOCAL char *g_bump_end[kClassCount];
// Set once the first chunk is registered. While it is clear no block can be
// a pool block, so `release` is a bare `free` -- the whole story with the
// pool off (see `poolDisabled`).
inline bool g_any_chunk = false;

/**
 * Per-size-class allocation census, printed at exit under `GEA_ALLOC_CENSUS=1`.
 *
 * The pool makes each allocation cheap -- a free-list pop is two loads and a
 * store -- so when `allocate` shows up in a profile it is saying the request
 * path allocates OFTEN, not slowly. A profile cannot say how many times or in
 * what sizes, and that is exactly what decides whether the answer is a faster
 * allocator or fewer objects. Counting is off unless asked for, and the
 * counters are thread-local, so the measured path pays one predictable branch.
 */
inline void installAllocCensusExit();

inline int allocCensusLevel() {
  static const int level = [] {
    const char *value = std::getenv("GEA_ALLOC_CENSUS");
    const int n = value == nullptr ? 0 : value[0] - '0';
    if (n >= 1) installAllocCensusExit();
    return n >= 1 && n <= 2 ? n : 0;
  }();
  return level;
}

inline bool allocCensusEnabled() { return allocCensusLevel() >= 1; }

/**
 * Caller census (`GEA_ALLOC_CENSUS=2`): which call sites allocate, not just
 * how big. A size histogram says "18 blocks of <=32B per request" without
 * saying whether that is one std::string per header or eighteen short-lived
 * runtime objects, and those two findings point at opposite fixes. Frames are
 * captured with backtrace() and bucketed by the whole 4-frame prefix, so two
 * paths reaching the same leaf stay distinguishable. Reported as
 * binary-relative offsets, which addr2line resolves against the executable.
 */
inline constexpr std::size_t kCallerFrames = 4;
inline constexpr std::size_t kCallerSlots = 4096;

struct CallerSite {
  void *frames[kCallerFrames];
  std::uint64_t count;
};
inline thread_local CallerSite g_callers[kCallerSlots];
inline thread_local std::uint64_t g_callers_lost;

inline void recordCallerSite() {
  void *raw[kCallerFrames + 3];
  const int depth = platform::captureStack(raw, static_cast<int>(kCallerFrames + 3));
  // Frame 0 is backtrace itself; 1 is allocate (or the operator new it was
  // inlined into). Start at 2 so the first recorded frame is request-path code.
  const int first = depth > 2 ? 2 : depth;
  std::uint64_t hash = 1469598103934665603ull;
  void *frames[kCallerFrames] = {};
  for (std::size_t i = 0; i < kCallerFrames; ++i) {
    const int index = first + static_cast<int>(i);
    frames[i] = index < depth ? raw[index] : nullptr;
    hash = (hash ^ reinterpret_cast<std::uintptr_t>(frames[i])) * 1099511628211ull;
  }
  std::size_t slot = static_cast<std::size_t>(hash) & (kCallerSlots - 1);
  for (std::size_t probe = 0; probe < 64; ++probe) {
    CallerSite &site = g_callers[slot];
    if (site.count == 0) {
      for (std::size_t i = 0; i < kCallerFrames; ++i) site.frames[i] = frames[i];
      site.count = 1;
      return;
    }
    bool same = true;
    for (std::size_t i = 0; i < kCallerFrames; ++i)
      if (site.frames[i] != frames[i]) { same = false; break; }
    if (same) { ++site.count; return; }
    slot = (slot + 1) & (kCallerSlots - 1);
  }
  ++g_callers_lost;
}

inline void reportCallerCensus() {
  if (allocCensusLevel() < 2) return;
  const std::uintptr_t base = platform::imageBaseOf(reinterpret_cast<const void *>(&reportCallerCensus));
  std::vector<const CallerSite *> sites;
  for (std::size_t i = 0; i < kCallerSlots; ++i)
    if (g_callers[i].count != 0) sites.push_back(&g_callers[i]);
  std::sort(sites.begin(), sites.end(),
            [](const CallerSite *a, const CallerSite *b) { return a->count > b->count; });
  std::fprintf(stderr, "gea-alloc-callers: sites=%zu lost=%llu base=0x%llx\n", sites.size(),
               static_cast<unsigned long long>(g_callers_lost),
               static_cast<unsigned long long>(base));
  const std::size_t shown = sites.size() < 30 ? sites.size() : 30;
  for (std::size_t i = 0; i < shown; ++i) {
    std::fprintf(stderr, "  %8llu", static_cast<unsigned long long>(sites[i]->count));
    for (std::size_t f = 0; f < kCallerFrames; ++f) {
      if (sites[i]->frames[f] == nullptr) break;
      std::fprintf(stderr, " %llx",
                   static_cast<unsigned long long>(
                       reinterpret_cast<std::uintptr_t>(sites[i]->frames[f]) - base));
    }
    std::fputc('\n', stderr);
  }
}

inline thread_local std::uint64_t g_census[kClassCount + 1];
inline thread_local std::uint64_t g_census_large;

inline void reportAllocCensus();

/**
 * Make the census survive how a server actually stops.
 *
 * A benchmark harness ends the server with SIGTERM, whose default action
 * terminates the process without running `atexit`, so a report registered
 * there alone would never print. Installed only when the census is enabled,
 * so the ordinary signal behaviour is untouched.
 */
inline void installAllocCensusExit() {
  ::atexit(reportAllocCensus);
  struct Handler {
    static void onSignal(int) { std::exit(0); }
  };
  platform::onTermination(&Handler::onSignal);
}

inline void reportAllocCensus() {
  if (!allocCensusEnabled()) return;
  std::uint64_t total = 0;
  for (std::size_t klass = 1; klass <= kClassCount; ++klass) total += g_census[klass];
  total += g_census_large;
  std::fprintf(stderr, "gea-alloc-census: total=%llu\n", static_cast<unsigned long long>(total));
  for (std::size_t klass = 1; klass <= kClassCount; ++klass) {
    if (g_census[klass] == 0) continue;
    std::fprintf(stderr, "  <=%zuB %llu\n", klass * kGranule,
                 static_cast<unsigned long long>(g_census[klass]));
  }
  if (g_census_large != 0)
    std::fprintf(stderr, "  large %llu\n", static_cast<unsigned long long>(g_census_large));
  reportCallerCensus();
}

#if defined(GEA_PROFILE_ALLOCATIONS)
/**
 * The runtime's own allocation profile (`gea::detail::allocationProfile`),
 * printed at exit under a `-DGEA_PROFILE_ALLOCATIONS=1` build: which TYPES
 * are allocated, buffered as cycle candidates and destroyed by the collector.
 * The caller census above says where allocations come from; this says what
 * the collector spends its tracing on, per struct, which is the question an
 * acyclic-type or buffering change has to be measured against.
 */
inline void reportRuntimeAllocationProfile() {
  const auto &all = gea::detail::allocationProfile();
  std::fprintf(stderr,
               "gea-runtime-profile: created=%llu destroyed=%llu bytes=%llu collections=%llu full=%llu candidates=%llu "
               "visited=%llu edges=%llu retained=%llu unreachable=%llu matureSkipped=%llu deferrals=%llu\n",
               static_cast<unsigned long long>(all.created), static_cast<unsigned long long>(all.destroyed),
               static_cast<unsigned long long>(all.bytes), static_cast<unsigned long long>(all.collections),
               static_cast<unsigned long long>(all.fullCollections), static_cast<unsigned long long>(all.candidates),
               static_cast<unsigned long long>(all.visited), static_cast<unsigned long long>(all.edges),
               static_cast<unsigned long long>(all.retained), static_cast<unsigned long long>(all.unreachable),
               static_cast<unsigned long long>(all.matureSkipped), static_cast<unsigned long long>(all.deferrals));
  std::vector<const gea::detail::AllocationTypeProfile *> types;
  for (const auto *type = all.types; type != nullptr; type = type->next) types.push_back(type);
  std::sort(types.begin(), types.end(), [](const auto *left, const auto *right) {
    return left->created + left->candidates > right->created + right->candidates;
  });
  for (const auto *type : types) {
    if (type->created == 0 && type->candidates == 0) continue;
    std::fprintf(stderr, "  created=%llu live=%llu candidates=%llu cycleDestroyed=%llu deferrals=%llu block=%zu %s\n",
                 static_cast<unsigned long long>(type->created), static_cast<unsigned long long>(type->created - type->destroyed),
                 static_cast<unsigned long long>(type->candidates), static_cast<unsigned long long>(type->cycleDestroyed),
                 static_cast<unsigned long long>(type->deferrals), type->blockBytes, type->name);
  }
  // What is still alive at exit, by bytes held: a steady-state program's live
  // set is its working set, so anything here that scales with the operation
  // count is a leak (a database driver kept ~2 objects per operation).
  std::sort(types.begin(), types.end(), [](const auto *left, const auto *right) {
    return (left->created - left->destroyed) * left->blockBytes > (right->created - right->destroyed) * right->blockBytes;
  });
  std::size_t shown = 0;
  for (const auto *type : types) {
    const std::uint64_t live = type->created - type->destroyed;
    if (live == 0 || shown++ == 40) break;
    std::fprintf(stderr, "  live-at-exit=%llu bytes=%llu block=%zu %s\n", static_cast<unsigned long long>(live),
                 static_cast<unsigned long long>(live * type->blockBytes), type->blockBytes, type->name);
  }
  const void *base = reinterpret_cast<const void *>(
      platform::imageBaseOf(reinterpret_cast<const void *>(&reportRuntimeAllocationProfile)));
  for (const auto *census = gea::detail::allocationSiteCensuses(); census != nullptr; census = census->next) {
    std::vector<std::pair<std::uint64_t, const gea::detail::AllocationSiteKey *>> rows;
    for (const auto &entry : census->sites) rows.emplace_back(entry.second, &entry.first);
    std::sort(rows.begin(), rows.end(), [](const auto &left, const auto &right) { return left.first > right.first; });
    std::fprintf(stderr, "gea-runtime-sites: base=0x%llx sites=%zu %s\n", static_cast<unsigned long long>(reinterpret_cast<std::uintptr_t>(base)),
                 rows.size(), census->type);
    std::size_t printed = 0;
    for (const auto &row : rows) {
      if (printed++ == 60) break;
      std::fprintf(stderr, "    %llu", static_cast<unsigned long long>(row.first));
      for (const void *frame : row.second->frames)
        std::fprintf(stderr, " %llx", static_cast<unsigned long long>(reinterpret_cast<std::uintptr_t>(frame) - reinterpret_cast<std::uintptr_t>(base)));
      std::fprintf(stderr, "\n");
    }
  }
}
inline const bool g_runtime_profile_installed = (::atexit(&reportRuntimeAllocationProfile), true);
#endif

inline bool poolDisabled() {
  // On macOS the pool is only safe because `registerPoolZone` below hands the
  // system allocator a way to recognize pool blocks; on Linux the exported
  // operators alone are consistent for every image in the process.
  static const bool disabled = [] {
    const char *p = std::getenv("GEA_NO_POOL_ALLOC");
    return p && *p == '1';
  }();
  return disabled;
}

inline std::size_t chunkSlot(std::uintptr_t base) {
  return (static_cast<std::size_t>(base >> 20) * 0x9E3779B97F4A7C15ull) & (kChunkTableSize - 1);
}

inline bool registerChunk(std::uintptr_t base, std::size_t klass) {
  const std::uint64_t entry = (static_cast<std::uint64_t>(base >> 20) << 12) | klass;
  std::size_t slot = chunkSlot(base);
  for (std::size_t probe = 0; probe < kMaxProbe; ++probe) {
    std::uint64_t expected = 0;
    if (g_chunk_table[slot].compare_exchange_strong(expected, entry, std::memory_order_release, std::memory_order_relaxed))
      return true;
    slot = (slot + 1) & (kChunkTableSize - 1);
  }
  return false;  // table saturated — caller falls back to malloc
}

inline std::size_t lookupChunkClass(std::uintptr_t base) {
  const std::uint64_t key = static_cast<std::uint64_t>(base >> 20) << 12;
  std::size_t slot = chunkSlot(base);
  for (std::size_t probe = 0; probe < kMaxProbe; ++probe) {
    const std::uint64_t entry = g_chunk_table[slot].load(std::memory_order_acquire);
    if (entry == 0) return 0;
    if ((entry & ~0xFFFull) == key) return static_cast<std::size_t>(entry & 0xFFF);
    slot = (slot + 1) & (kChunkTableSize - 1);
  }
  return 0;
}

// A chunk is carved out of memory the system allocator has never owned. On
// macOS that is what makes the zone below sound: libmalloc resolves a freed
// pointer by asking every registered zone for its size, starting with the
// default zone, and a chunk taken from that zone with posix_memalign would let
// it claim the block first as an interior pointer of the 1 MB allocation.
inline void *mapChunk() { return platform::reserveAlignedChunk(kChunkSize); }

inline void unmapChunk(void *chunk) { platform::releaseAlignedChunk(chunk, kChunkSize); }

inline void release(void *ptr) noexcept;

/**
 * macOS: a malloc zone that claims pool blocks, registered before the first
 * one is handed out.
 *
 * Apple's prebuilt libc++.dylib binds its own `operator new`/`delete` and
 * `free` through two-level namespaces, so the exported override never reaches
 * it: a `std::string` grown inside the dylib is freed there with the raw
 * libc `free`, and a pool block reaching that call aborted with "pointer being
 * freed was not allocated". libmalloc's `free` does not know the block's
 * origin either -- it asks every registered zone `size(ptr)` and frees through
 * the one that answers. This zone answers for pool blocks by the same chunk
 * lookup `release` uses, which routes every foreign `free`, `realloc` and
 * `malloc_size` of a pool block back to the free lists. It never becomes the
 * default zone: nothing is allocated through it except by way of `operator
 * new`, so the system allocator keeps serving its own callers.
 */
#ifdef __APPLE__
inline std::size_t poolBlockSize(const void *ptr) {
  if (!g_any_chunk || ptr == nullptr) return 0;
  const std::uintptr_t base = reinterpret_cast<std::uintptr_t>(ptr) & ~(kChunkSize - 1);
  const std::size_t klass = lookupChunkClass(base);
  return klass == 0 ? 0 : klass * kGranule;
}

inline void *allocate(std::size_t size) noexcept;

inline void registerPoolZone() {
  static const bool registered = [] {
    struct Callbacks {
      static size_t size(malloc_zone_t *, const void *ptr) { return poolBlockSize(ptr); }
      static void *malloc(malloc_zone_t *, size_t size) { return allocate(size); }
      static void *calloc(malloc_zone_t *, size_t count, size_t size) {
        const std::size_t bytes = count * size;
        void *block = allocate(bytes);
        std::memset(block, 0, bytes);
        return block;
      }
      static void *valloc(malloc_zone_t *, size_t size) {
        void *block = nullptr;
        return posix_memalign(&block, static_cast<std::size_t>(::getpagesize()), size) == 0 ? block : nullptr;
      }
      static void free(malloc_zone_t *, void *ptr) { release(ptr); }
      static void freeDefiniteSize(malloc_zone_t *, void *ptr, size_t) { release(ptr); }
      static void *realloc(malloc_zone_t *, void *ptr, size_t size) {
        const std::size_t old = poolBlockSize(ptr);
        if (old == 0) return std::realloc(ptr, size);
        void *block = allocate(size);
        std::memcpy(block, ptr, old < size ? old : size);
        release(ptr);
        return block;
      }
      static void destroy(malloc_zone_t *) {}
      static size_t goodSize(malloc_zone_t *, size_t size) { return (size + kGranule - 1) / kGranule * kGranule; }
      static boolean_t check(malloc_zone_t *) { return 1; }
      static void print(malloc_zone_t *, boolean_t) {}
      static void log(malloc_zone_t *, void *) {}
      static void forceLock(malloc_zone_t *) {}
      static void forceUnlock(malloc_zone_t *) {}
      static void statistics(malloc_zone_t *, malloc_statistics_t *stats) { *stats = malloc_statistics_t{}; }
      static boolean_t zoneLocked(malloc_zone_t *) { return 0; }
      static kern_return_t enumerator(task_t, void *, unsigned, vm_address_t, memory_reader_t, vm_range_recorder_t) {
        return KERN_FAILURE;
      }
    };
    static malloc_introspection_t introspection{};
    introspection.enumerator = &Callbacks::enumerator;
    introspection.good_size = &Callbacks::goodSize;
    introspection.check = &Callbacks::check;
    introspection.print = &Callbacks::print;
    introspection.log = &Callbacks::log;
    introspection.force_lock = &Callbacks::forceLock;
    introspection.force_unlock = &Callbacks::forceUnlock;
    introspection.statistics = &Callbacks::statistics;
    introspection.zone_locked = &Callbacks::zoneLocked;
    static malloc_zone_t zone{};
    zone.size = &Callbacks::size;
    zone.malloc = &Callbacks::malloc;
    zone.calloc = &Callbacks::calloc;
    zone.valloc = &Callbacks::valloc;
    zone.free = &Callbacks::free;
    zone.realloc = &Callbacks::realloc;
    zone.destroy = &Callbacks::destroy;
    zone.zone_name = "gea-node-pool";
    zone.introspect = &introspection;
    zone.version = 9;
    zone.free_definite_size = &Callbacks::freeDefiniteSize;
    malloc_zone_register(&zone);
    return true;
  }();
  (void)registered;
}
#else
inline void registerPoolZone() {}
#endif

// Free-list links are stored in the first bytes of recycled blocks — memory
// that previously held arbitrary object types. Read/write them with memcpy
// (the blessed type-punning form): a raw `*(void **)block` violates strict
// aliasing, and clang at -O2 reorders those accesses against the surrounding
// object stores (observed as a SIGABRT on the first request).
inline void *loadFreeLink(const void *block) noexcept {
  void *next;
  std::memcpy(&next, block, sizeof next);
  return next;
}
inline void storeFreeLink(void *block, void *next) noexcept { std::memcpy(block, &next, sizeof next); }

inline void *allocate(std::size_t size) noexcept {
  const std::size_t klass = (size + kGranule - 1) / kGranule;  // 1-based class index
  // Both answers are fixed for the process; read them once rather than
  // through two guarded static initializers per allocation.
  static const bool census = allocCensusEnabled();
  static const bool pooled = !poolDisabled();
  if (census) {
    if (klass >= 1 && klass <= kClassCount) ++g_census[klass];
    else ++g_census_large;
    if (allocCensusLevel() >= 2) recordCallerSite();
  }
  if (pooled && klass >= 1 && klass <= kClassCount) {
    void *&head = g_free[klass - 1];
    if (head) {
      void *block = head;
      head = loadFreeLink(block);
      return block;
    }
    const std::size_t bytes = klass * kGranule;
    char *&bump = g_bump[klass - 1];
    char *&end = g_bump_end[klass - 1];
    if (bump == nullptr || static_cast<std::size_t>(end - bump) < bytes) {
      void *chunk = mapChunk();
      if (chunk) {
        if (registerChunk(reinterpret_cast<std::uintptr_t>(chunk), klass)) {
          registerPoolZone();
          g_any_chunk = true;
          bump = static_cast<char *>(chunk);
          end = bump + kChunkSize;
        } else {
          unmapChunk(chunk);  // table saturated — stop pooling this class
        }
      }
    }
    if (bump != nullptr && static_cast<std::size_t>(end - bump) >= bytes) {
      void *block = bump;
      bump += bytes;
      return block;
    }
  }
  void *block = std::malloc(size ? size : 1);
  if (!block) std::abort();
  return block;
}

// The steady-state request path frees from the same couple of hot chunks, so a
// one-entry cache in front of the table probe catches almost every release.
inline GEA_NODE_ALLOC_THREAD_LOCAL std::uintptr_t g_last_chunk_base;
inline GEA_NODE_ALLOC_THREAD_LOCAL std::size_t g_last_chunk_class;

inline void release(void *ptr) noexcept {
  if (!ptr) return;
  if (!g_any_chunk) {
    std::free(ptr);
    return;
  }
  const std::uintptr_t base = reinterpret_cast<std::uintptr_t>(ptr) & ~(kChunkSize - 1);
  std::size_t klass;
  if (base == g_last_chunk_base && g_last_chunk_class != 0) {
    klass = g_last_chunk_class;
  } else {
    klass = lookupChunkClass(base);
    if (klass != 0) {
      g_last_chunk_base = base;
      g_last_chunk_class = klass;
    }
  }
  if (klass != 0) {
    void *&head = g_free[klass - 1];
    storeFreeLink(ptr, head);
    head = ptr;
    return;
  }
  std::free(ptr);
}

}  // namespace gea::node::alloc

// Exported so every image in the process resolves `operator new` to the pool.
// A Windows executable has no such interposition: the replacement simply is
// the program's `operator new`, linked statically.
#if defined(_WIN32)
#define GEA_ALLOC_EXPORT
#else
#define GEA_ALLOC_EXPORT __attribute__((visibility("default")))
#endif
#ifdef GEA_NODE_IMPLEMENTATION
GEA_ALLOC_EXPORT void *operator new(std::size_t size) { return gea::node::alloc::allocate(size); }
GEA_ALLOC_EXPORT void *operator new(std::size_t size, const std::nothrow_t &) noexcept { return gea::node::alloc::allocate(size); }
GEA_ALLOC_EXPORT void *operator new[](std::size_t size) { return gea::node::alloc::allocate(size); }
GEA_ALLOC_EXPORT void *operator new[](std::size_t size, const std::nothrow_t &) noexcept { return gea::node::alloc::allocate(size); }
GEA_ALLOC_EXPORT void operator delete(void *ptr) noexcept { gea::node::alloc::release(ptr); }
GEA_ALLOC_EXPORT void operator delete(void *ptr, std::size_t) noexcept { gea::node::alloc::release(ptr); }
GEA_ALLOC_EXPORT void operator delete(void *ptr, const std::nothrow_t &) noexcept { gea::node::alloc::release(ptr); }
GEA_ALLOC_EXPORT void operator delete[](void *ptr) noexcept { gea::node::alloc::release(ptr); }
GEA_ALLOC_EXPORT void operator delete[](void *ptr, std::size_t) noexcept { gea::node::alloc::release(ptr); }
GEA_ALLOC_EXPORT void operator delete[](void *ptr, const std::nothrow_t &) noexcept { gea::node::alloc::release(ptr); }
#endif

namespace gea::node {

// ---------------------------------------------------------------------------
// A file descriptor the event loop watches. Concrete watchers (listener,
// connection) implement the readiness callback.
// ---------------------------------------------------------------------------
class IoWatcher {
public:
  virtual ~IoWatcher() = default;
  virtual platform::Fd fd() const noexcept = 0;
  virtual short interestedEvents() const noexcept = 0;  // POLLIN | POLLOUT
  virtual void onReady(short revents) = 0;
  virtual bool isReferenced() const noexcept { return true; }
};

// ---------------------------------------------------------------------------
// A minimal single-threaded poll() reactor. Borrowed watchers outlive the loop
// (e.g. a listener owned by its server); adopted watchers are owned by the loop
// (e.g. accepted connections) and destroyed when closed.
// ---------------------------------------------------------------------------
class EventLoop {
public:
#ifdef __linux__
  EventLoop() : epollFd_(::epoll_create1(EPOLL_CLOEXEC)) {}
  ~EventLoop() {
    if (epollFd_ >= 0) ::close(epollFd_);
  }
#elif defined(__APPLE__)
  EventLoop() : kqueueFd_(::kqueue()) {}
  ~EventLoop() {
    if (kqueueFd_ >= 0) ::close(kqueueFd_);
  }
#endif

  void addBorrowed(IoWatcher *watcher) {
    borrowed_.push_back(watcher);
    watcherAdded(watcher);
  }
  void adopt(std::unique_ptr<IoWatcher> watcher) {
    watcherAdded(watcher.get());
    owned_.push_back(std::move(watcher));
  }

  // Defer removal until the current dispatch pass completes, so a watcher can
  // close itself from inside its own onReady().
  void close(IoWatcher *watcher) {
    if (std::find(closing_.begin(), closing_.end(), watcher) != closing_.end()) return;
    closing_.push_back(watcher);
#ifdef __linux__
    // Stop readiness delivery immediately but retain the C++ object until no
    // outer or nested dispatch snapshot can still contain its pointer.
    if (epollFd_ >= 0) {
      ::epoll_ctl(epollFd_, EPOLL_CTL_DEL, watcher->fd(), nullptr);
      epollInterest_.erase(watcher);
    }
#elif defined(__APPLE__)
    if (kqueueFd_ >= 0) {
      const auto found = kqueueInterest_.find(watcher);
      if (found != kqueueInterest_.end()) {
        struct kevent changes[2];
        int count = 0;
        if (found->second & POLLIN) EV_SET(&changes[count++], watcher->fd(), EVFILT_READ, EV_DELETE, 0, 0, nullptr);
        if (found->second & POLLOUT) EV_SET(&changes[count++], watcher->fd(), EVFILT_WRITE, EV_DELETE, 0, 0, nullptr);
        if (count != 0) (void)::kevent(kqueueFd_, changes, count, nullptr, 0, nullptr);
        kqueueInterest_.erase(found);
      }
    }
#endif
  }

  void run() {
#ifdef __linux__
    if (epollFd_ >= 0) {
      (void)runEpoll(nullptr);
      return;
    }
#elif defined(__APPLE__)
    if (kqueueFd_ >= 0) {
      (void)runKqueue(nullptr);
      return;
    }
#endif
    (void)runPoll(nullptr);
  }

  // Drive the same reactor used by the normal Node run loop until an awaited
  // host operation settles. This deliberately does not wait for the reactor to
  // become empty: a connected client socket remains watched for future reads,
  // so "run everything" would never return from an individual operation.
  //
  // Returns false only when no referenced timer or live watcher can make
  // progress (or the platform wait fails) while the predicate is still false.
  bool runUntil(const std::function<bool()> &done) {
    if (!done || done()) return true;
#ifdef __linux__
    if (epollFd_ >= 0) return runEpoll(&done);
#elif defined(__APPLE__)
    if (kqueueFd_ >= 0) return runKqueue(&done);
#endif
    return runPoll(&done);
  }

private:
  using StopPredicate = const std::function<bool()> *;

  static bool stopRequested(StopPredicate done) {
    return done != nullptr && static_cast<bool>(*done) && (*done)();
  }

  // An await may pump the reactor from inside an existing watcher callback
  // (for example, outbound I/O issued by an HTTP handler). Watchers closed
  // by that nested pass must remain alive until the outer callback and its
  // ready-watcher snapshot have both finished. The closing_ membership check
  // prevents any such watcher from being dispatched again in the meantime.
  void dispatchWatcher(IoWatcher *watcher, short revents) {
    gea::detail::HostCallbackScope scope;
    ++dispatchDepth_;
    watcher->onReady(revents);
    --dispatchDepth_;
  }

  void purgeClosedIfSafe() {
    if (dispatchDepth_ == 0) purgeClosed();
  }

  // poll() backend — the portable fallback (epoll_create failure), and the
  // Windows backend (`platform::poll` is WSAPoll there).
  bool runPoll(StopPredicate done) {
    std::vector<pollfd> pfds;
    std::vector<IoWatcher *> polledWatchers;
    std::vector<std::pair<IoWatcher *, short>> ready;
    for (;;) {
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        purgeClosedIfSafe();
        return true;
      }
      pfds.clear();
      polledWatchers.clear();
      bool haveReferencedWatcher = false;
      forEachWatcher([&](IoWatcher *w) {
        if (isClosing(w)) return;
        pfds.push_back(pollfd{w->fd(), w->interestedEvents(), 0});
        polledWatchers.push_back(w);
        if (w->isReferenced()) haveReferencedWatcher = true;
      });
      const bool timersPending = gea::node::timers::hasPending();
      if ((pfds.empty() || (done == nullptr && !haveReferencedWatcher)) &&
          !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) {
        purgeClosedIfSafe();
        return stopRequested(done);  // only unref work remains
      }

      // About to block: the request path's garbage is dead, its live state is
      // at its smallest, and the queues are empty. The collector's own
      // safepoint (in `makeRef`) would otherwise run mid-request, several
      // times per request, re-tracing the request's live objects each time.
      gea::collectCyclesAtQuiescence();
      if (!gea::node::microtasks().empty() || !gea::node::next_ticks().empty()) continue;
      const int timeout = gea::node::immediates::due(haveReferencedWatcher || gea::node::timers::hasReferenced()) ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1;
      const int events = platform::poll(pfds.data(), pfds.size(), timeout);
      if (events < 0) {
        if (errno == EINTR) continue;
        purgeClosedIfSafe();
        return stopRequested(done);
      }

      // Keep the watcher set alive for the whole post-poll pass. Timer and
      // microtask callbacks may themselves await host work and recursively
      // pump this loop; any watcher they close must remain valid until the
      // pollfd snapshot below is no longer in use.
      ++dispatchDepth_;
      gea::node::timers::fireDue();
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        --dispatchDepth_;
        purgeClosedIfSafe();
        return true;
      }

      // Snapshot ready watchers before dispatching (a handler may add/close).
      ready.clear();
      for (std::size_t i = 0; i < pfds.size(); ++i) {
        IoWatcher *watcher = polledWatchers[i];
        if (!isClosing(watcher) && pfds[i].revents != 0) ready.emplace_back(watcher, pfds[i].revents);
      }
      for (const auto &[watcher, revents] : ready) {
        if (isClosing(watcher)) continue;
        if (revents != 0) dispatchWatcher(watcher, revents);
        if (stopRequested(done)) break;
      }
      // Handler promises progress here: continuations queued during dispatch
      // (async handlers, body-delivery microtasks) run before the next poll.
      gea::node::drain_microtasks();
      gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced());
      const bool stopped = stopRequested(done);
      --dispatchDepth_;
      purgeClosedIfSafe();
      if (stopped) return true;
    }
  }

#ifdef __linux__
  // epoll backend: the kernel wait is O(ready) instead of poll()'s O(watchers)
  // array copy + scan per wakeup. Watchers register on add; interest changes
  // (POLLOUT wanted only while a response is buffered) are synced with
  // EPOLL_CTL_MOD each pass — a cheap userland compare, a syscall only when
  // the interest actually changed.
  bool runEpoll(StopPredicate done) {
    epoll_event events[128];
    for (;;) {
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        purgeClosedIfSafe();
        return true;
      }
      bool haveWatchers = false;
      bool haveReferencedWatcher = false;
      forEachWatcher([&](IoWatcher *w) {
        if (isClosing(w)) return;
        haveWatchers = true;
        if (w->isReferenced()) haveReferencedWatcher = true;
        const short want = w->interestedEvents();
        auto found = epollInterest_.find(w);
        if (found != epollInterest_.end() && found->second != want) {
          epoll_event ev{};
          ev.events = static_cast<uint32_t>(want);
          ev.data.ptr = w;
          ::epoll_ctl(epollFd_, EPOLL_CTL_MOD, w->fd(), &ev);
          found->second = want;
        }
      });
      const bool timersPending = gea::node::timers::hasPending();
      if ((!haveWatchers || (done == nullptr && !haveReferencedWatcher)) &&
          !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) {
        purgeClosedIfSafe();
        return stopRequested(done);
      }

      gea::collectCyclesAtQuiescence();  // see `runPoll`
      if (!gea::node::microtasks().empty() || !gea::node::next_ticks().empty()) continue;
      const int timeout = gea::node::immediates::due(haveReferencedWatcher || gea::node::timers::hasReferenced()) ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1;
      const int count = ::epoll_wait(epollFd_, events, 128, timeout);
      if (count < 0) {
        if (errno == EINTR) continue;
        purgeClosedIfSafe();
        return stopRequested(done);
      }

      // epoll stores raw watcher pointers in its event snapshot. Preserve
      // those objects across timer/microtask callbacks and nested pumps until
      // this entire batch has been consumed.
      ++dispatchDepth_;
      gea::node::timers::fireDue();
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        --dispatchDepth_;
        purgeClosedIfSafe();
        return true;
      }

      for (int i = 0; i < count; ++i) {
        auto *watcher = static_cast<IoWatcher *>(events[i].data.ptr);
        if (isClosing(watcher)) continue;
        // Linux defines the EPOLL* event bits to coincide with the POLL* bits.
        dispatchWatcher(
            watcher,
            static_cast<short>(events[i].events & (POLLIN | POLLOUT | POLLERR | POLLHUP | POLLNVAL)));
        if (stopRequested(done)) break;
      }
      gea::node::drain_microtasks();
      gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced());
      const bool stopped = stopRequested(done);
      --dispatchDepth_;
      purgeClosedIfSafe();
      if (stopped) return true;
    }
  }
#endif

#if defined(__APPLE__)
  // kqueue backend, the macOS counterpart of `runEpoll`. `poll()` on XNU is
  // itself built on kqueue: every call registers a knote per descriptor,
  // waits, and tears them all down again, so a client blocking once per
  // round trip paid that kernel setup per operation -- a database driver's
  // system time was a quarter above Node's, whose libuv keeps its kqueue
  // registrations. Here each watcher's read and write filters stay
  // registered (level-triggered, as poll is) and only interest CHANGES reach
  // the kernel, batched into the wait itself. Readiness is translated back
  // to poll bits exactly as XNU's own poll does over the same knotes: EOF is
  // POLLHUP, alongside POLLIN for a reader; a write filter adds POLLOUT only
  // before hang-up; an error is POLLNVAL.
  bool runKqueue(StopPredicate done) {
    struct kevent events[128];
    std::vector<struct kevent> changes;
    std::vector<std::pair<IoWatcher *, short>> ready;
    for (;;) {
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        purgeClosedIfSafe();
        return true;
      }
      changes.clear();
      bool haveWatchers = false;
      bool haveReferencedWatcher = false;
      forEachWatcher([&](IoWatcher *w) {
        if (isClosing(w)) return;
        haveWatchers = true;
        if (w->isReferenced()) haveReferencedWatcher = true;
        syncKqueueInterest(w, changes);
      });
      const bool timersPending = gea::node::timers::hasPending();
      if ((!haveWatchers || (done == nullptr && !haveReferencedWatcher)) &&
          !gea::node::timers::hasReferenced() && !gea::node::immediates::hasReferenced()) {
        applyKqueueChanges(changes);
        purgeClosedIfSafe();
        return stopRequested(done);
      }

      gea::collectCyclesAtQuiescence();  // see `runPoll`
      if (!gea::node::microtasks().empty() || !gea::node::next_ticks().empty()) {
        applyKqueueChanges(changes);
        continue;
      }
      struct timespec wait{};
      const int delay = gea::node::immediates::due(haveReferencedWatcher || gea::node::timers::hasReferenced()) ? 0 : timersPending ? gea::node::timers::nextDelayMs() : -1;
      if (delay >= 0) {
        wait.tv_sec = delay / 1000;
        wait.tv_nsec = static_cast<long>(delay % 1000) * 1000000L;
      }
      const int count = ::kevent(kqueueFd_, changes.data(), static_cast<int>(changes.size()), events, 128,
                                 delay >= 0 ? &wait : nullptr);
      if (count < 0) {
        if (errno == EINTR) continue;
        purgeClosedIfSafe();
        return stopRequested(done);
      }

      // Merge each watcher's read and write readiness into one poll mask
      // before dispatching, as a poll pass hands each descriptor one revents.
      ready.clear();
      for (int i = 0; i < count; ++i) {
        const struct kevent &event = events[i];
        auto *watcher = static_cast<IoWatcher *>(event.udata);
        if (watcher == nullptr) continue;
        short bits = 0;
        if (event.flags & EV_ERROR) {
          // A change that could not be applied is reported in the event list:
          // only a real error on a live descriptor is readiness.
          if (event.data == ENOENT || event.data == 0) continue;
          bits = POLLNVAL;
        } else {
          if (event.flags & EV_EOF) bits |= POLLHUP;
          if (event.filter == EVFILT_READ) bits |= POLLIN;
          else if (event.filter == EVFILT_WRITE) bits |= POLLOUT;
        }
        auto merged = std::find_if(ready.begin(), ready.end(), [&](const auto &entry) { return entry.first == watcher; });
        if (merged == ready.end()) ready.emplace_back(watcher, bits);
        else merged->second = static_cast<short>(merged->second | bits);
      }
      for (auto &entry : ready) {
        if (entry.second & POLLHUP) entry.second = static_cast<short>(entry.second & ~POLLOUT);
      }

      // As in `runEpoll`: the snapshot holds raw watcher pointers, so they
      // stay alive across timer/microtask callbacks and nested pumps.
      ++dispatchDepth_;
      gea::node::timers::fireDue();
      gea::node::drain_microtasks();
      if (stopRequested(done)) {
        --dispatchDepth_;
        purgeClosedIfSafe();
        return true;
      }
      for (const auto &[watcher, revents] : ready) {
        if (isClosing(watcher)) continue;
        dispatchWatcher(watcher, revents);
        if (stopRequested(done)) break;
      }
      gea::node::drain_microtasks();
      gea::node::immediates::runAll(haveReferencedWatcher || gea::node::timers::hasReferenced());
      const bool stopped = stopRequested(done);
      --dispatchDepth_;
      purgeClosedIfSafe();
      if (stopped) return true;
    }
  }

  // Queue the filter additions/removals that bring `watcher`'s registration
  // to its current interest; a cheap compare, a change only on a difference.
  void syncKqueueInterest(IoWatcher *watcher, std::vector<struct kevent> &changes) {
    const short want = static_cast<short>(watcher->interestedEvents() & (POLLIN | POLLOUT));
    short &have = kqueueInterest_[watcher];
    if (have == want) return;
    const auto toggle = [&](short bit, int16_t filter) {
      if ((have & bit) == (want & bit)) return;
      struct kevent change;
      EV_SET(&change, watcher->fd(), filter, (want & bit) ? EV_ADD : EV_DELETE, 0, 0, watcher);
      changes.push_back(change);
    };
    toggle(POLLIN, EVFILT_READ);
    toggle(POLLOUT, EVFILT_WRITE);
    have = want;
  }

  void applyKqueueChanges(const std::vector<struct kevent> &changes) {
    if (!changes.empty()) (void)::kevent(kqueueFd_, changes.data(), static_cast<int>(changes.size()), nullptr, 0, nullptr);
  }
#endif

  void watcherAdded(IoWatcher *watcher) {
#ifdef __linux__
    if (epollFd_ >= 0) {
      const short want = watcher->interestedEvents();
      epoll_event ev{};
      ev.events = static_cast<uint32_t>(want);
      ev.data.ptr = watcher;
      ::epoll_ctl(epollFd_, EPOLL_CTL_ADD, watcher->fd(), &ev);
      epollInterest_[watcher] = want;
    }
#else
    (void)watcher;
#endif
  }

public:

private:
  template <typename Fn>
  void forEachWatcher(Fn &&fn) {
    for (IoWatcher *w : borrowed_) fn(w);
    for (const std::unique_ptr<IoWatcher> &w : owned_) fn(w.get());
  }

  bool isClosing(IoWatcher *watcher) const {
    return std::find(closing_.begin(), closing_.end(), watcher) != closing_.end();
  }

  void purgeClosed() {
    for (IoWatcher *watcher : closing_) {
      borrowed_.erase(std::remove(borrowed_.begin(), borrowed_.end(), watcher), borrowed_.end());
      owned_.erase(std::remove_if(owned_.begin(), owned_.end(),
                                  [&](const std::unique_ptr<IoWatcher> &owned) { return owned.get() == watcher; }),
                   owned_.end());
    }
    closing_.clear();
  }

  std::vector<IoWatcher *> borrowed_;
  std::vector<std::unique_ptr<IoWatcher>> owned_;
  std::vector<IoWatcher *> closing_;
  std::size_t dispatchDepth_ = 0;
#ifdef __linux__
  int epollFd_ = -1;
  std::unordered_map<IoWatcher *, short> epollInterest_;
#elif defined(__APPLE__)
  int kqueueFd_ = -1;
  // The poll bits each watcher's registered filters currently cover.
  std::unordered_map<IoWatcher *, short> kqueueInterest_;
#endif
};

// Every Node client and HTTP server in a process shares one reactor. This is
// essential for lazy clients created inside an HTTP handler: the server's run
// loop must also drive the newly-created outbound socket.
inline EventLoop &clientEventLoop() {
  static EventLoop loop;
  return loop;
}

// ---------------------------------------------------------------------------
// A non-blocking TCP socket wrapper (RAII: closes its fd on destruction).
// ---------------------------------------------------------------------------
class Socket {
public:
  explicit Socket(platform::Fd fd) noexcept : fd_(fd) {
    if (platform::valid(fd_)) platform::configureStream(fd_);
  }
  Socket(const Socket &) = delete;
  Socket &operator=(const Socket &) = delete;
  Socket(Socket &&other) noexcept : fd_(other.fd_) { other.fd_ = platform::kInvalidFd; }
  ~Socket() {
    if (platform::valid(fd_)) platform::closeSocket(fd_);
  }

  platform::Fd fd() const noexcept { return fd_; }

  void makeNonBlocking() const { platform::setNonBlocking(fd_); }

  void setTcpNoDelay(bool enabled = true) const { platform::setIntOption(fd_, IPPROTO_TCP, TCP_NODELAY, enabled ? 1 : 0); }

private:
  platform::Fd fd_;
};

// ---------------------------------------------------------------------------
// Outbound TCP connections for node:net.
//
// The callback is deliberately zero-argument. TS drains event codes and pulls
// bytes through net_read() as gea_node_buffer, keeping database wire payloads on
// the native byte carrier instead of boxing them as gea_cpp_value.
// ---------------------------------------------------------------------------
class ClientConnection;

inline std::unordered_map<std::uint64_t, ClientConnection *> &clientConnectionRegistry() {
  static std::unordered_map<std::uint64_t, ClientConnection *> map;
  return map;
}

inline std::uint64_t nextClientConnectionId() {
  static std::uint64_t id = 0;
  return ++id;
}

enum class ClientEvent : int { connect = 1, data = 2, error = 3, close = 4, drain = 5, finish = 6, end = 7 };

class ClientConnection final : public IoWatcher {
public:
  ClientConnection(EventLoop &loop, Socket socket, std::uint64_t id, std::string remote_address,
                   std::string remote_family, int port,
                   std::function<void()> notify, int initial_error, bool initially_connected = false,
                   std::function<void()> on_closed = {})
      : loop_(loop), socket_(std::move(socket)), id_(id), remoteAddress_(std::move(remote_address)),
        remoteFamily_(std::move(remote_family)), port_(port),
        notify_(std::move(notify)), initialError_(initial_error), connecting_(!initially_connected),
        connected_(initially_connected), onClosed_(std::move(on_closed)) {
    clientConnectionRegistry()[id_] = this;
  }

  ~ClientConnection() override {
    clientConnectionRegistry().erase(id_);
    if (onClosed_) onClosed_();
  }

  platform::Fd fd() const noexcept override { return socket_.fd(); }
  bool isReferenced() const noexcept override { return referenced_; }

  short interestedEvents() const noexcept override {
    if (connecting_) return POLLOUT;
    return static_cast<short>(((paused_ || peerEnded_) ? 0 : POLLIN) |
                              (outSent_ < outBuffer_.size() ? POLLOUT : 0));
  }

  void onReady(short revents) override {
    if (closing_) return;

    if (connecting_) completeConnect();
    if (!closing_ && connected_ && (revents & POLLOUT)) flushWrites();
    if (!closing_ && connected_ && (revents & POLLIN)) readAvailable();

    if (!closing_ && (revents & (POLLERR | POLLNVAL))) {
      int socketError = 0;
      (void)platform::readSocketError(socket_.fd(), socketError);
      fail(socketError == 0 ? EIO : socketError, "read");
    } else if (!closing_ && (revents & POLLHUP)) {
      closeFromPeer();
    }

    deliverPending();
    if (closing_) loop_.close(this);
  }

  int nextEvent() {
    if (events_.empty()) return 0;
    const int event = events_.front();
    events_.pop_front();
    return event;
  }

  std::vector<std::uint8_t> readChunk() {
    if (chunks_.empty()) return {};
    auto chunk = std::move(chunks_.front());
    chunks_.pop_front();
    return chunk;
  }

  const std::string &errorText() const { return errorText_; }
  // The failing errno (positive POSIX value) and the syscall node names in
  // its error (`connect`/`read`/`write`), so TypeScript can build node's
  // `connect ECONNREFUSED 127.0.0.1:27017` shape instead of strerror text.
  int errorNumber() const { return errorNumber_; }
  const char *errorSyscall() const { return errorSyscall_; }
  const std::string &remoteAddress() const { return remoteAddress_; }
  const std::string &remoteFamily() const { return remoteFamily_; }
  int remotePort() const { return port_; }

  std::string localAddress() const {
    sockaddr_storage address{};
    socklen_t length = sizeof(address);
    if (::getsockname(socket_.fd(), reinterpret_cast<sockaddr *>(&address), &length) != 0) return {};
    char host[NI_MAXHOST]{};
    if (::getnameinfo(reinterpret_cast<sockaddr *>(&address), length, host, sizeof(host), nullptr, 0,
                      NI_NUMERICHOST) != 0)
      return {};
    return host;
  }

  int localPort() const {
    sockaddr_storage address{};
    socklen_t length = sizeof(address);
    if (::getsockname(socket_.fd(), reinterpret_cast<sockaddr *>(&address), &length) != 0) return 0;
    if (address.ss_family == AF_INET)
      return ntohs(reinterpret_cast<const sockaddr_in *>(&address)->sin_port);
    if (address.ss_family == AF_INET6)
      return ntohs(reinterpret_cast<const sockaddr_in6 *>(&address)->sin6_port);
    return 0;
  }

  std::string localFamily() const {
    sockaddr_storage address{};
    socklen_t length = sizeof(address);
    if (::getsockname(socket_.fd(), reinterpret_cast<sockaddr *>(&address), &length) != 0) return {};
    if (address.ss_family == AF_INET) return "IPv4";
    if (address.ss_family == AF_INET6) return "IPv6";
    return {};
  }

  bool write(const std::uint8_t *data, std::size_t size) {
    if (closing_ || endRequested_) return false;
    if (size != 0) outBuffer_.append(reinterpret_cast<const char *>(data), size);
    if (connected_) {
      const bool ok = flushWrites();
      if (!ok) {
        notifySoon();
        if (closing_) loop_.close(this);
      } else if (!events_.empty()) {
        notifySoon();
      }
    }
    // A database driver's command writes are much smaller than Node's default writable
    // high-water mark. We accept them into outBuffer_ and report writable.
    return !closing_;
  }

  std::size_t bufferSize() const { return outBuffer_.size() - outSent_; }

  void end() {
    if (closing_ || endRequested_) return;
    endRequested_ = true;
    finishWriteSideIfReady();
    if (!events_.empty()) notifySoon();
  }

  void destroy() {
    if (closing_) return;
    closing_ = true;
    loop_.close(this);
  }

  void resetAndDestroy() {
    if (closing_) return;
    platform::setLingerReset(socket_.fd());
    destroy();
  }

  void setPaused(bool paused) { paused_ = paused; }
  void setReferenced(bool referenced) { referenced_ = referenced; }
  void setNotify(std::function<void()> notify) {
    notify_ = std::move(notify);
    if (!events_.empty()) notifySoon();
  }

  void setKeepAlive(bool enabled, int initialDelayMs) { platform::setKeepAlive(socket_.fd(), enabled, initialDelayMs); }

  void setNoDelay(bool enabled) { socket_.setTcpNoDelay(enabled); }

private:
  void completeConnect() {
    int socketError = initialError_;
    if (socketError == 0) {
      if (!platform::readSocketError(socket_.fd(), socketError)) socketError = errno;
    }
    connecting_ = false;
    if (socketError != 0) {
      fail(socketError, "connect");
      return;
    }
    connected_ = true;
    events_.push_back(static_cast<int>(ClientEvent::connect));
    finishWriteSideIfReady();
  }

  bool flushWrites() {
    while (outSent_ < outBuffer_.size()) {
      const platform::ssize count = platform::sendBytes(socket_.fd(), outBuffer_.data() + outSent_, outBuffer_.size() - outSent_);
      if (count > 0) {
        outSent_ += static_cast<std::size_t>(count);
        continue;
      }
      if (count < 0 && errno == EINTR) continue;
      if (count < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) return true;
      fail(count < 0 ? errno : EPIPE, "write");
      return false;
    }
    if (!outBuffer_.empty()) events_.push_back(static_cast<int>(ClientEvent::drain));
    outBuffer_.clear();
    outSent_ = 0;
    finishWriteSideIfReady();
    return true;
  }

  void finishWriteSideIfReady() {
    if (!connected_ || !endRequested_ || finishEmitted_ || outSent_ < outBuffer_.size()) return;
    platform::shutdownWrite(socket_.fd());
    finishEmitted_ = true;
    events_.push_back(static_cast<int>(ClientEvent::finish));
    if (peerEnded_) {
      events_.push_back(static_cast<int>(ClientEvent::close));
      closing_ = true;
    }
  }

  void readAvailable() {
    std::uint8_t bytes[65536];
    for (;;) {
      const platform::ssize count = platform::recvBytes(socket_.fd(), bytes, sizeof(bytes));
      if (count > 0) {
        chunks_.emplace_back(bytes, bytes + count);
        events_.push_back(static_cast<int>(ClientEvent::data));
        // A short read drained the socket: asking again only buys an EAGAIN,
        // one wasted syscall per readable event (libuv stops here too). The
        // loop's poll is level-triggered, so bytes that land meanwhile report
        // readable on the next turn.
        if (static_cast<std::size_t>(count) < sizeof(bytes)) return;
        continue;
      }
      if (count == 0) {
        closeFromPeer();
        return;
      }
      if (errno == EINTR) continue;
      if (errno == EAGAIN || errno == EWOULDBLOCK) return;
      fail(errno, "read");
      return;
    }
  }

  void closeFromPeer() {
    if (closing_ || peerEnded_) return;
    peerEnded_ = true;
    events_.push_back(static_cast<int>(ClientEvent::end));
    if (finishEmitted_) {
      events_.push_back(static_cast<int>(ClientEvent::close));
      closing_ = true;
    }
  }

  void fail(int error, const char *syscall) {
    if (closing_) return;
    errorText_ = platform::errorText(error);
    errorNumber_ = error;
    errorSyscall_ = syscall;
    events_.push_back(static_cast<int>(ClientEvent::error));
    events_.push_back(static_cast<int>(ClientEvent::close));
    closing_ = true;
  }

  void deliverPending() {
    if (!events_.empty() && notify_) notify_();
  }

  void notifySoon() {
    if (!notify_) return;
    auto notify = notify_;
    gea::node::queue_microtask([notify = std::move(notify)]() mutable { notify(); });
  }

  EventLoop &loop_;
  Socket socket_;
  std::uint64_t id_;
  std::string remoteAddress_;
  std::string remoteFamily_;
  int port_;
  std::function<void()> notify_;
  int initialError_ = 0;
  bool connecting_ = true;
  bool connected_ = false;
  bool closing_ = false;
  bool paused_ = false;
  bool referenced_ = true;
  bool endRequested_ = false;
  bool finishEmitted_ = false;
  bool peerEnded_ = false;
  std::deque<int> events_;
  std::deque<std::vector<std::uint8_t>> chunks_;
  std::string errorText_;
  int errorNumber_ = 0;
  const char *errorSyscall_ = "";
  std::string outBuffer_;
  std::size_t outSent_ = 0;
  std::function<void()> onClosed_;
};

inline ClientConnection *findClientConnection(double id) {
  const auto found = clientConnectionRegistry().find(static_cast<std::uint64_t>(id));
  return found == clientConnectionRegistry().end() ? nullptr : found->second;
}

inline std::string &lastNetCreateError() {
  static thread_local std::string error;
  return error;
}

// errno of the last synchronous net_create failure (0 when the failure was not
// a socket syscall, e.g. a resolver error). node reports a refused or
// unreachable connect as `connect <CODE> <address>:<port>` whether it failed
// synchronously or on completion, so TypeScript needs the number either way.
inline int &lastNetCreateErrno() {
  static thread_local int error = 0;
  return error;
}

inline bool bindClientSocket(platform::Fd fd, int family, int socketType, int protocol, const std::string &localAddress,
                      int localPort, std::string &error) {
  addrinfo query{};
  query.ai_family = family;
  query.ai_socktype = socketType;
  query.ai_protocol = protocol;
  query.ai_flags = localAddress.empty() ? AI_PASSIVE : 0;
  addrinfo *resolved = nullptr;
  const std::string service = std::to_string(localPort);
  const int status = ::getaddrinfo(localAddress.empty() ? nullptr : localAddress.c_str(), service.c_str(), &query,
                                   &resolved);
  if (status != 0) {
    error = ::gai_strerror(status);
    return false;
  }
  bool bound = false;
  int lastError = EADDRNOTAVAIL;
  for (addrinfo *candidate = resolved; candidate; candidate = candidate->ai_next) {
    if (platform::bindSocket(fd, candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen)) == 0) {
      bound = true;
      break;
    }
    lastError = errno;
  }
  ::freeaddrinfo(resolved);
  if (!bound) error = platform::errorText(lastError);
  return bound;
}

inline double net_create_native(std::string host, double requestedPort, double requestedFamily, double requestedHints,
                         std::string localAddress, double requestedLocalPort, bool bindLocal,
                         std::function<void()> notify) {
  platform::ensureSockets();
  lastNetCreateError().clear();
  lastNetCreateErrno() = 0;
  if (requestedPort < 0 || requestedPort > 65535 || requestedLocalPort < 0 || requestedLocalPort > 65535) {
    lastNetCreateError() = "Invalid TCP port";
    return 0;
  }
  const int port = static_cast<int>(requestedPort);
  const int localPort = static_cast<int>(requestedLocalPort);
  const int family = requestedFamily == 4 ? AF_INET : requestedFamily == 6 ? AF_INET6 : AF_UNSPEC;
  addrinfo query{};
  query.ai_family = family;
  query.ai_socktype = SOCK_STREAM;
  query.ai_protocol = IPPROTO_TCP;
  query.ai_flags = static_cast<int>(requestedHints);
  addrinfo *resolved = nullptr;
  const std::string service = std::to_string(port);
  const int resolveStatus = ::getaddrinfo(host.c_str(), service.c_str(), &query, &resolved);
  if (resolveStatus != 0) {
    lastNetCreateError() = ::gai_strerror(resolveStatus);
    return 0;
  }

  platform::Fd selectedFd = platform::kInvalidFd;
  int selectedFamily = AF_UNSPEC;
  std::string selectedAddress;
  int lastError = ECONNREFUSED;
  for (addrinfo *candidate = resolved; candidate; candidate = candidate->ai_next) {
    const platform::Fd fd = platform::openSocket(candidate->ai_family, candidate->ai_socktype, candidate->ai_protocol);
    if (!platform::valid(fd)) {
      lastError = errno;
      continue;
    }
    platform::setNonBlocking(fd);
    if (bindLocal &&
        !bindClientSocket(fd, candidate->ai_family, candidate->ai_socktype, candidate->ai_protocol,
                          localAddress, localPort, lastNetCreateError())) {
      platform::closeSocket(fd);
      continue;
    }
    if (platform::connectSocket(fd, candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen)) != 0 &&
        errno != EINPROGRESS) {
      lastError = errno;
      platform::closeSocket(fd);
      continue;
    }
    char numericHost[NI_MAXHOST]{};
    if (::getnameinfo(candidate->ai_addr, candidate->ai_addrlen, numericHost, sizeof(numericHost), nullptr, 0,
                      NI_NUMERICHOST) == 0)
      selectedAddress = numericHost;
    else
      selectedAddress = host;
    selectedFamily = candidate->ai_family;
    selectedFd = fd;
    break;
  }
  ::freeaddrinfo(resolved);
  if (!platform::valid(selectedFd)) {
    if (lastNetCreateError().empty()) {
      lastNetCreateError() = platform::errorText(lastError);
      lastNetCreateErrno() = lastError;
    }
    return 0;
  }

  const std::uint64_t id = nextClientConnectionId();
  Socket socket(selectedFd);
  auto connection = std::make_unique<ClientConnection>(
      clientEventLoop(), std::move(socket), id, std::move(selectedAddress),
      selectedFamily == AF_INET6 ? "IPv6" : "IPv4", port, std::move(notify), 0);
  clientEventLoop().adopt(std::move(connection));
  return static_cast<double>(id);
}

inline std::string net_create_error() { return lastNetCreateError(); }

// Both errno accessors answer in libuv's numbering (`platform::uvErrorNumber`):
// TypeScript negates the number into node's `error.errno` and names it with
// `net_errno_name`, so the two agree on every platform.
inline double net_create_errno() { return static_cast<double>(platform::uvErrorNumber(lastNetCreateErrno())); }

inline double net_error_errno(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? static_cast<double>(platform::uvErrorNumber(connection->errorNumber())) : 0.0;
}

inline std::string net_error_syscall(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? std::string(connection->errorSyscall()) : std::string();
}

// libuv's error names for the errnos a TCP client can see. node spells a
// system error's `code` with these (`ECONNREFUSED`), and its `errno` is the
// negated libuv number -- the negated POSIX value on every unix, libuv's own
// numbering on Windows -- which TypeScript computes from the same number.
inline std::string net_errno_name(double requested) {
  const int wanted = static_cast<int>(requested);
#define GEA_NODE_ERRNO_NAME(name) \
  if (platform::uvErrorNumber(name) == wanted) return #name;
  GEA_NODE_ERRNO_NAME(EACCES)
  GEA_NODE_ERRNO_NAME(EADDRINUSE)
  GEA_NODE_ERRNO_NAME(EADDRNOTAVAIL)
  GEA_NODE_ERRNO_NAME(EAFNOSUPPORT)
  GEA_NODE_ERRNO_NAME(EAGAIN)
  GEA_NODE_ERRNO_NAME(EALREADY)
  GEA_NODE_ERRNO_NAME(EBADF)
  GEA_NODE_ERRNO_NAME(ECANCELED)
  GEA_NODE_ERRNO_NAME(ECONNABORTED)
  GEA_NODE_ERRNO_NAME(ECONNREFUSED)
  GEA_NODE_ERRNO_NAME(ECONNRESET)
#ifdef EHOSTDOWN
  GEA_NODE_ERRNO_NAME(EHOSTDOWN)
#endif
  GEA_NODE_ERRNO_NAME(EHOSTUNREACH)
  GEA_NODE_ERRNO_NAME(EINTR)
  GEA_NODE_ERRNO_NAME(EINVAL)
  GEA_NODE_ERRNO_NAME(EIO)
  GEA_NODE_ERRNO_NAME(EISCONN)
  GEA_NODE_ERRNO_NAME(ELOOP)
  GEA_NODE_ERRNO_NAME(EMFILE)
  GEA_NODE_ERRNO_NAME(EMSGSIZE)
  GEA_NODE_ERRNO_NAME(ENAMETOOLONG)
  GEA_NODE_ERRNO_NAME(ENETDOWN)
  GEA_NODE_ERRNO_NAME(ENETUNREACH)
  GEA_NODE_ERRNO_NAME(ENFILE)
  GEA_NODE_ERRNO_NAME(ENOBUFS)
  GEA_NODE_ERRNO_NAME(ENOENT)
  GEA_NODE_ERRNO_NAME(ENOMEM)
  GEA_NODE_ERRNO_NAME(ENOTCONN)
  GEA_NODE_ERRNO_NAME(ENOTDIR)
  GEA_NODE_ERRNO_NAME(ENOTSOCK)
  GEA_NODE_ERRNO_NAME(EOVERFLOW)
  GEA_NODE_ERRNO_NAME(EPERM)
  GEA_NODE_ERRNO_NAME(EPIPE)
  GEA_NODE_ERRNO_NAME(EPROTO)
  GEA_NODE_ERRNO_NAME(EPROTONOSUPPORT)
  GEA_NODE_ERRNO_NAME(ETIMEDOUT)
#undef GEA_NODE_ERRNO_NAME
  return "UNKNOWN";
}

// ---------------------------------------------------------------------------
// Non-blocking name resolution for node:net's lookup (node's dns.lookup).
//
// getaddrinfo has no non-blocking form on any platform, so -- like libuv's
// threadpool -- each lookup runs on a short-lived detached thread that writes
// one byte to a wake pipe when it is done. The read end is a watcher on the
// shared reactor, so completion is delivered on the loop thread exactly like a
// socket event: `notify` runs from onReady and TypeScript pulls the results
// with primitive accessors. A failed thread start falls back to resolving
// inline.
//
// The lookup thread touches nothing but the state's plain fields and the
// resolver's own `addrinfo` list: it allocates nothing through `operator new`
// and frees nothing, because the pool allocator's free lists belong to the
// loop thread. The numeric strings are built on the loop thread (`collect`),
// and the state is only ever destroyed there: a lookup released before it
// finished keeps its watcher -- unreferenced, so it holds no process open --
// until the thread's byte arrives.
// ---------------------------------------------------------------------------
struct ResolveState {
  platform::Fd readFd = platform::kInvalidFd;
  platform::Fd writeFd = platform::kInvalidFd;
  std::string host;
  int family = AF_UNSPEC;
  int hints = 0;
  int status = 0;       // getaddrinfo status (0 = ok)
  int systemError = 0;  // errno when status == EAI_SYSTEM
  addrinfo *resolved = nullptr;  // the lookup thread's answer, until `collect`
  std::vector<std::pair<std::string, int>> addresses;
  std::atomic<bool> done{false};

  ~ResolveState() {
    if (resolved != nullptr) ::freeaddrinfo(resolved);
    platform::closeWake(readFd);
    platform::closeWake(writeFd);
  }

  // The lookup thread's whole job.
  void resolve() {
    addrinfo query{};
    query.ai_family = family;
    query.ai_socktype = SOCK_STREAM;
    query.ai_flags = hints;
    status = ::getaddrinfo(host.c_str(), nullptr, &query, &resolved);
#ifdef EAI_SYSTEM
    if (status == EAI_SYSTEM) systemError = errno;
#endif
    if (status != 0) resolved = nullptr;
    done.store(true, std::memory_order_release);
    platform::wakeWrite(writeFd);
  }

  // Loop thread: the answer as numeric addresses, once.
  void collect() {
    if (resolved == nullptr || !done.load(std::memory_order_acquire)) return;
    for (addrinfo *candidate = resolved; candidate; candidate = candidate->ai_next) {
      if (candidate->ai_family != AF_INET && candidate->ai_family != AF_INET6) continue;
      char numeric[NI_MAXHOST]{};
      if (::getnameinfo(candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen), numeric, sizeof(numeric),
                        nullptr, 0, NI_NUMERICHOST) != 0)
        continue;
      addresses.emplace_back(numeric, candidate->ai_family == AF_INET6 ? 6 : 4);
    }
    ::freeaddrinfo(resolved);
    resolved = nullptr;
  }
};

class ResolveWatcher;

struct ResolveEntry {
  std::shared_ptr<ResolveState> state;
  ResolveWatcher *watcher = nullptr;  // null once completed or released
};

inline std::unordered_map<std::uint64_t, ResolveEntry> &resolveRegistry() {
  static std::unordered_map<std::uint64_t, ResolveEntry> map;
  return map;
}

class ResolveWatcher final : public IoWatcher {
public:
  ResolveWatcher(EventLoop &loop, std::uint64_t id, std::shared_ptr<ResolveState> state, std::function<void()> notify)
      : loop_(loop), id_(id), state_(std::move(state)), notify_(std::move(notify)) {}

  platform::Fd fd() const noexcept override { return state_->readFd; }
  short interestedEvents() const noexcept override { return finished_ ? 0 : POLLIN; }
  bool isReferenced() const noexcept override { return !cancelled_; }

  void onReady(short revents) override {
    if (finished_) return;
    if (!state_->done.load(std::memory_order_acquire) && !(revents & (POLLERR | POLLHUP | POLLNVAL))) return;
    char byte = 0;
    (void)platform::wakeRead(state_->readFd, &byte, 1);
    finished_ = true;
    loop_.close(this);
    state_->collect();
    if (cancelled_) return;
    const auto found = resolveRegistry().find(id_);
    if (found != resolveRegistry().end()) found->second.watcher = nullptr;
    auto notify = std::move(notify_);
    if (notify) notify();
  }

  // Released before completion: nobody is told, but the watcher (and with it
  // the state the lookup thread is still writing) stays until the thread's
  // byte arrives.
  void cancel() {
    if (finished_) return;
    cancelled_ = true;
    notify_ = nullptr;
  }

private:
  EventLoop &loop_;
  std::uint64_t id_;
  std::shared_ptr<ResolveState> state_;
  std::function<void()> notify_;
  bool finished_ = false;
  bool cancelled_ = false;
};

inline void resolveThreadMain(void *argument) { static_cast<ResolveState *>(argument)->resolve(); }

inline double net_resolve_native(std::string host, double requestedFamily, double requestedHints,
                                 std::function<void()> notify) {
  static std::uint64_t nextId = 0;
  platform::ensureSockets();
  auto state = std::make_shared<ResolveState>();
  state->host = std::move(host);
  state->family = requestedFamily == 4 ? AF_INET : requestedFamily == 6 ? AF_INET6 : AF_UNSPEC;
  state->hints = static_cast<int>(requestedHints);
  platform::Fd fds[2] = {platform::kInvalidFd, platform::kInvalidFd};
  if (!platform::openWakePipe(fds)) {
    // No pipe: resolve inline and complete on the next microtask, so the
    // callback still never runs synchronously inside the caller.
    state->resolve();
    state->collect();
  } else {
    state->readFd = fds[0];
    state->writeFd = fds[1];
  }
  const std::uint64_t id = ++nextId;
  ResolveEntry &entry = resolveRegistry()[id];
  entry.state = state;
  if (!platform::valid(state->readFd)) {
    gea::node::queue_microtask([notify = std::move(notify)]() mutable {
      if (notify) notify();
    });
    return static_cast<double>(id);
  }
  auto watcher = std::make_unique<ResolveWatcher>(clientEventLoop(), id, state, std::move(notify));
  entry.watcher = watcher.get();
  clientEventLoop().adopt(std::move(watcher));
  if (!platform::startThread<&resolveThreadMain>(state.get())) {
    state->resolve();  // writes the completion byte; the watcher fires next pass
  }
  return static_cast<double>(id);
}

inline ResolveState *findResolveState(double id) {
  const auto found = resolveRegistry().find(static_cast<std::uint64_t>(id));
  if (found == resolveRegistry().end()) return nullptr;
  found->second.state->collect();
  return found->second.state.get();
}

// -1 when the lookup failed; the error accessors below then describe it.
inline double net_resolve_count(double id) {
  ResolveState *state = findResolveState(id);
  if (state == nullptr || !state->done.load(std::memory_order_acquire) || state->status != 0) return -1;
  return static_cast<double>(state->addresses.size());
}

inline std::string net_resolve_address(double id, double index) {
  ResolveState *state = findResolveState(id);
  if (state == nullptr || index < 0 || index >= static_cast<double>(state->addresses.size())) return {};
  return state->addresses[static_cast<std::size_t>(index)].first;
}

inline double net_resolve_family(double id, double index) {
  ResolveState *state = findResolveState(id);
  if (state == nullptr || index < 0 || index >= static_cast<double>(state->addresses.size())) return 0;
  return static_cast<double>(state->addresses[static_cast<std::size_t>(index)].second);
}

// node's `code` for a failed dns.lookup: EAI_NONAME/EAI_NODATA are reported
// as ENOTFOUND, every other resolver error by its libuv name.
inline std::string net_resolve_error_code(double id) {
  ResolveState *state = findResolveState(id);
  if (state == nullptr) return "ECANCELLED";
  switch (state->status) {
    case 0:
      return state->addresses.empty() ? "ENOTFOUND" : "";
    case EAI_NONAME:
#if defined(EAI_NODATA) && EAI_NODATA != EAI_NONAME
    case EAI_NODATA:
#endif
      return "ENOTFOUND";
    case EAI_AGAIN:
      return "EAI_AGAIN";
    case EAI_BADFLAGS:
      return "EAI_BADFLAGS";
    case EAI_FAIL:
      return "EAI_FAIL";
    case EAI_FAMILY:
      return "EAI_FAMILY";
    case EAI_MEMORY:
      return "EAI_MEMORY";
    case EAI_SERVICE:
      return "EAI_SERVICE";
    case EAI_SOCKTYPE:
      return "EAI_SOCKTYPE";
#ifdef EAI_SYSTEM
    case EAI_SYSTEM:
      return net_errno_name(static_cast<double>(platform::uvErrorNumber(state->systemError)));
#endif
    default:
      return "EAI_FAIL";
  }
}

// node's `errno` for the same error: libuv's UV_EAI_* numbers.
inline double net_resolve_errno(double id) {
  ResolveState *state = findResolveState(id);
  if (state == nullptr) return -3003;  // UV_EAI_CANCELED
  switch (state->status) {
    case 0:
      return state->addresses.empty() ? -3008 : 0;
    case EAI_NONAME:
      return -3008;
#if defined(EAI_NODATA) && EAI_NODATA != EAI_NONAME
    case EAI_NODATA:
      return -3007;
#endif
    case EAI_AGAIN:
      return -3001;
    case EAI_BADFLAGS:
      return -3002;
    case EAI_FAIL:
      return -3004;
    case EAI_FAMILY:
      return -3005;
    case EAI_MEMORY:
      return -3006;
    case EAI_SERVICE:
      return -3010;
    case EAI_SOCKTYPE:
      return -3011;
#ifdef EAI_SYSTEM
    case EAI_SYSTEM:
      return -static_cast<double>(platform::uvErrorNumber(state->systemError));
#endif
    default:
      return -3004;
  }
}

inline void net_resolve_release(double id) {
  const auto found = resolveRegistry().find(static_cast<std::uint64_t>(id));
  if (found == resolveRegistry().end()) return;
  if (found->second.watcher != nullptr) found->second.watcher->cancel();
  resolveRegistry().erase(found);
}

// dns.ADDRCONFIG, the hint node adds to an unpinned-family lookup.
inline double net_addrconfig_hint() { return static_cast<double>(AI_ADDRCONFIG); }

inline double net_next_event(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? static_cast<double>(connection->nextEvent()) : 0.0;
}

inline std::vector<std::uint8_t> net_read(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->readChunk() : std::vector<std::uint8_t>{};
}

inline std::string net_error(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->errorText() : std::string("Socket is closed");
}

inline bool net_write(double id, const gea::TypedArray<std::uint8_t> &buffer) {
  ClientConnection *connection = findClientConnection(id);
  return connection && connection->write(buffer.data(), buffer.size());
}

inline double net_buffer_size(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? static_cast<double>(connection->bufferSize()) : 0.0;
}

inline void net_end(double id) {
  if (ClientConnection *connection = findClientConnection(id)) connection->end();
}

inline void net_destroy(double id) {
  if (ClientConnection *connection = findClientConnection(id)) connection->destroy();
}

inline void net_reset_and_destroy(double id) {
  if (ClientConnection *connection = findClientConnection(id)) connection->resetAndDestroy();
}

inline void net_set_paused(double id, bool paused) {
  if (ClientConnection *connection = findClientConnection(id)) connection->setPaused(paused);
}

inline void net_set_referenced(double id, bool referenced) {
  if (ClientConnection *connection = findClientConnection(id)) connection->setReferenced(referenced);
}

inline void net_set_notify_native(double id, std::function<void()> notify) {
  if (ClientConnection *connection = findClientConnection(id)) connection->setNotify(std::move(notify));
}

inline void net_set_keep_alive(double id, bool enabled, double initialDelayMs) {
  if (ClientConnection *connection = findClientConnection(id)) {
    connection->setKeepAlive(enabled, static_cast<int>(std::max(0.0, initialDelayMs)));
  }
}

inline void net_set_no_delay(double id, bool enabled) {
  if (ClientConnection *connection = findClientConnection(id)) connection->setNoDelay(enabled);
}

inline std::string net_remote_address(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->remoteAddress() : std::string();
}

inline double net_remote_port(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? static_cast<double>(connection->remotePort()) : 0.0;
}

inline std::string net_remote_family(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->remoteFamily() : std::string();
}

inline std::string net_local_address(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->localAddress() : std::string();
}

inline double net_local_port(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? static_cast<double>(connection->localPort()) : 0.0;
}

inline std::string net_local_family(double id) {
  ClientConnection *connection = findClientConnection(id);
  return connection ? connection->localFamily() : std::string();
}

// ---------------------------------------------------------------------------
// TCP servers for node:net. The listener and every accepted socket share the
// client reactor, so a compiled process can host servers and clients without a
// second event loop. TypeScript drains lightweight event codes and adopts
// accepted ClientConnection ids into ordinary node:net Socket facades.
// ---------------------------------------------------------------------------
class NetServerListener;

enum class NetServerEvent : int { listening = 1, connection = 2, error = 3, close = 4, drop = 5 };

struct NetServerState : std::enable_shared_from_this<NetServerState> {
  std::uint64_t id = 0;
  std::function<void()> notify;
  NetServerListener *listener = nullptr;
  std::deque<int> events;
  std::deque<std::uint64_t> accepted;
  std::string errorText;
  std::string address;
  std::string family;
  int port = 0;
  std::size_t connections = 0;
  std::size_t maxConnections = 0;
  bool listening = false;
  bool closing = false;
  bool referenced = true;
  bool closeEmitted = false;

  void notifySoon() {
    if (!notify) return;
    auto callback = notify;
    gea::node::queue_microtask([callback = std::move(callback)]() mutable { callback(); });
  }

  void push(NetServerEvent event) {
    events.push_back(static_cast<int>(event));
    notifySoon();
  }

  void maybeEmitClose() {
    if (!closing || connections != 0 || closeEmitted) return;
    closeEmitted = true;
    push(NetServerEvent::close);
  }

  void connectionClosed() {
    if (connections > 0) --connections;
    maybeEmitClose();
  }
};

inline std::unordered_map<std::uint64_t, std::shared_ptr<NetServerState>> &netServerRegistry() {
  static std::unordered_map<std::uint64_t, std::shared_ptr<NetServerState>> map;
  return map;
}

inline std::uint64_t nextNetServerId() {
  static std::uint64_t id = 0;
  return ++id;
}

inline std::shared_ptr<NetServerState> findNetServer(double id) {
  const auto found = netServerRegistry().find(static_cast<std::uint64_t>(id));
  return found == netServerRegistry().end() ? nullptr : found->second;
}

inline std::string &lastNetServerCreateError() {
  static thread_local std::string error;
  return error;
}

class NetServerListener final : public IoWatcher {
public:
  NetServerListener(EventLoop &loop, Socket socket, std::shared_ptr<NetServerState> state)
      : loop_(loop), socket_(std::move(socket)), state_(std::move(state)) {}

  ~NetServerListener() override {
    if (state_->listener == this) state_->listener = nullptr;
  }

  platform::Fd fd() const noexcept override { return socket_.fd(); }
  short interestedEvents() const noexcept override { return POLLIN; }
  bool isReferenced() const noexcept override { return state_->referenced; }

  void onReady(short revents) override {
    if (state_->closing) return;
    if (revents & POLLIN) acceptAvailable();
    if (!state_->closing && (revents & (POLLERR | POLLHUP | POLLNVAL))) {
      int socketError = 0;
      (void)platform::readSocketError(socket_.fd(), socketError);
      state_->errorText = platform::errorText(socketError == 0 ? EIO : socketError);
      state_->push(NetServerEvent::error);
      stop();
    }
  }

  void stop() {
    if (state_->listener == this) state_->listener = nullptr;
    state_->listening = false;
    state_->closing = true;
    loop_.close(this);
    state_->maybeEmitClose();
  }

private:
  void acceptAvailable() {
    for (;;) {
      sockaddr_storage peer{};
      socklen_t peerLength = sizeof(peer);
      const platform::Fd clientFd = platform::acceptSocket(socket_.fd(), reinterpret_cast<sockaddr *>(&peer), &peerLength);
      if (!platform::valid(clientFd)) {
        if (errno == EINTR) continue;
        if (errno == EAGAIN || errno == EWOULDBLOCK) return;
        state_->errorText = platform::errorText(errno);
        state_->push(NetServerEvent::error);
        return;
      }

      platform::setNonBlocking(clientFd);

      if (state_->maxConnections != 0 && state_->connections >= state_->maxConnections) {
        platform::closeSocket(clientFd);
        state_->push(NetServerEvent::drop);
        continue;
      }

      char host[NI_MAXHOST]{};
      char service[NI_MAXSERV]{};
      const int nameStatus = ::getnameinfo(reinterpret_cast<sockaddr *>(&peer), peerLength, host, sizeof(host),
                                           service, sizeof(service), NI_NUMERICHOST | NI_NUMERICSERV);
      const std::string remoteAddress = nameStatus == 0 ? std::string(host) : std::string();
      const int remotePort = nameStatus == 0 ? std::atoi(service) : 0;
      const std::string remoteFamily = peer.ss_family == AF_INET6 ? "IPv6" : "IPv4";
      const std::uint64_t connectionId = nextClientConnectionId();
      ++state_->connections;
      std::weak_ptr<NetServerState> weakState = state_;
      auto connection = std::make_unique<ClientConnection>(
          loop_, Socket(clientFd), connectionId, remoteAddress, remoteFamily, remotePort, std::function<void()>{}, 0,
          true, [weakState]() {
            if (const auto state = weakState.lock()) state->connectionClosed();
          });
      loop_.adopt(std::move(connection));
      state_->accepted.push_back(connectionId);
      state_->push(NetServerEvent::connection);
    }
  }

  EventLoop &loop_;
  Socket socket_;
  std::shared_ptr<NetServerState> state_;
};

inline double net_server_listen_native(std::string host, double requestedPort, double requestedFamily,
                                double requestedBacklog, bool reusePort, bool ipv6Only,
                                double requestedMaxConnections, std::function<void()> notify) {
  platform::ensureSockets();
  lastNetServerCreateError().clear();
  if (requestedPort < 0 || requestedPort > 65535) {
    lastNetServerCreateError() = "Invalid TCP port";
    return 0;
  }

  const int family = requestedFamily == 4 ? AF_INET : requestedFamily == 6 ? AF_INET6 : AF_UNSPEC;
  addrinfo query{};
  query.ai_family = family;
  query.ai_socktype = SOCK_STREAM;
  query.ai_protocol = IPPROTO_TCP;
  query.ai_flags = AI_PASSIVE;
  addrinfo *resolved = nullptr;
  const std::string service = std::to_string(static_cast<int>(requestedPort));
  const int resolveStatus = ::getaddrinfo(host.empty() ? nullptr : host.c_str(), service.c_str(), &query, &resolved);
  if (resolveStatus != 0) {
    lastNetServerCreateError() = ::gai_strerror(resolveStatus);
    return 0;
  }

  platform::Fd selectedFd = platform::kInvalidFd;
  int selectedFamily = AF_UNSPEC;
  int lastError = EADDRNOTAVAIL;
  for (addrinfo *candidate = resolved; candidate; candidate = candidate->ai_next) {
    // A cluster worker on Windows listens on the primary's socket.
    platform::Fd shared = platform::kInvalidFd;
    if (platform::sharedClusterListener(candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen),
                                        std::max(1, static_cast<int>(requestedBacklog)), ipv6Only, shared)) {
      if (!platform::valid(shared)) {
        lastError = errno;
        continue;
      }
      selectedFd = shared;
      selectedFamily = candidate->ai_family;
      break;
    }
    const platform::Fd fd = platform::openSocket(candidate->ai_family, candidate->ai_socktype, candidate->ai_protocol);
    if (!platform::valid(fd)) {
      lastError = errno;
      continue;
    }
    // A cluster worker shares its port with its siblings the way makeListener
    // already does for node:http: the kernel distributes the connections.
    platform::configureListener(fd, reusePort || std::getenv("GEA_CLUSTER_WORKER_ID") != nullptr);
    if (candidate->ai_family == AF_INET6) platform::setIntOption(fd, IPPROTO_IPV6, IPV6_V6ONLY, ipv6Only ? 1 : 0);
    if (platform::bindSocket(fd, candidate->ai_addr, static_cast<socklen_t>(candidate->ai_addrlen)) != 0) {
      lastError = errno;
      platform::closeSocket(fd);
      continue;
    }
    const int backlog = std::max(1, static_cast<int>(requestedBacklog));
    if (platform::listenSocket(fd, backlog) != 0) {
      lastError = errno;
      platform::closeSocket(fd);
      continue;
    }
    platform::setNonBlocking(fd);
    selectedFd = fd;
    selectedFamily = candidate->ai_family;
    break;
  }
  ::freeaddrinfo(resolved);
  if (!platform::valid(selectedFd)) {
    lastNetServerCreateError() = platform::errorText(lastError);
    return 0;
  }

  sockaddr_storage bound{};
  socklen_t boundLength = sizeof(bound);
  char boundHost[NI_MAXHOST]{};
  char boundService[NI_MAXSERV]{};
  if (::getsockname(selectedFd, reinterpret_cast<sockaddr *>(&bound), &boundLength) != 0 ||
      ::getnameinfo(reinterpret_cast<sockaddr *>(&bound), boundLength, boundHost, sizeof(boundHost), boundService,
                    sizeof(boundService), NI_NUMERICHOST | NI_NUMERICSERV) != 0) {
    lastNetServerCreateError() = platform::errorText(errno);
    platform::closeSocket(selectedFd);
    return 0;
  }

  const std::uint64_t id = nextNetServerId();
  auto state = std::make_shared<NetServerState>();
  state->id = id;
  state->notify = std::move(notify);
  state->address = boundHost;
  state->port = std::atoi(boundService);
  state->family = selectedFamily == AF_INET6 ? "IPv6" : "IPv4";
  state->maxConnections = requestedMaxConnections > 0
                              ? static_cast<std::size_t>(requestedMaxConnections)
                              : 0;
  state->listening = true;
  netServerRegistry()[id] = state;
  auto listener = std::make_unique<NetServerListener>(clientEventLoop(), Socket(selectedFd), state);
  state->listener = listener.get();
  clientEventLoop().adopt(std::move(listener));
  state->push(NetServerEvent::listening);
  return static_cast<double>(id);
}

inline std::string net_server_create_error() { return lastNetServerCreateError(); }

inline std::string net_server_error(double id) {
  const auto state = findNetServer(id);
  return state ? state->errorText : std::string("Server is closed");
}

inline double net_server_next_event(double id) {
  const auto state = findNetServer(id);
  if (!state || state->events.empty()) return 0.0;
  const int event = state->events.front();
  state->events.pop_front();
  return static_cast<double>(event);
}

inline double net_server_take_connection(double id) {
  const auto state = findNetServer(id);
  if (!state || state->accepted.empty()) return 0.0;
  const std::uint64_t connection = state->accepted.front();
  state->accepted.pop_front();
  return static_cast<double>(connection);
}

inline void net_server_close(double id) {
  const auto state = findNetServer(id);
  if (!state || state->closing) return;
  if (state->listener) state->listener->stop();
  else {
    state->listening = false;
    state->closing = true;
    state->maybeEmitClose();
  }
}

inline void net_server_set_referenced(double id, bool referenced) {
  if (const auto state = findNetServer(id)) state->referenced = referenced;
}

inline void net_server_set_max_connections(double id, double maxConnections) {
  if (const auto state = findNetServer(id)) {
    state->maxConnections = maxConnections > 0 ? static_cast<std::size_t>(maxConnections) : 0;
  }
}

inline double net_server_connections(double id) {
  const auto state = findNetServer(id);
  return state ? static_cast<double>(state->connections) : 0.0;
}

inline std::string net_server_address(double id) {
  const auto state = findNetServer(id);
  return state ? state->address : std::string();
}

inline double net_server_port(double id) {
  const auto state = findNetServer(id);
  return state ? static_cast<double>(state->port) : 0.0;
}

inline std::string net_server_family(double id) {
  const auto state = findNetServer(id);
  return state ? state->family : std::string();
}

inline double net_is_ip(const std::string &input) {
  // Darwin's inet_pton accepts legacy leading-zero IPv4 components
  // ("01.2.3.4"), while Node's net.isIP intentionally requires strict
  // dotted-decimal input. Parse IPv4 before consulting the platform IPv6
  // parser so behavior is identical across macOS and Linux.
  int componentCount = 0;
  std::size_t componentStart = 0;
  bool validIpv4 = !input.empty();
  for (std::size_t index = 0; validIpv4 && index <= input.size(); ++index) {
    if (index < input.size() && input[index] != '.') continue;
    const std::size_t length = index - componentStart;
    if (length == 0 || length > 3 || (length > 1 && input[componentStart] == '0')) {
      validIpv4 = false;
      break;
    }
    int value = 0;
    for (std::size_t digit = componentStart; digit < index; ++digit) {
      const char character = input[digit];
      if (character < '0' || character > '9') {
        validIpv4 = false;
        break;
      }
      value = value * 10 + (character - '0');
    }
    if (!validIpv4 || value > 255) {
      validIpv4 = false;
      break;
    }
    ++componentCount;
    componentStart = index + 1;
  }
  if (validIpv4 && componentCount == 4) return 4.0;
  // Node accepts a `%zone` suffix of one or more [0-9a-zA-Z-.:] characters on an IPv6 address without
  // consulting the interface table; glibc's inet_pton rejects every zone and Darwin's resolves it, so
  // validate the zone here and hand only the address to the platform parser.
  std::string address = input;
  const std::size_t zoneStart = input.find('%');
  if (zoneStart != std::string::npos) {
    if (zoneStart + 1 >= input.size()) return 0.0;
    for (std::size_t index = zoneStart + 1; index < input.size(); ++index) {
      const char character = input[index];
      const bool allowed = (character >= '0' && character <= '9') || (character >= 'a' && character <= 'z') ||
                           (character >= 'A' && character <= 'Z') || character == '-' || character == '.' || character == ':';
      if (!allowed) return 0.0;
    }
    address = input.substr(0, zoneStart);
  }
  in6_addr ipv6{};
  if (::inet_pton(AF_INET6, address.c_str(), &ipv6) == 1) return 6.0;
  return 0.0;
}

inline std::string net_normalize_ip(const std::string &input, double family) {
  if (family == 4.0) return net_is_ip(input) == 4.0 ? input : std::string();
  if (family != 6.0) return std::string();

  // SocketAddress accepts an IPv6 scope suffix but exposes only the canonical
  // numeric address. Keep that behavior separate from net.isIP(), which must
  // continue rejecting scoped input.
  const std::size_t scope = input.find('%');
  const std::string numeric = scope == std::string::npos ? input : input.substr(0, scope);
  in6_addr address{};
  if (::inet_pton(AF_INET6, numeric.c_str(), &address) != 1) return std::string();
  char normalized[INET6_ADDRSTRLEN]{};
  if (::inet_ntop(AF_INET6, &address, normalized, sizeof(normalized)) == nullptr) return std::string();
  return normalized;
}

namespace {
inline bool parseComparableIp(const std::string &input, double family, std::uint8_t (&bytes)[16]) {
  std::memset(bytes, 0, sizeof(bytes));
  if (family == 4.0) {
    if (net_is_ip(input) != 4.0) return false;
    in_addr address{};
    if (::inet_pton(AF_INET, input.c_str(), &address) != 1) return false;
    bytes[10] = 0xff;
    bytes[11] = 0xff;
    std::memcpy(bytes + 12, &address, 4);
    return true;
  }
  if (family != 6.0) return false;
  const std::size_t scope = input.find('%');
  const std::string numeric = scope == std::string::npos ? input : input.substr(0, scope);
  in6_addr address{};
  if (::inet_pton(AF_INET6, numeric.c_str(), &address) != 1) return false;
  std::memcpy(bytes, &address, 16);
  return true;
}
}  // namespace

inline double net_ip_compare(const std::string &left, double leftFamily, const std::string &right, double rightFamily) {
  std::uint8_t leftBytes[16]{};
  std::uint8_t rightBytes[16]{};
  if (!parseComparableIp(left, leftFamily, leftBytes) || !parseComparableIp(right, rightFamily, rightBytes)) return 2.0;
  const int result = std::memcmp(leftBytes, rightBytes, 16);
  return result < 0 ? -1.0 : result > 0 ? 1.0 : 0.0;
}

inline bool net_ip_in_subnet(const std::string &address, double addressFamily, const std::string &network,
                      double networkFamily, double prefixValue) {
  std::uint8_t addressBytes[16]{};
  std::uint8_t networkBytes[16]{};
  if (!parseComparableIp(address, addressFamily, addressBytes) ||
      !parseComparableIp(network, networkFamily, networkBytes))
    return false;
  int prefix = static_cast<int>(prefixValue);
  if (networkFamily == 4.0) prefix += 96;
  if (prefix < 0 || prefix > 128) return false;
  const int wholeBytes = prefix / 8;
  const int remainingBits = prefix % 8;
  if (wholeBytes > 0 && std::memcmp(addressBytes, networkBytes, static_cast<std::size_t>(wholeBytes)) != 0) return false;
  if (remainingBits == 0) return true;
  const std::uint8_t mask = static_cast<std::uint8_t>(0xffu << (8 - remainingBits));
  return (addressBytes[wholeBytes] & mask) == (networkBytes[wholeBytes] & mask);
}

// ---------------------------------------------------------------------------
// Connection registry. Connections are template-instantiated on the concrete
// responder type, so the TS-side intrinsics reach them through this
// type-erased interface. Ids are monotonically increasing and never reused;
// an op on a dead connection is a registry miss and a safe no-op.
// ---------------------------------------------------------------------------
class HttpConnectionBase : public IoWatcher {
public:
  virtual void enqueueResponseBytes(std::string_view data) = 0;
  /** Both parts appended, then ONE flush. Two `enqueueResponseBytes` calls
   *  would flush twice and so could cost a second `send` for a write issued
   *  outside the reactor pass. */
  virtual void enqueueResponseParts(std::string_view head, std::string_view body) = 0;
  /**
   * The response header block, serialized straight into the connection's own
   * retained output buffer instead of into a string the response object owns.
   *
   * A `ServerResponse` is a fresh object per request, so a header block built
   * in one of its fields starts from an empty `std::string` every time and
   * regrows through 15 -> 30 -> 60 -> 120 -> 240 bytes: four allocations and
   * four copies per response, for bytes whose only destination is this buffer.
   * The buffer here is warm -- `flush()` clears it and keeps its capacity --
   * so the same appends cost nothing but the memcpy.
   *
   * `headBegin` writes the status line, `headField` one `Name: value` line,
   * and `headEnd` the lines Node computes itself (Date, Connection,
   * Keep-Alive, the framing header) plus the blank line. Nothing is flushed:
   * the block leaves with the first body bytes, exactly as Node holds
   * `_header` until the first write. `headEnd` answers the two facts the
   * caller still needs -- see `kHeadChunked` / `kHeadKeepAlive`.
   */
  virtual void headBegin(int status, std::string_view message) = 0;
  virtual void headField(std::string_view name, std::string_view value) = 0;
  virtual int headEnd(int flags, double autoContentLength) = 0;
  /** Header-block bytes the caller has already spelled; appended, not flushed. */
  virtual void headRaw(std::string_view text) = 0;
  /** A whole chunked body in one piece: size line, bytes, terminator; one flush. */
  virtual void enqueueFinalChunk(std::string_view data) = 0;
  virtual void responseComplete(bool keepAlive) = 0;
  virtual void hardDestroy() = 0;
  virtual std::string peerName() const = 0;
  /** Run the request `HandlerDispatchGate` parked on this connection. */
  virtual void dispatchDeferred() = 0;
};

// Request handlers never nest.
//
// Historically geatsc had no suspension primitive for `await`: an awaiting
// handler completed the await by pumping this reactor until the promise
// settled (`gea::detail::waitForPromise`), so its frame stayed on the C++
// stack for as long as it waited. An `await` is now a real suspension
// (`co_await`, and a nested pump from inside a callback aborts), so a handler
// leaves the stack at its first await and this gate is only ever open while a
// handler runs synchronously; it is kept as the fail-safe for that window. A request parsed from
// inside that pump used to be dispatched right there, ON TOP of the waiting
// handler -- and the stack makes the waits LIFO: the outer handler cannot
// resume, even once its promise has settled, until the inner one returns. The
// moment the inner handler waits on something only the outer can release, the
// process spins forever. A database driver hits this on the second concurrent
// request: its connection pool's check-out queues behind `maxConnecting`, while the
// handshake that would free a slot is an `await` in the frame underneath.
//
// So a request that completes while any handler is on the stack is parked
// here, in arrival order, and dispatched as soon as the last handler frame
// returns. Requests run one after another -- which is what a blocking await
// makes of them anyway -- instead of starving each other. I/O, timers and
// microtasks still run inside the pump: only NEW handler entries wait.
struct HandlerDispatchGate {
  int active = 0;  // handler frames currently on the C++ stack
  bool draining = false;
  std::deque<std::uint64_t> deferred;  // connection ids holding a parsed request
};

inline HandlerDispatchGate &handlerDispatchGate() {
  static HandlerDispatchGate gate;
  return gate;
}

inline std::unordered_map<std::uint64_t, HttpConnectionBase *> &connectionRegistry();

// Dispatch parked requests once no handler frame remains. A dispatched handler
// may park further requests from its own pump; they join the same queue and
// this loop takes them next, so the stack never grows with the queue.
inline void drainDeferredHandlers() {
  HandlerDispatchGate &gate = handlerDispatchGate();
  if (gate.draining) return;
  gate.draining = true;
  while (gate.active == 0 && !gate.deferred.empty()) {
    const std::uint64_t id = gate.deferred.front();
    gate.deferred.pop_front();
    // A connection destroyed while its request waited left the registry.
    auto found = connectionRegistry().find(id);
    if (found != connectionRegistry().end()) found->second->dispatchDeferred();
  }
  gate.draining = false;
}

// `headEnd` inputs. The response object knows these; the reactor does not.
inline constexpr int kHeadNoBody = 1;        // HEAD, 1xx, 204, 304: no framing header, no body
inline constexpr int kHeadHttp10 = 2;        // cannot chunk: an unframed body runs to connection close
inline constexpr int kHeadKeepAliveAsked = 4;
inline constexpr int kHeadSendDate = 8;
// `headEnd` outputs.
inline constexpr int kHeadChunked = 1;
inline constexpr int kHeadKeepAlive = 2;

inline std::unordered_map<std::uint64_t, HttpConnectionBase *> &connectionRegistry() {
  static std::unordered_map<std::uint64_t, HttpConnectionBase *> map;
  return map;
}

inline std::uint64_t nextConnectionId() {
  static std::uint64_t id = 0;
  return ++id;
}

// One-entry cache: a request's write/done ops hit the same connection
// back-to-back; skip the hash lookup for the repeat.
inline std::uint64_t g_lastConnectionId = 0;
inline HttpConnectionBase *g_lastConnection = nullptr;

inline HttpConnectionBase *findConnection(double id) {
  const auto key = static_cast<std::uint64_t>(id);
  if (key == g_lastConnectionId && g_lastConnection) return g_lastConnection;
  auto &registry = connectionRegistry();
  const auto found = registry.find(key);
  if (found == registry.end()) return nullptr;
  g_lastConnectionId = key;
  g_lastConnection = found->second;
  return found->second;
}

inline void invalidateConnectionCache(std::uint64_t id) {
  if (g_lastConnectionId == id) {
    g_lastConnectionId = 0;
    g_lastConnection = nullptr;
  }
}

// server.close() hook — set by serve() for the active listener.
inline std::function<void()> &listenerCloser() {
  static std::function<void()> fn;
  return fn;
}

// RFC 7231 IMF-fixdate for the Date response header, cached per second.
struct HttpDateCache {
  std::string text;
  time_t second = 0;
};

inline HttpDateCache &httpDateCache() {
  static HttpDateCache cache;
  return cache;
}

inline const std::string &cachedHttpDate() {
  HttpDateCache &cache = httpDateCache();
  const time_t now = ::time(nullptr);
  if (now != cache.second) {
    cache.second = now;
    tm parts{};
    platform::utcTime(now, parts);
    char buffer[64];
    const std::size_t n = ::strftime(buffer, sizeof(buffer), "%a, %d %b %Y %H:%M:%S GMT", &parts);
    cache.text.assign(buffer, n);
  }
  return cache.text;
}

/**
 * Which second the cached date belongs to.
 *
 * `cachedHttpDate` reformats once a second, but its RESULT crosses the host
 * boundary by value, and twenty-nine characters is past what any std::string
 * keeps inline -- so reading the date cost an allocation and a free on every
 * single response, for a string that changes once a second. Handing out the
 * second lets the caller keep the whole `Date: ...` line it already built and
 * rebuild it only when this number changes.
 */
inline time_t cachedHttpDateSecond() {
  cachedHttpDate();
  return httpDateCache().second;
}

inline bool asciiEqualsIgnoreCase(std::string_view a, const char *b) {
  const std::size_t bLength = std::strlen(b);
  if (a.size() != bLength) return false;
  for (std::size_t i = 0; i < a.size(); ++i) {
    char ca = a[i];
    char cb = b[i];
    if (ca >= 'A' && ca <= 'Z') ca = static_cast<char>(ca - 'A' + 'a');
    if (cb >= 'A' && cb <= 'Z') cb = static_cast<char>(cb - 'A' + 'a');
    if (ca != cb) return false;
  }
  return true;
}

// Case-insensitive "does the comma-separated header value contain this token".
inline bool headerValueHasToken(std::string_view value, const char *token) {
  std::size_t pos = 0;
  while (pos < value.size()) {
    std::size_t comma = value.find(',', pos);
    if (comma == std::string_view::npos) comma = value.size();
    std::string_view item = value.substr(pos, comma - pos);
    while (!item.empty() && (item.front() == ' ' || item.front() == '\t')) item.remove_prefix(1);
    while (!item.empty() && (item.back() == ' ' || item.back() == '\t')) item.remove_suffix(1);
    if (asciiEqualsIgnoreCase(item, token)) return true;
    pos = comma + 1;
  }
  return false;
}

template <typename Responder>
class HttpServer;  // forward

// ---------------------------------------------------------------------------
// One accepted HTTP/1.x connection. Parses full requests (request line,
// headers, content-length and chunked bodies, limits, 100-continue) and
// dispatches each into the geatsc-compiled responder with a connection id.
// The TS side streams response bytes back via enqueueResponseBytes and
// signals completion via responseComplete — so a response can finish
// synchronously, from a microtask (async handler), or from a timer.
// Pipelining stays ordered: the next buffered request is not dispatched
// until the current response completes.
// ---------------------------------------------------------------------------
template <typename Responder>
class HttpConnection final : public HttpConnectionBase {
public:
  HttpConnection(EventLoop &loop, HttpServer<Responder> &server, platform::Fd fd)
      : loop_(loop), server_(server), socket_(fd), id_(nextConnectionId()) {
    socket_.makeNonBlocking();
    socket_.setTcpNoDelay();
    connectionRegistry()[id_] = this;
  }

  ~HttpConnection() override {
    connectionRegistry().erase(id_);
    invalidateConnectionCache(id_);
  }

  platform::Fd fd() const noexcept override { return socket_.fd(); }
  short interestedEvents() const noexcept override {
    return static_cast<short>(POLLIN | (outsent_ >= outbuf_.size() ? 0 : POLLOUT));
  }

  void onReady(short revents) override {
    if (revents & (POLLERR | POLLNVAL)) {
      loop_.close(this);
      return;
    }
    if (revents & (POLLIN | POLLHUP)) {
      if (!readAvailable()) {
        peerClosed_ = true;
        if (!busy_ && outsent_ >= outbuf_.size()) {
          loop_.close(this);
          return;
        }
      }
      processing_ = true;
      processInput();
      processing_ = false;
    }
    flushAndMaybeClose();
  }

  // ---- ops reachable from TypeScript through the registry ----

  void enqueueResponseBytes(std::string_view data) override {
    // Append into the retained buffer. `flush()` clears it but keeps its
    // capacity, so in steady state this is a memcpy into warm storage and the
    // connection allocates nothing per response.
    //
    // This deliberately replaces an earlier `outbuf_ = std::move(data)` fast
    // path taken when the buffer was empty. Stealing the caller's string saved
    // the memcpy, but it DISCARDED the warm capacity every single response --
    // the buffer `flush()` had just cleared for reuse was freed and replaced
    // by a fresh allocation, so the saving was paid for with an allocate/free
    // pair per response. A few hundred bytes of memcpy is cheaper than that.
    headMark_ = std::string::npos;
    outbuf_.append(data);
    // Streaming writes issued outside the reactor pass (async handlers,
    // timers) flush eagerly; writes during dispatch batch into the pass flush.
    if (!processing_) flushAndMaybeClose();
  }

  void enqueueResponseParts(std::string_view head, std::string_view body) override {
    headMark_ = std::string::npos;
    outbuf_.append(head);
    outbuf_.append(body);
    if (!processing_) flushAndMaybeClose();
  }

  void headBegin(int status, std::string_view message) override {
    headMark_ = outbuf_.size();
    headHasContentLength_ = false;
    headHasTransferEncoding_ = false;
    headHasDate_ = false;
    headHasConnection_ = false;
    headConnectionClose_ = false;
    if (status == 200 && message == "OK") {
      outbuf_.append("HTTP/1.1 200 OK\r\n");
      return;
    }
    outbuf_.append("HTTP/1.1 ");
    char digits[16];
    const auto converted = std::to_chars(digits, digits + sizeof(digits), status);
    outbuf_.append(digits, static_cast<std::size_t>(converted.ptr - digits));
    outbuf_.push_back(' ');
    outbuf_.append(message);
    outbuf_.append("\r\n");
  }

  void headField(std::string_view name, std::string_view value) override {
    // The four names the framing decision reads. Length first: almost every
    // header an application sets fails all four on that alone.
    switch (name.size()) {
      case 4:
        if (asciiEqualsIgnoreCase(name, "date")) headHasDate_ = true;
        break;
      case 10:
        if (asciiEqualsIgnoreCase(name, "connection")) {
          headHasConnection_ = true;
          if (asciiEqualsIgnoreCase(value, "close")) headConnectionClose_ = true;
        }
        break;
      case 14:
        if (asciiEqualsIgnoreCase(name, "content-length")) headHasContentLength_ = true;
        break;
      case 17:
        if (asciiEqualsIgnoreCase(name, "transfer-encoding")) headHasTransferEncoding_ = true;
        break;
      default:
        break;
    }
    outbuf_.append(name);
    outbuf_.append(": ");
    outbuf_.append(value);
    outbuf_.append("\r\n");
  }

  void headRaw(std::string_view text) override {
    if (headMark_ == std::string::npos) headMark_ = outbuf_.size();
    outbuf_.append(text);
  }

  int headEnd(int flags, double autoContentLength) override {
    const bool noBody = (flags & kHeadNoBody) != 0;
    bool chunked = false;
    bool closeDelimited = false;
    bool writeContentLength = false;
    if (noBody || headHasContentLength_ || headHasTransferEncoding_) {
      // No body, or the application framed it itself: serialize as given.
    } else if (autoContentLength >= 0) {
      writeContentLength = true;
    } else if ((flags & kHeadHttp10) != 0) {
      closeDelimited = true;
    } else {
      chunked = true;
    }
    bool keepAlive = (flags & kHeadKeepAliveAsked) != 0;
    if (headConnectionClose_ || closeDelimited) keepAlive = false;
    // The overwhelmingly common tail -- Date, keep-alive, chunked -- is the
    // same bytes for every response in a given second, so it is assembled once
    // per second and appended as one piece instead of as seven.
    if (chunked && keepAlive && !headHasConnection_ && !headHasDate_ && (flags & kHeadSendDate) != 0) {
      static thread_local time_t tailSecond = 0;
      static thread_local std::string tail;
      const std::string &date = cachedHttpDate();
      const time_t second = httpDateCache().second;
      if (second != tailSecond || tail.empty()) {
        tailSecond = second;
        tail.assign("Date: ");
        tail.append(date);
        tail.append("\r\nConnection: keep-alive\r\nKeep-Alive: timeout=5\r\nTransfer-Encoding: chunked\r\n\r\n");
      }
      outbuf_.append(tail);
      return kHeadChunked | kHeadKeepAlive;
    }
    if ((flags & kHeadSendDate) != 0 && !headHasDate_) {
      outbuf_.append("Date: ");
      outbuf_.append(cachedHttpDate());
      outbuf_.append("\r\n");
    }
    if (!headHasConnection_) {
      if (keepAlive) outbuf_.append("Connection: keep-alive\r\nKeep-Alive: timeout=5\r\n");
      else outbuf_.append("Connection: close\r\n");
    }
    if (writeContentLength) {
      outbuf_.append("Content-Length: ");
      char digits[24];
      const auto converted = std::to_chars(digits, digits + sizeof(digits), static_cast<std::uint64_t>(autoContentLength));
      outbuf_.append(digits, static_cast<std::size_t>(converted.ptr - digits));
      outbuf_.append("\r\n");
    } else if (chunked) {
      outbuf_.append("Transfer-Encoding: chunked\r\n");
    }
    outbuf_.append("\r\n");
    return (chunked ? kHeadChunked : 0) | (keepAlive ? kHeadKeepAlive : 0);
  }

  void enqueueFinalChunk(std::string_view data) override {
    headMark_ = std::string::npos;
    if (!data.empty()) {
      char digits[24];
      const auto converted = std::to_chars(digits, digits + sizeof(digits), data.size(), 16);
      outbuf_.append(digits, static_cast<std::size_t>(converted.ptr - digits));
      outbuf_.append("\r\n");
      outbuf_.append(data);
      outbuf_.append("\r\n");
    }
    outbuf_.append("0\r\n\r\n");
    if (!processing_) flushAndMaybeClose();
  }


  void responseComplete(bool keepAlive) override {
    busy_ = false;
    if (!keepAlive) closing_ = true;
    if (!processing_) {
      processing_ = true;
      processInput();
      processing_ = false;
      flushAndMaybeClose();
    }
  }

  void hardDestroy() override {
    closing_ = true;
    loop_.close(this);
  }

  std::string peerName() const override {
    sockaddr_in addr{};
    socklen_t length = sizeof(addr);
    if (::getpeername(socket_.fd(), reinterpret_cast<sockaddr *>(&addr), &length) != 0) return std::string();
    char ip[INET_ADDRSTRLEN] = {0};
    ::inet_ntop(AF_INET, &addr.sin_addr, ip, sizeof(ip));
    std::string out(ip);
    out += ' ';
    out += std::to_string(static_cast<unsigned>(ntohs(addr.sin_port)));
    return out;
  }

private:
  enum class ReadState { kHead, kBodyFixed, kChunkSize, kChunkData, kChunkCrlf, kTrailers };

  static constexpr std::size_t kMaxHeadBytes = 16384;              // Node --max-http-header-size default
  static constexpr std::size_t kMaxBodyBytes = 64 * 1024 * 1024;   // hard cap; Node itself is unbounded
  // How much output capacity one connection may retain between responses.
  // Above this, `flush()` releases it rather than pinning it for the life of
  // a keep-alive connection that sent one large body.
  static constexpr std::size_t kMaxRetainedOutBuf = 64 * 1024;
  // The same cap on the way in. `inbuf_` is consumed with `erase(0, n)`, which
  // never shrinks, so without this a keep-alive connection that once carried a
  // large upload pins that much for the rest of its life -- per connection.
  static constexpr std::size_t kMaxRetainedInBuf = 64 * 1024;

  bool readAvailable() {
    char buffer[65536];
    for (;;) {
      const platform::ssize n = platform::recvBytes(socket_.fd(), buffer, sizeof(buffer));
      if (n > 0) {
        inbuf_.append(buffer, static_cast<std::size_t>(n));
        if (n < static_cast<platform::ssize>(sizeof(buffer))) return true;  // likely drained
        continue;
      }
      if (n == 0) return false;  // peer closed
      return errno == EAGAIN || errno == EWOULDBLOCK;
    }
  }

  void sendErrorAndClose(int status, const char *reason) {
    // Shape matches Node's parser-level error responses: status line +
    // Connection: close, nothing else (no Date, no Content-Length).
    std::string response = "HTTP/1.1 ";
    response += std::to_string(status);
    response += ' ';
    response += reason;
    response += "\r\nConnection: close\r\n\r\n";
    enqueueResponseBytes(response);
    busy_ = true;  // stop parsing anything further on this connection
    closing_ = true;
  }

  // Parse buffered input; dispatch each complete request while idle.
  void processInput() {
    for (;;) {
      if (busy_ || closing_) return;
      if (state_ == ReadState::kHead) {
        const std::size_t headEnd = inbuf_.find("\r\n\r\n");
        if (headEnd == std::string::npos) {
          if (inbuf_.size() > kMaxHeadBytes) sendErrorAndClose(431, "Request Header Fields Too Large");
          // Everything buffered has been consumed and the next request has not
          // begun: the one moment releasing capacity costs nothing. Gated on
          // EMPTY so the steady state -- a warm buffer sized for ordinary
          // requests -- keeps its storage and allocates nothing per request.
          if (inbuf_.empty() && inbuf_.capacity() > kMaxRetainedInBuf) inbuf_.shrink_to_fit();
          return;
        }
        if (headEnd + 4 > kMaxHeadBytes) {
          sendErrorAndClose(431, "Request Header Fields Too Large");
          return;
        }
        const int parseStatus = parseHead(std::string_view(inbuf_.data(), headEnd));
        inbuf_.erase(0, headEnd + 4);
        if (parseStatus == 400) {
          sendErrorAndClose(400, "Bad Request");
          return;
        }
        if (parseStatus == 501) {
          sendErrorAndClose(501, "Not Implemented");
          return;
        }
        if (expectContinue_) outbuf_ += "HTTP/1.1 100 Continue\r\n\r\n";
        if (chunked_) {
          state_ = ReadState::kChunkSize;
        } else if (bodyRemaining_ > 0) {
          state_ = ReadState::kBodyFixed;
        } else {
          dispatchCurrent();
        }
        continue;
      }
      if (state_ == ReadState::kBodyFixed) {
        const std::size_t take = std::min(bodyRemaining_, inbuf_.size());
        body_.append(inbuf_.data(), take);
        inbuf_.erase(0, take);
        bodyRemaining_ -= take;
        if (body_.size() > kMaxBodyBytes) {
          sendErrorAndClose(413, "Payload Too Large");
          return;
        }
        if (bodyRemaining_ > 0) return;  // need more bytes
        state_ = ReadState::kHead;
        dispatchCurrent();
        continue;
      }
      if (state_ == ReadState::kChunkSize) {
        const std::size_t eol = inbuf_.find("\r\n");
        if (eol == std::string::npos) {
          if (inbuf_.size() > 1024) sendErrorAndClose(400, "Bad Request");
          return;
        }
        std::size_t size = 0;
        bool sawDigit = false;
        for (std::size_t i = 0; i < eol; ++i) {
          const char c = inbuf_[i];
          if (c == ';') break;  // chunk extensions — ignored
          std::size_t digit;
          if (c >= '0' && c <= '9') digit = static_cast<std::size_t>(c - '0');
          else if (c >= 'a' && c <= 'f') digit = static_cast<std::size_t>(c - 'a' + 10);
          else if (c >= 'A' && c <= 'F') digit = static_cast<std::size_t>(c - 'A' + 10);
          else {
            sendErrorAndClose(400, "Bad Request");
            return;
          }
          size = size * 16 + digit;
          sawDigit = true;
          if (size > kMaxBodyBytes) {
            sendErrorAndClose(413, "Payload Too Large");
            return;
          }
        }
        if (!sawDigit) {
          sendErrorAndClose(400, "Bad Request");
          return;
        }
        inbuf_.erase(0, eol + 2);
        chunkRemaining_ = size;
        state_ = size == 0 ? ReadState::kTrailers : ReadState::kChunkData;
        continue;
      }
      if (state_ == ReadState::kChunkData) {
        const std::size_t take = std::min(chunkRemaining_, inbuf_.size());
        body_.append(inbuf_.data(), take);
        inbuf_.erase(0, take);
        chunkRemaining_ -= take;
        if (body_.size() > kMaxBodyBytes) {
          sendErrorAndClose(413, "Payload Too Large");
          return;
        }
        if (chunkRemaining_ > 0) return;
        state_ = ReadState::kChunkCrlf;
        continue;
      }
      if (state_ == ReadState::kChunkCrlf) {
        if (inbuf_.size() < 2) return;
        if (inbuf_[0] != '\r' || inbuf_[1] != '\n') {
          sendErrorAndClose(400, "Bad Request");
          return;
        }
        inbuf_.erase(0, 2);
        state_ = ReadState::kChunkSize;
        continue;
      }
      if (state_ == ReadState::kTrailers) {
        const std::size_t eol = inbuf_.find("\r\n");
        if (eol == std::string::npos) return;
        inbuf_.erase(0, eol + 2);
        if (eol == 0) {
          // Empty line: trailer section over (trailers parsed + discarded).
          state_ = ReadState::kHead;
          dispatchCurrent();
        }
        continue;
      }
      return;
    }
  }

  // Returns 0 (ok), 400, or 501. Fills the current-request fields.
  int parseHead(std::string_view head) {
    rawHead_.clear();
    body_.clear();
    bodyRemaining_ = 0;
    chunkRemaining_ = 0;
    chunked_ = false;
    expectContinue_ = false;

    std::size_t lineEnd = head.find("\r\n");
    const std::string_view line = head.substr(0, lineEnd == std::string_view::npos ? head.size() : lineEnd);
    const std::size_t sp1 = line.find(' ');
    if (sp1 == std::string_view::npos || sp1 == 0) return 400;
    const std::size_t sp2 = line.find(' ', sp1 + 1);
    if (sp2 == std::string_view::npos || sp2 == sp1 + 1) return 400;
    method_.assign(line.data(), sp1);
    target_.assign(line.data() + sp1 + 1, sp2 - sp1 - 1);
    version_.assign(line.data() + sp2 + 1, line.size() - sp2 - 1);
    if (version_ != "HTTP/1.1" && version_ != "HTTP/1.0") return 400;
    keepAlive_ = version_ == "HTTP/1.1";
    // Hand the raw header BLOCK to TS as one buffer; the [name, value, ...]
    // array materializes lazily only if the app reads req.rawHeaders/headers.
    if (lineEnd != std::string_view::npos && lineEnd + 2 < head.size()) {
      rawHead_.assign(head.data() + lineEnd + 2, head.size() - lineEnd - 2);
    }

    bool haveContentLength = false;
    std::string_view contentLengthValue;
    std::size_t pos = lineEnd == std::string_view::npos ? head.size() : lineEnd + 2;
    while (pos < head.size()) {
      std::size_t eol = head.find("\r\n", pos);
      if (eol == std::string_view::npos) eol = head.size();
      const std::string_view header = head.substr(pos, eol - pos);
      pos = eol + 2;
      if (header.empty()) continue;
      const std::size_t colon = header.find(':');
      if (colon == std::string_view::npos || colon == 0) return 400;
      const std::string_view name = header.substr(0, colon);
      for (const char c : name) {
        if (c == ' ' || c == '\t') return 400;  // smuggling guard: no WS in field names
      }
      std::size_t valueStart = colon + 1;
      while (valueStart < header.size() && (header[valueStart] == ' ' || header[valueStart] == '\t')) ++valueStart;
      std::size_t valueEnd = header.size();
      while (valueEnd > valueStart && (header[valueEnd - 1] == ' ' || header[valueEnd - 1] == '\t')) --valueEnd;
      const std::string_view value = header.substr(valueStart, valueEnd - valueStart);

      if (asciiEqualsIgnoreCase(name, "content-length")) {
        if (haveContentLength && value != contentLengthValue) return 400;
        haveContentLength = true;
        contentLengthValue = value;
      } else if (asciiEqualsIgnoreCase(name, "transfer-encoding")) {
        if (headerValueHasToken(value, "chunked")) chunked_ = true;
        else return 400;  // matches Node: unsupported transfer-encoding → 400
      } else if (asciiEqualsIgnoreCase(name, "connection")) {
        if (headerValueHasToken(value, "close")) keepAlive_ = false;
        else if (headerValueHasToken(value, "keep-alive")) keepAlive_ = true;
      } else if (asciiEqualsIgnoreCase(name, "expect")) {
        if (asciiEqualsIgnoreCase(value, "100-continue")) expectContinue_ = true;
      }
    }

    if (!chunked_ && haveContentLength) {
      // RFC 9112: Transfer-Encoding wins over Content-Length when both appear.
      std::size_t parsed = 0;
      if (contentLengthValue.empty()) return 400;
      for (const char c : contentLengthValue) {
        if (c < '0' || c > '9') return 400;
        parsed = parsed * 10 + static_cast<std::size_t>(c - '0');
        if (parsed > kMaxBodyBytes) return 400;
      }
      bodyRemaining_ = parsed;
    }
    return 0;
  }

  void dispatchCurrent() {
    busy_ = true;
    // Another handler is waiting on the stack: park this request (the parsed
    // fields stay put -- `busy_` stops the parser) until it returns.
    HandlerDispatchGate &gate = handlerDispatchGate();
    if (gate.active > 0) {
      deferred_ = true;
      gate.deferred.push_back(id_);
      return;
    }
    ++gate.active;
    runHandler();
    --gate.active;
    if (gate.active == 0) drainDeferredHandlers();
  }

  void dispatchDeferred() override {
    if (!deferred_) return;
    deferred_ = false;
    const bool wasProcessing = processing_;
    processing_ = true;
    dispatchCurrent();
    processInput();  // a pipelined request buffered behind the parked one
    processing_ = wasProcessing;
    if (!processing_) flushAndMaybeClose();
  }

  // The handler call itself; never throws (a throw becomes this request's 500).
  void runHandler() {
    double flags = keepAlive_ ? 1.0 : 0.0;
    if (firstRequest_) flags += 2.0;
    firstRequest_ = false;
    // A handler that throws is isolated to ITS request -- 500 and close the
    // connection -- rather than taking the process down, which is the
    // documented deviation from real Node (`runtime/node/http.ts`).
    //
    // How a throw arrives differs by runtime and only that: v1 sets a global
    // flag and returns normally, v2 propagates a real C++ exception carrying a
    // `gea::Value`. Each arm reads its own runtime's answer; `handlerThrew`
    // then says the same thing to the same recovery below.
    bool handlerThrew = false;
    try {
      server_.responder()(static_cast<double>(id_), flags, std::move(method_), std::move(target_),
                          std::move(version_), std::move(rawHead_), std::move(body_));
    } catch (...) {
      handlerThrew = true;
    }
    method_.clear();
    target_.clear();
    version_.clear();
    rawHead_.clear();
    body_.clear();
    if (handlerThrew) {
      if (busy_) {
        // A header block the handler committed but never released (it threw
        // between `writeHead` and the first body write) is still sitting at
        // the tail of the buffer. It was never the response; the 500 is.
        if (headMark_ != std::string::npos && headMark_ >= outsent_ && headMark_ <= outbuf_.size()) outbuf_.resize(headMark_);
        enqueueResponseBytes(
            "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
        busy_ = false;
        closing_ = true;
      }
    }
    if (busy_ && peerClosed_) closing_ = true;  // half-closed peer: finish this response, take no more
  }

  bool flush() {
    while (outsent_ < outbuf_.size()) {
      const platform::ssize n = platform::sendBytes(socket_.fd(), outbuf_.data() + outsent_, outbuf_.size() - outsent_);
      if (n > 0) {
        outsent_ += static_cast<std::size_t>(n);
        continue;
      }
      if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) return true;  // retry on writable
      return false;
    }
    outbuf_.clear();
    headMark_ = std::string::npos;
    // `clear()` keeps capacity, which is the point -- the next response reuses
    // it. Cap what a connection may hold onto so a single large body does not
    // pin that much per connection for the rest of its life.
    if (outbuf_.capacity() > kMaxRetainedOutBuf) outbuf_.shrink_to_fit();
    outsent_ = 0;
    return true;
  }

  void flushAndMaybeClose() {
    if (!flush()) {
      loop_.close(this);
      return;
    }
    const bool drained = outsent_ >= outbuf_.size();
    if (drained && !busy_ && (closing_ || peerClosed_)) loop_.close(this);
  }

  EventLoop &loop_;
  HttpServer<Responder> &server_;
  Socket socket_;
  std::uint64_t id_;
  std::string inbuf_;
  std::string outbuf_;
  std::size_t outsent_ = 0;  // bytes of outbuf_ already written to the socket

  ReadState state_ = ReadState::kHead;
  bool busy_ = false;         // a dispatched request is awaiting responseComplete
  bool closing_ = false;      // no further requests; close once drained
  bool peerClosed_ = false;   // read side saw EOF
  bool processing_ = false;   // inside the reactor's processInput pass
  bool firstRequest_ = true;
  bool deferred_ = false;     // parked in `HandlerDispatchGate` with a parsed request

  // The header block being serialized by `headBegin`/`headField`/`headEnd`.
  // `headMark_` is where it starts in `outbuf_` until body bytes release it.
  std::size_t headMark_ = std::string::npos;
  bool headHasContentLength_ = false;
  bool headHasTransferEncoding_ = false;
  bool headHasDate_ = false;
  bool headHasConnection_ = false;
  bool headConnectionClose_ = false;

  // Current request being parsed.
  std::string method_;
  std::string target_;
  std::string version_;
  std::string rawHead_;
  std::string body_;
  std::size_t bodyRemaining_ = 0;
  std::size_t chunkRemaining_ = 0;
  bool chunked_ = false;
  bool keepAlive_ = true;
  bool expectContinue_ = false;
};

// ---------------------------------------------------------------------------
// The listening server. Owns the responder invoked once per parsed request.
// ---------------------------------------------------------------------------
template <typename Responder>
class HttpServer final : public IoWatcher {
public:
  HttpServer(EventLoop &loop, int port, Responder responder)
      : loop_(loop), listener_(makeListener(port)), responder_(std::move(responder)) {}

  bool valid() const noexcept { return platform::valid(listener_.fd()); }
  const Responder &responder() const noexcept { return responder_; }
  Responder &responder() noexcept { return responder_; }

  platform::Fd fd() const noexcept override { return listener_.fd(); }
  short interestedEvents() const noexcept override { return POLLIN; }

  void onReady(short revents) override {
    if ((revents & POLLIN) == 0) return;
    for (;;) {
      const platform::Fd clientFd = platform::acceptSocket(listener_.fd(), nullptr, nullptr);
      if (!platform::valid(clientFd)) break;  // EAGAIN: drained
      loop_.adopt(std::make_unique<HttpConnection<Responder>>(loop_, *this, clientFd));
    }
  }

private:
  static Socket makeListener(int port) {
    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(static_cast<uint16_t>(port));
    // A cluster worker on Windows listens on the primary's socket.
    platform::Fd shared = platform::kInvalidFd;
    if (platform::sharedClusterListener(reinterpret_cast<sockaddr *>(&addr), sizeof(addr), 1024, false, shared))
      return Socket(shared);
    const platform::Fd fd = platform::openSocket(AF_INET, SOCK_STREAM, 0);
    Socket socket(fd);
    if (!platform::valid(fd)) return socket;
    platform::configureListener(fd, true);
    if (platform::bindSocket(fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) < 0) return Socket(platform::kInvalidFd);
    if (platform::listenSocket(fd, 1024) < 0) return Socket(platform::kInvalidFd);
    socket.makeNonBlocking();
    return socket;
  }

  EventLoop &loop_;
  Socket listener_;
  Responder responder_;
};

// Build a complete raw HTTP/1.1 response (used by the __gea_serve_hello demo).
inline std::string makeResponse(const char *contentType, const std::string &body) {
  char header[256];
  const int n = std::snprintf(header, sizeof(header),
                              "HTTP/1.1 200 OK\r\nContent-Type: %s\r\nContent-Length: %zu\r\nConnection: keep-alive\r\n\r\n",
                              contentType, body.size());
  std::string out;
  out.append(header, static_cast<std::size_t>(n));
  out.append(body);
  return out;
}

// Run a server on `port` with the given responder until the loop drains.
// Templated on the concrete responder type — the whole listener→connection→
// dispatch chain monomorphizes on it, so the responder call is direct.
//
// GEA_WORKERS=N (N > 1) forks into N single-threaded worker processes BEFORE
// the listen socket exists. Each worker then opens its OWN SO_REUSEPORT
// socket (makeListener already sets the option), so the kernel hash-balances
// incoming connections across N independent reactors — the nginx / node
// `cluster` deployment shape. Forking pre-socket means no shared accept
// queue and no thundering herd; forking post-app-setup means every worker
// inherits the fully-built JS world (routes, closures) copy-on-write and
// nothing is shared afterwards, so the runtime's single-thread assumptions
// (pool allocator free lists, microtask queue, boxed values) hold per
// process exactly as they do today. The parent serves as worker 0.
template <typename Responder>
inline void serve(int port, Responder responder) {
  platform::ensureSockets();
  int workers = 1;
  if (const char *env = std::getenv("GEA_WORKERS")) {
    const int parsed = std::atoi(env);
    if (parsed > 1 && parsed <= 256) workers = parsed;
  }
  const bool isParent = platform::forkWorkers(workers);  // 1 where the host cannot fork
  EventLoop &loop = clientEventLoop();
  HttpServer<Responder> server(loop, port, std::move(responder));
  if (!server.valid()) {
    std::fprintf(stderr, "gea-node: failed to bind port %d: %s\n", port, platform::errorText(errno));
    return;
  }
  if (isParent) {
    // Formatted first and written once: an unbuffered stream would otherwise
    // reach the file in pieces, interleaved with every other process writing
    // to the same stderr (cluster workers).
    char line[96];
    if (workers > 1) std::snprintf(line, sizeof(line), "gea-node: listening on http://127.0.0.1:%d (%d workers)\n", port, workers);
    else std::snprintf(line, sizeof(line), "gea-node: listening on http://127.0.0.1:%d\n", port);
    std::fputs(line, stderr);
  }
  loop.addBorrowed(&server);
  listenerCloser() = [&loop, &server]() { loop.close(&server); };
  loop.run();
  listenerCloser() = nullptr;
}

// OS byte operations use one native byte vector internally. Each compiler's
// boundary below wraps it in its own typed Buffer carrier without boxing.
static void byteOperationError(const std::string &name, const std::string &message) {
  gea::Value error = gea::Value::object();
  error.setProperty(gea::PropertyKey::string("name"), gea::Value::box(gea::Value::Tag::String, name));
  error.setProperty(gea::PropertyKey::string("message"), gea::Value::box(gea::Value::Tag::String, message));
  throw error;
}

static std::vector<std::uint8_t> randomByteVector(double requested_size) {
  if (!std::isfinite(requested_size) || requested_size < 0 || requested_size > 2147483647.0) {
    byteOperationError("RangeError", "randomBytes size is out of range");
    return {};
  }
  std::vector<std::uint8_t> out(static_cast<std::size_t>(requested_size));
  if (out.empty()) return out;
  if (!platform::fillRandom(out.data(), out.size())) {
    byteOperationError("Error", "Unable to obtain secure random bytes");
    return {};
  }
  return out;
}

static std::vector<std::uint8_t> readFileByteVector(const std::string &path) {
  std::vector<std::uint8_t> out;
  std::string failure;
  if (!platform::readWholeFile(path, out, failure)) {
    byteOperationError("Error", failure);
    return {};
  }
  return out;
}

inline std::vector<std::uint8_t> crypto_random_bytes(double requested_size) { return randomByteVector(requested_size); }
inline std::vector<std::uint8_t> read_file(const std::string &path) { return readFileByteVector(path); }

}  // namespace gea::node

// ---- node:cluster ----------------------------------------------------------
// Node's cluster forks WORKER PROCESSES that re-run the program from the top
// with `cluster.isPrimary === false`, and the primary hears of each worker's
// exit. This is that shape without an IPC channel:
//
//  - `fork` re-executes THIS binary (posix_spawn of the executable path) with
//    `GEA_CLUSTER_WORKER_ID` in the environment -- the worker starts at the
//    top of the program exactly as Node's does, not at the fork call.
//  - worker exits reach the primary as SIGCHLD -> self-pipe -> reactor watcher
//    -> waitpid, so 'exit' fires from the event loop like every other event.
//    The watcher is referenced only while a worker is alive, so a primary
//    with nothing else to do exits when its last worker has.
//  - a liveness pipe whose ONLY write end is the primary's (CLOEXEC) stands in
//    for the channel: when the primary dies, every worker reads EOF and exits,
//    which is what Node's worker does on an unexpected 'disconnect'.
//
// Connection distribution is the kernel's: each worker's listener sets
// SO_REUSEPORT (makeListener; net listeners inherit it below), i.e. Node's
// SCHED_NONE, not the primary-side round robin of SCHED_RR. No IPC means no
// `worker.send`; runtime/node/cluster.ts refuses those loudly.
namespace gea::node::cluster {

struct ExitEvent {
  double pid = 0;
  double code = -1;    // -1 encodes Node's `null` (the worker was killed by a signal)
  std::string signal;  // empty encodes Node's `null` (the worker exited on its own)
};

struct State {
  std::deque<ExitEvent> exits;
  ExitEvent last;
  std::function<void()> notify;
  std::vector<std::string> arguments;
  std::string spawnError;
  platform::Fd exitNotifier = platform::kInvalidFd;
  int liveChildren = 0;
  bool watcherInstalled = false;
};

inline State &state() {
  static State instance;
  return instance;
}

inline const char *signalName(int number) {
  switch (number) {
    case SIGHUP: return "SIGHUP";
    case SIGINT: return "SIGINT";
    case SIGQUIT: return "SIGQUIT";
    case SIGABRT: return "SIGABRT";
    case SIGKILL: return "SIGKILL";
    case SIGSEGV: return "SIGSEGV";
    case SIGPIPE: return "SIGPIPE";
    case SIGALRM: return "SIGALRM";
    case SIGTERM: return "SIGTERM";
    case SIGUSR1: return "SIGUSR1";
    case SIGUSR2: return "SIGUSR2";
    case SIGSTOP: return "SIGSTOP";
    case SIGCONT: return "SIGCONT";
    default: return "";
  }
}

inline int signalNumber(const std::string &name) {
  static const std::pair<const char *, int> table[] = {
      {"SIGHUP", SIGHUP},   {"SIGINT", SIGINT},   {"SIGQUIT", SIGQUIT}, {"SIGABRT", SIGABRT}, {"SIGKILL", SIGKILL},
      {"SIGSEGV", SIGSEGV}, {"SIGPIPE", SIGPIPE}, {"SIGALRM", SIGALRM}, {"SIGTERM", SIGTERM}, {"SIGUSR1", SIGUSR1},
      {"SIGUSR2", SIGUSR2}, {"SIGSTOP", SIGSTOP}, {"SIGCONT", SIGCONT}};
  for (const auto &entry : table)
    if (name == entry.first) return entry.second;
  if (!name.empty() && std::all_of(name.begin(), name.end(), [](unsigned char c) { return std::isdigit(c) != 0; }))
    return std::atoi(name.c_str());
  return -1;
}

inline void reap() {
  State &s = state();
  bool any = false;
  platform::reapExitedChildren([&](double pid, double code, int signal) {
    ExitEvent event;
    event.pid = pid;
    if (signal != 0) event.signal = signalName(signal);
    else if (code >= 0) event.code = code;
    s.exits.push_back(std::move(event));
    if (s.liveChildren > 0) --s.liveChildren;
    any = true;
  });
  if (any && s.notify) s.notify();
}

class ChildWatcher final : public IoWatcher {
public:
  platform::Fd fd() const noexcept override { return state().exitNotifier; }
  short interestedEvents() const noexcept override { return POLLIN; }
  bool isReferenced() const noexcept override { return state().liveChildren > 0; }
  void onReady(short) override {
    platform::drainChildExitNotifier(state().exitNotifier);
    reap();
  }
};

inline bool ensureWatcher() {
  State &s = state();
  if (s.watcherInstalled) return true;
  if (!platform::installChildExitNotifier(s.exitNotifier, s.spawnError)) return false;
  static ChildWatcher watcher;
  clientEventLoop().addBorrowed(&watcher);
  s.watcherInstalled = true;
  return true;
}

inline std::string executablePath() { return platform::executablePath(); }

// `envPairs` is the caller's `fork(env)` as "k=v" entries separated by 0x1f.
inline double spawn(double id, const std::string &envPairs) {
  State &s = state();
  std::string channel;
  if (!platform::createLivenessChannel(channel, s.spawnError) || !ensureWatcher()) return 0;
  const std::string exe = executablePath();
  if (exe.empty()) {
    s.spawnError = "cannot resolve the executable path of this process";
    return 0;
  }
  const auto isReserved = [](std::string_view entry) {
    for (const std::string_view key : {"GEA_CLUSTER_WORKER_ID=", "NODE_UNIQUE_ID=", "GEA_CLUSTER_CHANNEL_FD=", "GEA_WORKERS="})
      if (entry.substr(0, key.size()) == key) return true;
    return false;
  };
  std::vector<std::string> env;
  for (std::string &entry : platform::environmentEntries())
    if (!isReserved(entry)) env.push_back(std::move(entry));
  std::size_t start = 0;
  while (start < envPairs.size()) {
    std::size_t end = envPairs.find('\x1f', start);
    if (end == std::string::npos) end = envPairs.size();
    std::string pair = envPairs.substr(start, end - start);
    start = end + 1;
    const std::size_t equals = pair.find('=');
    if (pair.empty() || equals == std::string::npos) continue;
    const std::string key = pair.substr(0, equals + 1);
    if (isReserved(key)) continue;
    std::erase_if(env, [&key](const std::string &existing) { return existing.compare(0, key.size(), key) == 0; });
    env.push_back(std::move(pair));
  }
  const std::string idText = std::to_string(static_cast<long long>(id));
  env.push_back("GEA_CLUSTER_WORKER_ID=" + idText);
  env.push_back("NODE_UNIQUE_ID=" + idText);
  env.push_back("GEA_CLUSTER_CHANNEL_FD=" + channel);
  std::vector<std::string> args = s.arguments.empty() ? std::vector<std::string>{exe} : s.arguments;
  const double pid = platform::spawnChild(exe, std::move(args), std::move(env), s.spawnError);
  if (pid == 0) return 0;
  ++s.liveChildren;
  return pid;
}

// Worker side: EOF on the inherited liveness pipe means the primary is gone.
class ChannelWatcher final : public IoWatcher {
public:
  ChannelWatcher(platform::Fd fd, std::function<void()> onDisconnect) : fd_(fd), onDisconnect_(std::move(onDisconnect)) {}
  platform::Fd fd() const noexcept override { return fd_; }
  short interestedEvents() const noexcept override { return POLLIN; }
  // Unreferenced: a worker with nothing to serve exits, as under Node.
  bool isReferenced() const noexcept override { return false; }
  void onReady(short revents) override {
    char buffer[16];
    const platform::ssize n = platform::wakeRead(fd_, buffer, sizeof buffer);
    const bool closed = n == 0 || (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) ||
                        (revents & (POLLHUP | POLLERR)) != 0;
    if (!closed) return;
    clientEventLoop().close(this);
    if (onDisconnect_) onDisconnect_();
  }

private:
  platform::Fd fd_;
  std::function<void()> onDisconnect_;
};

inline void watchChannel(std::function<void()> onDisconnect) {
  const char *env = std::getenv("GEA_CLUSTER_CHANNEL_FD");
  if (env == nullptr) return;
  const platform::Fd fd = platform::openLivenessChannel(env);
  if (!platform::valid(fd)) return;
  clientEventLoop().adopt(std::make_unique<ChannelWatcher>(fd, std::move(onDisconnect)));
}

}  // namespace gea::node::cluster

namespace gea::node {

inline void cluster_set_notify_native(std::function<void()> notify) { cluster::state().notify = std::move(notify); }
inline void cluster_watch_channel_native(std::function<void()> onDisconnect) { cluster::watchChannel(std::move(onDisconnect)); }
inline void set_process_arguments(int argc, char **argv) { cluster::state().arguments = platform::arguments(argc, argv); }

}  // namespace gea::node

inline double __gea_node_cluster_worker_id() {
  const char *env = std::getenv("GEA_CLUSTER_WORKER_ID");
  return env == nullptr ? 0.0 : std::atof(env);
}
inline double __gea_node_cluster_fork(double id, std::string env) { return gea::node::cluster::spawn(id, env); }
inline std::string __gea_node_cluster_spawn_error() { return gea::node::cluster::state().spawnError; }
inline double __gea_node_cluster_next_exit() {
  gea::node::cluster::State &s = gea::node::cluster::state();
  if (s.exits.empty()) return 0;
  s.last = std::move(s.exits.front());
  s.exits.pop_front();
  return s.last.pid;
}
inline double __gea_node_cluster_exit_code() { return gea::node::cluster::state().last.code; }
inline std::string __gea_node_cluster_exit_signal() { return gea::node::cluster::state().last.signal; }
inline bool __gea_node_cluster_kill(double pid, std::string signal) {
  const int number = gea::node::cluster::signalNumber(signal);
  if (number < 0) return false;
  return gea::node::platform::signalProcess(pid, number);
}
inline double __gea_node_process_pid() { return gea::node::platform::processId(); }
inline double __gea_node_process_ppid() { return gea::node::platform::parentProcessId(); }
inline void __gea_node_process_exit(double code) {
  std::fflush(nullptr);
  gea::node::platform::exitImmediately(static_cast<int>(code));
}
inline double __gea_node_os_available_parallelism() {
  return static_cast<double>(gea::node::platform::onlineProcessors());
}

// ---- intrinsics exposed to geatsc-generated code --------------------------
// Registered by the node-compat plugin as `embeddedHostFunctions`, so a call in
// the compiled TypeScript lowers to a direct call of these C++ symbols.

inline double __gea_answer() { return 42.0; }

inline void __gea_serve_hello(double port) {
  gea::node::serve(static_cast<int>(port),
                   [](double connId, double, const std::string &, const std::string &path, const std::string &,
                      const std::string &, const std::string &) {
                     auto *conn = gea::node::findConnection(connId);
                     if (!conn) return;
                     conn->enqueueResponseBytes(
                         path == "/json" ? gea::node::makeResponse("application/json; charset=utf-8", "{\"hello\":\"world\"}")
                                         : gea::node::makeResponse("text/plain; charset=utf-8", "Hello, World!"));
                     conn->responseComplete(true);
                   });
}

// Per-request dispatch into a geatsc-compiled closure. `onRequest` is the
// TYPED lambda geatsc emits for the TS request-dispatch arrow — the template
// deduces its concrete closure type, so the per-request call is a direct
// (inlinable) invocation: no boxing of the callable or its arguments, no
// std::function type erasure. There is deliberately no boxed overload.
template <typename OnRequest>
inline void __gea_http_serve(double port, OnRequest &&onRequest) {
  gea::node::serve(static_cast<int>(port), std::forward<OnRequest>(onRequest));
}

// There is no `gea_cpp_value` fallback: the dispatch arrow is emitted as a
// typed lambda, which the template above takes directly. A program that somehow
// reached this boundary boxed fails to compile rather than silently boxing
// every request.

inline void __gea_http_write(double connId, std::string data) {
  if (auto *conn = gea::node::findConnection(connId)) conn->enqueueResponseBytes(data);
}

// The same enqueue for a response delivered as two pieces -- in practice the
// serialized header block and the first body chunk.
//
// `ServerResponse.emitPayload` used to hand those over as `head + body`, which
// built a third string holding a copy of both just so the boundary could take
// one argument. Both parts are appended into the connection's own buffer here
// instead, so that intermediate string is never built. Taking each part by
// reference matters as much as the arity: by value, `head` is an lvalue at the
// call site and would be copied to form the argument, reintroducing the
// allocation this removes.
inline void __gea_http_write2(double connId, const std::string &head, const std::string &body) {
  auto *conn = gea::node::findConnection(connId);
  if (!conn) return;
  conn->enqueueResponseParts(head, body);
}

// The header block, serialized into the connection buffer piece by piece --
// see `HttpConnectionBase::headBegin`. Every string crosses by reference: by
// value, a field read at the call site would be copied to form the argument.
inline void __gea_http_head_begin(double connId, double status, const std::string &message) {
  if (auto *conn = gea::node::findConnection(connId)) conn->headBegin(static_cast<int>(status), message);
}

inline void __gea_http_head_field(double connId, const std::string &name, const std::string &value) {
  if (auto *conn = gea::node::findConnection(connId)) conn->headField(name, value);
}

inline void __gea_http_head_raw(double connId, const std::string &text) {
  if (auto *conn = gea::node::findConnection(connId)) conn->headRaw(text);
}

inline double __gea_http_head_end(double connId, double flags, double autoContentLength) {
  auto *conn = gea::node::findConnection(connId);
  // A dead connection frames nothing; "not chunked, not kept alive" keeps the
  // caller from building chunk framing for bytes that have nowhere to go.
  return conn ? static_cast<double>(conn->headEnd(static_cast<int>(flags), autoContentLength)) : 0.0;
}

// Body bytes by reference. `__gea_http_write` takes its string BY VALUE, so an
// lvalue body is copied to form the argument before it is copied again into
// the connection buffer.
inline void __gea_http_body(double connId, const std::string &data) {
  if (auto *conn = gea::node::findConnection(connId)) conn->enqueueResponseBytes(data);
}

// A string is UTF-8 here, so its length on the wire is its size.
inline double __gea_http_text_bytes(const std::string &text) { return static_cast<double>(text.size()); }

// `end(text)` on a chunked response: size line, text, and terminator appended
// in place. The caller used to build `hex + CRLF + text + CRLF + "0" CRLF CRLF`
// as one string first -- an allocation and a full copy of the body, to feed a
// buffer that could have taken the four pieces directly.
inline void __gea_http_final_chunk(double connId, const std::string &data) {
  if (auto *conn = gea::node::findConnection(connId)) conn->enqueueFinalChunk(data);
}

// The same enqueue, reached with OCTETS instead of text.
//
// `__gea_http_write` above is the only way a response body ever left this
// runtime, and its parameter is a TypeScript `string`: an application handing
// `res.end(uint8Array)` a PNG, a protobuf frame or any other non-UTF-8 payload
// had no path that preserved its bytes, so `ServerResponse.end` could only
// declare `string` and an HTTP framework adapter's `outgoing.end(body)`
// (body: `Uint8Array`) was a
// checker error rather than a wrong answer. This is the missing half: the
// carrier is `gea::TypedArray<std::uint8_t>`, the same one `net_write` already
// takes for `node:net`, so a `Uint8Array`/`Buffer` arrives as its own storage
// with no conversion on either side.
//
// Header blocks, chunk-size lines and the terminating `0\r\n\r\n` stay on the
// string call -- they are ASCII by construction -- so a chunked byte body is
// three appends into the one connection buffer rather than one.
inline void __gea_http_write_bytes(double connId, const gea::TypedArray<std::uint8_t> &data) {
  auto *conn = gea::node::findConnection(connId);
  if (!conn) return;
  const auto *bytes = data.data();
  if (!bytes) return;
  // A view, not a string. This used to copy the whole body into a temporary
  // `std::string` purely to match the old owning parameter -- an allocation
  // and a full copy of every byte response, on top of the copy into the
  // connection buffer that follows. `enqueueResponseBytes` takes a
  // `string_view` now, so the bytes go straight from the TypedArray's own
  // storage into the buffer. This is a fetch-style framework's response path.
  conn->enqueueResponseBytes(std::string_view(reinterpret_cast<const char *>(bytes), data.size()));
}

// The inbound mirror of `__gea_http_write_bytes`: the reactor hands a request
// body over as a `std::string` of raw octets, and every string-to-Buffer
// encoding in `gea_node_buffer.hpp` decodes its input as UTF-8 first, so a
// body byte like 0xff came out as U+FFFD. This copies the octets as they are.
inline gea::Ref<gea::TypedArray<std::uint8_t>> __gea_http_body_bytes(const std::string &body) {
  return gea::node::buffer::detail::bufferResult(std::vector<std::uint8_t>(body.begin(), body.end()));
}

inline void __gea_http_done(double connId, bool keepAlive) {
  if (auto *conn = gea::node::findConnection(connId)) conn->responseComplete(keepAlive);
}

inline void __gea_http_destroy(double connId) {
  if (auto *conn = gea::node::findConnection(connId)) conn->hardDestroy();
}

inline std::string __gea_http_peer(double connId) {
  if (auto *conn = gea::node::findConnection(connId)) return conn->peerName();
  return std::string();
}

// Whether the reactor still holds this connection. `IncomingMessage.socket`
// keeps one `Socket` per connection, as Node does, and prunes its table with
// this: the reactor tells JavaScript nothing when a connection closes.
inline bool __gea_http_connection_open(double connId) { return gea::node::findConnection(connId) != nullptr; }

inline const std::string &__gea_http_date() { return gea::node::cachedHttpDate(); }

inline double __gea_http_date_second() { return static_cast<double>(gea::node::cachedHttpDateSecond()); }

inline void __gea_http_stop() {
  if (gea::node::listenerCloser()) gea::node::listenerCloser()();
}

// Real delayed timers (plugin maps global setTimeout/setInterval/clearTimeout/
// clearInterval to these — the runtime's standalone gea::host fallback would
// otherwise degrade a delay to an immediate microtask).
inline double __gea_node_set_timeout(std::function<void()> callback, double delayMs) {
  if (!callback) return 0;
  return gea::node::timers::add(std::move(callback), delayMs, false);
}

inline double __gea_node_set_interval(std::function<void()> callback, double delayMs) {
  if (!callback) return 0;
  return gea::node::timers::add(std::move(callback), delayMs, true);
}

inline void __gea_node_clear_timer(double id) { gea::node::timers::remove(id); }

inline void __gea_node_timer_unref(double id) { gea::node::timers::unref(id); }

inline double __gea_node_immediate_start(std::function<void()> callback) {
  if (!callback) return 0;
  return gea::node::immediates::add(std::move(callback));
}

inline void __gea_node_immediate_clear(double id) { gea::node::immediates::remove(id); }

inline void __gea_node_immediate_ref(double id) { gea::node::immediates::setReferenced(id, true); }

inline void __gea_node_immediate_unref(double id) { gea::node::immediates::setReferenced(id, false); }

inline void __gea_node_run_pending() { gea::node::clientEventLoop().run(); }

inline bool __gea_node_pump_until(const std::function<bool()> &done) {
  return gea::node::clientEventLoop().runUntil(done);
}

inline std::string __gea_node_version() { return "v20.0.0-gea"; }

inline std::string __gea_node_platform() {
#if defined(__APPLE__)
  return "darwin";
#elif defined(__linux__)
  return "linux";
#elif defined(_WIN32)
  return "win32";
#else
  return "unknown";
#endif
}

inline std::string __gea_node_os_arch() { return gea::node::platform::machineArch(); }

inline std::string __gea_node_os_release() { return gea::node::platform::kernelRelease(); }

inline std::string __gea_node_os_type() {
#if defined(__APPLE__)
  return "Darwin";
#elif defined(__linux__)
  return "Linux";
#elif defined(_WIN32)
  return "Windows_NT";
#else
  return gea::node::platform::systemName();
#endif
}

inline void __gea_node_stdio_write(double fd, std::string data) {
  FILE *stream = static_cast<int>(fd) == 2 ? stderr : stdout;
  if (!data.empty()) (void)::fwrite(data.data(), 1, data.size(), stream);
  (void)::fflush(stream);
}

inline bool __gea_node_fs_access(std::string path, double mode) {
  return gea::node::platform::canAccess(path, static_cast<int>(mode));
}

static gea::node::platform::FileStatus gea_node_fs_last_stat;

inline double __gea_node_fs_stat(std::string path, bool followSymlinks) {
  return static_cast<double>(gea::node::platform::statFile(path, followSymlinks, gea_node_fs_last_stat));
}

inline double __gea_node_fs_stat_field(double field) {
  const gea::node::platform::FileStatus &st = gea_node_fs_last_stat;
  switch (static_cast<int>(field)) {
    case 0: return st.dev;
    case 1: return st.ino;
    case 2: return st.mode;
    case 3: return st.nlink;
    case 4: return st.uid;
    case 5: return st.gid;
    case 6: return st.rdev;
    case 7: return st.size;
    case 8: return st.blksize;
    case 9: return st.blocks;
    case 10: return st.atimeMs;
    case 11: return st.mtimeMs;
    case 12: return st.ctimeMs;
    case 13: return st.birthtimeMs;
    default: return 0.0;
  }
}

// The same refusal under both runtimes, raised the way each one raises: v1
// writes its global thrown slot and sets the flag, v2 throws a real C++
// exception carrying the identical `gea::Value` payload -- which is also what
// the request path above catches to isolate a handler throw to its request.
inline void __gea_node_not_implemented(std::string moduleName, std::string memberName) {
  const std::string target = "node@24";
  const std::string message = std::string("ERR_GEA_NODE_NOT_IMPLEMENTED: ") + moduleName + "." + memberName +
      " is not implemented for geastack target " + target;
  throw gea::Value::box(gea::Value::Tag::String, std::string("NodeNotImplementedError: ") + message);
}

#endif  // GEA_NODE_RUNTIME_INCLUDED