import { spawn, type ChildProcess } from 'node:child_process';
import { createRequire } from 'node:module';
import { randomBytes } from 'node:crypto';
import {
  existsSync,
  readFileSync,
  rmSync,
  writeFileSync,
  mkdirSync,
  statSync,
  openSync,
  readSync,
  closeSync,
} from 'node:fs';
import { join, resolve as resolvePath, sep } from 'node:path';
import debug from 'debug';
import pidusage from 'pidusage';
import {
  PortAllocator,
  type Orchestrator,
  type OrchestratorCreateRequest,
  type OrchestratorCreateResult,
  type OrchestratorStats,
} from '@serfab/cadre-provider';

import { defaultLogPath, rotateOnDisk } from './log-rotator.js';
import { ensureNodeIdentity } from './node-identity.js';
import {
  allocateNodePorts,
  releaseNodePorts,
  reserveNodePorts,
  reusedNodePorts,
} from './port-allocator.js';
import { StateStore, type PersistedHandle, type PersistedState } from './state-store.js';
import { isPidAlive } from './pid-liveness.js';
import { assertPortFree, type PortBinding } from './port-probe.js';
import type { PushCredentials } from '@serfab/cadre-core';
// The child's `cadre.json` is typed against the CLI's own config schema, so a key the CLI stops
// accepting (or starts requiring) is a compile error here rather than a child that fails to start.
import { strandFilterConfigFromText, type CliConfig } from '@serfab/cadre-cli';
import {
  encodeDockerId,
  type Handle,
  type HostProcessConfig,
  type HostedSpawnRequest,
  type ManagedNodeInfo,
  type NodePorts,
  type NodeStateListener,
  type PushCredentialsResolver,
} from './types.js';

const log = debug('cadre:host:orchestrator');

const DEFAULTS = {
  portRangeStart: 10000,
  portRangeEnd: 20000,
  stopTimeoutMs: 10000,
  logMaxBytes: 10 * 1024 * 1024,
  logMaxFiles: 5,
};

const STARTUP_TOKEN_FILE = '.startup-token';
const CONFIG_FILE = 'cadre.json';
const LIVENESS_POLL_MS = 100;

const CADRE_ENV_PREFIX = 'CADRE_';

/**
 * Manager's own `process.env` with every `CADRE_*` key removed. A child node
 * takes its whole configuration from the config file this orchestrator writes
 * plus the vars set explicitly below, and `@serfab/cadre-cli` treats inherited
 * `CADRE_*` vars as overrides that BEAT that config file — so inheriting them
 * would let a var set on the manager (e.g. CADRE_PARTY_ID) silently
 * reconfigure every spawned child. Scrubbing the whole prefix rather than a
 * list of known keys means a new cli env var can't reintroduce the leak by
 * being forgotten here. Non-CADRE vars (PATH, DEBUG, proxy/TLS settings, etc.)
 * still pass through untouched.
 */
function scrubbedParentEnv(): NodeJS.ProcessEnv {
  const env = { ...process.env };
  for (const key of Object.keys(env)) {
    if (key.startsWith(CADRE_ENV_PREFIX)) delete env[key];
  }
  return env;
}

/**
 * The env carrying how the child gets into its cadre, and nowhere else (see
 * `HostedSpawnRequest`): `CADRE_CLAIM_SECRET` makes `cadre-cli start` wait to be
 * claimed, `CADRE_INVITATION` makes it redeem the invitation once it is up.
 */
function joinEnv(request: HostedSpawnRequest): Record<string, string> | undefined {
  if (request.claimSecret) return { CADRE_CLAIM_SECRET: request.claimSecret };
  if (request.invitation) return { CADRE_INVITATION: request.invitation };
  return undefined;
}

/**
 * The libp2p listen addresses a managed child binds — its `CADRE_LISTEN_ADDRS`, one
 * entry per element: TCP on the `p2p` port and WebSocket on the `ws` port, both on
 * every interface.
 *
 * The WebSocket entry is what lets a phone reach the child. A phone's node carries
 * WebSocket, circuit-relay and WebRTC transports but no TCP
 * (`reference-app-rn/src/phone-node-config.ts`), and it listens on nothing, so the
 * phone has to be the one that dials. Naming a `/ws` address is all the child needs:
 * cadre-core adds the WebSocket transport for any listen entry that names one
 * (`resolveTransportOptions` in `cadre-core/src/relay-addrs.ts`).
 */
export function childListenAddrs(ports: Pick<NodePorts, 'p2p' | 'ws'>): string[] {
  return [`/ip4/0.0.0.0/tcp/${ports.p2p}`, `/ip4/0.0.0.0/tcp/${ports.ws}/ws`];
}

/**
 * Every TCP port a managed child binds, on the address it binds it to: health and
 * metrics (`cadre-cli`'s `HealthServer`) and the libp2p TCP and WebSocket listeners
 * ({@link childListenAddrs}), all on every interface. A key the set lacks (a handle
 * persisted by an older build) is skipped.
 */
function childPortBindings(ports: Partial<NodePorts>): PortBinding[] {
  const bindings: PortBinding[] = [];
  for (const port of [ports.health, ports.metrics, ports.p2p, ports.ws]) {
    if (port !== undefined) bindings.push({ port, host: '0.0.0.0' });
  }
  return bindings;
}

