import os from "node:os";
import { readFileSync } from "node:fs";
import { execFileSync } from "node:child_process";
import type { QueuedSpawn } from "agent-relay-sdk";

// #893 — memory/load-aware spawn admission. The #930/#880 spawnSlots cap bounds how many
// spawns can be MID-REGISTRATION at once (a port-collision fix); it does nothing about the
// STEADY-STATE count of already-registered running agents. The 2026-06-30 incident hung the
// relay by accumulating 24 running agents until the host swapped to 96% and load average hit
// 43 — no gate anywhere rejected or queued spawns past what the host could actually hold. This
// is the per-host resource check that closes that gap: a cheap, local memory/`loadavg()`
// read the orchestrator (which already has host access) can do before every spawn, no relay/
// protocol change required.
//
// #1516 — the gate must reflect *usable* memory, not raw `os.freemem()`. On darwin macOS keeps
// raw "free" low by holding pages as reclaimable file cache, so raw free under-reports true
// headroom. During the 2026-07-18 macmini incident raw free dipped to 146–217MB and the gate
// rejected a spawn at 185MB while GBs were idle-and-reclaimable and memory_pressure reported
// "normal". On linux we derive MemAvailable from /proc/meminfo and gate on that.
//
// #1516-r3 — on darwin, STOP deriving a usable-bytes number by summing `vm_stat` page classes.
// Two prior rounds (r1: "Pages inactive"; r2: "File-backed pages") both over-ADMITTED (OOM risk):
// every "reclaimable" page class secretly contains dirty content (dirty anonymous pages must swap;
// XNU counts the FULL external pageable population — including dirty/writeback external pages — in
// the file-backed figure). No page-class sum is a sound "immediately usable" number.
//
// Instead we gate darwin on the KERNEL'S OWN memory-pressure signal, which already accounts for
// dirty/reclaimable/compressed memory internally:
//   * `sysctl -n kern.memorystatus_vm_pressure_level` → 1 = NORMAL, 2 = WARN, 4 = CRITICAL (the
//     dispatch_source_memorypressure flag values; XNU raises WARN/CRITICAL precisely when it can no
//     longer reclaim enough — including the dirty-writeback case both prior rounds mis-admitted).
//   * NORMAL  → the kernel reports headroom → report usable at physical RAM so the memory gate
//     passes (this is what fixes macmini's false-low flicker: normal pressure despite low raw-free).
//   * WARN / CRITICAL / any non-normal level → HOLD: no reclaimable boost, fall back to raw
//     `os.freemem()` so a genuinely tight host still blocks. Never harder-closed than raw-free.
//   * usable is sanity-capped at os.totalmem() and floored at a SINGLE os.freemem() sample threaded
//     through the whole computation, so `freeMemMb <= usableMemMb <= totalMemMb` holds deterministically.
//   * The sysctl probe uses a short timeout and a short-TTL cache (keyed on a MONOTONIC clock, so a
//     wall-clock rollback can't keep a stale high probe alive) so a synchronous execFileSync can't
//     serialize into multi-second event-loop stalls across admission polls.
// The probe stays best-effort: any failure (sysctl missing, timeout, unparseable, unknown platform)
// falls back to os.freemem() (fail-open) — the gate is never harder-closed than raw-free today.

const DEFAULT_MIN_FREE_MEM_MB = 512;
const DEFAULT_MAX_LOAD_PER_CPU = 4;
const DEFAULT_ADMISSION_MAX_WAIT_MS = 60_000;
const DEFAULT_ADMISSION_POLL_MS = 2_000;

// #1516-r2 — execFileSync(sysctl) synchronously blocks the event loop. Keep the per-probe timeout
// short and cache the result briefly so admission polling (every ~2s) doesn't spawn the probe on
// every iteration and can't serialize failing probes into multi-second stalls.
const PROBE_TIMEOUT_MS = 400;
const PROBE_CACHE_TTL_MS = 3_000;

