// Phase C6 / Phase G — hot spares, warmth and reservation, split out of
// automanager.ts as a pure move. The reconciler drives every configured
// `hotSpares` entry toward its target (provisions short, terminates excess
// coldest first, reaps orphans), keeps each spare's pulls in step with its
// spec, and owns the spare bookkeeping (pending provisions, backoffs,
// prefetch signatures, runtime template images). It reads the fleet through
// the narrow view below; the AutoManager owns the timers that call it and
// delegates its public hot-spare surface here.

import * as uuid from "uuid";
import type * as ManagerPB from "@norskvideo/norsk-api/lib/manager_pb";
import type { NodeId, NorskManager } from "@norskvideo/norsk-manager-sdk";
import type { AutoSettings, HotSpareConfig, HotSpareSpecsDiff, NodeSummary, SpareSpec } from "./automanager";
import type { Clock } from "./clock";
import { debuglog } from "./debuglog";
import type { JobBindings } from "./jobBinding";
import { HeldImage, ImageRef, NodeView, PlacementPool, PlacementTier, warmth } from "./placement";
import { AutoSettingsValidationError, validateAutoSettings } from "./settingsValidation";
import type { BundleJobSpec, NodeInventory as TypedNodeInventory } from "./types";
import { imageRefsFrom, WarmNodeTransport, warmNodeTransport } from "./warmAdapter";

/** How long a spare's failed prefetch waits before the reconciler asks
 *  again. Five minutes: long enough that a registry outage or a stale
 *  credential is not hammered, short enough that a fixed one is picked up
 *  without an operator's nudge. */
export const PREFETCH_RETRY_BACKOFF_MS = 5 * 60 * 1000;

/**
 * How long a pool's spare provisioning waits after a provision fails — the
 * create is refused, or the node stops before it ever runs. Every failure
 * used to reconcile straight back into another create: norsk-e2e
 * 2026-09-15, an AMI still `pending`, 29 RunInstances calls in 77 s. The
 * periodic tick retries once the backoff has passed.
 * @public
 */
export const SPARE_PROVISION_RETRY_BACKOFF_MS = 60 * 1000;

/** What the hot-spare collaborator needs from the AutoManager, handed in
 *  explicitly. `settings` is the live object: `updateHotSpareSpecs` writes
 *  the new list into it, which is what every other reader sees. */
export interface HotSpareDeps {
  settings: AutoSettings;
  clock: Clock;
  norsk: Pick<NorskManager, "createAwsNode" | "createOciNode">;
  bindings: JobBindings;
  /** False before hydration and after close: no reconcile runs then. */
  active: () => boolean;
  nodes: () => Iterable<NodeSummary>;
  nodeOf: (nodeId: NodeId) => NodeSummary | undefined;
  inventoryOf: (nodeId: NodeId) => TypedNodeInventory | undefined;
  poolNameForNode: (nodeId: NodeId) => string;
  /** Record a provision as a `starting` node (and tell the embedder). */
  nodeStarting: (nodeId: NodeId) => void;
  /** Drop a node that never came to exist (a create that failed). */
  deleteNode: (nodeId: NodeId) => void;
  /** Apply a spec edit's tags to the node map now, ahead of the daemon's announcement. */
  mergeNodeTags: (nodeId: NodeId, tags: Record<string, string>) => void;
  /** The AutoManager's terminate path (fires onNodeTerminated once per node). */
  issueTerminate: (nodeId: NodeId) => Promise<void>;
}

export class HotSpares {
  // Spare provisions dispatched but not yet arrived, by nodeId, with the
  // pool each was for — counted toward the target so a slow boot does not
  // provision twice.
  private readonly pendingSpareProvisions: Map<NodeId, { pool: string }> = new Map();
  // Phase G — the transport for prefetch / idle-expiry / tag writes and
  // how a node's held images are read (settings.warmNodes overrides).
  private readonly warm: WarmNodeTransport;
  // Phase G — template images registered at runtime (`setTemplateImages`),
  // overlaying `settings.templateImages`.
  private readonly runtimeTemplateImages: Map<string, string[]> = new Map();
  // Phase G — the prefetch signature last sent to each spare, so a
  // reconcile only re-sends when the spec's image set changed.
  private readonly sparePrefetched: Map<NodeId, string> = new Map();
  // When each pool's last spare provision failed (ms since epoch): the
  // reconciler waits out SPARE_PROVISION_RETRY_BACKOFF_MS before trying.
  private readonly spareProvisionFailedAt: Map<string, number> = new Map();
  // When each node's last prefetch failed: retried after PREFETCH_RETRY_BACKOFF_MS.
  private readonly sparePrefetchFailedAt: Map<NodeId, Date> = new Map();