/**
 * Runs cadre nodes as plain Node child processes on the host.
 *
 * Each container gets its own workdir under `rootDir`. Handles are mirrored
 * to `<rootDir>/state.json` so cadre-host can re-attach after a restart.
 *
 * ## Child stdio: direct-fd inheritance
 *
 * Spawned children inherit a file descriptor opened on `<workdir>/node.log`
 * directly as their stdout and stderr. The orchestrator closes its own copy
 * of the fd immediately after spawn, so the child's write path no longer
 * touches any handle owned by this process. This is what lets children
 * survive — and keep logging through — an orchestrator restart. Were the
 * orchestrator instead piping stdout/stderr through an in-process Writable,
 * a parent exit would deliver SIGPIPE to the child (POSIX) or fail its
 * writes with EPIPE (Windows), silently breaking the documented "init()
 * re-attach" flow.
 *
 * ## Log rotation: spawn-time only
 *
 * Because the child holds a stable fd open against the inode, the
 * orchestrator cannot safely rename `node.log` while the child is running.
 * Rotation therefore happens only at spawn time, when no live child is
 * attached to the workdir (`createContainer` is the single producer per
 * workdir). If `node.log` already exceeds `logMaxBytes`, the rotation
 * cascade runs against the on-disk paths before the fresh fd is opened.
 * A child whose log grows past `logMaxBytes` mid-run will keep appending
 * to the active file until it (or its successor) is restarted; v1 traffic
 * is low enough that this is acceptable.
 */
export class HostProcessOrchestrator implements Orchestrator {
  private readonly rootDir: string;
  private readonly portAllocator: PortAllocator;
  private readonly stateStore: StateStore;
  private readonly stopTimeoutMs: number;
  private readonly logMaxBytes: number;
  private readonly logMaxFiles: number;
  private readonly defaultMemoryLimit?: string;
  private readonly handles = new Map<string, Handle>();
  private readonly cliEntrypoint: string;
  /** Resolves push credentials fresh on every spawn (see HostProcessConfig). */
  private readonly pushResolver?: PushCredentialsResolver;
  /** The public addresses a node announces, asked at every spawn (see HostProcessConfig). */
  private readonly announceAddrs?: HostProcessConfig['announceAddrs'];
  /** State-change listeners — invoked when a handle's alive state changes. */
  private readonly stateListeners = new Set<NodeStateListener>();

  constructor(private readonly cfg: HostProcessConfig) {
    this.rootDir = resolvePath(cfg.rootDir);
    mkdirSync(this.rootDir, { recursive: true });
    this.portAllocator = new PortAllocator(
      cfg.portRange?.start ?? DEFAULTS.portRangeStart,
      cfg.portRange?.end ?? DEFAULTS.portRangeEnd,
    );
    this.stateStore = new StateStore(this.rootDir);
    this.stopTimeoutMs = cfg.stopTimeoutMs ?? DEFAULTS.stopTimeoutMs;
    this.logMaxBytes = cfg.logMaxBytes ?? DEFAULTS.logMaxBytes;
    this.logMaxFiles = cfg.logMaxFiles ?? DEFAULTS.logMaxFiles;
    this.defaultMemoryLimit = cfg.defaultResources?.memoryLimit;
    this.cliEntrypoint = cfg.spawn?.entrypoint ?? resolveCadreCliBin();
    this.pushResolver = cfg.pushResolver;
    this.announceAddrs = cfg.announceAddrs;
    log('HostProcessOrchestrator rootDir=%s cli=%s', this.rootDir, this.cliEntrypoint);
  }

  /**
   * Read persisted state and re-attach to surviving children. Not part of
   * `Orchestrator` — call once before handing the instance to a container
   * service.
   */
  async init(): Promise<void> {
    const state = this.stateStore.load();
    for (const persisted of state.handles) {
      const handle: Handle = {
        containerId: persisted.containerId,
        dockerId: persisted.dockerId,
        pid: persisted.pid,
        startupToken: persisted.startupToken,
        workdir: persisted.workdir,
        ports: persisted.ports,
        spawnedAt: persisted.spawnedAt,
        partyId: persisted.partyId,
        profile: persisted.profile,
        announcedAddrs: persisted.announcedAddrs ?? [],
        alive: false,
      };
      handle.alive = isHandleLive(handle);
      this.handles.set(handle.dockerId, handle);
      // Always reserve ports — even for dead handles, until the caller cleans them up
      this.reservePorts(persisted.ports);
    }
    log('init: re-attached %d handles', state.handles.length);
  }

  /**
   * Snapshot of currently-known managed nodes. Used by the local UI's
   * `/api/nodes` route.
   */
  listNodes(): ManagedNodeInfo[] {
    return [...this.handles.values()].map((h) => toNodeInfo(h));
  }

  /**
   * Look up a single managed node by either containerId (friendly id) or
   * dockerId (opaque `pid:token`). Returns undefined when neither matches.
   */
  getNode(idOrDockerId: string): ManagedNodeInfo | undefined {
    const direct = this.handles.get(idOrDockerId);
    if (direct) return toNodeInfo(direct);
    for (const h of this.handles.values()) {
      if (h.containerId === idOrDockerId) return toNodeInfo(h);
    }
    return undefined;
  }

  /**
   * Resolve the opaque dockerId for a friendly containerId so callers can
   * pass it into stopContainer / getLogs / getStats.
   */
  resolveDockerId(idOrDockerId: string): string | undefined {
    if (this.handles.has(idOrDockerId)) return idOrDockerId;
    for (const h of this.handles.values()) {
      if (h.containerId === idOrDockerId) return h.dockerId;
    }
    return undefined;
  }

  /**
   * Register a state-change listener. Invoked when a handle's `alive`
   * transitions (create, child exit, manual stop) and once more, as `stopped`,
   * when a handle is removed. Returns an unsubscribe fn.
   */
  onStateChange(listener: NodeStateListener): () => void {
    this.stateListeners.add(listener);
    return () => { this.stateListeners.delete(listener); };
  }