// #1516-r3 — darwin `kern.memorystatus_vm_pressure_level` values. These are the XNU
// dispatch_source_memorypressure flag values (NOTE_MEMORYSTATUS_PRESSURE_{NORMAL,WARN,CRITICAL}).
export const VM_PRESSURE_NORMAL = 1;
export const VM_PRESSURE_WARN = 2;
export const VM_PRESSURE_CRITICAL = 4;

// #1516-r3 — the byte value the darwin NORMAL-pressure branch reports: "the kernel says there is
// headroom, impose no memory-class ceiling." readUsableMemMb clamps this down to the SINGLE
// os.totalmem() sample threaded through checkHostHeadroom, so it resolves to exactly physical RAM
// without the probe taking a second totalmem sample (preserves the free<=usable<=total invariant).
const PRESSURE_HEADROOM_SENTINEL_BYTES = Number.MAX_SAFE_INTEGER;

// #1513 — stable marker prefixed onto the rejection reason when a spawn is held back for the full
// admission wait ceiling and then fails. Lets the spawner (and the failed-command notification)
// distinguish a genuinely-can't-be-admitted queue timeout from other spawn failures.
export const HOST_HEADROOM_TIMEOUT_REASON = "host-headroom-timeout";

export interface HostHeadroomResult {
  ok: boolean;
  /** Present when ok is false — a human-readable reason suitable for a failed-command message. */
  reason?: string;
  /**
   * #1516 — usable memory the gate actually decides on. On darwin (#1516-r3) this is driven by the
   * kernel's own VM pressure level: NORMAL reports physical RAM (headroom), WARN/CRITICAL falls back
   * to raw free (hold); on linux it is MemAvailable. Floored at raw `os.freemem()` and capped at
   * `os.totalmem()`, so `freeMemMb <= usableMemMb <= totalMemMb` always holds.
   */
  usableMemMb: number;
  /** Raw `os.freemem()`, retained for observability alongside the usable figure the gate uses. */
  freeMemMb: number;
  totalMemMb: number;
  loadAvgPerCpu: number;
}

export function resolveMinFreeMemMb(env: NodeJS.ProcessEnv = process.env): number {
  const raw = env.AGENT_RELAY_SPAWN_MIN_FREE_MEM_MB;
  if (raw === undefined || raw.trim() === "") return DEFAULT_MIN_FREE_MEM_MB;
  const parsed = Number(raw);
  if (!Number.isFinite(parsed) || parsed < 0) return DEFAULT_MIN_FREE_MEM_MB;
  return parsed;
}

export function resolveMaxLoadPerCpu(env: NodeJS.ProcessEnv = process.env): number {
  const raw = env.AGENT_RELAY_SPAWN_MAX_LOAD_PER_CPU;
  if (raw === undefined || raw.trim() === "") return DEFAULT_MAX_LOAD_PER_CPU;
  const parsed = Number(raw);
  if (!Number.isFinite(parsed) || parsed <= 0) return DEFAULT_MAX_LOAD_PER_CPU;
  return parsed;
}

export function resolveAdmissionMaxWaitMs(env: NodeJS.ProcessEnv = process.env): number {
  const raw = env.AGENT_RELAY_SPAWN_ADMISSION_MAX_WAIT_MS;
  if (raw === undefined || raw.trim() === "") return DEFAULT_ADMISSION_MAX_WAIT_MS;
  const parsed = Number(raw);
  if (!Number.isFinite(parsed) || parsed < 0) return DEFAULT_ADMISSION_MAX_WAIT_MS;
  return parsed;
}

export function resolveAdmissionPollMs(env: NodeJS.ProcessEnv = process.env): number {
  const raw = env.AGENT_RELAY_SPAWN_ADMISSION_POLL_MS;
  if (raw === undefined || raw.trim() === "") return DEFAULT_ADMISSION_POLL_MS;
  const parsed = Number(raw);
  if (!Number.isFinite(parsed) || parsed <= 0) return DEFAULT_ADMISSION_POLL_MS;
  return parsed;
}