  constructor(private readonly deps: HotSpareDeps) {
    this.warm = warmNodeTransport(
      deps.norsk as NorskManager,
      (what, detail) =>
        debuglog("Phase G: Manager does not support this call yet — skipped", {
          call: what,
          ...detail,
        }),
      deps.settings.warmNodes
    );
  }

  // ---- bookkeeping the AutoManager's event handlers drive ---------------

  /** A pending spare provision landed (`nodeStarted`). */
  spareLanded(nodeId: NodeId): void {
    this.pendingSpareProvisions.delete(nodeId);
  }

  /** A node stopped. A spare that stops while still pending never ran: its
   *  create failed, and the replacement waits out the backoff rather than
   *  repeating the failure. */
  spareStopped(nodeId: NodeId): void {
    const pendingSpare = this.pendingSpareProvisions.get(nodeId);
    if (pendingSpare) this.spareProvisionFailedAt.set(pendingSpare.pool, this.deps.clock.now().getTime());
    this.pendingSpareProvisions.delete(nodeId);
  }

  /** Hydration: the fleet view is being rebuilt from the initial state. */
  clear(): void {
    this.pendingSpareProvisions.clear();
  }

  /**
   * Phase C5 — is the named node currently tagged as a hot spare?
   * Spares are warm-running nodes provisioned (by the C6 reconciler)
   * with `NodeMetadata.tags["spare"] = "true"`.
   *
   * There is deliberately no per-spare dedication. A `forPolicy` field and a
   * matching `spareBand` node tag used to exist, but placement never read
   * them — they only partitioned the reconciler's own accounting — so a spare
   * "reserved" for one policy was reserved from nobody. Which work prefers or
   * avoids a spare is decided on the JOB, by its onRuntimeFailure.
   * @public
   */
  isHotSpareNode(nodeId: NodeId): boolean {
    return this.deps.nodeOf(nodeId)?.nodeMetadata?.tags?.["spare"] === "true";
  }

  /**
   * A LOCAL SERVER (cluster-provider node) that is a hot spare — marked by
   * an operator through `setLocalSpare`, not provisioned by the reconciler
   * (review 2026-09-16, e2e-14). For placement it is a spare like any
   * other: prefer / avoid / reservation / warmth read the tag and the
   * running-jobs list, and consumption is the daemon dropping the tag.
   * For the RECONCILER it is nobody's: never counted toward an elastic
   * entry's target, never reaped as excess or as an orphan, never replaced
   * when consumed (there is nothing to provision), never re-specced by a
   * `hotSpares` edit. Every reconciler read goes through
   * `spareNodesMatching` / `reapOrphanedSpares`, which skip these.
   * @public
   */
  isLocalSpare(nodeId: NodeId): boolean {
    return this.isHotSpareNode(nodeId) && this.isLocalServer(nodeId);
  }

  private isLocalServer(nodeId: NodeId): boolean {
    return this.deps.nodeOf(nodeId)?.nodeMetadata?.providerMetadata?.provider === "cluster";
  }

  // ---- Phase G — warmth, reservation, pre-warming -------------------

  /**
   * Register (or replace) the container images a product template needs.
   * Overlays `settings.templateImages`; the embedder that reads the
   * template manifest calls this when a template is added or rebuilt.
   * @public
   */
  setTemplateImages(templateName: string, images: string[]): void {
    this.runtimeTemplateImages.set(templateName, [...images]);
    // A spare warmed for this template may now have more to pull.
    this.reconcile();
  }

  /**
   * The images a product template needs — runtime registration first, then
   * settings. Unknown ⇒ empty, which means "no warmth signal" rather than
   * "needs nothing": `warmth` scores 0 for every node, so placement is
   * unbiased rather than wrong.
   * @public
   */
  templateImages(templateName: string): string[] {
    return (
      this.runtimeTemplateImages.get(templateName) ??
      this.deps.settings.templateImages?.[templateName] ??
      []
    );
  }