  /**
   * Spawn a managed node — a hosted node for some cadre, started waiting to be
   * claimed when the request carries a `claimSecret`, or redeeming the cadre
   * invitation it carries as `invitation`. The node is given its own
   * protobuf identity key inside its workdir (`ensureNodeIdentity`, reused across
   * re-spawns of the same containerId), which is what makes its peer id stable
   * across restarts AND what makes its node-local stores durable: `cadre-cli
   * start` opens the file-backed bootstrap-peer, trusted-owner and strand
   * network-state stores in the node's state directory, which defaults to the
   * directory holding the `cadre.json` written here — the same workdir.
   * Removing the node (`removeContainer`) deletes the workdir, so the key and
   * every store go with it.
   */
  async createContainer(request: HostedSpawnRequest): Promise<OrchestratorCreateResult> {
    // Before anything is allocated: `cadre-cli start` refuses the pair anyway, and a child
    // that exits at once would only surface as a crash loop.
    if (request.claimSecret && request.invitation) {
      throw new Error(`container ${request.containerId}: a spawn carries a claim secret or an invitation, not both`);
    }
    const workdir = this.workdirFor(request.containerId);
    // Sampled BEFORE `ensureNodeIdentity`, which brings the directory into
    // existence as a side effect of writing the key. Only a spawn that found
    // nothing here may delete anything on its way out — a re-spawn is REUSING
    // the identity key and node-local stores this directory holds, and they are
    // the whole reason the node comes back as the same peer. `false` is
    // therefore the safe default: a directory that exists for any reason
    // (respawn, an earlier failed provision, an operator's file) is never
    // touched by the unwind below.
    const isNewWorkdir = !existsSync(workdir);
    let dropped: Handle[] = [];
    let ports: NodePorts | undefined;
    try {
      // Identity BEFORE any port is reserved: it is the one step here that can
      // fail on its own (an unreadable or undecodable identity.key throws rather
      // than silently re-keying), and everything past the allocation is either
      // infallible or releases the ports itself. Allocating first would leak a
      // node's worth of ports from a bounded range on every failed provision attempt.
      const identity = await ensureNodeIdentity(workdir);
      log('container %s identity peerId=%s', request.containerId, identity.peerId);
      // Only storage-profile nodes participate in strands and thus fan out push
      // wakes — a transaction-only node need not carry credentials. Resolved
      // BEFORE the drop below, which keeps the drop → launch window synchronous
      // (see restoreDroppedHandles).
      const push = request.profile === 'storage' ? await this.resolvePush() : undefined;
      const extraEnv = joinEnv(request);

      // The last `await` on this path: checked as close to the drop as it can be.
      await this.refuseRespawnOverLiveChild(request.containerId);

      // Nothing from here to the launch is `await`ed — which is what keeps
      // `restoreDroppedHandles`'s documented precondition true even though the
      // `try` now opens further up. Do not add an `await` past this line.
      dropped = this.dropStaleHandle(request.containerId);
      // A re-spawn comes back on the ports the dropped handle held, not merely the
      // lowest free ones: a requester that dials the node itself (a phone) knows it
      // only by the address it was given. The drop above just released them, and
      // nothing can have taken them since — see reusedNodePorts.
      ports = allocateNodePorts(this.portAllocator, reusedNodePorts(dropped));
      return this.launchChild({
        containerId: request.containerId,
        partyId: request.partyId,
        profile: request.profile,
        ports,
        buildConfig: () => this.buildChildConfig(request, workdir, push),
        extraArgs: ['--identity-file', identity.path],
        ...(extraEnv ? { extraEnv } : {}),
        ...(request.resources?.memoryLimit ? { memoryLimit: request.resources.memoryLimit } : {}),
      });
    } catch (err) {
      // Release BEFORE restoring: a re-spawn allocates the very ports the drop
      // just freed, so restoring first and releasing second would hand the
      // restored handle's own ports straight back to the allocator.
      if (ports) this.releasePorts(ports);
      // A no-op for the failures that happen before the drop (`dropped` is
      // still empty).
      this.restoreDroppedHandles(dropped);
      // Safe because a throw out of `launchChild` means NO child is running:
      // its throw surface ends where the child is spawned and its handle stored
      // (everything past that line is best-effort), and the `if (!child.pid)`
      // throw is on the no-process side too. So nothing is ever holding the
      // directory this deletes.
      if (isNewWorkdir) this.discardWorkdir(workdir);
      throw err;
    }
  }

  /**
   * Remove the working directory a spawn created for `containerId` when no
   * handle owns it — the cleanup of last resort for a hosted-node record that
   * never got a `dockerId` (the host died between `ensureNodeIdentity` and the
   * handle write, so `createContainer`'s own unwind never ran). Returns whether
   * anything was removed.
   *
   * Synchronous, mirroring {@link resolveDockerId}: nothing on this path needs
   * to await.
   */
  reclaimWorkdir(containerId: string): boolean {
    // A live handle owns that directory, and `removeContainer` — which stops
    // the child first — is the only thing allowed to delete it.
    if (this.resolveDockerId(containerId)) return false;
    const workdir = this.workdirFor(containerId);
    // Defence, not a live case: hosted-node ids are `hn_<base64url>` and so can
    // hold neither a path separator nor a dot. A caller that ever passes
    // something else must not be able to walk out of `rootDir`.
    if (!workdir.startsWith(this.rootDir + sep)) {
      log('refusing to reclaim workdir outside rootDir: %s', workdir);
      return false;
    }
    // Must be a DIRECTORY, not merely something that exists: `rootDir` also
    // holds `state.json` (this orchestrator's own persisted handles), and
    // `discardWorkdir`'s `rmSync` deletes files as happily as trees. Same
    // defence class as the escape check above — no real containerId names a
    // file — but the blast radius if one ever did is the whole handle map.
    if (!statSync(workdir, { throwIfNoEntry: false })?.isDirectory()) return false;
    this.discardWorkdir(workdir);
    return true;
  }