export function isHostAdmissionDisabled(env: NodeJS.ProcessEnv = process.env): boolean {
  const raw = env.AGENT_RELAY_SPAWN_ADMISSION_DISABLED;
  return raw === "1" || raw?.toLowerCase() === "true";
}

export interface WaitForHostHeadroomOptions {
  disabled: boolean;
  maxWaitMs: number;
  pollMs: number;
  check: () => HostHeadroomResult;
  /** Called once per failed poll with a human-readable "still queued" message. */
  log?: (message: string) => void;
  sleep?: (ms: number) => Promise<void>;
  // #1513 — called on each poll while the spawn is held back for headroom, with the current headroom
  // snapshot (its reason/freeMem update as the host recovers). Lets the caller surface a "queued"
  // dashboard row and notify the spawner that the spawn is queued rather than lost.
  onQueued?: (result: HostHeadroomResult) => void;
  // #1513 — called exactly once when a wait that ever queued ends (admitted OR timed out), so the
  // caller can clear the queued row it published via onQueued. Not called for an immediate admit.
  onSettled?: () => void;
}

/**
 * Poll host headroom until it's admissible or `maxWaitMs` elapses. A spawn racing an agent
 * that's about to exit gets admitted once headroom frees; only past the deadline does this
 * throw — a clear, actionable rejection rather than silently oversubscribing the host.
 *
 * #1513 — while it holds a spawn back it drives onQueued/onSettled so the queued state is
 * observable (dashboard row + spawner notification) instead of a silent wait, and the timeout
 * rejection is tagged with HOST_HEADROOM_TIMEOUT_REASON so a can't-ever-admit queue is unambiguous.
 */
export async function waitForHostHeadroom(opts: WaitForHostHeadroomOptions): Promise<void> {
  if (opts.disabled) return;
  const sleep = opts.sleep ?? ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
  const deadline = Date.now() + opts.maxWaitMs;
  let queued = false;
  try {
    for (;;) {
      const result = opts.check();
      if (result.ok) return;
      if (Date.now() >= deadline) {
        throw new Error(`spawn rejected: ${HOST_HEADROOM_TIMEOUT_REASON} — insufficient host headroom after waiting ${opts.maxWaitMs}ms (${result.reason})`);
      }
      queued = true;
      opts.onQueued?.(result);
      opts.log?.(`spawn queued: waiting for host headroom (${result.reason})`);
      await sleep(Math.min(opts.pollMs, Math.max(0, deadline - Date.now())));
    }
  } finally {
    // Only fire onSettled when we actually published a queued row (queued===true); an immediate
    // admit never surfaced one. Runs on both the admit (return) and timeout (throw) paths.
    if (queued) opts.onSettled?.();
  }
}