  /** The template a placed bundle job launches, when it launches one. */
  /**
   * The name warmth and reservation key a job's template by: the product
   * TEMPLATE's registered name (`cmaf-cdn`), which is what `warmFor` and
   * `reservedFor` name and what the operator registers image sets under.
   * Keying by product (`norsk-studio`) — as this did until 2026-09-15 —
   * made every template of a product one "template": a reservation could
   * not tell cmaf-cdn from studio-dev, and a warm list pulled the union of
   * every template's images (task 90, fix-list 45). A ref written before
   * the name existed carries none and keys by product, as before.
   */
  templateNameFor(spec: BundleJobSpec | undefined): string | undefined {
    if (spec?.launch?.kind !== "productTemplate") return undefined;
    const ref = spec.launch.ref;
    return ref.templateName && ref.templateName.length > 0 ? ref.templateName : ref.productName;
  }

  /** The image set placement should score warmth against, for one job. */
  jobImagesFor(spec: BundleJobSpec | undefined): ImageRef[] {
    const name = this.templateNameFor(spec);
    return name === undefined ? [] : imageRefsFrom(this.templateImages(name));
  }

  /** What a node advertises holding, through the G1 adapter. */
  heldImagesFor(nodeId: NodeId): HeldImage[] {
    return this.warm.heldImages(nodeId, this.deps.inventoryOf(nodeId));
  }

  /**
   * A spare's live spec, read from its node tags: what it was provisioned
   * with, as edited since through `updateHotSpareNode`. The tags are the
   * only copy — the daemon persists an edit and re-applies it when it
   * rediscovers the node, so the spec survives a reconnect and a restart
   * on either side (task 90 fix-list 46: an in-memory override here was
   * lost on every norsk-mgr restart). A node that is not a spare has no
   * spec.
   * @public
   */
  spareSpec(nodeId: NodeId): SpareSpec | undefined {
    if (!this.isHotSpareNode(nodeId)) return undefined;
    return spareSpecFromTags(this.deps.nodeOf(nodeId)?.nodeMetadata?.tags ?? {});
  }

  /** The reservation half of a node's placement view. */
  reservationViewFor(
    nodeId: NodeId
  ): Pick<NodeView, "reservedFor" | "exclusiveReservation"> {
    const spec = this.spareSpec(nodeId);
    if (!spec) return {};
    return {
      ...(spec.reservedFor.length > 0 ? { reservedFor: spec.reservedFor } : {}),
      ...(spec.exclusive ? { exclusiveReservation: true } : {}),
    };
  }

  /**
   * Edit a running spare's spec. Adding to `warmFor` starts the pulls
   * (Steve's case: two spares warmed for `xyz`, a migration of `abc`
   * needed, so `abc` is added and the pull happens while the migration is
   * still proposing). Removing from it does NOT delete images — it only
   * stops us claiming the node is kept warm for that template; the images
   * stay until the node's own eviction reclaims them.
   *
   * Omitted fields keep their current value. Throws if the node is not a
   * hot spare.
   * @public
   */
  async updateHotSpareNode(
    nodeId: NodeId,
    edit: { warmFor?: string[]; reservedFor?: string[]; exclusive?: boolean }
  ): Promise<void> {
    const current = this.spareSpec(nodeId);
    if (!current) {
      throw new Error(
        `updateHotSpareNode: node ${nodeId} is not a hot spare — nothing to edit`
      );
    }
    const next: SpareSpec = {
      warmFor: edit.warmFor ?? current.warmFor,
      reservedFor: edit.reservedFor ?? current.reservedFor,
      exclusive: edit.exclusive ?? current.exclusive,
    };
    const tags = spareSpecTags(next);
    // The daemon owns the edit from here: it persists the tags, pushes
    // them to the provider and announces the node's new metadata.
    await this.warm.setNodeTags(nodeId, tags);
    // Apply locally now rather than waiting for that announcement, so a
    // read straight after the edit sees it (and an older Manager without
    // the RPC still honours the edit for as long as this process lives).
    this.deps.mergeNodeTags(nodeId, tags);
    debuglog("Hot-spare spec edited", { nodeId, from: current, to: next });
    await this.ensurePrefetched(nodeId, next.warmFor);
  }