  /**
   * Best-effort removal of a workdir no handle owns. Logs, never throws: every
   * caller is either unwinding a failure it must report instead of this one, or
   * tearing a container down along a path where a surviving directory is a
   * nuisance rather than a fault.
   *
   * `retries` sizes the wait for the OS to release a cwd handle — Windows drops
   * it asynchronously after a child exits. `createContainer`'s unwind needs none
   * of that (its throw surface ends before any child is spawned), but
   * `reclaimWorkdir` can still meet an orphan process the previous host left
   * running with this directory as its cwd, so it is not a *guaranteed*
   * no-process path either — it simply logs and leaves the directory when the
   * removal fails.
   */
  private discardWorkdir(workdir: string, retries = { maxRetries: 5, retryDelay: 100 }): void {
    try {
      rmSync(workdir, { recursive: true, force: true, ...retries });
      log('discarded workdir %s', workdir);
    } catch (err) {
      log('failed to discard workdir %s: %o', workdir, err);
    }
  }

  /**
   * Resolve the push credentials to inject into the node about to spawn. Called
   * on every spawn so a re-spawn re-reads the secret store — raw private keys are
   * never persisted. A resolver failure (e.g. a partial credential set) is
   * logged and degrades to "no push" rather than failing the spawn: the node
   * stays up and reachable, just without platform push-wake.
   */
  private async resolvePush(): Promise<PushCredentials | undefined> {
    if (!this.pushResolver) return undefined;
    try {
      const push = await this.pushResolver();
      if (push) {
        log('resolved push credentials: fcm=%s apns=%s cooldownMs=%s', !!push.fcm, !!push.apns, push.cooldownMs);
      }
      return push;
    } catch (err) {
      log('push resolver failed; spawning node without push: %s', (err as Error).message);
      return undefined;
    }
  }

  /**
   * The public addresses the node about to spawn announces. A hook that throws costs
   * the node its public addresses, never its start: it still binds and is reachable
   * on its LAN addresses, and `NatService` restarts it once the addresses it should
   * announce differ from these.
   */
  private announceAddrsFor(containerId: string, ports: NodePorts): string[] {
    if (!this.announceAddrs) return [];
    try {
      return this.announceAddrs(containerId, ports);
    } catch (err) {
      log('announce hook failed for %s; it starts with no public addresses: %s', containerId, (err as Error).message);
      return [];
    }
  }