/**
 * #1516-r3/r4/r5 — parse `sysctl -n kern.memorystatus_vm_pressure_level` into the kernel's VM
 * pressure level (1 = NORMAL, 2 = WARN, 4 = CRITICAL). `-n` prints just the integer, e.g. "1\n";
 * this also tolerates the `name: value` form (`sysctl` without `-n`), but ONLY when `name` is the
 * exact literal key `kern.memorystatus_vm_pressure_level` — e.g. "kern.memorystatus_vm_pressure_level: 1".
 * Returns null on anything else — including empty/whitespace-only output, a leading sign or decimal,
 * extra tokens/words, embedded diagnostics, multiple numbers, a `name: value` form with any other
 * name, or an integer that isn't a known pressure level — so the caller falls back to
 * os.freemem() (HOLD, never a false NORMAL).
 *
 * #1516-r4 — this MUST be strict. The r3 parser took the LAST `/\d+/` token in the output, which
 * stripped signs and picked digits out of garbage: `"-1\n"` -> 1, `"warning: -1\n"` -> 1,
 * `"2 warning 1\n"` -> 1. Any of those misread as NORMAL inflates usable memory to physical RAM
 * under real memory pressure (over-open/OOM) — ambiguous/unexpected probe output must always HOLD.
 *
 * #1516-r5 — the r4 `name: value` branch still accepted ANY identifier before the colon
 * (`/^[A-Za-z0-9_.]+:\s*(\d+)$/`), so diagnostic/garbage output like `"warning: 1"` or `"foo: 1"`
 * parsed as pressure level 1 (NORMAL) even though no real sysctl emits those keys. The name must
 * match the literal sysctl key exactly — anything else is unrecognized output and must HOLD.
 *
 * We gate on this signal — not a sum of `vm_stat` page classes — because every "reclaimable" page
 * class (inactive, file-backed) secretly holds dirty content that isn't immediately usable, so any
 * such sum over-admits (the r1 and r2 OOM bugs). The kernel's pressure level already accounts for
 * dirty/reclaimable/compressed memory internally and raises WARN/CRITICAL when it genuinely can't
 * reclaim — including the dirty-writeback-to-slow-storage case both prior rounds mis-admitted.
 */
export function parseVmPressureLevel(output: string): number | null {
  if (!output) return null;
  const trimmed = output.trim();
  if (trimmed === "") return null;

  let valueStr: string;
  if (/^\d+$/.test(trimmed)) {
    valueStr = trimmed;
  } else {
    const match = trimmed.match(/^kern\.memorystatus_vm_pressure_level:\s*(\d+)$/);
    if (!match) return null;
    valueStr = match[1]!;
  }

  const level = Number(valueStr);
  if (!Number.isSafeInteger(level)) return null;
  if (level !== VM_PRESSURE_NORMAL && level !== VM_PRESSURE_WARN && level !== VM_PRESSURE_CRITICAL) {
    return null;
  }
  return level;
}

/**
 * #1516 — parse `/proc/meminfo` on linux for `MemAvailable` (kernel's own estimate of memory
 * available for new work without swapping, which already discounts page-cache that can't be
 * reclaimed). Returns bytes, or null if the field is absent (caller falls back to os.freemem()).
 */
export function parseMemAvailableBytes(meminfo: string): number | null {
  if (!meminfo) return null;
  const m = meminfo.match(/^MemAvailable:\s*(\d+)\s*kB/im);
  if (!m) return null;
  const kb = Number(m[1]);
  if (!Number.isSafeInteger(kb) || kb < 0) return null;
  const bytes = kb * 1024;
  if (!Number.isSafeInteger(bytes) || bytes < 0) return null;
  return bytes;
}

/**
 * #1516-r5 — the darwin raw-sysctl-output → usable-bytes mapping, factored out of
 * {@link computeUsableMemBytes} so tests can drive this EXACT production mapping (parse, then
 * NORMAL→sentinel/else→hold) over injected sysctl text, instead of hand-reimplementing it.
 */
export function darwinUsableMemBytesFromSysctlOutput(out: string): number | null {
  const level = parseVmPressureLevel(out);
  if (level === null) return null; // unparseable → fail OPEN (raw free)
  if (level === VM_PRESSURE_NORMAL) {
    // Kernel reports headroom → no memory-class ceiling; readUsableMemMb clamps this to the
    // threaded os.totalmem() sample and floors at raw free, so the memory gate passes. This is
    // what fixes macmini's false-low flicker (normal pressure despite a low raw-free reading).
    return PRESSURE_HEADROOM_SENTINEL_BYTES;
  }
  // WARN / CRITICAL / any unexpected non-normal level → HOLD: no reclaimable boost. Falling back
  // to raw free keeps a genuinely tight host blocked while never being harder-closed than
  // raw-free, and is safe against the dirty-writeback OOM path (the kernel is why we're here).
  return null;
}

