// RunnerPool Durable Object (docs/COVALENCE_RUNNER.md §7–8): one per runner pool (one pool per
// owning device; `poolId = sha256(provisioning device static key)`). Owns the runner-instance
// registry, the `maxRunners` cap the user sets in Settings ▸ Covalence ▸ Covalence Cloud, epoch
// fencing, and idle reaping / scale-to-zero.
//
// The scaling unit is a whole **runner** — a full `nucleicd` host with its own mesh identity /
// relay enrollment / pairing, appearing as its own `.cloud` peer in the roster. Instances are
// `runner-<n>`. The pool scales the NUMBER of runners up (on demand — the owning device calls
// `provision` when a mesh dispatch is parked and the live count is below the cap) and down (idle
// reap). It does NOT place sessions: the owning device's mesh dispatcher spreads chats across the
// runner peers directly. Mirrors the macOS-VM ceiling pattern (`nucleic.macvm.maxConcurrent`): a
// fan-out of sessions past the cap runs in-place on the devices/host rather than failing.
//
// Shape notes:
//   • The DO is a plain class over structural interfaces (storage/alarm), so the whole thing is
//     unit-testable under node:test with fakes — the same discipline as the relay's Room DO.
//   • Containers are driven through the RUNNER_CONTAINERS namespace via `ContainerDriver`, an
//     injected seam so tests never need the Containers runtime — and so a self-hosted supervisor
//     (docker/podman) can implement the same two calls (docs/COVALENCE_RUNNER.md §8).

/** How a runner replaces Apple Foundation Models (docs/COVALENCE_RUNNER.md §5, item 6).
 * Mirrors Swift's `RunnerIntelligenceMode` (keep in lockstep). Reaches the container as
 * `NUCLEIC_RUNNER_INTELLIGENCE_MODE` on its next boot — a settings change never restarts a
 * running container; the scale-to-zero lifecycle re-boots it soon enough. */
export type IntelligenceMode = "mesh" | "agent" | "heuristic";

export interface PoolSettings {
  /** The most concurrent runner peers the pool may run. Each runner is a full `nucleicd` host
   * with its own mesh identity; the fleet scales up on demand and reaps to zero. 1 ⇒ a single
   * runner. Mirrors `nucleic.macvm.maxConcurrent`. */
  maxRunners: number;
  /** Runners idle (zero sessions) longer than this are reaped by the alarm. */
  idleTimeoutSeconds: number;
  intelligenceMode: IntelligenceMode;
}

export const DEFAULT_SETTINGS: PoolSettings = {
  maxRunners: 4,
  idleTimeoutSeconds: 600,
  intelligenceMode: "mesh",
};

export const MAX_RUNNERS_RANGE = { min: 1, max: 16 } as const;
export const IDLE_TIMEOUT_RANGE = { min: 60, max: 7200 } as const;

/** Server-side clamp: the Settings UI clamps too, but the DO never trusts a client. Unknown
 * fields are dropped; missing fields keep their current value. */
export function clampSettings(current: PoolSettings, patch: Partial<PoolSettings> | null | undefined): PoolSettings {
  const next = { ...current };
  if (patch && typeof patch.maxRunners === "number" && Number.isFinite(patch.maxRunners)) {
    next.maxRunners = Math.min(MAX_RUNNERS_RANGE.max, Math.max(MAX_RUNNERS_RANGE.min, Math.floor(patch.maxRunners)));
  }
  if (patch && typeof patch.idleTimeoutSeconds === "number" && Number.isFinite(patch.idleTimeoutSeconds)) {
    next.idleTimeoutSeconds = Math.min(IDLE_TIMEOUT_RANGE.max, Math.max(IDLE_TIMEOUT_RANGE.min, Math.floor(patch.idleTimeoutSeconds)));
  }
  if (
    patch &&
    (patch.intelligenceMode === "mesh" ||
      patch.intelligenceMode === "agent" ||
      patch.intelligenceMode === "heuristic")
  ) {
    next.intelligenceMode = patch.intelligenceMode;
  }
  return next;
}

export type InstanceState = "starting" | "ready" | "stopped";