  /**
   * Core spawn mechanics behind `createContainer`.
   *
   * The caller has already allocated `ports` and **owns their lifetime**: this
   * method never releases them, so the caller must release on any throw out
   * of here (it has to unwind its handle drop on the same path anyway — see
   * `restoreDroppedHandles`).
   *
   * That contract holds because this method throws only *before* the child is
   * spawned and its handle stored — everything after that point is
   * best-effort. Anything new added at the tail must keep it that way.
   *
   * Entirely synchronous, which is what makes that unwind sound. Do not add an
   * `await` here.
   */
  private launchChild(opts: {
    containerId: string;
    partyId: string;
    profile: 'storage' | 'transaction';
    ports: NodePorts;
    buildConfig: (workdir: string) => CliConfig;
    extraArgs: string[];
    /** Extra env vars merged into the child's environment (e.g. CADRE_CLAIM_SECRET). */
    extraEnv?: Record<string, string>;
    memoryLimit?: string;
  }): OrchestratorCreateResult {
    const { containerId, ports } = opts;
    const workdir = this.workdirFor(containerId);
    mkdirSync(workdir, { recursive: true });
    mkdirSync(join(workdir, 'storage'), { recursive: true });

    const token = randomBytes(16).toString('hex');
    const tokenPath = join(workdir, STARTUP_TOKEN_FILE);
    // Ensure stale token from a previous container with the same id is cleared.
    if (existsSync(tokenPath)) {
      try { rmSync(tokenPath); } catch { /* ignore */ }
    }

    const configPath = join(workdir, CONFIG_FILE);
    writeFileSync(configPath, JSON.stringify(opts.buildConfig(workdir), null, 2), 'utf8');

    const memoryLimit = opts.memoryLimit ?? this.defaultMemoryLimit;
    const heapBytes = parseMemoryLimit(memoryLimit);
    const heapMB = heapBytes !== undefined ? Math.max(64, Math.floor(heapBytes / (1024 * 1024))) : undefined;
    const announcedAddrs = this.announceAddrsFor(containerId, ports);

    const env: NodeJS.ProcessEnv = {
      // Scrubbed, not raw process.env: every CADRE_* config-override key is
      // stripped so a var set on the manager can't silently reconfigure every
      // child (see scrubbedParentEnv). Caller-supplied extras come next — the
      // fixed per-child vars below always win so extraEnv can never clobber
      // the startup token or port wiring.
      ...scrubbedParentEnv(),
      ...opts.extraEnv,
      CADRE_STARTUP_TOKEN: token,
      CADRE_HEALTH_PORT: String(ports.health),
      CADRE_METRICS_PORT: String(ports.metrics),
      CADRE_LISTEN_ADDRS: childListenAddrs(ports).join(','),
      // The append form, not `CADRE_ANNOUNCE_ADDRS`: that one replaces the advertised
      // set and would drop the LAN addresses a phone at home dials. Every entry must be
      // a valid multiaddr: cadre-core refuses to start on one that is not.
      // The TCP and WebSocket ports are the CHILD'S CONTROL NODE's alone: each strand
      // node the child runs binds the same two entries with OS-assigned ports instead,
      // since one port cannot be held twice, and inherits neither announce field
      // (`cadre-core/src/strand-network-config.ts`). Those ports are not mapped, and
      // strand nodes are reached through observed addresses only, since no
      // `CADRE_RELAY_ADDRS` is set here and a child therefore holds no relay reservation.
      ...(announcedAddrs.length > 0 ? { CADRE_APPEND_ANNOUNCE_ADDRS: announcedAddrs.join(',') } : {}),
      // Pin each child's node-local state (trusted-owner anchor, retained
      // cold-start dial targets) to its OWN workdir. This is the same value the
      // cli would derive by default (the config file lives here too), but stated
      // explicitly rather than relying on the scrub above (belt and suspenders).
      CADRE_NODE_STATE_DIR: workdir,
    };
    if (heapMB !== undefined) {
      env.NODE_OPTIONS = `${process.env.NODE_OPTIONS ?? ''} --max-old-space-size=${heapMB}`.trim();
    }

    // Rotate the on-disk log files BEFORE opening the fd — once the child
    // is running it holds the inode open and any later rename would leave
    // it writing to the rotated file. See class docstring.
    const logPath = defaultLogPath(workdir);
    maybeRotateAtSpawn(logPath, this.logMaxBytes, this.logMaxFiles);
    const logFd = openSync(logPath, 'a');

    const args = [
      this.cliEntrypoint,
      'start',
      '-c', configPath,
      '--health-port', String(ports.health),
      '--metrics-port', String(ports.metrics),
      '--startup-token-file', tokenPath,
      ...opts.extraArgs,
    ];

    let child: ChildProcess;
    try {
      child = spawn(process.execPath, args, {
        cwd: workdir,
        detached: true,
        // Inherit the log fd directly. OS duplicates it into the child;
        // closing our copy below is safe and severs the parent's only
        // tie to the child's write path.
        stdio: ['ignore', logFd, logFd],
        env,
      });
    } catch (err) {
      // Synchronous spawn failure — close the fd before propagating. This is
      // the ONLY fd-leak guard on this path: nothing between the `openSync`
      // above and the `spawn` can throw, so no other catch is needed. Keep it
      // that way, or add one. (The ports are the caller's to release.)
      closeSync(logFd);
      throw err;
    }
    // Always release the parent's copy of the fd — the child has its own.
    closeSync(logFd);

    if (!child.pid) {
      throw new Error(`Failed to spawn child for container ${containerId}`);
    }

    child.unref();

    const dockerId = encodeDockerId(child.pid, token);
    const handle: Handle = {
      containerId,
      dockerId,
      pid: child.pid,
      startupToken: token,
      workdir,
      ports,
      spawnedAt: new Date().toISOString(),
      partyId: opts.partyId,
      profile: opts.profile,
      announcedAddrs,
      child,
      alive: true,
    };

    child.once('exit', () => {
      const h = this.handles.get(dockerId);
      if (h && h.alive) {
        h.alive = false;
        this.emitStateChange(h);
      }
    });

    this.handles.set(dockerId, handle);
    // Past this line the launch has SUCCEEDED: the child is running and this
    // handle owns its ports. So the state mirror is best-effort — a throw here
    // would reach the caller's unwind, which would release ports the live child
    // is bound to and restore the handle this spawn just replaced, leaving two
    // handles for one container and one of them holding unreserved ports. A
    // lost write self-heals: the next persist() (any stop/remove/spawn)
    // rewrites the whole map. Until then a manager restart would fail to
    // re-attach this child, hence the log.
    try {
      this.persist();
    } catch (err) {
      log('failed to persist state after launching container %s: %o', containerId, err);
    }
    this.emitStateChange(handle);

    log('created container %s -> pid %d ports=%j', containerId, child.pid, handle.ports);

    return {
      dockerId,
      healthEndpoint: `http://localhost:${ports.health}/health`,
      metricsEndpoint: `http://localhost:${ports.metrics}/metrics`,
      p2pPort: ports.p2p,
    };
  }