  /**
   * Replace the live `hotSpares` list without a reconnect. The embedder
   * used to reload settings for this, which tore the AutoManager down and
   * — since `close()` aborts every in-flight migration — killed whatever
   * was migrating (the IBC-demo landmine). Now: the list is validated as
   * it would be at boot (the `hotSpares` slice of `validateAutoSettings`,
   * plus one pool per entry, since the pool is the key) and rejected whole
   * with the codes; otherwise it becomes the live list and the reconciler
   * converges on it on its next tick, which runs at once. A removed entry's
   * unconsumed spares are reaped as orphans; an added one is provisioned; a
   * lowered target terminates the excess coldest first; a changed
   * `warmFor` / `reservedFor` / `exclusive` is written to the live spares'
   * tags (only the fields that changed, so a per-node edit of the others
   * survives), and the pulls follow. Nothing else is touched — a migration
   * in flight stays in flight.
   * @public
   */
  async updateHotSpareSpecs(specs: HotSpareConfig[]): Promise<HotSpareSpecsDiff> {
    const errors = validateAutoSettings({ ...this.deps.settings, hotSpares: specs }).filter(
      (e) => e.path === "hotSpares" || e.path.startsWith("hotSpares[")
    );
    if (errors.length > 0) throw new AutoSettingsValidationError(errors);

    const before = new Map((this.deps.settings.hotSpares ?? []).map((e) => [e.pool, e]));
    const after = new Map(specs.map((e) => [e.pool, e]));
    const diff: HotSpareSpecsDiff = {
      added: [...after.keys()].filter((k) => !before.has(k)),
      removed: [...before.keys()].filter((k) => !after.has(k)),
      changed: [...after.keys()].filter((k) => {
        const prev = before.get(k);
        return prev !== undefined && !sameHotSpareConfig(prev, after.get(k) as HotSpareConfig);
      }),
    };
    this.deps.settings.hotSpares = specs.map((e) => ({ ...e, spec: { ...e.spec } }));
    debuglog("Hot-spare specs updated", diff);

    // A spec edit reaches the spares that already exist; the count and the
    // shape of future provisions are the reconciler's.
    for (const pool of diff.changed) {
      const prev = before.get(pool) as HotSpareConfig;
      const next = after.get(pool) as HotSpareConfig;
      const edit: { warmFor?: string[]; reservedFor?: string[]; exclusive?: boolean } = {
        ...(sameList(prev.warmFor, next.warmFor) ? {} : { warmFor: dedupe(next.warmFor ?? []) }),
        ...(sameList(prev.reservedFor, next.reservedFor) ? {} : { reservedFor: dedupe(next.reservedFor ?? []) }),
        ...((prev.exclusive ?? false) === (next.exclusive ?? false) ? {} : { exclusive: next.exclusive ?? false }),
      };
      if (Object.keys(edit).length === 0) continue;
      for (const node of this.spareNodesMatching(next)) {
        try {
          await this.updateHotSpareNode(node.nodeId, edit);
        } catch (err) {
          this.deps.settings.onError?.(err);
        }
      }
    }
    this.reconcile();
    return diff;
  }

  /**
   * Mark a LOCAL SERVER as a hot spare, or clear it (review 2026-09-16,
   * e2e-14: "no reason why we wouldn't support a local server as a hot
   * spare"). The daemon owns the tag: it is asked through `setNodeTags`
   * with the same tag set an elastic spare is provisioned with (`spare`,
   * `warmFor`, `reservedFor`, `exclusive`), persists it, announces it, and
   * drops `spare` the moment a job binds — so from then on the node is a
   * spare exactly as an elastic one is, to placement and to norsk-mgr.
   *
   * Refused here, before the wire, for a node that is not a local server
   * (an elastic spare is the reconciler's to make) and for a node that
   * holds a job (a spare is empty; the daemon refuses the same, and the
   * local check also covers a placement this AutoManager has made that no
   * report has confirmed yet). `warmFor` / `reservedFor` / `exclusive`
   * omitted on a mark keep whatever the node's tags say; on a clear they
   * are left alone. The view is updated at once rather than waiting for
   * the daemon's announcement, and a mark starts the pulls for `warmFor`.
   * @public
   */
  async setLocalSpare(
    nodeId: NodeId,
    edit: { spare: boolean; warmFor?: string[]; reservedFor?: string[]; exclusive?: boolean }
  ): Promise<void> {
    const node = this.deps.nodeOf(nodeId);
    if (node === undefined) throw new Error(`setLocalSpare: unknown node ${nodeId}`);
    if (!this.isLocalServer(nodeId)) {
      throw new Error(
        `setLocalSpare: node ${nodeId} is not a local server — an elastic spare is the hot-spare reconciler's to provision`
      );
    }
    if (edit.spare && this.deps.bindings.anyOn(nodeId)) {
      throw new Error(`setLocalSpare: node ${nodeId} holds a job — a spare must be empty`);
    }
    const tags: Record<string, string> = {
      spare: edit.spare ? "true" : "false",
      ...(edit.warmFor !== undefined ? { warmFor: joinTagList(edit.warmFor) } : {}),
      ...(edit.reservedFor !== undefined ? { reservedFor: joinTagList(edit.reservedFor) } : {}),
      ...(edit.exclusive !== undefined ? { exclusive: edit.exclusive ? "true" : "false" } : {}),
    };
    await this.warm.setNodeTags(nodeId, tags);
    this.deps.mergeNodeTags(nodeId, tags);
    debuglog(edit.spare ? "Local server marked as a hot spare" : "Local server is a hot spare no longer", {
      nodeId,
      tags,
    });
    if (edit.spare) {
      const spec = this.spareSpec(nodeId);
      if (spec && spec.warmFor.length > 0) await this.ensurePrefetched(nodeId, spec.warmFor);
    }
  }