export interface InstanceRecord {
  instanceId: string; // "runner-<n>"
  state: InstanceState;
  /** Live agent sessions the runner reported on its last heartbeat (for the status list and idle
   * reaping — NOT placement; the owning device's mesh dispatcher decides where a chat lands). */
  sessions: number;
  /** Sessions with active or pending work in that same heartbeat — the complement of "Done and
   * quiet". A session that finished its turn (Done) with no queued follow-up counts 0 here even
   * though it still counts in `sessions`; a Done session still holding a queued message counts as
   * active (its message is waiting to dispatch). Drives the all-done sleep (`DONE_SLEEP_MS`).
   * Absent on records from a runner/Worker that predates the field ⇒ read as `sessions` (all
   * active), so an older runner never sleeps down this path. */
  activeSessions?: number;
  /** Pool-wide monotonic fencing epoch stamped at boot; a heartbeat carrying an older epoch is
   * a zombie (the pool has since re-booted that slot) and is rejected. */
  epoch: number;
  /** Last accepted heartbeat (ms since epoch). While a runner is still `starting` (no first
   * heartbeat yet) this stays its boot time, so the boot-grace window measures against it. */
  lastSeenAt: number;
  /** Last moment the runner had ≥1 session (ms) — drives idle reaping. */
  lastActiveAt: number;
  /** Last moment the runner had ≥1 *active* session (ms) — drives the all-done sleep. Distinct
   * from `lastActiveAt` (≥1 session of any kind): once every session goes Done this freezes
   * while `lastActiveAt` keeps counting the still-present finished chats, so `now - lastBusyAt`
   * measures how long the runner has held nothing but Done work. Absent on older records ⇒ falls
   * back to `lastActiveAt`. */
  lastBusyAt?: number;
  /** The runner's relay roomId (= its mesh HostID), from its heartbeats. This is how a device
   * holding the pool credential tells a mesh-joined runner from an orphan (e.g. a wake-hook
   * reboot whose fresh identity nobody has paired yet): a `ready` instance whose roomId isn't a
   * connected mesh peer needs its pairing pulled + joined. Absent until the first heartbeat
   * that carries one (and on records from older Workers). */
  roomId?: string;
}

/** A `ready` runner silent this long is gone or wedged — reap it (dead capacity otherwise holds a
 * `maxRunners` slot forever). */
export const HEARTBEAT_STALE_MS = 90_000;

/** A runner still `starting` gets this much longer before it's declared a failed boot — a cold
 * image pull can take a couple of minutes (docs/COVALENCE_RUNNER.md §11.1), and reaping it at the
 * ready-staleness window would kill a slow first boot. */
export const BOOT_GRACE_MS = 300_000;

/** All of a runner's sessions are Done (finished, or an interactive turn classified `completed`)
 * with none still holding a queued follow-up — sleep the runner once they've been settled this
 * long. The delay keeps a container from being torn down the instant an agent returns, since the
 * user often sends a follow-up within a few minutes. Deliberately a fixed floor, not the user's
 * `idleTimeoutSeconds` knob: that governs a *sessionless* runner (scale-to-zero); this governs one
 * still holding finished chats (docs/COVALENCE_RUNNER.md §7). The pool learns "all Done" from the
 * runner's per-heartbeat `activeSessions` count. */
export const DONE_SLEEP_MS = 300_000;

// --- Durable Object ----------------------------------------------------------------------------

/** The two lifecycle calls the pool makes against a container slot. In production this wraps the
 * RUNNER_CONTAINERS namespace (container.ts); in tests, a fake; on a self-hosted runner, a local
 * docker/podman supervisor. */
export interface ContainerDriver {
  start(poolId: string, instanceId: string, env: Record<string, string>): Promise<void>;
  stop(poolId: string, instanceId: string): Promise<void>;
}

interface StorageLike {
  get<T>(key: string): Promise<T | undefined>;
  put(key: string, value: unknown): Promise<void>;
  delete(key: string): Promise<boolean>;
  setAlarm(time: number): Promise<void>;
}

export interface PoolStateLike {
  storage: StorageLike;
}

function json(obj: unknown, status = 200): Response {
  return new Response(JSON.stringify(obj), { status, headers: { "content-type": "application/json" } });
}

const K_SETTINGS = "settings";
const K_INSTANCES = "instances";
const K_EPOCH = "epoch";

export interface PoolDeps {
  driver: ContainerDriver;
  mintInstanceToken: (poolId: string, instanceId: string) => Promise<string>;
  now?: () => number;
}

/** The RunnerPool DO logic, constructed over injectable state + deps so node:test drives it
 * directly. `RunnerPool` (in index.ts) is the thin Workers-runtime binding. */
export class RunnerPoolCore {
  private readonly state: PoolStateLike;
  private readonly poolId: string;
  private readonly deps: PoolDeps;

  // Node's strip-types runner rejects constructor parameter properties (non-erasable syntax),
  // so fields are declared explicitly — same reason the edge Worker's DO classes do this.
  constructor(state: PoolStateLike, poolId: string, deps: PoolDeps) {
    this.state = state;
    this.poolId = poolId;
    this.deps = deps;
  }