/**
 * The raw, un-cached probe: run the platform-specific headroom read (darwin: kernel VM pressure
 * level → headroom-sentinel/hold; linux: /proc/meminfo MemAvailable). Never throws — any failure
 * (binary missing, timeout, permissions, unexpected format, unknown platform) → null (fail-open).
 */
function computeUsableMemBytes(platform: NodeJS.Platform): number | null {
  try {
    if (platform === "darwin") {
      // #1516-r3 — gate on the kernel's own memory-pressure signal, not a page-class sum.
      const out = execFileSync("sysctl", ["-n", "kern.memorystatus_vm_pressure_level"], {
        encoding: "utf8",
        timeout: PROBE_TIMEOUT_MS,
        // Capture stdout; discard stderr so a failing/absent sysctl fails silently (fail-open) rather
        // than spamming the orchestrator's stderr on every re-probe.
        stdio: ["ignore", "pipe", "ignore"],
      });
      return darwinUsableMemBytesFromSysctlOutput(out);
    }
    if (platform === "linux") {
      const meminfo = readFileSync("/proc/meminfo", "utf8");
      return parseMemAvailableBytes(meminfo);
    }
  } catch {
    // Probe failed — fail OPEN (caller floors at os.freemem()).
    return null;
  }
  return null;
}

interface ProbeCacheEntry {
  atMs: number;
  value: number | null;
}
const probeCache = new Map<string, ProbeCacheEntry>();

export interface ProbeUsableMemOptions {
  /**
   * Injectable clock for tests. Defaults to a MONOTONIC source (`performance.now`) — #1516-r2/r3:
   * the cache TTL must not be measured on the wall clock, or a backwards clock step (NTP/DST) makes
   * the age negative and keeps a stale high probe alive past the TTL.
   */
  now?: () => number;
  /** Injectable raw probe for tests (defaults to the real platform read). */
  compute?: (platform: NodeJS.Platform) => number | null;
}

// #1516-r3 — monotonic default clock for the probe cache TTL. `performance.now()` never runs
// backwards, so a wall-clock rollback can't extend the cache the way Date.now() could.
const monotonicNowMs = (): number => performance.now();

/** Clear the probe cache — for tests, so cached results don't leak across cases. */
export function resetUsableMemProbeCache(): void {
  probeCache.clear();
}

/**
 * #1516 — best-effort probe of *usable* host memory in bytes. Prefers genuinely-reclaimable
 * platform signals (darwin vm_stat, linux MemAvailable); on any probe failure, or an unrecognized
 * platform, returns null so the caller falls back to raw `os.freemem()`. Never throws.
 *
 * #1516-r2 — the result is cached per-platform for {@link PROBE_CACHE_TTL_MS} so a synchronous
 * execFileSync(sysctl) isn't spawned on every admission poll and can't serialize into long
 * event-loop stalls. #1516-r3 — the TTL is measured on a MONOTONIC clock, and a negative age (a
 * clock that ran backwards) is treated as expired, so a stale high probe can't survive a rollback.
 */
export function probeUsableMemBytes(
  platform: NodeJS.Platform = process.platform,
  opts: ProbeUsableMemOptions = {},
): number | null {
  const now = opts.now ?? monotonicNowMs;
  const compute = opts.compute ?? computeUsableMemBytes;
  const t = now();
  const cached = probeCache.get(platform);
  if (cached) {
    const age = t - cached.atMs;
    if (age >= 0 && age < PROBE_CACHE_TTL_MS) {
      return cached.value;
    }
  }
  const value = compute(platform);
  probeCache.set(platform, { atMs: t, value });
  return value;
}

/**
 * #1516 — usable memory in MB the admission gate decides on. Takes the platform probe when
 * available (darwin: pressure-level → physical RAM on NORMAL / null on WARN+CRITICAL; linux:
 * MemAvailable), floors it at raw `os.freemem()` so the gate is NEVER stricter than the historical
 * raw-free behavior even if the probe under-reports or holds, and caps it at physical RAM so a
 * headroom signal can't inflate the figure past `os.totalmem()`.
 *
 * #1516-r2 — `rawFreeBytes` and `totalBytes` are threaded in from a SINGLE sample by the caller,
 * so `usableMemMb >= freeMemMb` holds deterministically even while free memory churns.
 */