  /**
   * Override how long this node lingers idle before the daemon expires it
   * (fix-list item 8: the node a handover vacated was destroyed before you
   * could migrate back onto it). `0` restores the global `emptyNodeLinger`.
   * @public
   */
  async setNodeIdleExpiry(nodeId: NodeId, ms: number): Promise<void> {
    if (!Number.isFinite(ms) || ms < 0) {
      throw new Error(`setNodeIdleExpiry: ms must be a non-negative number, got ${ms}`);
    }
    debuglog("Per-node idle expiry override", { nodeId, ms });
    await this.warm.setNodeIdleExpiry(nodeId, ms);
  }

  /**
   * A prefetch batch settled. Warmth itself comes from the node's
   * re-advertised `heldImages` (which follows this event), so nothing is
   * driven off it — it is logged, and handed to the operator, because a
   * ref that would not pull is exactly what somebody needs to be told:
   * the spare will stay cold for that template until it is fixed.
   */
  onImagesPrefetched(raw: ManagerPB.ImagesPrefetchedEvent): void {
    const nodeId = raw.nodeId?.nodeId ?? "";
    const results = raw.results.map((r) => ({
      imageRef: r.imageRef,
      digest: r.digest,
      ok: r.ok,
      error: r.error,
    }));
    const failed = results.filter((r) => !r.ok);
    debuglog("Images prefetched", { nodeId, ok: results.length - failed.length, failed });
    if (failed.length > 0) {
      // Let a later reconcile try the whole set again — after the backoff.
      this.sparePrefetched.delete(nodeId);
      this.sparePrefetchFailedAt.set(nodeId, this.deps.clock.now());
    } else {
      this.sparePrefetchFailedAt.delete(nodeId);
    }
    this.deps.settings.onImagesPrefetched?.(nodeId, results);
  }

  /**
   * Send the node a prefetch for the images of `warmFor`'s templates, if
   * that set has changed since the last one we sent it. Idempotent, so
   * every reconcile can call it.
   */
  /**
   * Pull the images of some product templates onto ANY node, spare or not
   * (task 90, fix-list 39).
   *
   * A spare's `warmFor` edit does this and also records the declaration;
   * an ordinary node — the one a handover just vacated, which is the fastest
   * target the next handover could have — has no spec to edit, but the pull
   * is the same pull, and warmth is OBSERVED (`heldImages`) so placement
   * scores it the moment the images land. Nothing is declared or reserved.
   * Resolves with the image refs asked for; empty when no template resolves
   * to any image (no warmth signal exists for it yet), which is not an
   * error. Rejects for a node this AutoManager does not know.
   * @public
   */
  async prefetchTemplates(nodeId: NodeId, templateNames: string[]): Promise<string[]> {
    if (this.deps.nodeOf(nodeId) === undefined) {
      throw new Error(`prefetchTemplates: unknown node ${nodeId}`);
    }
    const refs = dedupe(templateNames.flatMap((t) => this.templateImages(t)));
    if (refs.length === 0) return refs;
    debuglog("Prefetching images onto a node", { nodeId, templateNames, refs });
    await this.warm.prefetchImages(nodeId, refs);
    return refs;
  }

  private async ensurePrefetched(nodeId: NodeId, warmFor: string[]): Promise<void> {
    const refs = dedupe(warmFor.flatMap((t) => this.templateImages(t)));
    const signature = refs.join("\u0000");
    if (this.sparePrefetched.get(nodeId) === signature) return;
    const failedAt = this.sparePrefetchFailedAt.get(nodeId);
    if (failedAt !== undefined && this.deps.clock.now().getTime() - failedAt.getTime() < PREFETCH_RETRY_BACKOFF_MS) {
      return;
    }
    this.sparePrefetched.set(nodeId, signature);
    if (refs.length === 0) return;
    debuglog("Prefetching images onto a spare", { nodeId, warmFor, refs });
    try {
      await this.warm.prefetchImages(nodeId, refs);
    } catch (err) {
      // A failed pull is not fatal: the node is merely still cold, and the
      // next reconcile will try again.
      this.sparePrefetched.delete(nodeId);
      this.deps.settings.onError?.(err);
    }
  }