  /**
   * Refuse to re-spawn `containerId` while a previous child of it may still be
   * running. `createContainer` calls this as its last `await` before the handle
   * drop, so a refusal happens while nothing has been released, no token rotated
   * and no child launched — the caller sees an ordinary failed spawn.
   *
   * Why it exists: the drop hands the old handle's ports back and the launch takes
   * those same ports (`reusedNodePorts`), so a child started over a live one dies
   * on a port clash — but only after `launchChild` has returned success. The
   * caller then records a dead child, and the live one keeps running with no
   * handle naming it.
   *
   * - A handle this process spawned is checked exactly ({@link isHandleLive}); a
   *   live one refuses outright.
   * - A handle re-attached from `state.json` whose pid is alive but whose token
   *   does not match is either our child still starting or an unrelated process
   *   that inherited the pid. The ports decide: if any the re-spawn would take back
   *   is already bound, refuse. The pid is never killed on that evidence — it may
   *   not be ours.
   *
   * NOTE: the probe cannot see a re-attached child still loading its modules —
   * `cadre-cli start` writes its token and only then binds, so a child in that
   * window shows neither, and a re-spawn launched then races it for the ports.
   * Reachable only when the host restarts within a child's first seconds. If it
   * is ever seen, treat a re-attached handle with a live pid as starting for a
   * grace period after its `spawnedAt`.
   *
   * NOTE: the probe holds each port for an instant, so a child binding that port
   * in the same instant fails. It would have had to write its token between the
   * token check above and the probe, against the several awaited steps
   * `cadre-cli start` takes between writing its token and binding; if a re-attached
   * child is ever seen dying on its own port right after a host restart, probe by
   * connecting instead of binding.
   */
  private async refuseRespawnOverLiveChild(containerId: string): Promise<void> {
    const previous = this.handlesFor(containerId);
    if (previous.some(isHandleLive)) {
      throw new Error(`container ${containerId} is still running`);
    }
    if (!previous.some((h) => isPidAlive(h.pid))) return;
    const bindings = childPortBindings(reusedNodePorts(previous));
    try {
      await Promise.all(bindings.map(assertPortFree));
    } catch (err) {
      throw new Error(
        `container ${containerId} may still be running: ${(err as Error).message}`,
        { cause: err },
      );
    }
  }

  /** Every handle for `containerId`, in map order — the order `dropStaleHandle` drops them in. */
  private handlesFor(containerId: string): Handle[] {
    return [...this.handles.values()].filter((h) => h.containerId === containerId);
  }

  /**
   * Drop any handle left over from a previous spawn of the same `containerId`
   * and release its ports. Handles are keyed by the per-spawn `dockerId`, so
   * without this a re-spawn (a hosted node's respawn) would strand the prior
   * handle in the map forever, leaking its ports from a bounded range each
   * time.
   *
   * DO NOT delete the workdir: the identity key and node-local stores
   * (trusted-owner anchor, retained cold-start dial targets) live there and are
   * the whole reason a respawn comes back as the same node. `launchChild`
   * reuses the same workdir.
   *
   * Releasing the ports hands them straight back to the allocator, and the
   * re-spawn then takes those same ports as overrides (`reusedNodePorts`), so a
   * re-spawn while the *previous* child is still listening would bind-clash.
   * The caller must therefore never drop a live child, and does so only after
   * {@link refuseRespawnOverLiveChild} has passed. A caller that needs to replace
   * a live child must stop it first.
   *
   * **The drop is reversible.** It happens before the launch can fail, so the
   * dropped handles are returned and the caller restores them via
   * {@link restoreDroppedHandles} when the launch throws — otherwise the caller
   * would be left holding a dockerId this orchestrator no longer knows, and the
   * node's workdir could never be reclaimed. On a *successful* spawn the drop
   * stands: `HostedNodeService.abandonRespawn` depends on the old handle being
   * gone by then.
   */
  private dropStaleHandle(containerId: string): Handle[] {
    const dropped = this.handlesFor(containerId);
    for (const h of dropped) {
      log('dropping stale handle %s for container %s', h.dockerId, containerId);
      this.releasePorts(h.ports);
      this.handles.delete(h.dockerId);
    }
    return dropped;
  }

  /**
   * Inverse of {@link dropStaleHandle}, for a spawn that then failed: put each
   * handle back in the map and re-reserve its ports, leaving host state
   * exactly as the failed attempt found it.
   *
   * **Precondition: the drop → launch window is synchronous.** Restoring is
   * sound only because nothing can have taken those ports or run cleanup
   * against the handle in between — `launchChild` contains no `await` and the
   * caller resolves push credentials *before* its drop. `launchChild`'s half
   * is enforced: its declared return type is not a Promise, so adding an
   * `await` there is a compile error. The caller's half is not — an `await`
   * slipped between a drop and its launch silently breaks this.
   *
   * No `persist()` here: `state.json` is rewritten only by `launchChild` and
   * only *after* the new handle is stored, so throughout the failure window the
   * on-disk state still lists the dropped handle — this restore realigns memory
   * to disk rather than diverging from it.
   *
   * NOTE: a restored handle keeps `alive: true` even if its child exited while
   * it was out of the map (`launchChild`'s exit listener looks the handle up by
   * dockerId and finds nothing). Only reachable by dropping a handle whose child
   * is still live, which {@link refuseRespawnOverLiveChild} refuses before any
   * drop. (Its start-up blind spot concerns re-attached handles only, and those
   * have no exit listener to miss.)
   */
  private restoreDroppedHandles(dropped: Handle[]): void {
    for (const h of dropped) {
      log('restoring dropped handle %s for container %s after failed launch', h.dockerId, h.containerId);
      this.handles.set(h.dockerId, h);
      this.reservePorts(h.ports);
    }
  }

  /** A managed node's own directory under `rootDir` — config, log, storage, identity. */
  private workdirFor(containerId: string): string {
    return join(this.rootDir, containerId);
  }

  /**
   * Inverse of {@link releasePorts} — hold a handle's whole port set against
   * the allocator. Used both to rehydrate on `init` and to undo a drop whose
   * launch then failed.
   */
  private reservePorts(ports: NodePorts): void {
    reserveNodePorts(this.portAllocator, ports);
  }

  private releasePorts(ports: NodePorts): void {
    releaseNodePorts(this.portAllocator, ports);
  }