  private now(): number {
    return this.deps.now ? this.deps.now() : Date.now();
  }

  private async settings(): Promise<PoolSettings> {
    // Merge over defaults so a record persisted before a field existed reads the default rather
    // than `undefined`, and drop any legacy keys (an older pool stored `maxContainers` /
    // `balancerPolicy`) by projecting onto the known shape.
    const stored = await this.state.storage.get<Partial<PoolSettings>>(K_SETTINGS);
    const merged = { ...DEFAULT_SETTINGS, ...(stored ?? {}) };
    return {
      maxRunners: merged.maxRunners,
      idleTimeoutSeconds: merged.idleTimeoutSeconds,
      intelligenceMode: merged.intelligenceMode,
    };
  }

  private async instances(): Promise<InstanceRecord[]> {
    return (await this.state.storage.get<InstanceRecord[]>(K_INSTANCES)) ?? [];
  }

  private async saveInstances(list: InstanceRecord[]): Promise<void> {
    await this.state.storage.put(K_INSTANCES, list);
  }

  private async bumpEpoch(): Promise<number> {
    const epoch = ((await this.state.storage.get<number>(K_EPOCH)) ?? 0) + 1;
    await this.state.storage.put(K_EPOCH, epoch);
    return epoch;
  }

  /** Routes an internal request (the Worker has already authenticated the caller and stamped
   * `x-nucleic-auth: pool | instance:<instanceId>`). */
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const auth = request.headers.get("x-nucleic-auth") ?? "";
    const isPool = auth === "pool";
    const instanceAuth = auth.startsWith("instance:") ? auth.slice("instance:".length) : null;