  // ---- Phase C6 — hot-spare reconciliation -------------------------

  /**
   * Drive every configured `hotSpares` entry toward its `targetCount`.
   * Counts live + pending spares matching the pool; provisions
   * short, terminates excess (oldest live first). Cluster pools are
   * non-elastic — under-target just fires `onSpareUnderflow` and
   * leaves provisioning to the operator. Idempotent and cheap; safe
   * to call on every lifecycle event plus the periodic tick.
   */
  reconcile(): void {
    if (!this.deps.active()) return;
    const entries = this.deps.settings.hotSpares ?? [];
    this.reapOrphanedSpares(entries);
    if (entries.length === 0) return;

    for (const cfg of entries) {
      const pool = (this.deps.settings.placementPools ?? []).find(
        (p) => p.name === cfg.pool
      );
      if (!pool) {
        debuglog("hotSpares entry references unknown pool", { entry: cfg });
        continue;
      }

      const matching = this.spareNodesMatching(cfg);
      const pending = this.pendingSparesMatching(cfg);
      const effective = matching.length + pending;
      const target = cfg.targetCount;

      // Phase G — keep every live spare's pulls in step with its spec.
      // Idempotent (`ensurePrefetched` compares a signature), so this
      // covers the provision that has just arrived, an edited entry, and a
      // template whose image set was registered after the spare came up.
      for (const node of matching) {
        const spec = this.spareSpec(node.nodeId);
        if (!spec || spec.warmFor.length === 0) continue;
        void this.ensurePrefetched(node.nodeId, spec.warmFor).catch((err) =>
          this.deps.settings.onError?.(err)
        );
      }

      if (effective < target) {
        const short = target - effective;
        this.deps.settings.onSpareUnderflow?.(cfg.pool, effective, target);
        const failedAt = this.spareProvisionFailedAt.get(cfg.pool);
        if (failedAt !== undefined) {
          const waited = this.deps.clock.now().getTime() - failedAt;
          if (waited < SPARE_PROVISION_RETRY_BACKOFF_MS) {
            debuglog("Under spare target, but the last provision failed recently — waiting", {
              pool: cfg.pool,
              effective,
              target,
              retryInMs: SPARE_PROVISION_RETRY_BACKOFF_MS - waited,
            });
            continue;
          }
        }
        // Warm the cheapest provisionable tier (first elastic tier in
        // cost order). A pool with no elastic tier can't grow spares.
        const provisionTier = pool.tiers.find((t) => t.scaleOut === "elastic");
        if (!provisionTier) {
          debuglog("Pool has no elastic tier — under spare target, no provision", {
            cfg,
            effective,
            target,
          });
          continue;
        }
        for (let i = 0; i < short; i++) {
          void this.provisionHotSpare(pool, provisionTier, cfg);
        }
      } else if (matching.length > target) {
        // Only terminate from the truly-live (non-pending) set.
        //
        // Phase G (fix-list item 10) — COLDEST first. This used to key on
        // `lastActivity` alone, "to support rotation for security
        // patching", and so reliably destroyed the node that had most
        // recently run the work — the warmest one in the pool, the one
        // worth keeping. Warmth against the pool's own `warmFor` templates
        // is the key now; a spare that declares nothing is scored by how
        // little it advertises holding, so an empty node still goes before
        // a laden one. `lastActivity` survives as the TIE-BREAK, which is
        // where rotation belongs: among equally cold nodes, rotate.
        const excess = matching.length - target;
        const coldness = new Map<NodeId, number>(
          matching.map((n) => [n.nodeId, this.spareWarmth(n.nodeId, cfg)])
        );
        const sorted = [...matching].sort((a, b) => {
          const wa = coldness.get(a.nodeId) ?? 0;
          const wb = coldness.get(b.nodeId) ?? 0;
          if (wa !== wb) return wa - wb;
          return a.lastActivity.getTime() - b.lastActivity.getTime();
        });
        for (let i = 0; i < excess; i++) {
          const victim = sorted[i];
          debuglog("Hot-spare excess — terminating the coldest", {
            nodeId: victim.nodeId,
            warmth: coldness.get(victim.nodeId),
            cfg,
          });
          void this.deps.issueTerminate(victim.nodeId).catch((err) => this.deps.settings.onError?.(err));
        }
      }
    }
  }