  async stopContainer(dockerId: string): Promise<void> {
    const handle = this.requireHandle(dockerId);
    if (!isPidAlive(handle.pid)) {
      handle.alive = false;
      this.persist();
      return;
    }

    // Capture the exit event before we send signals so we can await full
    // teardown — important on Windows where cwd handles release lazily.
    const exitPromise = handle.child
      ? new Promise<void>((resolve) => handle.child!.once('exit', () => resolve()))
      : Promise.resolve();

    log('SIGTERM container %s (pid=%d)', handle.containerId, handle.pid);
    try {
      process.kill(handle.pid, 'SIGTERM');
    } catch (err) {
      const code = (err as NodeJS.ErrnoException).code;
      if (code !== 'ESRCH') {
        throw err;
      }
    }

    const deadline = Date.now() + this.stopTimeoutMs;
    while (Date.now() < deadline) {
      if (!isPidAlive(handle.pid)) break;
      await sleep(LIVENESS_POLL_MS);
    }

    if (isPidAlive(handle.pid)) {
      log('SIGKILL container %s (pid=%d)', handle.containerId, handle.pid);
      try {
        process.kill(handle.pid, 'SIGKILL');
      } catch (err) {
        const code = (err as NodeJS.ErrnoException).code;
        if (code !== 'ESRCH') {
          throw err;
        }
      }
      const killDeadline = Date.now() + 2000;
      while (Date.now() < killDeadline && isPidAlive(handle.pid)) {
        await sleep(LIVENESS_POLL_MS);
      }
    }

    // Wait briefly for the OS to fully tear down the child (Windows cwd lock).
    await Promise.race([exitPromise, sleep(1000)]);

    const wasAlive = handle.alive;
    handle.alive = false;
    this.persist();
    if (wasAlive) this.emitStateChange(handle);
  }

  async removeContainer(dockerId: string): Promise<void> {
    const handle = this.requireHandle(dockerId);
    if (isPidAlive(handle.pid)) {
      await this.stopContainer(dockerId);
    }

    this.releasePorts(handle.ports);

    // Longer retries than the other callers': this is the one path that has
    // just killed a child which held this directory as its cwd, and on Windows
    // the OS releases that handle asynchronously.
    this.discardWorkdir(handle.workdir, { maxRetries: 20, retryDelay: 200 });

    this.handles.delete(dockerId);
    this.persist();
    // Removal is a state change too: `NatService` releases the node's port
    // mappings on it at once rather than at its next timer pass. A stopped
    // event for a node no longer in `listNodes()` is how listeners tell a
    // removal from a stop.
    this.emitStateChange(handle);
  }

  /**
   * Whether the child `dockerId` names is running — see {@link isHandleLive}. An
   * id with no handle is not running: without the handle's workdir there is no
   * token to verify a live pid against.
   */
  async isRunning(dockerId: string): Promise<boolean> {
    const handle = this.handles.get(dockerId);
    if (!handle) return false;
    if (isHandleLive(handle)) return true;
    if (!isPidAlive(handle.pid)) handle.alive = false;
    return false;
  }

  async getStats(dockerId: string): Promise<OrchestratorStats> {
    const handle = this.requireHandle(dockerId);
    if (!isPidAlive(handle.pid)) {
      return { cpuPercent: 0, memoryBytes: 0, networkRxBytes: 0, networkTxBytes: 0 };
    }
    try {
      const usage = await pidusage(handle.pid);
      return {
        cpuPercent: usage.cpu,
        memoryBytes: usage.memory,
        networkRxBytes: 0,
        networkTxBytes: 0,
      };
    } catch (err) {
      log('pidusage failed for %d: %o', handle.pid, err);
      return { cpuPercent: 0, memoryBytes: 0, networkRxBytes: 0, networkTxBytes: 0 };
    }
  }

  async getLogs(dockerId: string, tail = 100): Promise<string> {
    const handle = this.requireHandle(dockerId);
    const logPath = defaultLogPath(handle.workdir);
    return tailFile(logPath, tail);
  }

  private requireHandle(dockerId: string): Handle {
    const handle = this.handles.get(dockerId);
    if (!handle) {
      throw new Error(`Container not found: ${dockerId}`);
    }
    return handle;
  }

  private buildChildConfig(
    req: OrchestratorCreateRequest,
    workdir: string,
    push?: PushCredentials,
  ): CliConfig {
    const cfg: CliConfig = {
      controlNetwork: {
        partyId: req.partyId,
        // A hosted node is dialed (a claim) or dials the members its invitation names.
        bootstrapNodes: [],
      },
      profile: req.profile,
      // The request carries the filter as text (the provider hands the same text to
      // CADRE_STRAND_FILTER); the file form is `all`/`none` or the parsed JSON object, and a
      // filter the node would refuse fails here, before anything is spawned.
      // NOTE: an empty string is refused here, where cadre-provider's container-env treats it as
      // unset. Nothing in cadre-host populates `req.strandFilter` today; if the provision route
      // starts accepting one, pick one rule for both.
      strandFilter: req.strandFilter === undefined ? 'all' : strandFilterConfigFromText(req.strandFilter),
    };
    if (req.profile === 'storage') {
      cfg.storage = {
        type: 'file',
        path: join(workdir, 'storage'),
        ...(req.resources?.storageQuotaBytes !== undefined
          ? { quotaBytes: req.resources.storageQuotaBytes }
          : {}),
      };
    }
    // Push credentials land in the child's cadre.json on the host filesystem —
    // the same trust boundary as the control-DB storage in this workdir. The
    // private keys are NOT logged (the resolver redacts; we log only presence).
    if (push) cfg.push = push;
    return cfg;
  }

  private persist(): void {
    const state: PersistedState = {
      version: 1,
      handles: [...this.handles.values()].map((h) => this.toPersisted(h)),
    };
    this.stateStore.save(state);
  }