    switch (`${request.method} ${url.pathname}`) {
      case "GET /status":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.status();
      case "PATCH /settings":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.patchSettings(await readJson(request));
      case "POST /provision":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.provision();
      case "POST /wake":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.wake(await readJson(request));
      case "POST /stop":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.stopAll();
      case "POST /prune":
        if (!isPool) return new Response("forbidden", { status: 403 });
        return this.prune();
      case "POST /heartbeat":
        if (!instanceAuth) return new Response("forbidden", { status: 403 });
        return this.heartbeat(instanceAuth, await readJson(request));
      default:
        return new Response("not found", { status: 404 });
    }
  }

  private async status(): Promise<Response> {
    const [settings, instances, epoch] = await Promise.all([
      this.settings(),
      this.instances(),
      this.state.storage.get<number>(K_EPOCH),
    ]);
    const now = this.now();
    return json({
      poolId: this.poolId,
      settings,
      epoch: epoch ?? 0,
      instances: instances.map((i) => ({
        ...i,
        stale: i.state !== "stopped" && now - i.lastSeenAt > this.staleWindow(i),
      })),
    });
  }

  private async patchSettings(patch: Partial<PoolSettings> | null): Promise<Response> {
    const next = clampSettings(await this.settings(), patch);
    await this.state.storage.put(K_SETTINGS, next);
    // A lowered cap doesn't kill running work; the alarm reaps idle runners down to it.
    await this.scheduleReap();
    return json({ settings: next });
  }

  /** The lowest free `runner-<n>` slot (a stopped slot is reusable — a fresh boot mints a new
   * epoch and, inside the container, a new mesh identity). */
  private nextSlot(instances: InstanceRecord[]): string {
    const live = new Set(instances.filter((i) => i.state !== "stopped").map((i) => i.instanceId));
    let n = 1;
    while (live.has(`runner-${n}`)) n += 1;
    return `runner-${n}`;
  }

  /** Boot one more runner if the fleet is under the cap. Demand-driven: the owning device calls
   * this when a mesh dispatch is parked and `liveCount < maxRunners`. `at_capacity` is a normal,
   * non-error outcome — the caller keeps the chat queued (or runs it elsewhere on the mesh). */
  private async provision(): Promise<Response> {
    const [settings, instances] = await Promise.all([this.settings(), this.instances()]);
    const liveCount = instances.filter((i) => i.state !== "stopped").length;
    if (liveCount >= settings.maxRunners) {
      return json({ decision: "at_capacity", liveCount, maxRunners: settings.maxRunners });
    }
    return this.boot(this.nextSlot(instances), settings, instances);
  }

  /** Handle a wake targeting a specific runner slot — the relay-presence hook forwards
   * `{instanceId}` when a device dials a room with no host in it. A live-and-fresh slot is a
   * no-op (`alreadyRunning` — the dialer's ordinary reconnect will land). Any other state is
   * **declined**, not rebooted: a runner's mesh identity is minted per boot inside the
   * container, so a reboot can never serve the room the device is dialing — it would only
   * manufacture a mesh-orphan that holds a slot until reaped, gets re-woken by the next dial,
   * and burns containers in a loop. Capacity returns through the device dispatch path
   * (`provision`, which completes the mesh join and adopts unjoined runners). A bodyless wake
   * (a pool-credential holder's explicit verb) still behaves like `provision`. */
  private async wake(body: { instanceId?: unknown } | null): Promise<Response> {
    if (typeof body?.instanceId !== "string") return this.provision();
    const instanceId = body.instanceId;
    const instances = await this.instances();
    const existing = instances.find((i) => i.instanceId === instanceId);
    const isLive =
      existing &&
      existing.state !== "stopped" &&
      this.now() - existing.lastSeenAt <= this.staleWindow(existing);
    if (isLive) {
      return json({ instanceId, epoch: existing.epoch, alreadyRunning: true });
    }
    return json({ instanceId, decision: "identityRotated" });
  }

  /** Start (or restart) a runner slot: bump the epoch, mint its instance token, drive the
   * container up, and record it `starting`. */
  private async boot(instanceId: string, settings: PoolSettings, instances: InstanceRecord[]): Promise<Response> {
    const epoch = await this.bumpEpoch();
    const token = await this.deps.mintInstanceToken(this.poolId, instanceId);
    await this.deps.driver.start(this.poolId, instanceId, {
      NUCLEIC_RUNNER_POOL_ID: this.poolId,
      NUCLEIC_RUNNER_INSTANCE_ID: instanceId,
      NUCLEIC_RUNNER_EPOCH: String(epoch),
      NUCLEIC_RUNNER_INSTANCE_TOKEN: token,
      NUCLEIC_RUNNER_INTELLIGENCE_MODE: settings.intelligenceMode,
    });
    const now = this.now();
    const record: InstanceRecord = {
      instanceId,
      state: "starting",
      sessions: 0,
      activeSessions: 0,
      epoch,
      lastSeenAt: now,
      lastActiveAt: now,
      lastBusyAt: now,
    };
    await this.saveInstances([...instances.filter((i) => i.instanceId !== instanceId), record]);
    await this.scheduleReap();
    return json({ decision: "boot", instanceId, epoch });
  }

  private async stopAll(): Promise<Response> {
    const instances = await this.instances();
    for (const i of instances) {
      if (i.state !== "stopped") await this.deps.driver.stop(this.poolId, i.instanceId);
    }
    await this.saveInstances(
      instances.map((i) => ({ ...i, state: "stopped" as const, sessions: 0, activeSessions: 0 })),
    );
    return json({ stopped: instances.length });
  }

  /** Delete every stopped runner record. A stopped row holds no container — it's pure
   * bookkeeping (its slot is already reusable via `nextSlot`) — but rows otherwise accumulate
   * in the Settings list forever. Live records are untouched. */
  private async prune(): Promise<Response> {
    const instances = await this.instances();
    const kept = instances.filter((i) => i.state !== "stopped");
    if (kept.length !== instances.length) await this.saveInstances(kept);
    return json({ removed: instances.length - kept.length });
  }

  /** Runner liveness + fencing. A heartbeat whose epoch doesn't match the registry's record for
   * that slot is a zombie (the slot was re-booted since) and gets 409 — the container must exit,
   * never write. An accepted heartbeat re-mints the instance token. */
  private async heartbeat(
    instanceId: string,
    body:
      | { epoch?: unknown; sessions?: unknown; activeSessions?: unknown; state?: unknown; roomId?: unknown }
      | null,
  ): Promise<Response> {
    const instances = await this.instances();
    const record = instances.find((i) => i.instanceId === instanceId);
    if (!record || record.state === "stopped") return new Response("unknown instance", { status: 404 });
    if (typeof body?.epoch !== "number" || body.epoch !== record.epoch) {
      return new Response("stale epoch", { status: 409 });
    }
    const sessions =
      typeof body.sessions === "number" && Number.isFinite(body.sessions)
        ? Math.max(0, Math.floor(body.sessions))
        : record.sessions;
    // Sessions still doing something (running/blocked, or Done-but-holding-a-queued-message).
    // An older runner omits it — assume every session is active so it never sleeps down the
    // all-done path (the pool only sleeps a runner it's sure is quiet). Clamped to `sessions`:
    // there can't be more active sessions than sessions.
    const activeSessions =
      typeof body.activeSessions === "number" && Number.isFinite(body.activeSessions)
        ? Math.max(0, Math.min(sessions, Math.floor(body.activeSessions)))
        : sessions;
    // The runner's mesh identity (relay roomId), validated like wake.ts's mapping — status
    // exposes it so a device can tell a joined runner from an orphan needing adoption.
    const roomId =
      typeof body.roomId === "string" && /^[0-9a-f]{64}$/.test(body.roomId) ? body.roomId : record.roomId;
    const now = this.now();
    const next = instances.map((i) =>
      i.instanceId === instanceId
        ? {
            ...i,
            state: "ready" as const,
            sessions,
            activeSessions,
            roomId,
            lastSeenAt: now,
            lastActiveAt: sessions > 0 ? now : i.lastActiveAt,
            lastBusyAt: activeSessions > 0 ? now : (i.lastBusyAt ?? i.lastActiveAt),
          }
        : i,
    );
    await this.saveInstances(next);
    await this.scheduleReap();
    const token = await this.deps.mintInstanceToken(this.poolId, instanceId);
    return json({ ok: true, token });
  }

  /** How long a runner may be silent before it's declared dead — a longer grace while it's still
   * booting (never heartbeated) than once it has gone `ready`. */
  private staleWindow(record: InstanceRecord): number {
    return record.state === "starting" ? BOOT_GRACE_MS : HEARTBEAT_STALE_MS;
  }

  /** Alarm: uniform scale-to-zero. Reap every runner that is idle (zero sessions past the idle
   * timeout), all-done (still holds sessions but every one is Done and quiet past `DONE_SLEEP_MS`
   * — the runner's `activeSessions` heartbeat count reached 0), dead (silent past its staleness
   * window — a wedged/failed boot otherwise holds a `maxRunners` slot forever), or over a lowered
   * cap. There is no host-of-record: every runner is an equal, reapable mesh peer. A later
   * `provision`/`wake` (or the relay-presence hook) re-boots one when work returns. */
  async alarm(): Promise<void> {
    const [settings, instances] = await Promise.all([this.settings(), this.instances()]);
    const now = this.now();
    const live = instances.filter((i) => i.state !== "stopped");
    const runners = [...live].sort((a, b) => a.lastActiveAt - b.lastActiveAt); // oldest-idle first

    const overCap = Math.max(0, live.length - settings.maxRunners);
    const toStop = new Set<string>();
    for (const r of runners) {
      const dead = now - r.lastSeenAt > this.staleWindow(r);
      if (dead) {
        toStop.add(r.instanceId);
        continue;
      }
      // Every session Done and quiet (no active work, none holding a queued follow-up waiting to
      // dispatch) and settled past the done-sleep window: sleep the runner even though finished
      // chats still sit on it. `activeSessions` is absent from an older runner's heartbeat ⇒ read
      // as `sessions`, so this never fires for one that can't report it.
      const active = r.activeSessions ?? r.sessions;
      const doneSettled =
        r.sessions > 0 && active === 0 && now - (r.lastBusyAt ?? r.lastActiveAt) >= DONE_SLEEP_MS;
      if (doneSettled) {
        toStop.add(r.instanceId);
        continue;
      }
      const idle = r.sessions === 0 && now - r.lastActiveAt >= settings.idleTimeoutSeconds * 1000;
      // Trim toward a lowered cap by stopping idle runners oldest-first; never kill one with live
      // sessions (a lowered cap drains, it doesn't evict work).
      if (idle || toStop.size < overCap) {
        if (r.sessions === 0) toStop.add(r.instanceId);
      }
    }

    for (const id of toStop) await this.deps.driver.stop(this.poolId, id);
    if (toStop.size > 0) {
      await this.saveInstances(
        instances.map((i) => (toStop.has(i.instanceId) ? { ...i, state: "stopped" as const } : i)),
      );
    }
    // Keep the alarm cadence alive while anything is still live — including an idle runner that
    // hasn't yet crossed its scale-to-zero timeout, so a later pass can reap it.
    const remaining = instances.some(
      (i) => i.state !== "stopped" && !toStop.has(i.instanceId),
    );
    if (remaining) await this.scheduleReap();
  }

  private async scheduleReap(): Promise<void> {
    const settings = await this.settings();
    await this.state.storage.setAlarm(this.now() + Math.min(settings.idleTimeoutSeconds * 1000, 60_000));
  }
}

async function readJson(request: Request): Promise<Record<string, unknown> | null> {
  try {
    return (await request.json()) as Record<string, unknown>;
  } catch {
    return null;
  }
}