  /**
   * How warm a spare is for the work its pool exists to serve: the overlap
   * between what it advertises holding and the union of the images of the
   * pool entry's `warmFor` templates.
   *
   * When the entry declares no `warmFor`, there is no intended image set to
   * measure against, so the fallback is how much the node holds at all,
   * normalised into (0,1) so it can never outrank a genuine warmth score.
   * That keeps "coldest first" meaningful for an undeclared pool: an empty
   * node still goes before a laden one.
   */
  private spareWarmth(nodeId: NodeId, cfg: HotSpareConfig): number {
    const held = this.heldImagesFor(nodeId);
    const warmFor = dedupe(cfg.warmFor ?? []);
    const images = imageRefsFrom(dedupe(warmFor.flatMap((t) => this.templateImages(t))));
    if (images.length === 0) {
      // 0 images ⇒ 0; grows toward (but never reaches) 1 with the count.
      return held.length / (held.length + 1);
    }
    const view: NodeView = {
      nodeId,
      poolName: cfg.pool,
      inventory: {} as TypedNodeInventory,
      runningJobs: [],
      isHotSpare: true,
      heldImages: held,
    };
    return warmth(view, images);
  }

  /**
   * A spare whose pool no longer has a reserve entry (the entry was deleted,
   * or every entry was) has a target of zero and nobody else to reap it:
   * the daemon's empty-node sweep leaves a `spare`-tagged node alone by
   * design, and the per-entry loop below never looks at a pool it has no
   * entry for. Observed on norsk-e2e 2026-09-14: two spares outlived their
   * reserve indefinitely after DELETE /api/hot-spares. Terminate them.
   */
  private reapOrphanedSpares(entries: HotSpareConfig[]): void {
    const declared = new Set(entries.map((e) => e.pool));
    for (const node of this.deps.nodes()) {
      if (!this.isHotSpareNode(node.nodeId)) continue;
      // A local spare is the operator's, not a provision of ours to reap.
      if (this.isLocalServer(node.nodeId)) continue;
      if (node.nodeState !== "running") continue;
      if (declared.has(this.deps.poolNameForNode(node.nodeId))) continue;
      if (this.deps.bindings.anyOn(node.nodeId)) continue;
      debuglog("Spare has no reserve entry for its pool — terminating", {
        nodeId: node.nodeId,
        pool: this.deps.poolNameForNode(node.nodeId),
      });
      void this.deps.issueTerminate(node.nodeId).catch((err) => this.deps.settings.onError?.(err));
    }
  }

  /**
   * The ELASTIC spares an entry governs: tagged, running, in its pool. A
   * local spare in the same pool is not one of them — it is not counted
   * toward the target (so it never displaces a provision), not excess (so
   * it is never terminated), and not re-specced by an entry edit.
   */
  private spareNodesMatching(cfg: HotSpareConfig): NodeSummary[] {
    const out: NodeSummary[] = [];
    for (const node of this.deps.nodes()) {
      if (!this.isHotSpareNode(node.nodeId)) continue;
      if (this.isLocalServer(node.nodeId)) continue;
      if (node.nodeState !== "running") continue;
      if (this.deps.poolNameForNode(node.nodeId) !== cfg.pool) continue;
      // A consumed spare is no longer a spare: the daemon drops the `spare`
      // tag when a job binds (isHotSpareNode above), so nothing else is
      // needed to keep it out of the count.
      out.push(node);
    }
    return out;
  }

  private pendingSparesMatching(cfg: HotSpareConfig): number {
    let n = 0;
    for (const p of this.pendingSpareProvisions.values()) {
      if (p.pool !== cfg.pool) continue;
      n++;
    }
    return n;
  }