  private emitStateChange(handle: Handle): void {
    if (this.stateListeners.size === 0) return;
    const info = toNodeInfo(handle);
    for (const listener of this.stateListeners) {
      try {
        listener(info);
      } catch (err) {
        log('state listener threw for %s: %o', info.id, err);
      }
    }
  }

  private toPersisted(h: Handle): PersistedHandle {
    return {
      containerId: h.containerId,
      dockerId: h.dockerId,
      pid: h.pid,
      startupToken: h.startupToken,
      workdir: h.workdir,
      ports: h.ports,
      spawnedAt: h.spawnedAt,
      partyId: h.partyId,
      profile: h.profile,
      announcedAddrs: h.announcedAddrs,
    };
  }
}

function toNodeInfo(h: Handle): ManagedNodeInfo {
  return {
    id: h.containerId,
    dockerId: h.dockerId,
    partyId: h.partyId,
    profile: h.profile,
    status: h.alive ? 'running' : 'stopped',
    spawnedAt: h.spawnedAt,
    workdir: h.workdir,
    ports: { ...h.ports },
    announcedAddrs: [...h.announcedAddrs],
  };
}

/**
 * Locate the cadre-cli bin via the consumer-style module resolver. Works for
 * both yarn-managed workspaces and traditional npm installs.
 */
function resolveCadreCliBin(): string {
  const req = createRequire(import.meta.url);
  try {
    return req.resolve('@serfab/cadre-cli/bin/cadre.js');
  } catch (err) {
    throw new Error(
      `Unable to resolve @serfab/cadre-cli bin from cadre-host. ` +
      `Ensure cadre-cli is built (yarn workspace @serfab/cadre-cli build) ` +
      `and listed as a dependency. Underlying error: ${(err as Error).message}`,
      { cause: err },
    );
  }
}

/**
 * Whether the process a handle names is still the child it was launched as — the
 * one liveness rule behind `init`, `isRunning` and the pre-spawn guard, so they
 * cannot drift apart.
 *
 * A handle this process spawned carries its `ChildProcess`, and that answer is
 * exact from the moment of spawn: no exit seen, and the pid still answering. (The
 * pid check covers the moment between the OS ending the process and Node
 * delivering its exit; the pid cannot have been reused in that moment, since the
 * OS holds it until Node has reaped the child.) The startup-token file is no use
 * here — a child writes it only once it is running its own code, so reading it
 * would report every child dead for its first moments.
 *
 * A handle re-attached from `state.json` after a host restart has no
 * `ChildProcess`, so it falls back to the pid plus that token file, which tells our
 * child apart from an unrelated process that inherited its pid.
 */
function isHandleLive(handle: Handle): boolean {
  const { child } = handle;
  if (child) {
    return child.exitCode === null && child.signalCode === null && isPidAlive(handle.pid);
  }
  return isPidAlive(handle.pid) && tokenMatches(handle.workdir, handle.startupToken);
}

function tokenMatches(workdir: string, expected: string): boolean {
  const tokenPath = join(workdir, STARTUP_TOKEN_FILE);
  try {
    const contents = readFileSync(tokenPath, 'utf8').trim();
    return contents === expected;
  } catch {
    return false;
  }
}

/**
 * Rotate the on-disk log files for a workdir if the active file is already
 * at or above the byte threshold. Called from `createContainer` before the
 * child's stdio fd is opened — the only safe rotation point because the
 * caller guarantees no live child is attached to this workdir.
 */
function maybeRotateAtSpawn(logPath: string, maxBytes: number, maxFiles: number): void {
  if (maxBytes <= 0 || maxFiles < 1) return;
  let size: number;
  try {
    size = statSync(logPath).size;
  } catch {
    return;
  }
  if (size >= maxBytes) {
    rotateOnDisk(logPath, maxFiles);
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((res) => {
    const t = setTimeout(res, ms);
    if (typeof t.unref === 'function') t.unref();
  });
}

/**
 * Read the last `n` lines from a log file. Reads from the tail in chunks
 * rather than slurping the whole file.
 */
function tailFile(path: string, n: number): string {
  if (!existsSync(path)) return '';
  const fd = openSync(path, 'r');
  try {
    const size = statSync(path).size;
    const chunkSize = 8192;
    let pos = size;
    const buffers: Buffer[] = [];
    let newlineCount = 0;
    while (pos > 0 && newlineCount <= n) {
      const readLen = Math.min(chunkSize, pos);
      pos -= readLen;
      const buf = Buffer.alloc(readLen);
      readSync(fd, buf, 0, readLen, pos);
      buffers.unshift(buf);
      for (let i = 0; i < buf.length; i++) {
        if (buf[i] === 0x0a) newlineCount++;
      }
    }
    const joined = Buffer.concat(buffers).toString('utf8');
    const lines = joined.split('\n');
    // Drop any trailing empty line from a terminating newline before counting.
    const trimmed = lines[lines.length - 1] === '' ? lines.slice(0, -1) : lines;
    return trimmed.slice(-n).join('\n');
  } finally {
    closeSync(fd);
  }
}

function parseMemoryLimit(limit?: string): number | undefined {
  if (!limit) return undefined;
  const match = limit.match(/^(\d+(?:\.\d+)?)\s*(B|K|M|G|T)?$/i);
  if (!match) return undefined;
  const [, num, unit] = match;
  const multipliers: Record<string, number> = { B: 1, K: 1024, M: 1024 ** 2, G: 1024 ** 3, T: 1024 ** 4 };
  return Math.floor(parseFloat(num!) * (multipliers[unit?.toUpperCase() ?? 'B'] ?? 1));
}