export function readUsableMemMb(
  platform: NodeJS.Platform = process.platform,
  rawFreeBytes: number = os.freemem(),
  totalBytes: number = os.totalmem(),
): number {
  const probed = probeUsableMemBytes(platform);
  let usableBytes = probed === null ? rawFreeBytes : Math.max(probed, rawFreeBytes);
  // Sanity cap: usable can never exceed physical RAM (guards a plausible-but-wrong probe).
  if (Number.isSafeInteger(totalBytes) && totalBytes > 0) {
    usableBytes = Math.min(usableBytes, totalBytes);
  }
  return usableBytes / (1024 * 1024);
}

// #1513 — build the dashboard/notification descriptor for a spawn the headroom gate is holding back.
// `spawnedBy` (the routing target) and `spawnRequestId` (idempotency) come off the spawn command
// params; the reason/freeMem come from the live headroom poll so the row updates as the host recovers.
export function queuedSpawnFromCommand(
  params: Record<string, any>,
  result: HostHeadroomResult,
  queuedAt: number,
  minFreeMemMb: number,
  maxWaitMs: number,
): QueuedSpawn {
  return {
    provider: typeof params.provider === "string" ? params.provider : "unknown",
    ...(typeof params.spawnRequestId === "string" ? { spawnRequestId: params.spawnRequestId } : {}),
    ...(typeof params.label === "string" ? { label: params.label } : {}),
    ...(typeof params.cwd === "string" ? { cwd: params.cwd } : {}),
    ...(typeof params.spawnedBy === "string" ? { spawnedBy: params.spawnedBy } : {}),
    reason: result.reason ?? "waiting for host memory headroom",
    freeMemMb: Math.round(result.freeMemMb),
    minFreeMemMb,
    queuedAt,
    maxWaitMs,
  };
}

/** Read current host memory/load headroom and decide whether it can admit another spawn. */
export function checkHostHeadroom(opts: { minFreeMemMb?: number; maxLoadPerCpu?: number } = {}): HostHeadroomResult {
  // #1516-r2 — sample raw free / total ONCE and thread them through so the usable figure is
  // computed against the same free reading it is floored at (usableMemMb >= freeMemMb always).
  const rawFreeBytes = os.freemem();
  const totalBytes = os.totalmem();
  const freeMemMb = rawFreeBytes / (1024 * 1024);
  const usableMemMb = readUsableMemMb(process.platform, rawFreeBytes, totalBytes);
  const totalMemMb = totalBytes / (1024 * 1024);
  const cpuCount = Math.max(1, os.cpus().length);
  const loadAvgPerCpu = os.loadavg()[0]! / cpuCount;
  const minFreeMemMb = opts.minFreeMemMb ?? resolveMinFreeMemMb();
  const maxLoadPerCpu = opts.maxLoadPerCpu ?? resolveMaxLoadPerCpu();
  if (usableMemMb < minFreeMemMb) {
    return {
      ok: false,
      reason: `usable memory ${usableMemMb.toFixed(0)}MB is below the ${minFreeMemMb}MB minimum required to admit another agent`,
      usableMemMb,
      freeMemMb,
      totalMemMb,
      loadAvgPerCpu,
    };
  }
  if (loadAvgPerCpu > maxLoadPerCpu) {
    return {
      ok: false,
      reason: `load average ${loadAvgPerCpu.toFixed(2)} per CPU exceeds the ${maxLoadPerCpu} max allowed to admit another agent`,
      usableMemMb,
      freeMemMb,
      totalMemMb,
      loadAvgPerCpu,
    };
  }
  return { ok: true, usableMemMb, freeMemMb, totalMemMb, loadAvgPerCpu };
}