  private async provisionHotSpare(
    pool: PlacementPool,
    tier: PlacementTier,
    cfg: HotSpareConfig
  ): Promise<void> {
    const instanceType = cfg.spec.instanceType;
    if (!instanceType) {
      this.deps.settings.onError?.(
        new Error(
          `hotSpares entry for elastic pool '${cfg.pool}' needs spec.instanceType`
        )
      );
      return;
    }

    const nodeId = uuid.v4() + ".spare." + cfg.pool;
    // Phase G — the spare's spec rides on its node tags, so a replacement
    // provisioned for a consumed spare carries the SAME guarantee (the
    // entry is the same `cfg`), and a reconnect can read the spec back off
    // the node rather than losing it.
    const warmForTag = joinTagList(cfg.warmFor);
    const reservedForTag = joinTagList(cfg.reservedFor);
    const tags: Record<string, string> = {
      "auto-manager": "true",
      spare: "true",
      pool: cfg.pool,
      tier: tier.name,
      ...(warmForTag.length > 0 ? { warmFor: warmForTag } : {}),
      ...(reservedForTag.length > 0 ? { reservedFor: reservedForTag } : {}),
      ...(cfg.exclusive ? { exclusive: "true" } : {}),
    };

    this.pendingSpareProvisions.set(nodeId, { pool: cfg.pool });
    this.deps.nodeStarting(nodeId);
    this.deps.settings.onNodeStarting?.(nodeId);

    try {
      if (tier.kind === "aws") {
        await this.deps.norsk.createAwsNode({
          nodeId,
          region: cfg.spec.region ?? tier.region ?? "",
          instanceType,
          tags,
          writeFiles: [],
          ...(tier.rootVolume ? { rootVolume: tier.rootVolume } : {}),
        });
      } else if (tier.kind === "oci") {
        await this.deps.norsk.createOciNode({
          nodeId,
          availabilityDomain: cfg.spec.az ?? cfg.spec.region ?? tier.region ?? "",
          architecture: "x86_64",
          shape: instanceType,
          subnet: "",
          tags,
          writeFiles: [],
        });
      }
      // Cluster path falls through — non-elastic, no auto-provision.
    } catch (err) {
      // Failed to even submit. Drop the pending marker so the next
      // reconcile can try again instead of waiting forever — after the
      // backoff. And drop the node: it was inserted as `starting` above
      // (and onNodeStarting fired for it), but no instance exists, and
      // norsk-mgr serves nodes() verbatim — the same clean-up
      // abandonPendingProvision does for a job's provision.
      this.pendingSpareProvisions.delete(nodeId);
      this.spareProvisionFailedAt.set(cfg.pool, this.deps.clock.now().getTime());
      this.deps.deleteNode(nodeId);
      this.deps.settings.onNodeStopped?.(nodeId);
      this.deps.settings.onError?.(err);
    }
  }

}

// ---- Phase G helpers -------------------------------------------------

/** The spec a spare's tags describe (the provision-time tags, as edited). */
function spareSpecFromTags(tags: Record<string, string>): SpareSpec {
  return {
    warmFor: splitTagList(tags["warmFor"]),
    reservedFor: splitTagList(tags["reservedFor"]),
    exclusive: tags["exclusive"] === "true",
  };
}

/** The tags that describe a spec — the inverse of `spareSpecFromTags`,
 *  and the same shape a spare is provisioned with. Every key is written,
 *  an emptied list as "", so an edit that clears a field clears the tag. */
function spareSpecTags(spec: SpareSpec): Record<string, string> {
  return {
    warmFor: joinTagList(spec.warmFor),
    reservedFor: joinTagList(spec.reservedFor),
    exclusive: spec.exclusive ? "true" : "false",
  };
}

function splitTagList(v: string | undefined): string[] {
  if (!v) return [];
  return v
    .split(",")
    .map((s) => s.trim())
    .filter((s) => s.length > 0);
}

/** The node-tag form of a `warmFor` / `reservedFor` list. */
function joinTagList(v: string[] | undefined): string {
  return dedupe(v ?? []).join(",");
}

/** Order-preserving dedupe — a prefetch signature has to be stable. */
function dedupe(v: string[]): string[] {
  return [...new Set(v)];
}

/** Two template-name lists name the same set (order and repeats aside —
 *  that is how the node tag carries them). */
function sameList(a: string[] | undefined, b: string[] | undefined): boolean {
  const x = dedupe(a ?? []).sort();
  const y = dedupe(b ?? []).sort();
  return x.length === y.length && x.every((v, i) => v === y[i]);
}

/** Would these two entries have the reconciler do the same thing? */
function sameHotSpareConfig(a: HotSpareConfig, b: HotSpareConfig): boolean {
  return (
    a.pool === b.pool &&
    a.targetCount === b.targetCount &&
    (a.spec.instanceType ?? "") === (b.spec.instanceType ?? "") &&
    (a.spec.region ?? "") === (b.spec.region ?? "") &&
    (a.spec.az ?? "") === (b.spec.az ?? "") &&
    sameList(a.warmFor, b.warmFor) &&
    sameList(a.reservedFor, b.reservedFor) &&
    (a.exclusive ?? false) === (b.exclusive ?? false)
  );
}