import * as uuid from "uuid";
import * as ManagerPB from "@norskvideo/norsk-api/lib/manager_pb";
import * as CommonPB from "@norskvideo/norsk-api/lib/shared/common_pb";
import { CancellableTimer, Clock, RealClock } from "./clock";
import {
  Bundle as TypedBundle,
  BundleId,
  BundleJobSpec,
  JobRequirements,
  NodeInventory as TypedNodeInventory,
  NodeService,
  OnRuntimeFailure,
  replicaCount,
  ServiceId,
} from "./types";
import {
  NodeProfile,
  PoolDescriptor,
  ValidationError,
  ValidationResult,
  validateBundle as validateBundleFn,
} from "./validation";
import {
  AutoSettingsValidationError,
  validateAutoSettings,
} from "./settingsValidation";
import {
  fromPbBundle,
  fromPbNodeInventory,
  InventoryContractViolation,
  STOP_DATE_TIME_TAG,
  toPbBundle,
} from "./conversions";
import {
  defaultPlacementLeadMs,
  nodeMatchesTier,
  NodeView,
  PlacementFailureReason,
  PlacementPool,
  PlacementResult,
  PlacementTrace,
  placeWithTrace,
  RecoveryContext,
  ResilienceFlag,
  InstanceTypeOption,
  provisionTimeoutMs,
  RunningJob,
  tierReliability,
} from "./placement";
import { deriveReservation, overlayReservation, PlacedClaim } from "./inventory";
import { WarmNodeTransport } from "./warmAdapter";
import { JobBindings } from "./jobBinding";
import { debuglog } from "./debuglog";
import { NodeServiceReconciler } from "./nodeServices";
import { HotSpares } from "./hotSpares";
import { MigrationPlacement, MigrationSaga, roleFromReplicaIndex } from "./migrationSaga";
export { migrationTargetRole, newestHistoryEntryForRole, readApplySummary } from "./migrationSaga";
import {
  NorskManager,
  eventStream,
  JobId,
  JobInfo,
  JobWithHistory,
  NodeId,
  NodeMetadata,
  NodeStopping,
  Role,
  RunningNodeMetadata,
} from "@norskvideo/norsk-manager-sdk";
import type * as ManagerSdk from "@norskvideo/norsk-manager-sdk";

////////////////////////////////////////////////////////////////////////////////
// AutoManager — bundle-driven scheduler. The legacy AutoJob / region+
// instanceType API is gone; everything goes through createBundle and the
// placement engine. See .ai/steve/tasks/37-auto-manager-design.md.

export type NodeState =
  | "starting"
  | "running"
  | "stopping"
  | "stopped"
  | "terminating"
  | "terminated";

export type NodeSummary = {
  nodeId: NodeId;
  nodeMetadata?: NodeMetadata;
  runningNodeMetadata?: RunningNodeMetadata;
  nodeState: NodeState;
  lastActivity: Date;
  // Captured when AutoManager kicks off node creation. Lets us recover
  // jobId / role for nodeStopping events that fire before the node has
  // reported back nodeMetadata (and so before tags["jobId"] / tags["role"]
  // are available).
  provisional?: { jobId: JobId; role: Role };
};

export interface AutoSettings {
  url?: string;
  /**
   * Abort the connection / run. Aborting before `run()` connects rejects it
   * (so a connect-with-timeout retry loop can give up cleanly without
   * leaving an abandoned SDK connection alive); aborting a running
   * AutoManager tears down its Manager connection, ending the run loop.
   */
  signal?: AbortSignal;
  /** Seconds-from-now window for jobs to be considered pending in the eventStream. */
  pendingWindow: number;
  /** Seconds stopped nodes persist before being terminated. */
  removeStoppedNodesAfter: number;
  /** Configured pools; a bundle names one, placement walks its tiers. */
  placementPools: PlacementPool[];
  /**
   * Window (ms) for counting jobRejected events per-node. Default 5_000.
   * Once `rejectionEscalationCount` rejections fall inside this window the
   * node is marked "unhealthy" and excluded from placement until either a
   * job reaches `running` on it or `unhealthyCooldownMs` elapses. (Routine
   * inventory heartbeats deliberately do NOT clear the mark.)
   */
  rejectionBackoffMs?: number;
  /** Default 3. */
  rejectionEscalationCount?: number;
  /**
   * How long (ms) an "unhealthy" mark excludes a node from placement
   * before it is allowed to try again. Default 600_000 (10 min). A job
   * reaching `running` on the node clears the mark early.
   */
  unhealthyCooldownMs?: number;
  /**
   * After a node fails (`nodeStopping`), how long to keep it excluded from
   * placement. Default 30_000 ms. Expires on read; a job reaching
   * `running` on the node clears it early.
   */
  failureBackoffMs?: number;
  /**
   * Phase C2: per-replica restart-rate ceiling on the recovery path.
   * When a replica's restart count in the trailing hour exceeds this,
   * AutoManager stops re-placing it (and every other replica of the
   * same bundle), marks the bundle broken, and fires `onBundleBroken`.
   * Cleared when the bundle is updated or deleted, or when a fresh
   * `jobUpdated` arrives for the broken job. Default 6 (per design).
   */
  maxRestartsPerHour?: number;
  /**
   * Phase C3: node startup grace. If a node stays in `nodeStarting`
   * for longer than this, AutoManager terminates it — Manager emits a
   * `nodeStopping` and the existing policy-aware recovery path replaces
   * the job. Default 120_000 ms.
   *
   * (In v1, "job in starting" and "node in starting" collapse onto
   * the same signal because `running → ready` is auto-marked. The
   * design's distinction matters when explicit readiness lands.)
   */
  jobStartupGraceMs?: number;
  /**
   * Phase C4 — per-policy recovery tunables. When set, takes precedence
   * over the flat fields above. Only the two re-placing policies appear:
   * `drop` never restarts, so a restart ceiling is meaningless for it.
   * Whether a recovering job prefers or avoids hot spares follows from
   * the policy itself, not from config. Omitted entries fall back to the
   * flat top-level fields, then to defaults.
   */
  failurePolicies?: FailurePolicyConfig;
  /**
   * Phase C4 — placement / recovery tunables grouped into one typed
   * object. Same precedence as `failurePolicies`: object field beats flat field
   * beats default. The flat fields above are preserved for one
   * release; new code should prefer this shape.
   */
  placement?: PlacementConfig;
  /**
   * Phase C6 — hot-spare reconciliation configuration. One entry per
   * pool (plus its node spec). The reconciler runs on node lifecycle
   * events and on a periodic tick: counts live spares per entry,
   * provisions short, terminates excess. Empty/omitted = no spares.
   */
  hotSpares?: HotSpareConfig[];
  /**
   * Phase F — NodeService configurations. Each entry describes a
   * long-running service that consumes capabilities (typically
   * exclusive hardware) and advertises new ones for sibling jobs
   * to consume. AutoManager's reconciler places eager services on
   * every eligible node and tears them down when a node loses
   * eligibility.
   *
   * Worker-side capability advertisement is required for sibling
   * jobs to actually see the provided capabilities — see
   * `§Worker-side work` in the design doc.
   */
  nodeServices?: NodeService[];
  /**
   * Feature fences. A non-empty `nodeServices` is refused at boot unless
   * `experimental.nodeServices` is true: what exists is a working F1 (eager
   * placement on every eligible node) with none of the F2+ pieces — network
   * placement, identity across restarts, failure policy, `provides` into
   * inventory — so a yaml that happens to carry `nodeServices` must not
   * place a service on every node by accident. See norsk-mgr
   * `docs/internal/design/node-services.md` §"Meanwhile: fence what exists".
   */
  experimental?: { nodeServices?: boolean };
  /**
   * Phase F — how often the NodeService reconciler's periodic tick
   * fires. Event-driven reconciles handle most cases; the tick is a
   * safety net. Default 60_000 ms.
   */
  nodeServiceReconcileIntervalMs?: number;
  /**
   * Phase C6 — how often the spare reconciler's periodic tick fires.
   * Event-driven reconciles handle most cases; the tick is a safety
   * net for missed signals. Default 60_000 ms.
   */
  hotSpareReconcileIntervalMs?: number;
  /**
   * Phase G — which container images each product template needs, keyed by
   * the template's registered name (`ProductTemplateRef.templateName`; a
   * ref without one keys by `productName`). Placement scores warmth as the
   * overlap between this set and what a node advertises holding, and the
   * spare reconciler prefetches it for a `warmFor` template.
   *
   * Supplied by the embedder because the template tar is opaque to Manager:
   * norsk-ctl reads the images out of the manifest it already parses.
   * Entries may be `repository:tag`, `repository@sha256:...` or both
   * (`repository:tag@sha256:...`) — a digest matches exactly, a tag is the
   * documented fallback when the digest is not known at placement time.
   *
   * `setTemplateImages` adds/replaces one entry at runtime; unknown
   * templates simply have no warmth signal (every node scores 0), which is
   * the pre-Phase-G behaviour.
   */
  templateImages?: Record<string, string[]>;
  /**
   * Phase G — override the G1 transport (`src/warmAdapter.ts`): the image
   * prefetch, the per-node idle-expiry RPC, and how a node's advertised
   * image set is read out of its inventory. Production leaves this unset
   * and gets the duck-typed binding to the Manager SDK; tests and
   * embedders ahead of the wire supply their own.
   */
  warmNodes?: Partial<WarmNodeTransport>;
  /**
   * Phase G — one prefetch batch has settled on a node, one result per
   * requested ref. A `ok: false` entry is worth surfacing: that spare
   * stays cold for whatever template needed the ref.
   */
  onImagesPrefetched?: (
    nodeId: NodeId,
    results: { imageRef: string; digest: string; ok: boolean; error: string }[]
  ) => void;
  /**
   * Phase E — per-phase deadline configuration for in-flight
   * migrations. If a migration sits in a phase longer than the
   * configured window, AutoManager auto-aborts with reason
   * `phase timeout: <phase>`. Operators can still abort manually via
   * `abortMigration` before the deadline fires.
   */
  migration?: MigrationConfig;
  onError?: (err: unknown) => void;
  onNodeStarting?: (node: NodeId) => void;
  onNodeStarted?: (node: NodeSummary) => void;
  /** Fired when the daemon emits a NodeStopping event — the node is
   *  going away but may not have fully terminated yet. For cluster
   *  workers Ctrl-C'd at the keyboard this is the signal that fires
   *  (NodeStopped/NodeTerminated come later, after the daemon's
   *  keepalive timeout or cleanup loop). Consumers wanting prompt
   *  "worker dropped" UI updates listen here. */
  onNodeStopping?: (nodeId: NodeId) => void;
  onNodeStopped?: (nodeId: NodeId) => void;
  /**
   * Fired when THIS AutoManager issues a terminate for a node — from any
   * of its own paths: `terminateNode()` / `evictNode()` / a `drainNode()`
   * that emptied its node, the hot-spare
   * excess and orphan reapers, single-use release, startup-grace expiry
   * and the stopped-node sweep. Exactly once per node, however many of
   * those paths (or retries of one) issue the RPC. It fires when the
   * terminate is issued, not when the daemon confirms it: the daemon's
   * `NodeStopping` / `NodeStopped` follow when it acts. A node terminated
   * by anyone else (an operator going to Manager directly, the daemon's
   * own empty-node sweep) is reported ONLY through `onNodeStopping` /
   * `onNodeStopped` — this is not "a node was terminated" in general.
   */
  onNodeTerminated?: (nodeId: NodeId) => void;
  onBundleUpdated?: (bundle: TypedBundle) => void;
  onBundleDeleted?: (bundleId: BundleId) => void;
  /**
   * Fired when the daemon appends a new audit-log entry (any source).
   * Pure pass-through of the SDK's `AuditLogEntry` — AutoManager doesn't
   * own audit semantics, it just forwards the live append so consumers
   * can tail the log without re-polling GetAuditLog.
   */
  onAuditAppended?: (entry: ManagerSdk.AuditLogEntry) => void;
  /**
   * Fires when a node accumulates enough rejections within
   * `rejectionBackoffMs` to be marked unhealthy. Operators may want to
   * surface this in monitoring.
   */
  onNodeUnhealthy?: (nodeId: NodeId) => void;
  /**
   * Fires when a previously-unhealthy node's exclusion is cleared —
   * either a job reached `running` on it, or its mark aged past
   * `unhealthyCooldownMs`. Lets a consumer mirroring the unhealthy set
   * keep its count accurate.
   */
  onNodeHealthy?: (nodeId: NodeId) => void;
  /**
   * Fired after `jobMap` is updated in response to a `jobUpdated` or
   * `jobPending` event. Carries the full `JobWithHistory` so consumers
   * can render state transitions without a separate SDK round-trip.
   */
  onJobUpdated?: (jobWithHistory: JobWithHistory) => void;
  /**
   * Fired for an informational job message from the worker — launch
   * progress (pulling-images / starting-containers / health), container
   * events, service lifecycle. Lets consumers surface live launch feedback
   * tied to a job without changing its placement state.
   */
  onJobInfo?: (info: JobInfo) => void;
  /**
   * Fired after a job is removed from `jobMap` in response to a
   * `jobDeleted` or `jobOutOfWindow` event.
   */
  onJobDeleted?: (jobId: JobId) => void;
  /**
   * Fired on every `jobRejected` event after AutoManager has updated
   * its inventory cache. Note the node is not necessarily marked
   * unhealthy by this single occurrence — see `onNodeUnhealthy` for
   * escalation.
   */
  onJobRejected?: (info: JobRejectedInfo) => void;
  /**
   * Fires every time the placement engine runs, regardless of outcome.
   * Carries the full trace so consumers (e.g. an operator UI) can show
   * "why did my job land here?" without re-running placement
   * themselves.
   */
  onPlacementDecision?: (info: PlacementDecisionInfo) => void;
  /**
   * Fires when the placement engine returns `kind: "failure"`. Today
   * the failure also surfaces via `onError` as a stringified message;
   * this typed callback is the recommended path. The two are not
   * mutually exclusive — `onError` is preserved for back-compat.
   */
  onPlacementFailed?: (info: PlacementFailedInfo) => void;
  /**
   * Fired whenever the underlying gRPC channel changes connectivity
   * state. Maps the five gRPC connectivity states to the strings
   * `idle | connecting | ready | transientFailure | shutdown`.
   * Lets a consumer surface "connected / reconnecting / disconnected"
   * without holding a separate NorskManager handle.
   */
  onConnectionStateChange?: (state: ConnectionState) => void;
  /**
   * Fired when a bundle's restart rate exceeds its `maxRestartsPerHour`
   * (or a runtime ends terminally: a `drop` policy, an invalid spec) and
   * AutoManager marks it broken.
   */
  onBundleBroken?: (bundleId: BundleId, reason: string) => void;
  /**
   * Fired when a job fails and is dropped rather than restarted
   * (`onRuntimeFailure: drop`, a non-essential runtime past its ceiling,
   * or an invalid spec).
   */
  onJobDropped?: (jobId: JobId) => void;
  /**
   * Phase-C: fired when a hot-spare pool's live spare count is below
   * its configured target. Declared now; not fired in pre-Phase-C
   * code.
   */
  onSpareUnderflow?: (pool: string, current: number, target: number) => void;
  /**
   * Fired when a bundle's backup is best-effort placed inside a failure
   * domain its resilience policy asked to avoid (no compliant node had
   * capacity). The backup is running, but DR posture is reduced — surface
   * it so an operator can act / re-place when capacity frees.
   */
  onResilienceDegraded?: (info: ResilienceDegradedInfo) => void;
  /**
   * Phase E — fired on every migration phase transition. Use for
   * telemetry / ops UIs. The payload is the post-transition state;
   * `phaseHistory[0]` is launch, `phaseHistory.at(-1)` is the current
   * phase at the time of the callback.
   */
  onMigrationPhase?: (info: MigrationState) => void;
  /** Phase E — fired once when a migration reaches `done`. */
  onMigrationCompleted?: (migrationId: MigrationId) => void;
  /** Phase E — fired once when a migration enters `aborted`. */
  onMigrationAborted?: (migrationId: MigrationId, reason: string) => void;
  /**
   * Fired on every change to a node drain's state — a phase transition,
   * or a job's status within it. The payload is a snapshot; the same
   * state is readable at any time through `drainStatus(nodeId)`.
   */
  onDrainProgress?: (state: DrainState) => void;
  /**
   * Fired once per `handleNodeInterruption` call, after the node is
   * cordoned and its drain started (or tightened). The payload is a
   * snapshot; `nodeInterruption(nodeId)` reads the current one.
   */
  onNodeInterrupted?: (state: NodeInterruptionState) => void;
  /** Optional clock injection. Production unset (RealClock); tests pass MockClock. */
  clock?: Clock;
}

/** A drain as AutoManager runs it: its public state plus the timeout
 *  timer and the caller's strategy choice. */
interface DrainRun {
  state: DrainState;
  timer?: CancellableTimer;
  strategyFor?: (jobId: JobId) => MigrationStrategy | undefined;
}

function isDrainInFlight(state: DrainState): boolean {
  return state.phase === "migrating" || state.phase === "waiting" || state.phase === "terminating";
}

/** A copy the caller may keep: the drain keeps mutating its own. */
function snapshotDrain(state: DrainState): DrainState {
  return { ...state, jobs: state.jobs.map((j) => ({ ...j })) };
}

export { NODE_SERVICE_ROLE } from "./nodeServices";

/** @public */
export type MigrationId = string;

/**
 * How the two instances of a migrating job hand over.
 *  - `overlap`: start the target, and once it is running stop the
 *    source. Both run for a moment; no coordination with the job.
 *  - `stopThenStart`: create the target without starting it
 *    (`deferStart`), stop the source, then start the prepared target.
 *    Never two running at once; a gap of one container start.
 *  - `handover`: start the target and run the four-step handshake
 *    (target ready → source stops output → target applies) before
 *    stopping the source. The job owns the handover payload.
 * @public
 */
export type MigrationStrategy = "overlap" | "stopThenStart" | "handover";

/**
 * Phases, in order per strategy:
 * ```
 * overlap:       placing → launching → stoppingSource → done
 * stopThenStart: placing → launching → stoppingSource → startingTarget → done
 * handover:      placing → launching → awaitingTargetReady → awaitingSourceStop
 *                        → applying → stoppingSource → done
 * any:           … → aborted
 * ```
 * @public
 */
export type MigrationPhase =
  | "proposing"            // handover: GetState sent to the SOURCE, waiting for its answer.
                           //   FIRST, before any target is placed — a refusal here ends the
                           //   migration with nothing started anywhere.
  | "placing"              // placement running (synchronous; visible in phaseHistory)
  | "launching"            // target StartJob dispatched; waiting for its first signal:
                           //   overlap: JobRunning(target)  stopThenStart: JobPrepared(target)
                           //   handover: JobRunning(target) → settingStateOnTarget
  | "settingStateOnTarget" // handover: the source's state rode in on the target's StartJob and
                           //   the worker delivers it as `initial-state` right after hello-ack,
                           //   then says so with MigrationInitialStateDelivered, which ends this
                           //   phase (task 90, fix-list 27 — before that event existed the phase
                           //   ran on until MigrationTargetReady and absorbed the whole of the
                           //   target's start-up, so the next phase always read 0 ms).
                           //   MigrationTargetReady still completes it too, for a worker that
                           //   predates the event.
  | "awaitingTargetReady"  // handover: the target's participants applying that state and
                           //   running their ready round; waiting for MigrationTargetReady
  | "awaitingSourceStop"   // handover: cutover-ready sent to source, waiting for SourceStoppedOutput
  | "applying"             // handover: cutover-apply sent to target, waiting for TargetApplied
  | "stoppingSource"       // StopJob(source, migrationId) sent, waiting for JobStopped(source)
  | "startingTarget"       // stopThenStart: StartPreparedJob sent, waiting for JobRunning(target)
  | "done"
  | "aborted";

/**
 * What a handover target reports when it has applied: the envelope-level
 * summary the job's coordinator writes, carried opaque through the worker
 * and the Manager and read here. Every field optional on the wire; a job
 * that sent bytes this build cannot read yields `undefined` rather than a
 * failed handover.
 * @public
 */
export interface MigrationApplySummary {
  /** Participant ids whose apply completed, in the order they completed. */
  applied: string[];
  /** Participant ids whose apply rejected or ran out of budget, with why. */
  failed: { id: string; reason: string }[];
  /** ms between the first participant going live and the last. */
  skewMs: number;
}

/**
 * Migration state, kept entirely in AutoManager. "No concurrent
 * migrations per Job" — `sourceJobId` is the primary index;
 * `migrationId` is a stable opaque handle for ops tooling.
 *
 * `sourceRole` is the source's instance key (the slot for a job that
 * has never migrated, `primary\@xxxxxxxx` after one has); `targetRole`
 * is the key the target wears (`migrationTargetRole(slot, id)`).
 *
 * @public
 */
export interface MigrationState {
  migrationId: MigrationId;
  sourceJobId: JobId;
  sourceRole: Role;
  sourceNodeId: NodeId;
  strategy: MigrationStrategy;
  /** `migrationTargetRole(slot, migrationId)`, slot = the source's runtime slot. */
  targetRole: Role;
  /** Set once placement has chosen. */
  targetNodeId?: NodeId;
  startedAt: Date;
  phase: MigrationPhase;
  phaseHistory: { phase: MigrationPhase; at: Date }[];
  /** Set when phase becomes "aborted"; undefined while in flight or on done. */
  abortReason?: string;
  /**
   * Opaque, the SOURCE's answer to GetState in `proposing` (handover).
   * Rides out on the target's StartJob as `migrationInitialState`.
   * Undefined until the source has answered; an empty array is a
   * perfectly good answer — a job with nothing to hand over still
   * accepts, and that is not a refusal.
   */
  initialState?: Uint8Array;
  /** Opaque, from MigrationTargetReady (handover). Relayed to the source. */
  targetPayload?: Uint8Array;
  /** Opaque, from MigrationSourceStoppedOutput (handover). Relayed to the target. */
  sourcePayload?: Uint8Array;
  /**
   * The target's own account of its apply, from MigrationTargetApplied
   * (handover): which participants went live, which did not, and the spread
   * between the first and the last. Past the point of no return a failure
   * here is reported, not fatal — a handover is `done` with `failed`
   * non-empty, and this is the only record of that (task 90, fix-list 41).
   * Undefined for a target that sent none (a job predating the summary).
   */
  applySummary?: MigrationApplySummary;
  /**
   * True from the moment StopJob(source) is sent. Before it, an abort is
   * a clean rollback (target stopped, source untouched). After it the
   * saga only fails forward: it keeps waiting on the target, and if the
   * target is lost the migration is marked aborted and ordinary
   * recovery re-places the job.
   */
  pastPointOfNoReturn: boolean;
}

/**
 * Coarse connection state surfaced by `onConnectionStateChange`. Maps
 * to gRPC's `ChannelConnectivityState` values via NorskManager's
 * existing watcher.
 *
 * @public
 */
export type ConnectionState =
  | "idle"
  | "connecting"
  | "ready"
  | "transientFailure"
  | "shutdown";

/**
 * Phase C4 — per-policy recovery tunables, keyed by `OnRuntimeFailure`.
 *
 * Only the two re-placing policies appear. `drop` is absent by
 * construction: it never restarts, so a restart ceiling cannot apply to
 * it. Its old entry carried a lone `onFailure: "drop"` field that
 * nothing read — a tautology restating its own key — so it went with the
 * rename, along with the matching `onFailure: "restart"` on the other
 * two.
 *
 * Spare preference is likewise NOT configurable here: it follows from
 * the policy itself, which the recovery path reads directly.
 *
 * Omitted entries fall through to the flat `AutoSettings` fields, then
 * to compiled-in defaults.
 *
 * @public
 */
export interface FailurePolicyConfig {
  preferSpare?: {
    /** Default: 6 */
    maxRestartsPerHour?: number;
  };
  avoidSpare?: {
    /** Default: 6 */
    maxRestartsPerHour?: number;
  };
}

/**
 * Phase C4 — placement / recovery tunables. Same fields as the flat
 * top-level settings; this typed grouping is the recommended path.
 * Omitted fields fall through to the flat settings, then to
 * compiled-in defaults.
 *
 * @public
 */
export interface PlacementConfig {
  /** Default: 5_000 ms. */
  rejectionBackoffMs?: number;
  /** Default: 3. */
  rejectionEscalationCount?: number;
  /** Default: 600_000 ms. */
  unhealthyCooldownMs?: number;
  /** Default: 30_000 ms. */
  failureBackoffMs?: number;
  /** Default: 120_000 ms. */
  jobStartupGraceMs?: number;
}

/**
 * Phase C6 — keep `targetCount` warm-running nodes ready per pool.
 * The reconciler provisions short, terminates
 * excess. Cluster pools cannot auto-scale — `targetCount` there
 * means "refuse to place onto a spare-tagged node unless this is a
 * recovery placement"; reconciler emits `onSpareUnderflow` and leaves
 * the rest to operator IT.
 *
 * @public
 */
export interface HotSpareConfig {
  pool: string;
  targetCount: number;
  /** The shape to provision spares with (instance type, region, …). */
  spec: NodeSpec;
  /**
   * Phase G — product-template names this spare should be PRE-WARMED for.
   * On provision (and whenever the spec changes) the reconciler sends the
   * node's worker a prefetch of these templates' images, so the spare is
   * warm before anybody needs it. Warmth itself is still observed, never
   * declared: `warmFor` only says which pulls to start.
   *
   * Independent of `reservedFor`. A spare can be warm for a template it is
   * not reserved for, and reserved for one it was never asked to pull.
   */
  warmFor?: string[];
  /**
   * Phase G — product-template names this spare is set aside for. With
   * `exclusive: false` (the default) it is a preference: anyone may take
   * the spare, these templates just score it first. With `exclusive: true`
   * it is a guarantee: nobody else may have it.
   */
  reservedFor?: string[];
  /**
   * Phase G — is `reservedFor` a guarantee? Default false. The operator
   * opts in to exclusivity, because the cost of it is real: a spare nobody
   * else may take can sit idle while a job waits for a cold node.
   */
  exclusive?: boolean;
}

/**
 * Phase G — the editable part of a hot spare's spec: what it is warmed
 * for, who it is reserved for, and whether that reservation is a
 * guarantee. Set from a `HotSpareConfig` at provision time and editable
 * per node at runtime through `updateHotSpareNode`.
 *
 * @public
 */
export interface SpareSpec {
  warmFor: string[];
  reservedFor: string[];
  exclusive: boolean;
}

/**
 * What `updateHotSpareSpecs` did to the live list, keyed by pool (the
 * entry's key): entries that were not there before, entries that are gone,
 * and entries whose target, spec, warmth or reservation differs.
 *
 * @public
 */
export interface HotSpareSpecsDiff {
  added: string[];
  removed: string[];
  changed: string[];
}

/**
 * Phase C6 — what kind of node the spare reconciler should
 * provision. For elastic (aws/oci) pools, `instanceType` is
 * required. For cluster pools, the spec is only used to constrain
 * which existing cluster nodes are eligible to be tagged as spare
 * (future).
 *
 * @public
 */
export interface NodeSpec {
  /** AwsInstanceType | OciShape; required for elastic pools. */
  instanceType?: string;
  /** Cloud-specific region or AZ; pass through to provider. */
  region?: string;
  az?: string;
}

/**
 * How long `drainNode` waits for the jobs it could not migrate off the
 * node before giving up (`partial`) or, with `force`, terminating the
 * node with them still on it. Ten minutes: long enough for a handover to
 * run its course and an operator to notice a stuck job, short enough
 * that a drain nobody is watching does not hold a cordoned node all day.
 * @public
 */
export const DEFAULT_DRAIN_TIMEOUT_MS = 10 * 60_000;

/**
 * Where a node drain is. In flight: `migrating` (at least one job's
 * migration is running), `waiting` (no migration in flight, but jobs the
 * drain could not move are still on the node — until the timeout, or
 * until they leave by other means), `terminating` (the node is empty
 * and its terminate has been issued). Terminal: `done` (terminated),
 * `partial` (the timeout passed with jobs still on the node and no
 * `force`: nothing was terminated, the node stays cordoned, `jobs`
 * names the stuck ones), `failed` (the terminate RPC failed, or the
 * drain was cancelled by `uncordonNode` — `reason` says which).
 * @public
 */
export type DrainPhase = "migrating" | "waiting" | "terminating" | "done" | "partial" | "failed";

/**
 * One job's fate within a drain. `migrating`: its migration is in
 * flight; `migrated`: it reached the target; `moved`: it left the node
 * by other means (recovery re-placed it, it was deleted); `stuck`: it
 * could not be migrated — the saga refused it (not running, not an
 * AutoManager-placed job), placement found nowhere, or the migration
 * aborted — and it is still on the node; `terminated`: it was still on
 * the node when a `force` drain terminated it.
 * @public
 */
export type DrainJobStatus = "migrating" | "migrated" | "moved" | "stuck" | "terminated";

/** @public */
export interface DrainJobState {
  jobId: JobId;
  status: DrainJobStatus;
  /** The migration the drain started for this job, once it has one. */
  migrationId?: MigrationId;
  /** Why the job is `stuck`. */
  reason?: string;
}

/**
 * A node drain's state — the snapshot `drainStatus` returns and
 * `onDrainProgress` carries.
 * @public
 */
export interface DrainState {
  nodeId: NodeId;
  phase: DrainPhase;
  started: Date;
  timeoutMs: number;
  force: boolean;
  jobs: DrainJobState[];
  /** Why the drain ended `partial` or `failed`, or (with `force`) what it
   *  terminated over. */
  reason?: string;
}

/** @public */
export interface DrainOptions {
  /** How long to wait for jobs that cannot be migrated. Default
   *  `DEFAULT_DRAIN_TIMEOUT_MS`. */
  timeoutMs?: number;
  /** At the timeout, terminate the node with its remaining jobs on it
   *  (they recover per their failure policy, as `evictNode` does) rather
   *  than ending `partial`. */
  force?: boolean;
  /**
   * The migration strategy each job's migration runs with; undefined for
   * a job means `migrateJob`'s default (`stopThenStart`). AutoManager
   * knows nothing about product manifests, so a caller that does (norsk-mgr
   * resolves the declared strategy from the job's template) passes it here.
   */
  strategyFor?: (jobId: JobId) => MigrationStrategy | undefined;
}

/**
 * What a provider's notice says it is doing with a node. A
 * `spot-interruption` is the terminal warning (EC2 gives two minutes);
 * a `rebalance` is a recommendation that the capacity is at elevated risk
 * of interruption, with no time attached — it is usually followed by the
 * interruption notice.
 * @public
 */
export type NodeInterruptionKind = "spot-interruption" | "rebalance";

/** @public */
export interface NodeInterruptionNotice {
  kind: NodeInterruptionKind;
  /** When the provider reclaims the node. Absent for a rebalance
   *  recommendation, or a notice that carries no time. */
  deadline?: Date;
  /** The migration strategy per job, as `DrainOptions.strategyFor`. */
  strategyFor?: DrainOptions["strategyFor"];
}

/**
 * A node the provider is taking back, and the drain running it down —
 * the snapshot `nodeInterruption` returns and `onNodeInterrupted` carries.
 * @public
 */
export interface NodeInterruptionState {
  nodeId: NodeId;
  kind: NodeInterruptionKind;
  deadline?: Date;
  /** When the (latest) notice was handled. */
  noticedAt: Date;
  drain: DrainState;
}

/**
 * How far ahead of the provider's deadline an interruption drain gives
 * up on migrating and terminates the node with whatever is still on it:
 * the terminate has to be issued, and the survivors re-placed, before the
 * host is pulled from under them. Thirty seconds off EC2's two-minute
 * warning leaves the 90 s below.
 * @public
 */
export const SPOT_INTERRUPTION_DRAIN_MARGIN_MS = 30_000;

/** The drain timeout for a spot interruption notice that carries no
 *  deadline: EC2's two-minute warning, less the margin. @public */
export const DEFAULT_SPOT_INTERRUPTION_DRAIN_MS = 120_000 - SPOT_INTERRUPTION_DRAIN_MARGIN_MS;

/** The drain timeout for a rebalance recommendation, which names no
 *  time: long enough for handovers, short enough to be off the node
 *  before the interruption that usually follows. @public */
export const DEFAULT_REBALANCE_DRAIN_MS = 5 * 60_000;

/**
 * Per-phase deadlines for in-flight migrations. Set any field to
 * override its default; omitted fields use the compiled-in default
 * below. Expiry before the point of no return aborts the migration;
 * after it (`stoppingSource`, `startingTarget`) the saga fails forward.
 *
 * @public
 */
export interface MigrationConfig {
  /** Default 30_000 ms — handover: the source has this long to answer the
   *  propose (GetState) before the migration is abandoned. Expiry aborts
   *  with no target ever placed. */
  proposeDeadlineMs?: number;
  /** Default 120_000 ms — the target's first signal after StartJob (running / prepared), every strategy. */
  launchingDeadlineMs?: number;
  /** Default 120_000 ms — handover: workflow has this long after target running to signal ready. */
  awaitingTargetReadyDeadlineMs?: number;
  /** Default 30_000 ms — handover: source workflow has this long to stop output after the cutover signal. */
  awaitingSourceStopDeadlineMs?: number;
  /** Default 10_000 ms — handover: target workflow has this long to apply the handover. */
  applyingDeadlineMs?: number;
  /** Default 30_000 ms — the source's JobStopped after StopJob. Expiry proceeds as if stopped. */
  stoppingSourceDeadlineMs?: number;
  /** Default 60_000 ms — stopThenStart: the target's JobRunning after StartPreparedJob. Expiry fails forward. */
  startingTargetDeadlineMs?: number;
}

/**
 * One entry in a job's restart history: when a recovery placement was
 * attempted, and whether that attempt represents a NEW failure as far as
 * the restart-rate ceiling is concerned. See `restartHistory`.
 */
interface RestartAttempt {
  at: Date;
  counts: boolean;
}

/**
 * How a failed `startJob` dispatch is classified, which decides three
 * things at once: whether the refusing node is kept out of the immediate
 * re-place, whether a consecutive repeat of the SAME refusal is a new
 * failure for the ceiling, and the token that makes "the same refusal"
 * comparable across attempts.
 *
 * `standing` means "this answer is a condition that is still in force,
 * not an event that just happened" — the daemon insisting the job is
 * already on the node does not become true-r by asking again. An
 * unclassified dispatch error is NOT standing: we deliberately retry the
 * same node for those, and the ceiling count is the only thing that ever
 * terminates a node/transport fault that never clears.
 */
interface DispatchRefusal {
  reason: string;
  excludeNode: boolean;
  standing: boolean;
}

function classifyDispatchFailure(msg: string): DispatchRefusal {
  // ProvisionJobDuplicate. The daemon believes the job is already
  // running on this node and it has not been corroborated by the job's
  // own history (that case adopts, above) — so the binding is stale or
  // leaked and re-asking THIS node cannot do anything but be refused
  // again. Steer the re-place elsewhere, and count the run once.
  if (/already running on node/i.test(msg)) {
    return { reason: "duplicate", excludeNode: true, standing: true };
  }
  // The daemon says the node is stopping/terminated. Already handled by
  // a `recentFailures` mark (which excludes it fleet-wide for the
  // backoff), so `excludeNode` here is belt-and-braces.
  if (/not in (a )?running state/i.test(msg)) {
    return { reason: "nodeNotRunning", excludeNode: true, standing: true };
  }
  return { reason: "unclassified", excludeNode: false, standing: false };
}

/** `placeAndProvision`'s answer: a node now holds the job's claim
 *  (dispatched, or deferred to `bindProvisionedJob`), or nothing does. */
export type PlacementOutcome =
  | { kind: "bound"; nodeId: NodeId }
  | {
      kind: "notPlaced";
      reason: string;
      /** When the caller NAMED a target: why that node in particular was
       *  rejected (its FilterReason), or that it was in no pool at all.
       *  Placement's own reason is the generic `noCapacityInAnyPool`, which
       *  says nothing about a pinned node (task 90, fix-list 17). */
      targetReason?: string;
    };

/**
 * Phase C4 — result of `validateBundleAdmission`. `ok` means the
 * bundle passes static validation AND every (replica, jobName)
 * placeDryRun would succeed at the current moment. The two failure
 * modes are distinct: `validationErrors` is "the bundle spec itself
 * is broken" (operator must fix); `noCapacity` is "current cluster
 * state can't host this" (operator must add capacity or wait).
 *
 * @public
 */
export type AdmissionResult =
  | { kind: "ok" }
  | { kind: "validationErrors"; errors: ValidationError[] }
  | { kind: "noCapacity"; unplaceable: AdmissionUnplaceable[] };

/** @public */
export interface AdmissionUnplaceable {
  replicaIndex: number;
  jobName: string;
  reason: PlacementFailureReason;
}

/**
 * Phase D4 — operational metrics snapshot. Cumulative counters since
 * `startedAt`; per-pool gauges are point-in-time at call. Consumers
 * (telemetry pipelines, ops UIs) poll `AutoManager.metrics()` on
 * their own cadence.
 *
 * @public
 */
export interface AutoManagerMetrics {
  /** AutoManager.run() entry time; undefined before runInternal starts. */
  startedAt?: Date;
  /** Milliseconds since startedAt, 0 if not yet started. */
  uptimeMs: number;
  // Cumulative counters (since startedAt):
  placementsAttempted: number;
  /**
   * Placement that successfully dispatched its wire call (startJob /
   * createAwsNode / createOciNode). Counted post-RPC, so a successful
   * decision whose dispatch then fails does not bump this counter —
   * see `placementsDispatchFailed`.
   */
  placementsSucceeded: number;
  /** Placement engine returned `failure` (no suitable target). */
  placementsFailed: number;
  /**
   * Decision succeeded but the downstream RPC (startJob /
   * createAwsNode / createOciNode) rejected. The job is left
   * unplaced; AutoManager surfaces the error via onError.
   */
  placementsDispatchFailed: number;
  jobsRejected: number;
  bundlesBroken: number;
  /** Jobs dropped rather than restarted (`onRuntimeFailure: drop`,
   *  non-essential past the ceiling, invalid spec). */
  jobsDropped: number;
  // Point-in-time gauges:
  pools: PoolMetrics[];
  /** Bundles currently marked broken by the restart-rate ceiling. */
  brokenBundleIds: BundleId[];
}

/** @public */
export interface PoolMetrics {
  poolName: string;
  nodeCount: number;
  /** Spare-tagged, running, no jobs assigned. */
  liveSpares: number;
  /** Sum of `targetCount` across hotSpares entries for this pool. */
  spareTarget: number;
  /**
   * Mean reservedCapacity / totalCapacity across nodes with inventory.
   * 0 if there are no nodes with inventory.
   */
  usedCapacityFraction: number;
}

/** A job bound to a node, with the GPU device it holds there when known
 *  (see gpuIndexOfRecord). What the derived reservation is summed over. */
interface PlacedJobRef {
  jobId: JobId;
  gpuIndex?: number;
}

/**
 * Payload for `onPlacementDecision`. The `bundleId`/`replicaIndex`/
 * `jobName` triple identifies which bundle member the decision was made
 * for; `result` is the placement engine's verdict; `trace` is the
 * per-pool filter/score breakdown described in §6 of the design.
 *
 * @public
 */
export interface PlacementDecisionInfo {
  jobId: JobId;
  bundleId: BundleId;
  replicaIndex: number;
  jobName: string;
  result: PlacementResult;
  trace: PlacementTrace;
  at: Date;
}

/**
 * Payload for `onPlacementFailed`. `triedPools` is the ordered list of
 * tier names AutoManager attempted within the bundle's `pool` before
 * giving up.
 *
 * @public
 */
export interface PlacementFailedInfo {
  jobId: JobId;
  bundleId: BundleId;
  replicaIndex: number;
  jobName: string;
  reason: PlacementFailureReason;
  triedPools: string[];
  at: Date;
}

/**
 * Payload for `onResilienceDegraded`. Fired when a bundle's backup
 * (replicaIndex 1) is placed inside a failure domain its resilience policy
 * asked to avoid, because no compliant node had capacity. `violated` lists
 * the anti-affinity flags that couldn't be honoured. The placement still
 * happened — a degraded backup beats no backup — but DR posture is reduced.
 *
 * @public
 */
export interface ResilienceDegradedInfo {
  jobId: JobId;
  bundleId: BundleId;
  replicaIndex: number;
  jobName: string;
  violated: ResilienceFlag[];
  /** Set when the primary is on interruptible (spot) capacity with no
   *  durable backup — see ResilienceDegradation.interruptiblePrimary. */
  interruptiblePrimary?: boolean;
  at: Date;
}

/**
 * Payload for `onJobRejected`. `inventory` is the typed snapshot the
 * worker sent alongside the rejection so consumers can render "why" in
 * the same UI tick as the rejection itself.
 *
 * @public
 */
export interface JobRejectedInfo {
  jobId: JobId;
  nodeId: NodeId;
  reason: ManagerPB.JobRejected["reason"];
  inventory?: TypedNodeInventory;
  at: Date;
}

const DEFAULT_REJECTION_BACKOFF_MS = 5_000;
const DEFAULT_REJECTION_ESCALATION_COUNT = 3;
const DEFAULT_UNHEALTHY_COOLDOWN_MS = 600_000;
const DEFAULT_FAILURE_BACKOFF_MS = 30_000;
const DEFAULT_MAX_RESTARTS_PER_HOUR = 6;
const RESTART_WINDOW_MS = 60 * 60 * 1000;
// Minimum spacing between consecutive recovery placement attempts for
// the same JobId. Stops the "provisionJob → JobFailed → provisionJob"
// tight loop that JobStore state-transition events naturally drive
// when worker-side launches keep failing fast. 5s is short enough that
// transient failures still recover quickly but long enough that even a
// pathological failure pattern doesn't burn through `maxRestartsPerHour`
// in seconds. Operator-tunable later via AutoSettings.
export const MIN_RECOVERY_INTERVAL_MS = 5_000;
// How long `close()` holds the Manager channel open so the aborts it
// sends for in-flight migrations can reach the workers. The channel
// closes as soon as they settle; this is the cap for a Manager that
// never answers.
export const CLOSE_ABORT_GRACE_MS = 5_000;
// Minimum spacing between any `placeAndProvision` entries for the same
// JobId. Tighter than MIN_RECOVERY_INTERVAL_MS — the window we need is
// just "time for daemon's provisionJob to land + JobStarted to round-
// trip back into AutoManager's nodeMap". A second is comfortably more
// than typical local-host dispatch latency.
const MIN_PLACEMENT_INTERVAL_MS = 1_000;
const DEFAULT_JOB_STARTUP_GRACE_MS = 120_000;
const DEFAULT_HOT_SPARE_RECONCILE_INTERVAL_MS = 60_000;
/**
 * How long, after a (re)connect, a job stays bound to a node its history
 * names but the daemon has not (yet) reported. A Manager daemon restart
 * sends its initial state as soon as it is up, while the EC2 rediscovery
 * of the nodes it had — one hello every few seconds — is still running;
 * until this, a job whose node had not re-hello'd read as unplaced, was
 * re-placed onto whatever node HAD, and when its real node then hello'd
 * carrying the same job the daemon terminated that node as a duplicate.
 * Observed on norsk-e2e 2026-09-14 16:53Z: two productions and two nodes
 * lost to one Manager restart. Five minutes covers a slow rediscovery;
 * a node that never comes back is released for re-placement after it.
 */
export const HYDRATION_NODE_GRACE_MS = 5 * 60 * 1000;
export { PREFETCH_RETRY_BACKOFF_MS, SPARE_PROVISION_RETRY_BACKOFF_MS } from "./hotSpares";
const DEFAULT_NODE_SERVICE_RECONCILE_INTERVAL_MS = 60_000;

export class AutoManager {
  norsk: NorskManager;
  settings: AutoSettings;
  clock: Clock;
  // All AutoManager state is a mirror of what the event stream most
  // recently said — AutoManager owns no durable state. Maps are cleared
  // and re-hydrated on (re)connect from the event-stream initial state.
  jobMap: Map<JobId, JobWithHistory> = new Map();
  // In-flight provisions (machine-types.md Part B): nodeId → the declared
  // shape placement synthesizes a prospective NodeView from, until the
  // node's first real inventory (or its death/timeout) removes it. The
  // triggering job's binding is set alongside; other jobs may bind
  // too (deferred dispatch — see bindProvisionedJob).
  private pendingProvisions: Map<
    NodeId,
    { pool: string; tier: string; instanceType: string; shape: InstanceTypeOption; requestedAt: Date }
  > = new Map();
  private pendingProvisionTimers: Map<NodeId, CancellableTimer> = new Map();
  nodeMap: Map<NodeId, NodeSummary> = new Map();
  bundleMap: Map<BundleId, TypedBundle> = new Map();
  inventoryMap: Map<NodeId, TypedNodeInventory> = new Map();
  // Per-node rejection history (timestamps). Pruned to the configured
  // backoff window on read; entries trigger escalation to `unhealthyNodes`
  // once the threshold is crossed.
  private rejectionsPerNode: Map<NodeId, Date[]> = new Map();
  // Nodes excluded from placement, mapped to the time the mark was set.
  // Cleared by real recovery (a job reaching `running` on the node) or by
  // the mark ageing past `unhealthyCooldownMs` — NOT by routine inventory
  // heartbeats (see handleNodeInventoryUpdated).
  private unhealthyNodes: Map<NodeId, Date> = new Map();
  // Recently-failed nodes — excluded from placement for `failureBackoffMs`
  // after we observe their `nodeStopping`. Map value is the deadline.
  private recentFailures: Map<NodeId, Date> = new Map();
  // Phase C2: per-replica restart history. Each entry is one recovery
  // placement ATTEMPT; older-than-1h entries are pruned on read. Two
  // readers, deliberately different:
  //   - the maxRestartsPerHour ceiling counts only entries with
  //     `counts: true` — the attempts that represent a NEW failure;
  //   - the minimum-interval backoff paces off the newest entry
  //     whatever it is, so a non-counting attempt still spaces the
  //     next dispatch.
  // (`counts: false` is the "repeated identical refusal" rule — see
  // `repeatOfCountedCondition` on placeAndProvision's recovery arg.)
  private restartHistory: Map<JobId, RestartAttempt[]> = new Map();
  // jobId → `<nodeId>|<reason>` of the STANDING dispatch refusal most
  // recently seen for this job. A consecutive refusal with the same
  // signature is the same condition still in force, so it does not count
  // against the restart ceiling again. Cleared the moment anything else
  // happens to the job — a different refusal, a successful dispatch, a
  // duplicate we adopted, the job reaching `running`, deletion — so the
  // rule can only ever collapse an unbroken RUN of identical refusals.
  private readonly lastCountedDispatchRefusal: Map<JobId, string> = new Map();
  // "Where is job J": one record per job — its node (the placement of
  // record; set at placement time, rebuilt from history at hydration,
  // released on failure and deletion, so `nodeForJob` is an O(1) read
  // rather than a fleet scan), the instance key its last migration left
  // it under, its placement decision of record, the in-flight dispatch
  // gate, and its in-flight migration's claim (the TARGET's binding and
  // parked placement). See src/jobBinding.ts for what must hold between
  // those, and for why a migration target is bound beside the job's own
  // node rather than in its place.
  /** @internal */
  public readonly bindings: JobBindings = new JobBindings();
  // Nodes whose last inventory report broke the wire contract (an
  // absolute `utilizationCap`), so the operator is told once per node —
  // workers re-report inventory every ~30 s — and again only if the node
  // comes back into contract and then out of it. See typedInventoryFrom.
  private inventoryContractWarned: Set<NodeId> = new Set();
  // Pending placement timers — one per scheduled job. Set when
  // `handleJobChanged` sees a pre/active job whose `startDateTime`
  // (minus the pool's `placementLeadMs`) is still in the future;
  // fires placeAndProvision when the lead-time window opens.
  // Cancelled on jobDeleted and on close(), and replaced on subsequent
  // job-updates that change the schedule; a fresh instance re-evaluates
  // them from hydrated state. Tracks `scheduledFor` so `pendingPlacements()`
  // can render upcoming work in operator UIs.
  private pendingPlacementTimers: Map<JobId, { timer: CancellableTimer; scheduledFor: Date }> = new Map();
  // Pending stop timers — one per bundleId. Set when any job in the
  // bundle carries the `__stopDateTime` tag and the instant is still
  // in the future; on fire, calls `deleteBundle(bundleId)` and
  // removes the entry. Daemon has no first-class concept of stop
  // time so a fresh instance rebuilds this state from job tags at
  // hydration; close() cancels it.
  // Public surface: `setBundleStopTime(bundleId, newStopTime?)` for
  // the future Edit-bundle UI; `bundleStopTimes()` for read.
  private pendingStopTimers: Map<BundleId, { timer: CancellableTimer; stopAt: Date }> = new Map();
  // Outstanding deferred recovery placements. Each entry caps to one
  // clock-driven retry per JobId; subsequent triggers while a retry is
  // in flight are no-ops (the scheduled fire will see the latest state
  // — including a bundle marked broken in the meantime, which it then
  // does not place). Prevents the tight-loop "provisionJob → JobFailed →
  // provisionJob" pattern we observed when worker-side launch errors
  // were instantly retryable. See 04-e2e-todo's B-daemon-retry-storm.
  private pendingPlacementRetries: Map<JobId, CancellableTimer> = new Map();
  // Timestamp of the most recent `placeAndProvision` entry per JobId.
  // Used by both initial and recovery branches to skip duplicate
  // Phase C2: bundles that have tripped the restart ceiling. While in
  // this set, AutoManager refuses recovery placements for any of the
  // bundle's replicas. Cleared on bundleUpdated (operator pushed a
  // fresh spec) or bundleDeleted.
  private brokenBundles: Set<BundleId> = new Set();
  // Phase C3: pending nodeStarting → nodeStarted grace timers. If a
  // timer fires before the node reports running, AutoManager
  // terminates the node and the existing nodeStopping path runs
  // recovery per the job's onRuntimeFailure.
  private startupGraceTimers: Map<NodeId, CancellableTimer> = new Map();
  // Nodes a hydrated job is bound to that the daemon has not reported yet,
  // with the timer that releases the binding if they never appear. See
  // HYDRATION_NODE_GRACE_MS.
  private unseenBoundNodes: Map<NodeId, CancellableTimer> = new Map();
  // Phase C6 / G — the hot-spare collaborator (src/hotSpares.ts): the
  // reconciler and every piece of spare bookkeeping. Built in the
  // constructor over this instance's fleet view.
  private hotSpares!: HotSpares;
  private hotSpareReconcileInterval?: CancellableTimer;
  // Phase C8: nodes the operator has cordoned client-side. Merged
  // into NodeView.inventory.cordoned at placement time so existing
  // placement filter excludes them. (When Manager grows a cordon
  // API and emits `cordoned: true` on inventory updates, this local
  // set becomes redundant — until then it's the API.)
  private cordonedNodes: Set<NodeId> = new Set();
  /** Nodes `onNodeTerminated` has fired for — see `issueTerminate`. */
  private terminateSignalled: Set<NodeId> = new Set();
  // Node drains, one per node: the in-flight one, or the last one to end
  // (so `drainStatus` can answer for a drain that finished a moment ago).
  // The timer is the drain's timeout; the resolver is the caller's
  // strategy choice, kept for jobs that turn up on the node mid-drain.
  private drains: Map<NodeId, DrainRun> = new Map();
  // Nodes a provider has said it is reclaiming (`handleNodeInterruption`),
  // by the latest notice. A job lost with one of these is re-placed
  // without counting against the restart ceiling. Cleared when the node
  // stops, or when an operator uncordons it (the notice is withdrawn).
  private reclaimedNodes: Map<NodeId, Omit<NodeInterruptionState, "drain">> = new Map();
  // Phase D4 — cumulative metrics counters since startedAt. Read via
  // `metrics()`. Counters never reset; consumers compute deltas.
  private metricsCounters = {
    placementsAttempted: 0,
    placementsSucceeded: 0,
    placementsFailed: 0,
    placementsDispatchFailed: 0,
    jobsRejected: 0,
    bundlesBroken: 0,
    jobsDropped: 0,
  };
  // Phase E — the migration saga (src/migrationSaga.ts): every migration
  // this instance has run and the machinery that drives one. Built in the
  // constructor over this instance's placement path and fleet view.
  private migrationSaga!: MigrationSaga;
  // Phase F — the NodeService reconciler (src/nodeServices.ts): places
  // each configured service on every eligible node and tracks what it
  // placed. Built in the constructor over this instance's fleet view.
  private nodeServiceReconciler!: NodeServiceReconciler;
  private nodeServiceReconcileInterval?: CancellableTimer;
  /**
   * The moment `runInternal()` first entered (i.e. the eventStream was
   * established). Undefined until then. Powers "AutoManager up for N
   * minutes" affordances.
   * @public
   */
  public startedAt?: Date;
  private cleanupInterval?: CancellableTimer;
  private closed: boolean = false;
  // Set once the initial state has been hydrated. Until then this instance
  // has an EMPTY view of the fleet, and a reconcile run from a public entry
  // point (an embedder cordoning nodes the moment `run` resolves — norsk-mgr
  // restores its cordons then) would provision against nothing, only for
  // hydration to wipe the record of that provision and the startup kick to
  // provision again. norsk-e2e 2026-09-15: targetCount 1, two spares 37 ms
  // apart. The reconcilers wait for hydration; the startup kick follows it.
  private hydrated: boolean = false;

  /** @public */
  public static async run(
    settings: AutoSettings
  ): Promise<[AutoManager, Promise<void>]> {

    // Phase D1 — fail fast on configuration mistakes.
    const validationErrors = validateAutoSettings(settings);
    if (validationErrors.length > 0) {
      throw new AutoSettingsValidationError(validationErrors);
    }

    debuglog("Trying to connect", { url: settings.url });

    const fire = (state: ConnectionState) =>
      settings.onConnectionStateChange?.(state);

    try {
      const norsk: NorskManager = await NorskManager.connect({
        url: settings.url,
        signal: settings.signal,
        onAttemptingToConnect: () => fire("idle"),
        onConnecting: () => fire("connecting"),
        onReady: () => fire("ready"),
        onFailedToConnect: () => fire("transientFailure"),
        onShutdown: () => fire("shutdown"),
      });

      const autoManager = new AutoManager(norsk, settings);
      const promise = autoManager.runInternal();

      return [autoManager, promise];
    } catch (err) {
      debuglog("Exception in connection");
      throw err;
    }
  }

  /** @public */
  public async terminateNode(nodeId: NodeId): Promise<void> {
    await this.issueTerminate(nodeId);
  }

  /**
   * Every terminate AutoManager itself issues goes through here, so
   * `onNodeTerminated` fires exactly once per node whichever path (and
   * however many retries of it — the orphan-spare reaper re-issues on
   * every tick until the node stops) got there. The signal fires at issue
   * time; callers keep their own handling of the RPC's outcome. (No await
   * before the signal: the body runs synchronously up to the return, so the
   * signal still fires at issue time, before any caller's await resumes.)
   */
  private async issueTerminate(nodeId: NodeId): Promise<void> {
    const rpc = this.norsk.terminateNode(nodeId);
    if (!this.terminateSignalled.has(nodeId)) {
      this.terminateSignalled.add(nodeId);
      this.settings.onNodeTerminated?.(nodeId);
    }
    return rpc;
  }

  /**
   * Phase C8 — cordon a node. AutoManager projects this onto every
   * placement decision: cordoned nodes are filtered out, so no new
   * placements land on them. Existing jobs continue running.
   * Idempotent.
   * @public
   */
  public cordonNode(nodeId: NodeId): void {
    if (this.cordonedNodes.has(nodeId)) return;
    this.cordonedNodes.add(nodeId);
    debuglog("Node cordoned", { nodeId });
    // Reconcile in case a cordoned node was holding a spare slot —
    // its un-spare-ness now opens the underflow path. Same for any
    // NodeServices placed on the now-cordoned node.
    this.reconcileHotSpares();
    this.reconcileNodeServices();
  }

  /**
   * Phase C8 — un-cordon a node, restoring it to the placement pool.
   * Idempotent.
   * @public
   */
  public uncordonNode(nodeId: NodeId): void {
    if (!this.cordonedNodes.delete(nodeId)) return;
    debuglog("Node uncordoned", { nodeId });
    // A drain still deciding the node's fate is cancelled: the operator
    // has changed their mind. Its migrations in flight run to their end
    // (a source that answered a propose is locked until it hears the
    // cutover or an abort, and the job is better off wherever they land);
    // only the terminate that would have followed is off.
    this.cancelDrain(nodeId, "cancelled: node uncordoned");
    this.reclaimedNodes.delete(nodeId);
    this.reconcileHotSpares();
    this.reconcileNodeServices();
  }

  /**
   * Phase C8 — whether a node is currently cordoned by AutoManager.
   * @public
   */
  public isCordoned(nodeId: NodeId): boolean {
    return this.cordonedNodes.has(nodeId);
  }

  /**
   * Evict a node: cordon it (so no new placements land here) and
   * terminate it at once — Manager emits `nodeStopping` for every job
   * hosted there, and the policy-aware recovery path re-places them onto
   * other nodes (excluding this cordoned one), with a gap. `drop` jobs
   * are dropped; `preferSpare` prefers hot spares. This is what
   * `drainNode` did until 2026-09-16; `drainNode` now moves the jobs
   * first, and this is the blunt instrument for when that is not wanted.
   * @public
   */
  public async evictNode(nodeId: NodeId): Promise<void> {
    this.cordonNode(nodeId);
    await this.issueTerminate(nodeId);
  }

  /**
   * Drain a node: cordon it, migrate every job off it, and terminate it
   * once it is empty.
   *
   * Each job bound to the node is started on a migration through the
   * same saga `migrateJob` runs (placement elsewhere — the source node is
   * never a candidate, and it is cordoned besides; a new node may be
   * provisioned in an elastic pool). A job the saga refuses (not running,
   * not AutoManager-placed), one placement finds nowhere for, or one
   * whose migration later aborts is `stuck`: it stays running where it is
   * until the timeout (`DEFAULT_DRAIN_TIMEOUT_MS` unless given), and
   * then either the drain ends `partial` — nothing terminated, the node
   * still cordoned, `jobs` naming the stuck ones — or, with `force`, the
   * node is terminated with them on it and they recover per their
   * failure policy, as `evictNode` would have had them. A migration that
   * fails forward past its point of no return is re-placed by ordinary
   * recovery, which counts as the job leaving the node. The terminate goes
   * through `issueTerminate`, so `onNodeTerminated` fires exactly once.
   *
   * Resolves as soon as the migrations have been started (their placement
   * and dispatch, or a handover's propose, settle afterwards) with the
   * drain's state; progress is pushed through `onDrainProgress` and read
   * back through `drainStatus`. One drain per node: a second call while
   * one is in flight returns that one's state. `uncordonNode` cancels a
   * drain that has not yet issued its terminate.
   * @public
   */
  public async drainNode(nodeId: NodeId, opts: DrainOptions = {}): Promise<DrainState> {
    const existing = this.drains.get(nodeId);
    if (existing && isDrainInFlight(existing.state)) return snapshotDrain(existing.state);
    const state: DrainState = {
      nodeId,
      phase: "migrating",
      started: this.clock.now(),
      timeoutMs: opts.timeoutMs ?? DEFAULT_DRAIN_TIMEOUT_MS,
      force: opts.force ?? false,
      jobs: [],
    };
    const run: DrainRun = { state, ...(opts.strategyFor ? { strategyFor: opts.strategyFor } : {}) };
    this.drains.set(nodeId, run);
    this.cordonNode(nodeId);
    debuglog("Drain started", { nodeId, timeoutMs: state.timeoutMs, force: state.force });
    run.timer = this.clock.setTimeout(() => this.drainTimedOut(nodeId), state.timeoutMs);
    for (const jobId of this.jobIdsBoundTo(nodeId)) this.startDrainMigration(run, jobId);
    this.settings.onDrainProgress?.(snapshotDrain(state));
    this.reconcileDrain(nodeId);
    return snapshotDrain(state);
  }

  /**
   * The drain in flight for a node, or the last one that ran; undefined
   * for a node never drained by this instance.
   * @public
   */
  public drainStatus(nodeId: NodeId): DrainState | undefined {
    const run = this.drains.get(nodeId);
    return run ? snapshotDrain(run.state) : undefined;
  }

  /**
   * A provider has said it is taking a node back — a spot interruption
   * notice (EC2: two minutes' warning) or a rebalance recommendation (no
   * time given). Cordon the node and run it down with a FORCED drain
   * (`drainNode`) whose timeout fits the deadline: `deadline − now −
   * SPOT_INTERRUPTION_DRAIN_MARGIN_MS`, or with no deadline the kind's
   * default (`DEFAULT_SPOT_INTERRUPTION_DRAIN_MS`,
   * `DEFAULT_REBALANCE_DRAIN_MS`). Jobs that migrate in time land
   * elsewhere; whatever is still on the node at the timeout goes down with
   * it and is recovered per its failure policy — and none of it counts
   * against the restart ceiling: the node was reclaimed, the jobs did not
   * fail (`recoverLostNode` reads `reclaimedNodes`).
   *
   * A notice for a node already draining (the interruption that follows a
   * rebalance, say) keeps that drain — its migrations carry on — and
   * tightens its timeout when the new one ends sooner. `uncordonNode`
   * withdraws the notice along with the drain. Idempotent per notice;
   * `onNodeInterrupted` fires once per call.
   *
   * Who calls this: the event loop, on the daemon's `nodeInterruptionNotice`
   * (the worker on the instance watches IMDS — spot/instance-action and the
   * rebalance recommendation — and the daemon relays it); and norsk-mgr,
   * from any other source it has for a notice (an operator hook).
   * @public
   */
  public async handleNodeInterruption(nodeId: NodeId, notice: NodeInterruptionNotice): Promise<NodeInterruptionState> {
    const now = this.clock.now();
    const timeoutMs =
      notice.deadline !== undefined
        ? Math.max(0, notice.deadline.getTime() - now.getTime() - SPOT_INTERRUPTION_DRAIN_MARGIN_MS)
        : notice.kind === "rebalance"
          ? DEFAULT_REBALANCE_DRAIN_MS
          : DEFAULT_SPOT_INTERRUPTION_DRAIN_MS;
    this.reclaimedNodes.set(nodeId, {
      nodeId,
      kind: notice.kind,
      ...(notice.deadline !== undefined ? { deadline: notice.deadline } : {}),
      noticedAt: now,
    });
    debuglog("Node interruption notice", { nodeId, kind: notice.kind, deadline: notice.deadline, timeoutMs });
    const existing = this.drains.get(nodeId);
    if (existing && isDrainInFlight(existing.state)) {
      this.tightenDrain(existing, timeoutMs);
    } else {
      await this.drainNode(nodeId, {
        timeoutMs,
        force: true,
        ...(notice.strategyFor ? { strategyFor: notice.strategyFor } : {}),
      });
    }
    const state = this.nodeInterruption(nodeId);
    if (!state) throw new Error(`interruption of ${nodeId}: no drain recorded`);
    this.settings.onNodeInterrupted?.(state);
    return state;
  }

  /**
   * The interruption notice in force for a node, with its drain; undefined
   * for a node no provider has reclaimed (or one that has since stopped).
   * @public
   */
  public nodeInterruption(nodeId: NodeId): NodeInterruptionState | undefined {
    const notice = this.reclaimedNodes.get(nodeId);
    const drain = this.drainStatus(nodeId);
    if (!notice || !drain) return undefined;
    return { ...notice, drain };
  }

  /** A drain in flight gets a new, shorter time to go: it becomes forced
   *  (a reclaim leaves no `partial` to end in) and its timer is re-armed
   *  when `remainingMs` ends before the current timeout would. */
  private tightenDrain(run: DrainRun, remainingMs: number): void {
    const { state } = run;
    const elapsed = this.clock.now().getTime() - state.started.getTime();
    state.force = true;
    if (elapsed + remainingMs < state.timeoutMs) {
      state.timeoutMs = elapsed + remainingMs;
      run.timer?.cancel();
      run.timer = this.clock.setTimeout(() => this.drainTimedOut(state.nodeId), remainingMs);
      debuglog("Drain tightened", { nodeId: state.nodeId, timeoutMs: state.timeoutMs });
    }
    this.settings.onDrainProgress?.(snapshotDrain(state));
  }

  /** Jobs whose placement of record is this node (`bindings`). A
   *  migration TARGET bound here is not yet on the node's books. */
  private jobIdsBoundTo(nodeId: NodeId): JobId[] {
    return this.bindings.jobsOn(nodeId);
  }

  /** Start one job's migration for a drain and track it. The saga records
   *  its state before its first await, so the id is known on return; a
   *  refusal before that shows up as the promise's rejection. */
  private startDrainMigration(run: DrainRun, jobId: JobId): void {
    const entry: DrainJobState = { jobId, status: "migrating" };
    run.state.jobs.push(entry);
    const strategy = run.strategyFor?.(jobId);
    const started = this.migrateJob(jobId, strategy !== undefined ? { strategy } : {});
    const migrationId = this.bindings.migrationOf(jobId)?.migrationId;
    if (migrationId !== undefined) entry.migrationId = migrationId;
    started.catch((err) => {
      // A rejection after the saga recorded the migration has already
      // been through markMigrationAborted; this is for the refusals
      // before it (and says nothing new for the rest).
      if (entry.status !== "migrating") return;
      entry.status = "stuck";
      entry.reason = err instanceof Error ? err.message : String(err);
      debuglog("Drain: job cannot migrate", { nodeId: run.state.nodeId, jobId, reason: entry.reason });
      this.reconcileDrain(run.state.nodeId);
    });
  }

  /**
   * Bring a drain's state up to date with where its jobs are. Called
   * after every stream event and whenever a migration ends: jobs that
   * have left the node are marked so, jobs that turned up on it are
   * started on migrations of their own, and an empty node is
   * terminated.
   */
  private reconcileDrain(nodeId: NodeId): void {
    const run = this.drains.get(nodeId);
    if (!run) return;
    const { state } = run;
    if (state.phase !== "migrating" && state.phase !== "waiting") return;
    const onNode = new Set(this.jobIdsBoundTo(nodeId));
    let changed = false;
    for (const entry of state.jobs) {
      if (onNode.has(entry.jobId) || entry.status === "migrated" || entry.status === "moved") continue;
      const migration = entry.migrationId !== undefined ? this.migrationById(entry.migrationId) : undefined;
      entry.status = migration?.phase === "done" ? "migrated" : "moved";
      delete entry.reason;
      changed = true;
    }
    // A job bound here since the drain began (a migration that was
    // targeting this node has completed, say) is drained like the rest.
    const tracked = new Set(state.jobs.map((j) => j.jobId));
    for (const jobId of onNode) {
      if (tracked.has(jobId)) continue;
      this.startDrainMigration(run, jobId);
      changed = true;
    }
    // A migration's own target binding here keeps the node busy too.
    const targetsHere = this.bindings.targetsOn(nodeId).length > 0;
    if (onNode.size === 0 && !targetsHere) {
      this.terminateDrainedNode(run);
      return;
    }
    const phase: DrainPhase = state.jobs.some((j) => onNode.has(j.jobId) && j.status === "migrating")
      ? "migrating"
      : "waiting";
    if (phase !== state.phase) {
      state.phase = phase;
      changed = true;
    }
    if (changed) this.settings.onDrainProgress?.(snapshotDrain(state));
  }

  /** Every drain, after a stream event that may have moved a job. */
  private reconcileDrains(): void {
    for (const nodeId of this.drains.keys()) this.reconcileDrain(nodeId);
  }

  /** The node is empty: issue its terminate (once, whichever path — see
   *  `issueTerminate`) and end the drain on the RPC's outcome. */
  private terminateDrainedNode(run: DrainRun): void {
    const { state } = run;
    run.timer?.cancel();
    run.timer = undefined;
    state.phase = "terminating";
    this.settings.onDrainProgress?.(snapshotDrain(state));
    debuglog("Drain: node empty, terminating", { nodeId: state.nodeId });
    this.issueTerminate(state.nodeId).then(
      () => {
        if (state.phase !== "terminating") return;
        state.phase = "done";
        this.settings.onDrainProgress?.(snapshotDrain(state));
      },
      (err) => {
        if (state.phase !== "terminating") return;
        state.phase = "failed";
        state.reason = `terminate failed: ${err instanceof Error ? err.message : String(err)}`;
        this.settings.onDrainProgress?.(snapshotDrain(state));
        this.settings.onError?.(err);
      }
    );
  }

  private drainTimedOut(nodeId: NodeId): void {
    const run = this.drains.get(nodeId);
    if (!run) return;
    run.timer = undefined;
    const { state } = run;
    if (state.phase !== "migrating" && state.phase !== "waiting") return;
    const onNode = new Set(this.jobIdsBoundTo(nodeId));
    const remaining = state.jobs.filter((j) => onNode.has(j.jobId));
    if (remaining.length === 0) {
      this.reconcileDrain(nodeId);
      return;
    }
    const seconds = Math.round(state.timeoutMs / 1000);
    if (state.force) {
      for (const j of remaining) j.status = "terminated";
      state.reason =
        `timed out after ${seconds}s with ${remaining.length} job(s) still on the node; ` +
        `terminated anyway (force) — they recover per their failure policy`;
      debuglog("Drain timed out; terminating anyway (force)", { nodeId, jobs: remaining.map((j) => j.jobId) });
      this.terminateDrainedNode(run);
      return;
    }
    for (const j of remaining) {
      if (j.status !== "migrating") continue;
      j.status = "stuck";
      j.reason = "its migration was still in flight at the drain timeout";
    }
    state.phase = "partial";
    state.reason =
      `timed out after ${seconds}s with ${remaining.length} job(s) still on the node; ` +
      `nothing terminated, the node stays cordoned`;
    debuglog("Drain ended partial", { nodeId, jobs: remaining.map((j) => j.jobId) });
    this.settings.onDrainProgress?.(snapshotDrain(state));
  }

  /** End a drain that has not issued its terminate. No-op otherwise. */
  private cancelDrain(nodeId: NodeId, reason: string): void {
    const run = this.drains.get(nodeId);
    if (!run) return;
    const { state } = run;
    if (state.phase !== "migrating" && state.phase !== "waiting") return;
    run.timer?.cancel();
    run.timer = undefined;
    state.phase = "failed";
    state.reason = reason;
    debuglog("Drain cancelled", { nodeId, reason });
    this.settings.onDrainProgress?.(snapshotDrain(state));
  }

  /** A migration the drain started has ended: an abort leaves its job
   *  stuck if it is still on the node (a fail-forward re-place moves it,
   *  which the reconcile sees). */
  private noteDrainMigrationEnded(migration: MigrationState): void {
    const run = this.drains.get(migration.sourceNodeId);
    if (run) {
      const entry = run.state.jobs.find((j) => j.migrationId === migration.migrationId);
      if (entry && entry.status === "migrating" && migration.phase === "aborted") {
        entry.status = "stuck";
        entry.reason = `migration aborted: ${migration.abortReason ?? "unknown"}`;
      }
    }
    this.reconcileDrains();
  }

  /**
   * Shut down this AutoManager. Cancels every timer this instance owns,
   * stops the event loop, aborts every in-flight migration, and closes
   * the underlying NorskManager — which resolves the `Promise<void>`
   * returned from `run()`. Idempotent.
   *
   * An instance is never reconnected: norsk-mgr disposes it with
   * `close()` and runs a fresh one, so this is the only place an
   * in-flight migration can be wound down. Before the point of no
   * return that is the clean rollback (abort to both keys — a source
   * that answered a handover propose is locked until it hears it —
   * and StopJob for the target, which wears a key nobody will address
   * again); the source keeps running. Past it there is no rollback:
   * the source is stopping or stopped, so the migration is only marked
   * lost and the next instance rebuilds the job's placement from
   * history. A `migrateJob` still parked on its propose rejects.
   *
   * The channel stays open just long enough for those aborts to be
   * sent (`CLOSE_ABORT_GRACE_MS` at most); with nothing in flight it
   * closes immediately.
   * @public
   */
  public close(): void {
    if (this.closed) return;
    this.closed = true;
    this.cleanupInterval?.cancel();
    this.cleanupInterval = undefined;
    this.hotSpareReconcileInterval?.cancel();
    this.hotSpareReconcileInterval = undefined;
    this.nodeServiceReconcileInterval?.cancel();
    this.nodeServiceReconcileInterval = undefined;
    for (const t of this.startupGraceTimers.values()) t.cancel();
    for (const t of this.unseenBoundNodes.values()) t.cancel();
    this.unseenBoundNodes.clear();
    this.startupGraceTimers.clear();
    for (const t of this.pendingProvisionTimers.values()) t.cancel();
    this.pendingProvisionTimers.clear();
    for (const { timer } of this.pendingPlacementTimers.values()) timer.cancel();
    this.pendingPlacementTimers.clear();
    // A stop timer left live would fire deleteBundle on a closed client.
    for (const { timer } of this.pendingStopTimers.values()) timer.cancel();
    this.pendingStopTimers.clear();
    for (const t of this.pendingPlacementRetries.values()) t.cancel();
    this.pendingPlacementRetries.clear();
    this.bindings.clearAttempts();
    // A drain's timeout left live would terminate a node from a closed
    // client. The drain's state is left as it stands: the next instance
    // knows nothing of it, and its migrations are aborted just below.
    for (const run of this.drains.values()) {
      run.timer?.cancel();
      run.timer = undefined;
    }
    this.reclaimedNodes.clear();
    const aborts = this.migrationSaga.abortInFlightMigrations("AutoManager closed");
    // markMigrationAborted has disarmed each aborted migration's deadline;
    // this catches any other.
    this.migrationSaga.cancelTimers();
    if (aborts.length === 0) {
      this.norsk.close();
      return;
    }
    let finished = false;
    const finish = (): void => {
      if (finished) return;
      finished = true;
      grace.cancel();
      this.norsk.close();
    };
    const grace = this.clock.setTimeout(finish, CLOSE_ABORT_GRACE_MS);
    void Promise.allSettled(aborts).then(finish);
  }

  /**
   * Alias for `close()`. Provided for symmetry with the
   * `using`-style disposal conventions emerging in TypeScript.
   * @public
   */
  public dispose(): void {
    this.close();
  }

  /**
   * AutoManager v1 — submit a new bundle. The bundle is validated by
   * Manager (per §4.3) and expanded into N replica jobs server-side;
   * AutoManager mirrors the resulting state via the bundleUpdated +
   * jobUpdated events. The returned bundleId matches the one Manager
   * assigns (typically the same as bundle.bundleId, but Manager has
   * the final word).
   *
   * @public
   */
  public async createBundle(bundle: TypedBundle): Promise<BundleId> {
    return this.norsk.createBundle(toPbBundle(bundle));
  }

  /**
   * AutoManager v1 — update an existing bundle. Always sends
   * `state: live`: the state field is a daemon-owned verdict
   * (markJobFailed sets FAILED, the delete saga DELETING) that
   * read-modify-write callers naturally echo back — norsk-mgr's Retry
   * sent back the failed bundle it had just read, and the daemon
   * persisted FAILED straight over the revived production, leaving a
   * healthy re-expanded bundle labelled Failed forever (observed live
   * 2026-08-26). An operator pushing a spec update is reviving the
   * bundle by definition.
   * @public
   */
  public async updateBundle(bundle: TypedBundle): Promise<void> {
    await this.norsk.updateBundle(toPbBundle({ ...bundle, state: "live" }));
  }

  /**
   * AutoManager v1 — delete a bundle and all its replica jobs.
   * @public
   */
  public async deleteBundle(bundleId: BundleId): Promise<void> {
    await this.norsk.deleteBundle(bundleId);
  }

  /**
   * AutoManager v1 — activate a dormant (start_mode = "manual") bundle
   * job. Manager removes the dormant tag and re-emits jobUpdated;
   * AutoManager's placement path picks the job up via the standard
   * flow. See §10.2 of the design.
   * @public
   */
  public async startBundleJob(
    bundleId: BundleId,
    replicaIndex: number,
    jobName: string,
  ): Promise<void> {
    await this.norsk.startBundleJob(bundleId, replicaIndex, jobName);
  }

  /**
   * AutoManager v1 — deactivate a bundle job (tag-flip).
   *
   * Runtime termination of an in-flight workload and empty-node
   * release with linger are pre-go-live work tracked as Phase E.5
   * (§10.2.1). Until they land, calling stopBundleJob on an active
   * job marks it dormant for future placement but does not stop the
   * currently-running worker.
   * @public
   */
  public async stopBundleJob(
    bundleId: BundleId,
    replicaIndex: number,
    jobName: string,
  ): Promise<void> {
    await this.norsk.stopBundleJob(bundleId, replicaIndex, jobName);
  }

  /**
   * AutoManager v1 — typed view of the jobs Manager has told us about.
   * `JobWithHistory` carries the state-transition log for each job so
   * consumers can render lifecycle timelines without a separate
   * round-trip.
   * @public
   */
  public jobs(): JobWithHistory[] {
    return Array.from(this.jobMap.values());
  }

  public nodes(): NodeSummary[] {
    debuglog("Getting nodes", this.nodeMap);
    return Array.from(this.nodeMap.values());
  }

  /**
   * AutoManager v1 — typed view of the bundle definitions Manager has
   * told us about. Mirror only; AutoManager owns no durable state.
   * @public
   */
  public bundles(): TypedBundle[] {
    return Array.from(this.bundleMap.values());
  }

  /**
   * Phase D4 — snapshot of AutoManager's operational metrics.
   * Cumulative counters since `startedAt`; per-pool gauges computed
   * at call time. Cheap to call repeatedly — consumers poll on their
   * own cadence and compute deltas if they want rates.
   * @public
   */
  public metrics(): AutoManagerMetrics {
    const now = this.clock.now();
    const uptimeMs = this.startedAt
      ? now.getTime() - this.startedAt.getTime()
      : 0;
    const poolNames = new Set<string>();
    for (const p of this.settings.placementPools ?? []) poolNames.add(p.name);
    // Include any pool name we've seen on a node, in case nodes are
    // tagged with a pool that isn't in placementPools (rare but worth
    // surfacing — operator's likely misconfigured).
    for (const node of this.nodeMap.values()) {
      const pn = node.nodeMetadata?.tags?.["pool"];
      if (pn) poolNames.add(pn);
    }
    const pools: PoolMetrics[] = [];
    for (const poolName of poolNames) {
      pools.push(this.computePoolMetrics(poolName));
    }
    return {
      startedAt: this.startedAt,
      uptimeMs,
      ...this.metricsCounters,
      pools,
      brokenBundleIds: [...this.brokenBundles],
    };
  }

  private computePoolMetrics(poolName: string): PoolMetrics {
    let nodeCount = 0;
    let liveSpares = 0;
    let utilSum = 0;
    let utilN = 0;
    for (const node of this.nodeMap.values()) {
      if (this.poolNameForNode(node.nodeId) !== poolName) continue;
      nodeCount++;
      if (this.isHotSpareNode(node.nodeId) && node.nodeState === "running") {
        liveSpares++;
      }
      const inv = this.inventoryMap.get(node.nodeId);
      if (inv && inv.totalCapacity > 0) {
        utilSum += inv.reservedCapacity / inv.totalCapacity;
        utilN++;
      }
    }
    let spareTarget = 0;
    for (const cfg of this.settings.hotSpares ?? []) {
      if (cfg.pool === poolName) spareTarget += cfg.targetCount;
    }
    return {
      poolName,
      nodeCount,
      liveSpares,
      spareTarget,
      usedCapacityFraction: utilN > 0 ? utilSum / utilN : 0,
    };
  }

  /**
   * Look up a single bundle by id. Convenience over `bundles().find(...)`.
   * @public
   */
  public bundleById(bundleId: BundleId): TypedBundle | undefined {
    return this.bundleMap.get(bundleId);
  }

  /**
   * Every job (with history) that belongs to a given bundle. Derived
   * from `tags.bundleId` on the job entries in jobMap. Returns an empty
   * array if the bundle is unknown or has no jobs.
   * @public
   */
  public jobsForBundle(bundleId: BundleId): JobWithHistory[] {
    const out: JobWithHistory[] = [];
    for (const entry of this.jobMap.values()) {
      if ((entry.job.tags ?? {})["bundleId"] === bundleId) {
        out.push(entry);
      }
    }
    return out;
  }

  /**
   * The node the given job is currently placed on, or undefined if
   * the job hasn't been placed (or has been deleted). Useful for
   * operator UIs that render "job X is on node Y" and for proxy
   * routers that need to know which worker hosts each job.
   *
   * O(1) lookup against the job's binding — set at placement
   * time, deleted on jobDeleted, rebuilt from job history on R7
   * reconnect. Works regardless of whether the node hosts one job
   * (cloud, today) or many (cluster, or any future cloud N:1)
   * because the assignment is recorded against the JOB rather than
   * scanned out of NODE tags.
   * @public
   */
  public nodeForJob(jobId: JobId): NodeSummary | undefined {
    const nodeId = this.bindings.nodeOf(jobId);
    return nodeId !== undefined ? this.nodeMap.get(nodeId) : undefined;
  }

  /**
   * Current placement of every job AutoManager has bound to a node,
   * newest-first by decision timestamp. Mirrors `nodeForJob` but
   * returns the full `PlacementDecisionInfo` (so consumers get the
   * pool, trace, replicaIndex etc., not just the node) — and exposes
   * it as queryable live state rather than only via the
   * `onPlacementDecision` event stream.
   *
   * Hydrated jobs (placed in a previous AutoManager generation, picked
   * up on R7 reconnect) get an entry with an empty `trace.attempts`
   * and a decision inferred from job history. Consumers that *need*
   * a real trace should subscribe to `onPlacementDecision` for live
   * decisions; this getter is for "where is everything right now?"
   * operator views.
   *
   * @public
   */
  public placements(): PlacementDecisionInfo[] {
    return this.bindings.placements().sort((a, b) => b.at.getTime() - a.at.getTime());
  }

  /**
   * O(1) lookup for the most recent placement decision binding `jobId`
   * to a node. Returns undefined for jobs that have never been placed
   * (or whose placement attempts have only failed). @public
   */
  public placementForJob(jobId: JobId): PlacementDecisionInfo | undefined {
    return this.bindings.placementOf(jobId);
  }

  /**
   * Run the placement engine against AutoManager's current inventory
   * snapshot and configured pools, without actually placing or
   * provisioning anything. Powers preview affordances in operator
   * UIs ("if I submitted this bundle now, where would each replica
   * land?").
   *
   * `jobName` is optional; if omitted the *first* job in the bundle
   * is used. For multi-job bundles, supply the name to get a decision
   * specific to that job.
   *
   * Returns undefined if the bundle's named job is missing or the
   * replica index is out of range.
   * @public
   */
  public placeDryRun(
    bundle: TypedBundle,
    replicaIndex: number,
    jobName?: string
  ): { result: PlacementResult; trace: PlacementTrace } | undefined {
    if (replicaIndex < 0 || replicaIndex >= replicaCount(bundle)) return undefined;
    const job = jobName
      ? bundle.jobs.find((j) => j.jobName === jobName)
      : bundle.jobs[0];
    if (!job) return undefined;
    const pools = new Map(
      (this.settings.placementPools ?? []).map((p) => [p.name, p])
    );
    return placeWithTrace({
      bundle,
      job,
      replicaIndex,
      inventory: this.inventoriesAsNodeViews(),
      pools,
    });
  }

  /**
   * Static validation of a bundle against AutoManager's currently
   * configured pools. Equivalent to calling the module-level
   * `validateBundle` with `PoolDescriptor[]` derived from
   * `AutoSettings.placementPools` — elastic pools project their
   * `candidateInstanceTypes` into node profiles; cluster pools project
   * their currently-known live inventory.
   * @public
   */
  public validateBundle(bundle: TypedBundle): ValidationResult {
    return validateBundleFn(bundle, this.poolDescriptorsForValidation());
  }

  /**
   * Phase C4 — packaged admission check. Combines `validateBundle`
   * (static spec sanity) with `placeDryRun` for every (replica,
   * jobName) pair. Returns `ok` only when both pass for every
   * member; otherwise reports the structured failure. Use as a
   * pre-submit guard in operator-facing tooling.
   *
   * Every policy rejects at admission on "no capacity right now" —
   * there's no per-policy differentiation here. (Re-placing policies
   * get *queued* internally on capacity-after-
   * submit failures via the rejection re-place path, but that's a
   * runtime concern, not admission.)
   *
   * @public
   */
  public validateBundleAdmission(bundle: TypedBundle): AdmissionResult {
    const validation = this.validateBundle(bundle);
    if (validation.errors.length > 0) {
      return { kind: "validationErrors", errors: validation.errors };
    }
    const unplaceable: AdmissionUnplaceable[] = [];
    for (let replicaIndex = 0; replicaIndex < replicaCount(bundle); replicaIndex++) {
      for (const job of bundle.jobs) {
        const dry = this.placeDryRun(bundle, replicaIndex, job.jobName);
        if (!dry) {
          // placeDryRun returns undefined only for malformed input
          // (out-of-range replica, unknown jobName) — validateBundle
          // should have caught those already. Treat as unplaceable.
          unplaceable.push({
            replicaIndex,
            jobName: job.jobName,
            reason: { code: "noCapacityInAnyPool", triedPools: [] },
          });
          continue;
        }
        if (dry.result.kind === "failure") {
          unplaceable.push({
            replicaIndex,
            jobName: job.jobName,
            reason: dry.result.reason,
          });
        }
      }
    }
    if (unplaceable.length > 0) {
      return { kind: "noCapacity", unplaceable };
    }
    return { kind: "ok" };
  }

  private poolDescriptorsForValidation(): PoolDescriptor[] {
    const out: PoolDescriptor[] = [];
    for (const pool of this.settings.placementPools ?? []) {
      // Union the feasibility profiles across the pool's tiers: elastic
      // tiers contribute the shapes they can provision; fixed (cluster)
      // tiers contribute the live inventory projected for them.
      const profiles: NodeProfile[] = [];
      for (const tier of pool.tiers) {
        if (tier.scaleOut === "elastic") {
          for (const opt of tier.candidateInstanceTypes) {
            profiles.push({
              totalCapacity: opt.totalCapacity,
              totalCores: opt.totalCores,
              capabilities: opt.capabilities,
            });
          }
        } else {
          for (const view of this.inventoriesAsNodeViews()) {
            if (!nodeMatchesTier(view, pool, tier)) continue;
            profiles.push({
              totalCapacity: view.inventory.totalCapacity,
              totalCores: view.inventory.totalCores,
              capabilities: view.inventory.capabilities,
            });
          }
        }
      }
      // A pool that can provision is judged on what it can provision (a
      // miss is a hard error); a purely-fixed pool's misses are warnings —
      // a worker may join after submission (see validation.ts).
      const kind = pool.tiers.some((t) => t.scaleOut === "elastic")
        ? "elastic"
        : "cluster";
      const clouds = [...new Set(pool.tiers.map((t) => t.kind))];
      out.push({ poolName: pool.name, kind, nodeProfiles: profiles, clouds });
    }
    return out;
  }

  /**
   * AutoManager v1 — typed inventory snapshot for a single node, or
   * undefined if the worker hasn't (yet) advertised one.
   *
   * Reservations are the same derived overlay placement scores against
   * (see `inventoriesAsNodeViews`): the wire values arrive as 0 / [] —
   * workers advertise raw hardware — so without the overlay a consumer
   * (e.g. norsk-mgr's /api/inventory → the Nodes page) shows a node
   * hosting jobs as "0/64 reserved" while placement correctly sees it
   * as loaded. One view of the truth, both here and there.
   * @public
   */
  public inventory(nodeId: NodeId): TypedNodeInventory | undefined {
    const inv = this.inventoryMap.get(nodeId);
    if (!inv) return undefined;
    return this.withReservationOverlay(inv, this.placedJobsByNode().get(nodeId) ?? []);
  }

  /**
   * AutoManager v1 — typed inventory snapshots for every node we know
   * about. Iteration order is insertion order. Reservations carry the
   * same derived overlay as `inventory()` above.
   * @public
   */
  public inventories(): { nodeId: NodeId; inventory: TypedNodeInventory }[] {
    const placed = this.placedJobsByNode();
    return Array.from(this.inventoryMap.entries()).map(([nodeId, inventory]) => ({
      nodeId,
      inventory: this.withReservationOverlay(inventory, placed.get(nodeId) ?? []),
    }));
  }

  /**
   * The one place a wire inventory becomes a typed one. The converter
   * enforces the `utilizationCap` contract (a fraction of totalCapacity;
   * an absolute — the vCPU count a pre-fix worker sent — is read as 1 so
   * it cannot inflate the node) and reports the violation; this reports it
   * to the operator ONCE per node rather than on every poll.
   */
  private typedInventoryFrom(nodeId: NodeId, pb: CommonPB.NodeInventory): TypedNodeInventory {
    let violation: InventoryContractViolation | undefined;
    const typed = fromPbNodeInventory(pb, (v) => {
      violation = v;
    });
    if (violation === undefined) {
      this.inventoryContractWarned.delete(nodeId);
      return typed;
    }
    if (!this.inventoryContractWarned.has(nodeId)) {
      this.inventoryContractWarned.add(nodeId);
      const v: InventoryContractViolation = violation;
      const msg =
        `node ${nodeId} advertises ${v.field} ${v.advertised}; the wire contract is a fraction (0-1] ` +
        `of totalCapacity — a pre-fix norsk-ctl worker sends its vCPU count — reading it as ${v.used} ` +
        `(no soft cap) so the node cannot advertise more than its declared capacity. Upgrade the worker.`;
      debuglog("Inventory contract violation", { nodeId, violation: v });
      this.settings.onError?.(new Error(msg));
    }
    return typed;
  }

  // ---- AutoManager v1 event handlers ----
  private handleNodeInventoryUpdated(raw: ManagerPB.NodeInventoryUpdated): void {
    if (!raw.nodeId || !raw.inventory) return;
    const nodeId = raw.nodeId.nodeId;
    // A real inventory supersedes the prospective view wholesale; jobs that
    // no longer fit re-place via the reconcile below.
    if (this.pendingProvisions.has(nodeId)) {
      this.pendingProvisions.delete(nodeId);
      this.pendingProvisionTimers.get(nodeId)?.cancel();
      this.pendingProvisionTimers.delete(nodeId);
    }
    this.inventoryMap.set(nodeId, this.typedInventoryFrom(nodeId, raw.inventory));
    // NB: a routine inventory update does NOT clear an "unhealthy" /
    // recently-failed mark. Workers heartbeat inventory continuously, so
    // clearing here meant a node that reliably rejects (e.g. a host-port
    // it can never bind) shed its mark seconds later and was picked again —
    // the mark could never accumulate to escalation, and a rejecting node
    // was re-tried indefinitely (observed live 2026-08-25). The mark is
    // cleared only by real recovery: a job reaching `running` on the node
    // (see clearNodeHealthMarks), or the mark's own cooldown expiring
    // (currentExcludeSet). Inventory arriving ≠ "I'm healthy now".
    // Phase C6: inventory change can flip a spare to "consumed" (jobs
    // assigned > 0). Reconcile so a replacement is provisioned
    // promptly rather than waiting for the periodic tick.
    this.reconcileHotSpares();
    // Phase F: inventory change might newly satisfy / no-longer satisfy
    // a NodeService's `consumes` — reconcile placements.
    this.reconcileNodeServices();
    // Fresh inventory means capacity may have appeared (a node's first
    // report after Hello — the startup race — a freed reservation, or an
    // un-cordon). Re-attempt any job that couldn't be placed before.
    this.reconcilePendingPlacements();
  }

  private handleJobRejected(raw: ManagerPB.JobRejected): void {
    if (!raw.nodeId || !raw.jobId) return;
    const nodeId = raw.nodeId.nodeId;
    const jobId = raw.jobId.jobId;
    const typedInventory = raw.inventory
      ? this.typedInventoryFrom(nodeId, raw.inventory)
      : undefined;
    if (typedInventory) {
      // Carries the worker's latest view, but does NOT clear unhealthy —
      // a rejection is the opposite of "I'm fine". Update the cache only.
      this.inventoryMap.set(nodeId, typedInventory);
    }

    this.metricsCounters.jobsRejected++;
    this.settings.onJobRejected?.({
      jobId,
      nodeId,
      reason: raw.reason,
      inventory: typedInventory,
      at: this.clock.now(),
    });

    // `invalidSpec` is about the JOB and only the job: the worker read the
    // spec and found it unrunnable anywhere (a missing required parameter,
    // a bad product template, a tar sha mismatch, an unrenderable compose).
    // No node is to blame, and no node would do better — so it is not a
    // recovery at all. End the job the way the ceiling ends one: a durable
    // verdict on the wire, the bundle broken, the operator told, with the
    // worker's own words as the reason. Retrying would only re-run the
    // same rejection on every node in the pool until the ceiling said the
    // same thing less clearly.
    if (raw.reason === CommonPB.JobRejectedReason.JOB_REJECTED_INVALID_SPEC) {
      this.endJobForInvalidSpec(jobId, nodeId, raw.detail?.value);
      return;
    }

    // Does this rejection say anything about the NODE?
    //
    // `insufficientCapacity` and `hardwareUnavailable` do: they are facts
    // about where the job was sent — it has no room, or it lacks a device —
    // so somewhere else may work and this node has earned a mark.
    //
    // `unknown` does NOT. It is the worker's catch-all for a launch failure
    // it could not classify (the worker classifies device / host-port
    // faults as `hardwareUnavailable` and spec faults as `invalidSpec`;
    // everything else is `unknown`), and the common case is a fault
    // intrinsic to the JOB — an image that cannot be pulled, a compose that
    // cannot come up — which no node would run. Reading it as evidence
    // against the node is wrong twice over:
    //
    //  - it excludes the rejecting node from the re-place, so on a
    //    single-node pool the recovery fails `noCapacityInAnyPool`, nothing
    //    re-triggers, and the production sits `live` / `broken: false` with
    //    no container forever (measured 2026-09-04, e2e scenario 06);
    //  - three of them inside the rejection window would mark a perfectly
    //    healthy node unhealthy for the whole cooldown (10 min), evicting
    //    OTHER productions' placements over a fault in ours.
    //
    // Retrying the same node converges instead: either it starts, or the
    // repeated rejection trips the restart ceiling → markJobFailed →
    // BundleBroken, which is a visible ending. That is exactly the reasoning
    // `handleDispatchFailure` already applies to its own re-place.
    const blamesNode =
      raw.reason !== CommonPB.JobRejectedReason.JOB_REJECTED_UNKNOWN;

    if (blamesNode) {
      const escalated = this.recordRejection(nodeId);
      if (escalated) {
        debuglog("Node escalated to unhealthy", { nodeId });
        this.settings.onNodeUnhealthy?.(nodeId);
      }
    }

    // A rejection of the job's MIGRATION TARGET is the saga's to handle:
    // the target is lost (abort, or fail forward past the point of no
    // return). It says nothing about the source, which still runs where
    // its binding says — the ordinary re-place below must not touch it.
    const migration = this.migrationForJob(jobId);
    if (migration && this.bindings.isTargetOn(jobId, nodeId)) {
      void this.migrationSaga
        .handleTargetLost(migration, `target rejected by worker on ${nodeId}: ${CommonPB.JobRejectedReason[raw.reason]}`)
        .catch((err) => this.settings.onError?.(err));
      return;
    }

    // The daemon released the job's provision-time binding as it recorded the
    // rejection (NodeController.processJobRejected), so our mirror must drop
    // it too. Left set, `nodeForJob` reports a node for a runtime that is not
    // there — and `inventoriesAsNodeViews` derives each node's reservation
    // from the bindings, so the phantom goes on consuming capacity on a node
    // that is in fact free. The release also clears the dispatch-in-flight
    // gate: the attempt conclusively failed, so it must not hold the
    // re-place back.
    this.bindings.releaseFrom(jobId, nodeId);

    const existing = this.jobMap.get(jobId);
    const existingReq = existing?.job.autoPlacement?.requirements;
    if (existing && existingReq) {
      // Re-place the job, excluding the nodes currently considered unfit —
      // plus the rejecting one, but only when it is what was rejected.
      const exclude = this.currentExcludeSet();
      if (blamesNode) exclude.add(nodeId);
      void this.placeAndProvision(jobId, existing.job.tags ?? {}, { excludeNodes: exclude }).catch((err) =>
        this.settings.onError?.(err)
      );
    }
  }

  /**
   * The ending for a job whose spec the worker refused as unrunnable
   * (`invalidSpec`): release the provision-time binding the daemon has
   * already dropped, mark the job terminally failed on the wire, and end
   * the production — the same ending the restart ceiling gives, reached
   * on the first rejection because a spec fault cannot improve with
   * retries. A migration target rejected this way is the saga's: the
   * source still runs, and aborting the migration is the right ending.
   */
  private endJobForInvalidSpec(jobId: JobId, nodeId: NodeId, detail: string | undefined): void {
    const why = detail && detail.length > 0 ? detail : "the worker rejected the job spec as invalid";
    const migration = this.migrationForJob(jobId);
    if (migration && this.bindings.isTargetOn(jobId, nodeId)) {
      void this.migrationSaga
        .handleTargetLost(migration, `target rejected by worker on ${nodeId}: invalid spec — ${why}`)
        .catch((err) => this.settings.onError?.(err));
      return;
    }
    this.bindings.releaseFrom(jobId, nodeId);
    const bundleId = this.jobMap.get(jobId)?.job.tags?.["bundleId"];
    const reason = `runtime ${jobId} has an invalid spec (rejected by ${nodeId}): ${why}`;
    debuglog("Job rejected for an invalid spec — ending it, no re-place", { jobId, nodeId, detail });
    this.issueTerminalFailure(jobId, reason);
    // The same essential/non-essential split as the ceiling: a runtime the
    // production can live without takes itself out and leaves the rest
    // running; everything else ends the production.
    if (this.specForPlacedJob(jobId)?.requirements.essential === false) {
      this.metricsCounters.jobsDropped++;
      this.settings.onJobDropped?.(jobId);
      return;
    }
    if (bundleId !== undefined && !this.brokenBundles.has(bundleId)) {
      this.brokenBundles.add(bundleId);
      this.metricsCounters.bundlesBroken++;
      this.settings.onBundleBroken?.(bundleId, reason);
    }
  }

  private recordRejection(nodeId: NodeId): boolean {
    const now = this.clock.now();
    const window = this.cfgRejectionBackoffMs();
    const previous = this.rejectionsPerNode.get(nodeId) ?? [];
    const recent = previous.filter((d) => now.getTime() - d.getTime() < window);
    recent.push(now);
    this.rejectionsPerNode.set(nodeId, recent);
    const threshold = this.cfgRejectionEscalationCount();
    if (recent.length >= threshold && !this.unhealthyNodes.has(nodeId)) {
      this.unhealthyNodes.set(nodeId, now);
      return true;
    }
    return false;
  }

  /** Clear every placement-exclusion mark for a node that has demonstrably
   *  recovered (a job just reached `running` on it). Fires `onNodeHealthy`
   *  so consumers mirroring the unhealthy set (e.g. norsk-mgr's state
   *  counter) can drop it too. */
  private clearNodeHealthMarks(nodeId: NodeId): void {
    const wasMarked =
      this.unhealthyNodes.delete(nodeId);
    this.rejectionsPerNode.delete(nodeId);
    this.recentFailures.delete(nodeId);
    if (wasMarked) {
      debuglog("Node health marks cleared — job reached running", { nodeId });
      this.settings.onNodeHealthy?.(nodeId);
    }
  }

  private currentExcludeSet(): Set<NodeId> {
    const now = this.clock.now().getTime();
    const cooldown = this.cfgUnhealthyCooldownMs();
    const exclude = new Set<NodeId>();
    // Unhealthy marks age out after the cooldown so a node is retried
    // rather than excluded forever (routine inventory no longer clears
    // them — see handleNodeInventoryUpdated). A node that keeps rejecting
    // re-earns the mark on its next rejection.
    for (const [nodeId, markedAt] of this.unhealthyNodes) {
      if (now - markedAt.getTime() >= cooldown) {
        this.unhealthyNodes.delete(nodeId);
        this.settings.onNodeHealthy?.(nodeId);
      } else {
        exclude.add(nodeId);
      }
    }
    for (const [nodeId, deadline] of this.recentFailures) {
      if (deadline.getTime() < now) {
        this.recentFailures.delete(nodeId);
      } else {
        exclude.add(nodeId);
      }
    }
    return exclude;
  }

  /**
   * Phase C2 — restart-rate check. Records the current attempt; if the
   * trailing-hour count crosses the job's `maxRestartsPerHour`, marks
   * the bundle broken, fires `onBundleBroken`, and returns true
   * (caller must short-circuit). Returns false when within budget.
   *
   * WHAT THE CEILING MEASURES — three rules, all of them about not
   * counting the same trouble twice:
   *
   *  - It counts FAILURES, not attempts. A recovery re-entering after
   *    its backoff (`retryOfCountedFailure`) is the same failure still
   *    being worked on, and is not counted again.
   *  - A run of consecutive dispatch refusals for the same job, with the
   *    same reason, against the same node is ONE condition, not N
   *    crashes (`countsTowardCeiling: false` — see
   *    `repeatOfCountedCondition`). The daemon holding a leaked binding
   *    and answering `Job already running on node` every 5 s used to
   *    burn a ceiling of 100 in half an hour on a healthy production
   *    (2026-09-09). A DIFFERENT reason, or a different node, is a new
   *    count; so is the same refusal after the job has run in between.
   *  - It counts nothing a provider RECLAIM does. A node lost to a spot
   *    interruption (`handleNodeInterruption`) takes its jobs with it
   *    through no fault of theirs; their re-place is recorded for the
   *    backoff and not counted (`reclaimed` on the recovery arg).
   *  - It counts nothing a MIGRATION does. A handover is a planned move,
   *    not a restart: the target is placed through the `migration` arm of
   *    `placeAndProvision`, which is neither initial nor recovery, and
   *    the source's `JobStopped(migration)` is consumed by the saga. The
   *    one exception is deliberate — a fail-forward past the point of no
   *    return has genuinely lost the job, and its re-place counts once,
   *    like any other loss.
   *
   * A non-counting attempt is still RECORDED, because the minimum-interval
   * backoff paces off the newest entry — skipping the record would let a
   * repeat storm at wire speed.
   */
  private shouldBreakOnRestart(
    jobId: JobId,
    bundleId: BundleId,
    onFailure: OnRuntimeFailure,
    jobOverride?: number,
    essential?: boolean,
    countsTowardCeiling = true
  ): boolean {
    const now = this.clock.now();
    const previous = this.restartHistory.get(jobId) ?? [];
    const recent = previous.filter(
      (d) => now.getTime() - d.at.getTime() < RESTART_WINDOW_MS
    );
    recent.push({ at: now, counts: countsTowardCeiling });
    this.restartHistory.set(jobId, recent);
    if (!countsTowardCeiling) return false;
    const limit = this.cfgMaxRestartsPerHour(onFailure, jobOverride);
    const counted = recent.reduce((n, e) => (e.counts ? n + 1 : n), 0);
    if (counted > limit) {
      // A runtime the production can live without takes itself out and leaves
      // the rest running. Everything else ends the production: a production is
      // a set of cooperating runtimes, so losing one of them ends the job it
      // was cooperating on.
      if (essential === false) {
        debuglog("Non-essential runtime exhausted its ceiling — failing it alone", {
          jobId,
          bundleId,
        });
        this.metricsCounters.jobsDropped++;
        this.settings.onJobDropped?.(jobId);
        this.issueTerminalFailure(jobId, `runtime failed after exceeding its redeploy ceiling (${limit}/h)`);
        return true;
      }
      if (!this.brokenBundles.has(bundleId)) {
        this.brokenBundles.add(bundleId);
        debuglog("Bundle marked broken — restart rate exceeded", {
          bundleId,
          jobId,
          attempts: counted,
          limit,
        });
        this.metricsCounters.bundlesBroken++;
        const reason = `Restart rate ${counted}/h exceeds ceiling ${limit}/h for replica ${jobId}`;
        this.settings.onBundleBroken?.(bundleId, reason);
        // Persist the verdict back to the daemon so the poison job is not
        // re-placed after a restart (the in-process `brokenBundles` flag is
        // rebuilt-from-scratch on reconnect). Fire-and-forget; the live flag
        // above already suppresses further placement this session.
        this.issueTerminalFailure(jobId, reason);
      }
      return true;
    }
    return false;
  }

  /**
   * How many failures the restart ceiling currently attributes to this
   * job — the number `maxRestartsPerHour` is compared against, within
   * the trailing hour. Attempts the ceiling deliberately does not count
   * (a backoff retry of a failure already counted, a repeat of a standing
   * dispatch refusal) are excluded, and a migration contributes nothing;
   * see `shouldBreakOnRestart` for what that means.
   * @public
   */
  public restartCount(jobId: JobId): number {
    const now = this.clock.now().getTime();
    return (this.restartHistory.get(jobId) ?? []).reduce(
      (n, e) => (e.counts && now - e.at.getTime() < RESTART_WINDOW_MS ? n + 1 : n),
      0
    );
  }

  /**
   * Whether the named bundle has tripped the restart ceiling and is
   * being held by AutoManager. Cleared by `bundleUpdated` (operator
   * pushed a fresh spec) or `bundleDeleted`.
   * @public
   */
  public isBundleBroken(bundleId: BundleId): boolean {
    return this.brokenBundles.has(bundleId);
  }

  // ---- migration state machine --------------------------------------
  //
  // norsk-ctl docs/internal/design/job-migration.md §3. Three strategies,
  // one saga, owned entirely here: Manager is a pass-through and the
  // worker holds no protocol state. The target is PLACED through
  // `placeAndProvision` like any other job (source node excluded, an
  // explicit target pinned) and started with a plain StartJob carrying
  // the migration fields; from then on the saga advances on the job's
  // own history events — keyed on the INSTANCE KEY, never the slot —
  // plus the four handshake events for `handover`.
  //
  // Point of no return: the moment StopJob(source) is sent. Before it an
  // abort is a clean rollback (abort to both keys, stop the target,
  // release its reservation; the source never noticed). After it the
  // saga only fails forward — it keeps waiting on the target, and if the
  // target is lost the migration is marked aborted and ordinary recovery
  // re-places the job under its slot key.

  // ---- Phase E — job migration ----------------------------------------
  // The saga is src/migrationSaga.ts; these delegate.

  /**
   * Migrate a job's running instance to another node. `strategy` picks
   * the saga (`stopThenStart` by default, `overlap`, or `handover`);
   * `targetNodeId` pins the target; `sourceRole` names the instance when
   * it is not the job's current one; `overrideReservation` lets the
   * target take an exclusively reserved hot spare. Resolves with the
   * migration id once the target is placed and launching; rejects — with
   * the migration aborted — when the source refuses, nothing can host the
   * target, or the target's dispatch fails.
   * @public
   */
  public async migrateJob(
    sourceJobId: JobId,
    opts: {
      targetNodeId?: NodeId;
      strategy?: MigrationStrategy;
      sourceRole?: Role;
      overrideReservation?: boolean;
    } = {}
  ): Promise<MigrationId> {
    return this.migrationSaga.migrateJob(sourceJobId, opts);
  }

  /**
   * Abort an in-flight migration cleanly: MigrationAbort to both keys,
   * StopJob for the target, its reservation released, the source
   * untouched. No-op (with a debug log) for an unknown migration id or
   * one already done/aborted. Rejects once the migration is past the
   * point of no return — the source has already been told to stop, so
   * there is nothing to roll back to; the saga will complete into the
   * target, or fall into ordinary recovery if the target is lost.
   * @public
   */
  public async abortMigration(migrationId: MigrationId, reason: string): Promise<void> {
    await this.migrationSaga.abortMigration(migrationId, reason);
  }

  /** All known migrations (active + done + aborted). @public */
  public migrations(): MigrationState[] {
    return this.migrationSaga.migrations();
  }

  /** Look up a single migration by id. @public */
  public migrationById(migrationId: MigrationId): MigrationState | undefined {
    return this.migrationSaga.migrationById(migrationId);
  }

  /** Migration active for the given source jobId, if any. @public */
  public migrationForJob(jobId: JobId): MigrationState | undefined {
    return this.migrationSaga.migrationForJob(jobId);
  }

  /**
   * Forget a TERMINAL migration — `migrations()` keeps every migration
   * this AutoManager has ever run, which is an unbounded list an operator
   * has no other way to shorten. Refuses (throws) an in-flight one: abort
   * it first. Returns false for an id that is not (or is no longer)
   * known, so a double-clear is a no-op rather than an error.
   * @public
   */
  public forgetMigration(migrationId: MigrationId): boolean {
    return this.migrationSaga.forgetMigration(migrationId);
  }

  /**
   * Forget every terminal migration, leaving the in-flight ones alone.
   * Returns the ids forgotten. @public
   */
  public forgetTerminalMigrations(): MigrationId[] {
    return this.migrationSaga.forgetTerminalMigrations();
  }

  private inFlightMigrations(): MigrationState[] {
    return this.migrationSaga.inFlightMigrations();
  }

  /**
   * The instance KEY of the job's current instance: the key its last
   * completed migration left it under, else its slot. This is the role
   * every RPC that addresses the RUNNING instance must use.
   */
  private roleForJob(jobId: JobId): Role {
    return this.migrationSaga.roleForJob(jobId);
  }

  /** The job's runtime SLOT ("primary" / "backup"), from its replicaIndex tag. */
  private slotForJob(jobId: JobId): Role | undefined {
    return this.migrationSaga.slotForJob(jobId);
  }

  // ---- Phase C4 — config resolution ---------------------------------
  // Precedence for every tunable: settings.placement.X > settings.X
  // > compiled-in default. Per-policy tunables use settings.failurePolicies[p]
  // first, with the same flat/default fallback.

  private cfgRejectionBackoffMs(): number {
    return (
      this.settings.placement?.rejectionBackoffMs ??
      this.settings.rejectionBackoffMs ??
      DEFAULT_REJECTION_BACKOFF_MS
    );
  }

  private cfgRejectionEscalationCount(): number {
    return (
      this.settings.placement?.rejectionEscalationCount ??
      this.settings.rejectionEscalationCount ??
      DEFAULT_REJECTION_ESCALATION_COUNT
    );
  }

  private cfgUnhealthyCooldownMs(): number {
    return (
      this.settings.placement?.unhealthyCooldownMs ??
      this.settings.unhealthyCooldownMs ??
      DEFAULT_UNHEALTHY_COOLDOWN_MS
    );
  }

  private cfgFailureBackoffMs(): number {
    return (
      this.settings.placement?.failureBackoffMs ??
      this.settings.failureBackoffMs ??
      DEFAULT_FAILURE_BACKOFF_MS
    );
  }

  private cfgJobStartupGraceMs(): number {
    return (
      this.settings.placement?.jobStartupGraceMs ??
      this.settings.jobStartupGraceMs ??
      DEFAULT_JOB_STARTUP_GRACE_MS
    );
  }

  /** The ceiling for one job: its own override first, then the per-policy
   *  setting, then the flat one, then the compiled-in default.
   *
   *  The per-job override leads because it is the most specific statement
   *  anyone made — an operator who says "this encoder gets 2 tries an hour"
   *  means it regardless of what the deployment-wide default happens to be. */
  private cfgMaxRestartsPerHour(
    onFailure: OnRuntimeFailure,
    jobOverride?: number
  ): number {
    // `drop` never restarts — the ceiling is moot, but return a large
    // number so the rate-check path can be called uniformly without
    // crashing if anyone reaches it.
    if (onFailure === "drop") return Number.POSITIVE_INFINITY;
    const perPolicy =
      this.settings.failurePolicies?.[onFailure]?.maxRestartsPerHour;
    return (
      jobOverride ??
      perPolicy ??
      this.settings.maxRestartsPerHour ??
      DEFAULT_MAX_RESTARTS_PER_HOUR
    );
  }

  // ---- AutoManager v1 placement-driven path ----
  //
  // A "modern" job — one with `requirements` set — is a member of a
  // bundle that Manager has expanded into N replica jobs. The job
  // carries bundleId / replicaIndex / jobName tags so AutoManager can
  // recover its bundle context and run the placement engine. The
  // legacy createJob path (no requirements) is unaffected.

  private async placeAndProvision(
    jobId: JobId,
    tags: { [key: string]: string },
    recovery?: {
      excludeNodes: Set<NodeId>;
      retryOfCountedFailure?: boolean;
      repeatOfCountedCondition?: boolean;
      /** The job was lost with a node its provider reclaimed
       *  (`handleNodeInterruption`): recorded for the backoff, not
       *  counted by the ceiling. */
      reclaimed?: boolean;
    },
    migration?: MigrationPlacement
  ): Promise<PlacementOutcome> {
    const notPlaced = (reason: string): PlacementOutcome => ({ kind: "notPlaced", reason });
    const bundleId = tags["bundleId"];
    const replicaIndexStr = tags["replicaIndex"];
    const jobName = tags["jobName"];
    if (!bundleId || replicaIndexStr === undefined || !jobName) {
      debuglog("Job has requirements but missing bundle context tags", tags);
      return notPlaced("job has requirements but is missing bundle context tags");
    }
    const replicaIndex = parseInt(replicaIndexStr, 10);

    // In-flight-placement gate (initial placements only). handleJobChanged
    // can re-fire `placeAndProvision` for the same jobId while the
    // daemon's provisionJob is mid-dispatch — `nodesByJob().size` is
    // still 0 because JobStarted hasn't landed yet. Each re-fire ends
    // in `INVALID_ARGUMENT: Job already running on node` from the
    // daemon and a `placementsDispatchFailed++`. Suppress. Recovery
    // placements bypass this gate; they're driven by failure events
    // (JobRejected, NodeStopped) and have their own MIN_RECOVERY_INTERVAL_MS
    // backoff further down — and the drop / policy-aware-recovery
    // branch needs to run even when an initial placement just dispatched.
    // A migration target bypasses it too: it is neither an initial
    // placement nor a recovery, and the job's own binding (the source)
    // is untouched by it — the "no concurrent migrations per job" rule
    // in migrateJob is its gate.
    const now = this.clock.now();
    if (!recovery && !migration) {
      const lastAttempt = this.bindings.lastAttemptOf(jobId);
      if (lastAttempt && now.getTime() - lastAttempt.getTime() < MIN_PLACEMENT_INTERVAL_MS) {
        debuglog("Placement attempt suppressed (in-flight)", { jobId, bundleId });
        return notPlaced("placement attempt suppressed (in-flight)");
      }
      this.bindings.noteAttempt(jobId, now);
    }

    const bundle = this.bundleMap.get(bundleId);
    if (!bundle) {
      debuglog("Job's bundle not in map yet", { jobId, bundleId });
      return notPlaced(`bundle ${bundleId} not known`);
    }
    const bundleJobSpec = bundle.jobs.find((j) => j.jobName === jobName);
    if (!bundleJobSpec) {
      debuglog("Bundle has no job named", { bundleId, jobName });
      return notPlaced(`bundle ${bundleId} has no job named ${jobName}`);
    }
    // Placement scores the bundle's typed requirements (`bundleJobSpec`),
    // the canonical spec the operator submitted — not the job's own PB
    // copy, which callers only use to tell a modern job from a legacy one.

    // Policy-aware recovery (Phase C1). On a recovery path:
    //   - drop:        do not re-place; fire onJobDropped.
    //   - preferSpare: re-place, biasing scoring toward hot spares.
    //   - avoidSpare:  re-place, penalising unconsumed spares so the
    //                  reserve survives a routine failure.
    // Initial placements bypass this entirely.
    const onFailure = bundleJobSpec.requirements.onRuntimeFailure;
    if (recovery && onFailure === "drop") {
      debuglog("Job dropped (onRuntimeFailure: drop)", { jobId, bundleId });
      this.metricsCounters.jobsDropped++;
      this.settings.onJobDropped?.(jobId);
      // Terminal — persist the verdict so the job isn't re-placed after a
      // restart.
      this.issueTerminalFailure(jobId, "job dropped after failure (onRuntimeFailure: drop)");
      // …and the production ends with it, unless this runtime was one it can
      // live without. This route used to be silently non-fatal while the
      // ceiling route ended the production — the same question with two
      // different answers depending on how the runtime happened to be lost.
      if (bundleJobSpec.requirements.essential !== false && !this.brokenBundles.has(bundleId)) {
        this.brokenBundles.add(bundleId);
        this.metricsCounters.bundlesBroken++;
        this.settings.onBundleBroken?.(
          bundleId,
          `essential runtime ${jobId} was dropped after failure (onRuntimeFailure: drop)`
        );
      }
      return notPlaced("job dropped (onRuntimeFailure: drop)");
    }

    // Phase C2 — restart-rate ceiling + broken-bundle short-circuit.
    // Always run the rate check on every recovery trigger — the
    // ceiling tracks failure rate, which is real-time, even if we
    // defer the actual placement below. Without this, the
    // minimum-interval backoff would coalesce N rapid failures into
    // a single recorded attempt and the rate-broken trip would never
    // fire.
    // `retryOfCountedFailure` marks the backoff timer re-entering with the
    // SAME failure it already counted below. It is not a new failure, so it
    // skips both the count and the spacing gate — the delay it was waiting
    // for has by definition just elapsed.
    //
    // Without that distinction the two roles fed each other: every re-entry
    // pushed a fresh `restartHistory` entry, which is also what the backoff
    // reads for "previous attempt", so two interleaved retry chains kept each
    // other permanently inside the 5s window — deferring forever, never
    // placing, while the counter ratcheted to the ceiling. Measured
    // 2026-09-04: one container death produced deferrals in pairs 25ms apart
    // every 5s and marked the bundle broken in fifteen seconds.
    //
    // Counting stays on every FRESH recovery trigger, which is what keeps a
    // genuinely flapping runtime tripping the ceiling (three rapid rejections
    // are three failures, and must still trip a ceiling of two).
    //
    // `repeatOfCountedCondition` is the SECOND thing that is not a new
    // failure, and it is a different shape from the first. Where
    // `retryOfCountedFailure` says "this is us re-entering with the failure
    // we already counted", this says "the wire answered us with the same
    // standing refusal, from the same node, that we already counted" — a
    // leaked daemon binding insisting `Job already running on node`, once
    // per attempt, forever. One condition, one count. It still records the
    // attempt (so the backoff paces the next one) and still respects the
    // broken-bundle short-circuit; only the ceiling ignores it. Set by
    // `handleDispatchFailure`, which owns the comparison — see
    // `classifyDispatchFailure`.
    //
    // The broken-bundle short-circuit is outside both flags: a deferred
    // retry re-enters flagged `retryOfCountedFailure`, and the bundle can
    // have broken while it was parked — the operator has been told it
    // Failed and markJobFailed is on the wire, so placing it now (on an
    // elastic pool, a fresh createAwsNode) is exactly wrong.
    if (recovery && this.brokenBundles.has(bundleId)) {
      debuglog("Skipping recovery — bundle is marked broken", { jobId, bundleId });
      return notPlaced("bundle is marked broken");
    }
    if (recovery && !recovery.retryOfCountedFailure) {
      if (
        this.shouldBreakOnRestart(
          jobId,
          bundleId,
          onFailure,
          bundleJobSpec.requirements.maxRestartsPerHour,
          bundleJobSpec.requirements.essential,
          !recovery.repeatOfCountedCondition && !recovery.reclaimed
        )
      ) {
        return notPlaced("restart ceiling tripped");
      }
    }

    // Minimum-interval backoff between consecutive recovery placement
    // dispatches for the same job. The failure rate has already been
    // counted above; this gate only delays the next actual placement
    // attempt so we don't hammer the daemon/worker at 12 attempts/sec.
    // Coalescing: if a deferred retry is already pending, further
    // triggers in the window become no-ops (the timer will see the
    // latest state when it fires).
    if (recovery && !recovery.retryOfCountedFailure) {
      const history = this.restartHistory.get(jobId) ?? [];
      // The just-pushed entry is the last one; we want the entry
      // before it to determine spacing.
      const previousAttempt = history[history.length - 2]?.at;
      const now = this.clock.now();
      if (previousAttempt && now.getTime() - previousAttempt.getTime() < MIN_RECOVERY_INTERVAL_MS) {
        if (this.pendingPlacementRetries.has(jobId)) {
          debuglog("Recovery placement already pending (coalescing)", { jobId, bundleId });
          return notPlaced("recovery placement already pending");
        }
        const delay = MIN_RECOVERY_INTERVAL_MS - (now.getTime() - previousAttempt.getTime());
        debuglog("Recovery placement deferred (backoff)", { jobId, bundleId, delayMs: delay });
        const timer = this.clock.setTimeout(() => {
          this.pendingPlacementRetries.delete(jobId);
          // The world may have moved on while the retry was parked: the
          // instance closed, the job reached a terminal state (the
          // ceiling's markJobFailed echoed back), or the bundle broke.
          // placeAndProvision checks the broken mark again; the others
          // are the caller's checks, made here for the caller we are.
          if (this.closed) return;
          const current = this.jobMap.get(jobId);
          if (!current || current.job.terminalState || this.brokenBundles.has(bundleId)) {
            debuglog("Deferred recovery placement dropped — job terminal or bundle broken", {
              jobId,
              bundleId,
              terminalState: current?.job.terminalState,
              broken: this.brokenBundles.has(bundleId),
            });
            return;
          }
          void this.placeAndProvision(jobId, tags, {
            ...recovery,
            retryOfCountedFailure: true,
          }).catch((err) => this.settings.onError?.(err));
        }, delay);
        this.pendingPlacementRetries.set(jobId, timer);
        return notPlaced("recovery placement deferred (backoff)");
      }
    }

    const inventory = this.inventoriesAsNodeViews();
    const pools = new Map(
      (this.settings.placementPools ?? []).map((p) => [p.name, p])
    );
    if (pools.size === 0) {
      debuglog("placementPools not configured — cannot place", { jobId });
      this.settings.onError?.(
        new Error(
          `placementPools not configured; can't place job ${jobId}. ` +
            `Add to AutoSettings or this job will never start.`
        )
      );
      return notPlaced("placementPools not configured");
    }

    // Hard node pin, read off the job's own AutoPlacement (API-only escape
    // hatch; nothing populates it from bundles). Recovery semantics come
    // free: a rejection excludes the target, that placement attempt fails,
    // and the ordinary reconcile sweeps retry the same target fresh —
    // "retry the one eligible node or eventually fail", never elsewhere.
    // A migration's explicit target is the same kind of pin, for one
    // placement; a pinned job asked to migrate with no explicit target
    // keeps its pin (the source node) and so fails placement, which is
    // the honest answer for "this node or nowhere".
    const targetNode =
      migration?.targetNode ?? this.jobMap.get(jobId)?.job.autoPlacement?.targetNode?.value;
    // A migration is neither an initial placement nor a recovery. It
    // borrows the recovery context for its EXCLUSION SET only — the source
    // node, plus whatever recovery would skip.
    //
    // Phase G (fix-list item 2) — it no longer borrows the spare policy.
    // `onRuntimeFailure` answers "what should happen when this runtime
    // FAILS", and a planned move is not a failure; reading it here meant an
    // operator-ordered migration of an `avoidSpare` job pushed the target
    // away from the very nodes most likely to be warm, and a `preferSpare`
    // one pulled it onto a cold spare. A planned move is warm-first like
    // any other placement, and the reserve is protected the way it is
    // actually described — by RESERVATION (`exclusive` spares), which the
    // caller overrides explicitly with `overrideReservation` when it means
    // to. Recovery keeps the failure policy, which is whose question it is.
    const recoveryContext: RecoveryContext | undefined = recovery
      ? { excludeNodes: recovery.excludeNodes, onRuntimeFailure: onFailure }
      : migration
        ? { excludeNodes: migration.excludeNodes }
        : undefined;
    // Phase G — warmth and reservation inputs. The template's image set is
    // what warmth is scored against; the template NAME is what a
    // reservation is matched on. `overrideReservation` comes from the
    // migration when there is one, else from the bundle (the launch's).
    const jobImages = this.hotSpares.jobImagesFor(bundleJobSpec);
    const templateName = this.hotSpares.templateNameFor(bundleJobSpec);
    const overrideReservation =
      migration?.overrideReservation ?? bundle.overrideReservation ?? false;
    // A migration may need more of the node than the job declared (the
    // job socket for a handover): the same placement, over a spec that
    // carries the extra requirement, so nothing about the check is special.
    const placedJobSpec =
      migration?.extraCapabilities && migration.extraCapabilities.length > 0
        ? {
            ...bundleJobSpec,
            requirements: {
              ...bundleJobSpec.requirements,
              requiredCapabilities: [
                ...bundleJobSpec.requirements.requiredCapabilities,
                ...migration.extraCapabilities,
              ],
            },
          }
        : bundleJobSpec;
    const { result, trace }: { result: PlacementResult; trace: PlacementTrace } =
      placeWithTrace({
        bundle,
        job: placedJobSpec,
        replicaIndex,
        inventory,
        pools,
        recoveryContext,
        ...(jobImages.length > 0 ? { jobImages } : {}),
        ...(templateName !== undefined ? { templateName } : {}),
        ...(overrideReservation ? { overrideReservation: true } : {}),
        ...(targetNode !== undefined ? { targetNode } : {}),
      });

    debuglog("Placement decision", {
      jobId,
      bundleId,
      replicaIndex,
      migration: migration?.migrationId,
      decision: result,
      trace: summariseTrace(trace),
    });

    const at = this.clock.now();
    this.metricsCounters.placementsAttempted++;
    if (result.kind === "failure") {
      this.metricsCounters.placementsFailed++;
    }
    // R12: placementsSucceeded counted post-dispatch; see below.
    const info: PlacementDecisionInfo = { jobId, bundleId, replicaIndex, jobName, result, trace, at };
    // A pinned target's own verdict, for the caller's failure message.
    const targetReason =
      targetNode !== undefined && result.kind === "failure"
        ? (trace.attempts.flatMap((a) => a.filtered).find((f) => f.nodeId === targetNode)?.reason ??
          (trace.attempts.some((a) => a.scored.some((n) => n.nodeId === targetNode))
            ? "eligible, but the pool could not place onto it"
            : "in no pool this job may use"))
        : undefined;
    // Mirror into live-state map so consumers can query the current
    // placement without subscribing from process start. Failures don't
    // bind a placement, so they don't go in the map — `placementForJob`
    // returning undefined for "we tried and failed" is the correct
    // signal; failures still flow through `onPlacementFailed` for
    // consumers that care. A migration target's decision is parked
    // until `done`: the placement of record stays the source's while
    // the job still lives there.
    if (result.kind !== "failure" && migration) {
      this.bindings.parkTargetPlacement(jobId, migration.migrationId, info);
    }
    this.settings.onPlacementDecision?.(info);

    if (result.kind === "failure") {
      // A failed placement dispatched nothing, so the in-flight gate
      // (the binding's attempt + MIN_PLACEMENT_INTERVAL_MS) must not
      // hold this job back: a capacity-increasing event can arrive
      // moments later and `reconcilePendingPlacements` needs to be free
      // to retry. The gate exists to suppress duplicate dispatch while a
      // real placement is in flight — a failure is not in flight.
      if (!migration) this.bindings.clearAttempt(jobId);
      this.settings.onPlacementFailed?.({
        jobId,
        bundleId,
        replicaIndex,
        jobName,
        reason: result.reason,
        triedPools: result.reason.triedPools,
        at,
      });
      const reason = JSON.stringify(result.reason);
      // A migration's caller gets the failure as its rejection; an
      // ordinary placement has no caller waiting, so it is surfaced here.
      if (!migration) {
        this.settings.onError?.(new Error(`Placement failed for ${jobId}: ${reason}`));
      }
      return { kind: "notPlaced", reason, ...(targetReason !== undefined ? { targetReason } : {}) };
    }

    const role = migration ? migration.targetRole : roleFromReplicaIndex(replicaIndex);

    // A degraded backup still gets placed/provisioned (best-effort) — surface
    // the reduced DR posture so an operator can act / re-place later.
    if (result.degraded) {
      this.settings.onResilienceDegraded?.({
        jobId,
        bundleId,
        replicaIndex,
        jobName,
        violated: result.degraded.violated,
        ...(result.degraded.interruptiblePrimary ? { interruptiblePrimary: true } : {}),
        at,
      });
    }

    // Where the claim goes. A migration target is bound as the claim's
    // target so the job's own binding keeps naming the source (see
    // `MigrationClaim`); an ordinary placement is a fresh instance under
    // the slot key, so any key a past migration left the job under is
    // forgotten with it, and its decision becomes the placement of record.
    const bind = (nodeId: NodeId): void => {
      if (migration) this.bindings.bindTarget(jobId, migration.migrationId, nodeId);
      else this.bindings.place(jobId, nodeId, info);
    };

    if (result.kind === "place") {
      // Placed onto a node still launching (prospective view): reserve via
      // the binding exactly like a real placement, but DEFER the startJob —
      // bindProvisionedJob dispatches every bound job when the node
      // arrives, and abandonPendingProvision re-places us if it never does.
      if (this.pendingProvisions.has(result.nodeId)) {
        bind(result.nodeId);
        debuglog("Placed onto in-flight provision; dispatch deferred", {
          jobId,
          nodeId: result.nodeId,
          migration: migration?.migrationId,
        });
        this.metricsCounters.placementsSucceeded++;
        return { kind: "bound", nodeId: result.nodeId };
      }
      // Existing node already running on the chosen target. Manager's
      // StartJob RPC binds the job to that node.
      //
      // Reserve BEFORE the dispatch await: the binding is what the derived
      // reservation reads, so another placement interleaving across this
      // await must see the claim, not the pre-claim snapshot — for capacity
      // that window meant oversubscription, for exclusive resources (host
      // ports, per-card capabilities) a hard double-book. Released on
      // dispatch failure, guarded in case something re-placed the job
      // meanwhile.
      bind(result.nodeId);
      if (migration) {
        const state = this.migrationById(migration.migrationId);
        if (!state) return notPlaced("migration no longer known");
        // A failed dispatch aborts the migration (releasing this binding)
        // — never the job's own binding, which still names the source.
        const ok = await this.migrationSaga.dispatchMigrationTarget(state, result.nodeId);
        if (!ok) return notPlaced("startJob dispatch failed for migration target");
        this.metricsCounters.placementsSucceeded++;
        return { kind: "bound", nodeId: result.nodeId };
      }
      try {
        await this.norsk.startJob(jobId, role, result.nodeId);
        // A dispatch the daemon accepted ends any run of refusals.
        this.lastCountedDispatchRefusal.delete(jobId);
        this.metricsCounters.placementsSucceeded++;
      } catch (err) {
        this.handleDispatchFailure(jobId, result.nodeId, err, "place");
        return notPlaced("startJob dispatch failed");
      }
      return { kind: "bound", nodeId: result.nodeId };
    }

    // Provision in the chosen pool's chosen tier.
    const pool = pools.get(result.pool);
    if (!pool) return notPlaced(`pool ${result.pool} not configured`);
    const tier = pool.tiers.find((t) => t.name === result.tier);
    if (!tier) return notPlaced(`tier ${result.tier} not configured in pool ${result.pool}`);
    // Both hold by construction — placement only provisions in an elastic
    // tier, settings validation forbids an elastic cluster tier, and the
    // instance type is picked from the tier's own candidates — and both
    // are checked HERE, before any state is touched: a refusal after the
    // node-map entry, the binding and `onNodeStarting` would leave a
    // phantom node and a bound job behind it.
    const shape = tier.candidateInstanceTypes.find((it) => it.instanceType === result.instanceType);
    if (tier.kind === "local-servers") {
      return notPlaced(`cluster tier ${pool.name}/${tier.name} cannot provision (place-on-existing only)`);
    }
    if (!shape) {
      return notPlaced(`instance type ${result.instanceType} is not a candidate of tier ${pool.name}/${tier.name}`);
    }
    // `pool` + `tier` tags are load-bearing: poolNameForNode / tierNameForNode
    // read them back via nodeMetadata.tags to identify which pool+tier an
    // existing node belongs to. We must set them here, or every placement
    // after the node starts will filter it out as wrong-pool/wrong-tier.
    const provisionTags = {
      "auto-manager": "true",
      jobId,
      role,
      bundleId,
      replicaIndex: replicaIndexStr,
      jobName,
      pool: pool.name,
      tier: tier.name,
      // The hydration recovery key: a restart mid-boot rebuilds the
      // pending-provision entry (declared shape lookup) from this tag —
      // provider metadata carries instanceType for AWS but not OCI.
      instanceType: result.instanceType,
      // Pool-level SG opt-out rides as a tag the Manager's reconcile
      // paths check (only stamped when opted out, so existing nodes'
      // tag sets are untouched).
      ...(pool.dynamicSecurityGroups === false ? { dynamicSecurityGroups: "false" } : {}),
      // Failure-domain metadata for the resilience engine + observability.
      reliability: tierReliability(tier),
      ...(migration ? { migrationId: migration.migrationId } : {}),
    };
    // The key's "@" is kept out of the node id: it is used in provider
    // names and tags where the character is unwelcome.
    const nodeId = uuid.v4() + "." + jobId + "." + role.replace(/@/g, "-");
    this.updateNodeMap(nodeId, "starting", undefined, undefined, { jobId, role });
    bind(nodeId);
    // Prospective capacity: placement can now see and fill the incoming
    // node (its declared shape) instead of provisioning another.
    this.pendingProvisions.set(nodeId, {
      pool: pool.name,
      tier: tier.name,
      instanceType: result.instanceType,
      shape,
      requestedAt: this.clock.now(),
    });
    const timeoutMs = provisionTimeoutMs(tier);
    this.pendingProvisionTimers.set(
      nodeId,
      this.clock.setTimeout(() => this.expirePendingProvision(nodeId), timeoutMs)
    );
    this.settings.onNodeStarting?.(nodeId);

    if (tier.kind === "aws") {
      try {
        await this.norsk.createAwsNode({
          nodeId,
          region: tier.region ?? "",
          instanceType: result.instanceType,
          tags: provisionTags,
          writeFiles: [],
          launchMode: result.launchMode,
          ...(tier.rootVolume ? { rootVolume: tier.rootVolume } : {}),
        });
        this.metricsCounters.placementsSucceeded++;
      } catch (err) {
        this.abandonPendingProvision(nodeId, "createAwsNode dispatch failed");
        this.metricsCounters.placementsDispatchFailed++;
        this.settings.onError?.(err);
        return notPlaced("createAwsNode dispatch failed");
      }
    } else {
      try {
        await this.norsk.createOciNode({
          nodeId,
          availabilityDomain: tier.region ?? "",
          architecture: "x86_64",
          shape: result.instanceType,
          subnet: "",
          tags: provisionTags,
          writeFiles: [],
        });
        this.metricsCounters.placementsSucceeded++;
      } catch (err) {
        this.abandonPendingProvision(nodeId, "createOciNode dispatch failed");
        this.metricsCounters.placementsDispatchFailed++;
        this.settings.onError?.(err);
        return notPlaced("createOciNode dispatch failed");
      }
    }
    return { kind: "bound", nodeId };
  }

  /** Does any node other than `except` advertise the named capability at
   *  all (reserved or not)? Used to word a placement refusal. */
  private anyOtherNodeAdvertises(name: string, except: NodeId): boolean {
    for (const [nodeId, inv] of this.inventoryMap.entries()) {
      if (nodeId === except) continue;
      if (inv.capabilities.some((c) => c.name === name)) return true;
    }
    return false;
  }

  private inventoriesAsNodeViews(): NodeView[] {
    // Reservation (reservedCapacity / cores / capabilities, assignedJobs) is
    // not populated upstream: the worker advertises raw hardware and the
    // daemon forwards it verbatim, so the wire values arrive as 0 / []. Left
    // as-is, every node would look fully free regardless of what's running on
    // it — placement would over-subscribe, and the capacity-driven reconcile
    // could never observe a node fill up and later free up. AutoManager is the
    // one component that knows both which jobs it placed where (`bindings`)
    // and each job's requirement (the bundle's typed spec), so it derives the
    // reservation here. A non-zero wire reservation, if a producer ever sends
    // one, is treated as authoritative and the derivation defers to it.
    const placedJobsByNode = this.placedJobsByNode();

    const out: NodeView[] = [];
    for (const [nodeId, inv] of this.inventoryMap.entries()) {
      const placedHere = placedJobsByNode.get(nodeId) ?? [];

      // The jobs AutoManager itself placed on this node feed the affinity /
      // spread constraints in the placement engine. (The wire's
      // `assignedJobs` is not consulted: no worker populates it.)
      const runningJobs: NodeView["runningJobs"] = placedHere.flatMap((p) => this.runningJobFor(p.jobId));

      // Phase C8: merge the AutoManager-local cordon into the wire-reported
      // cordoned flag. Either side can mark a node off-limits.
      const cordoned = this.cordonedNodes.has(nodeId)
        ? { ...inv, cordoned: true }
        : inv;
      // Overlay derived reservation, deferring to a non-zero wire value.
      const effectiveInv = this.withReservationOverlay(cordoned, placedHere);

      // poolName / tierName are read from the node's NodeMetadata (nodeMap);
      // nodes we haven't seen there yet resolve to "" / undefined and never
      // match a pool filter.
      out.push({
        nodeId,
        poolName: this.poolNameForNode(nodeId),
        tierName: this.tierNameForNode(nodeId),
        cloud: this.cloudForNode(nodeId),
        inventory: effectiveInv,
        runningJobs,
        az: this.azForNode(nodeId),
        // Phase C5: spare-ness is advertised via NodeMetadata.tags["spare"].
        isHotSpare: this.isHotSpareNode(nodeId),
        // Phase G: warmth is OBSERVED (what the worker says it holds) and
        // reservation is DECLARED (the spare's spec). Two properties.
        heldImages: this.hotSpares.heldImagesFor(nodeId),
        ...this.hotSpares.reservationViewFor(nodeId),
      });
    }

    // Prospective views for in-flight provisions (machine-types.md Part B):
    // inventory is the machine type's DECLARED shape; reservations come from
    // the same binding derivation as real nodes, so the triggering job's
    // claim (and any join-ers') shows as consumed capacity. Superseded
    // wholesale by the node's first real inventory report.
    for (const [nodeId, pending] of this.pendingProvisions) {
      if (this.inventoryMap.has(nodeId)) continue;
      const placedHere = placedJobsByNode.get(nodeId) ?? [];
      const reservation = this.derivedReservation(placedHere);
      out.push({
        nodeId,
        poolName: pending.pool,
        tierName: pending.tier,
        cloud: this.cloudForNode(nodeId),
        inventory: {
          totalCapacity: pending.shape.totalCapacity,
          utilizationCap: 1,
          totalCores: pending.shape.totalCores,
          gpus: (pending.shape.gpus ?? []).map((g, index) => {
            const reserved = reservation.gpus.find((r) => r.index === index)?.reservedCapacity ?? 0;
            return {
              index,
              model: g.model,
              totalCapacity: g.totalCapacity,
              reservedCapacity: reserved,
              exclusivelyReserved: reserved > 0 && reserved >= g.totalCapacity,
            };
          }),
          capabilities: pending.shape.capabilities,
          reservedCapacity: reservation.capacity,
          reservedCores: reservation.cores,
          reservedCapabilities: reservation.capabilities,
          reservedHostPorts: reservation.hostPorts,
          reachable: true,
          lastSeen: pending.requestedAt,
          cordoned: false,
          assignedJobs: [],
        },
        runningJobs: placedHere.flatMap((p) => this.runningJobFor(p.jobId)),
        az: undefined,
        isHotSpare: false,
        prospective: true,
      });
    }
    return out;
  }

  /** Drop a pending provision (dispatch failure / node death / timeout):
   *  the prospective view disappears, every job bound to the never-born
   *  node is unbound, and the reconcile sweep re-places them — elsewhere,
   *  or via a fresh provision. */
  private abandonPendingProvision(nodeId: NodeId, reason: string): void {
    const pending = this.pendingProvisions.get(nodeId);
    this.pendingProvisions.delete(nodeId);
    const timer = this.pendingProvisionTimers.get(nodeId);
    if (timer) {
      timer.cancel();
      this.pendingProvisionTimers.delete(nodeId);
    }
    if (!pending) return;
    const orphaned = this.bindings.jobsOn(nodeId);
    // The prior attempt conclusively failed with the node — the unbind
    // also clears the dispatch-in-flight gate, so nothing delays the retry.
    for (const jobId of orphaned) this.bindings.unbind(jobId);
    // A migration target waiting on this node has lost it: the saga
    // decides (abort, or fail forward past the point of no return) —
    // never the orphan re-place above, whose job still runs on the source.
    for (const { jobId, migrationId } of this.bindings.targets()) {
      if (!this.bindings.isTargetOn(jobId, nodeId)) continue;
      const state = this.migrationForJob(jobId);
      if (state) {
        void this.migrationSaga
          .handleTargetLost(state, `target provision abandoned: ${reason}`)
          .catch((err) => this.settings.onError?.(err));
      } else {
        this.bindings.unbindTarget(jobId, migrationId);
      }
    }
    this.nodeMap.delete(nodeId);
    debuglog("Abandoned pending provision", { nodeId, reason, orphaned });
    if (orphaned.length > 0) void this.reconcilePendingPlacements();
  }

  /** Rebuild a pending-provision entry for a still-booting node found at
   *  hydration time: shape from the instanceType tag (providerMetadata
   *  fallback for AWS nodes predating the tag) resolved against the
   *  CURRENT settings' machine-type catalogue. The timeout re-arms fresh
   *  from reconnect rather than remaining-from-createdAt — conservative
   *  against clock skew across generations; a truly dead provision costs
   *  one extra window before abandonment re-places its jobs. Without a
   *  resolvable shape there is no prospective view to synthesize — the
   *  node still binds its job on arrival, but can't be joined meanwhile. */
  private hydratePendingProvision(meta: NodeMetadata): void {
    const nodeId = meta.nodeId;
    if (meta.tags["auto-manager"] !== "true") return;
    const poolName = meta.tags["pool"];
    const tierName = meta.tags["tier"];
    const instanceType =
      meta.tags["instanceType"] ??
      (meta.providerMetadata.provider === "aws" ? meta.providerMetadata.instanceType : undefined);
    const pool = (this.settings.placementPools ?? []).find((p) => p.name === poolName);
    const tier = pool?.tiers.find((t) => t.name === tierName);
    const shape =
      instanceType !== undefined
        ? tier?.candidateInstanceTypes.find((it) => it.instanceType === instanceType)
        : undefined;
    if (!poolName || !tierName || !tier || instanceType === undefined || !shape) {
      debuglog("Starting node not hydrated as pending provision", {
        nodeId,
        poolName,
        tierName,
        instanceType,
      });
      return;
    }
    this.pendingProvisions.set(nodeId, {
      pool: poolName,
      tier: tierName,
      instanceType,
      shape,
      requestedAt: meta.createdAt ?? this.clock.now(),
    });
    this.pendingProvisionTimers.set(
      nodeId,
      this.clock.setTimeout(() => this.expirePendingProvision(nodeId), provisionTimeoutMs(tier))
    );
  }

  /** Timeout fire: the node never reported. Same abandonment as a create
   *  failure — jobs must not wait on a node that is no longer coming. */
  private expirePendingProvision(nodeId: NodeId): void {
    if (!this.pendingProvisions.has(nodeId)) return;
    if (this.inventoryMap.has(nodeId)) {
      this.pendingProvisions.delete(nodeId);
      this.pendingProvisionTimers.delete(nodeId);
      return;
    }
    this.abandonPendingProvision(nodeId, "provision timeout");
  }

  /** Map a placed jobId to the RunningJob view the placement engine uses
   *  for affinity / spread / exclusivity, via the job's bundle-context
   *  tags. Empty when the job lacks bundle tags. `exclusive` comes from
   *  the occupant's bundle spec so the node-hosts-exclusive filter can
   *  see it without a second lookup inside the pure engine. */
  private runningJobFor(jobId: JobId): RunningJob[] {
    const tags = this.tagsForJob(jobId);
    if (!tags) return [];
    const bundleId = tags["bundleId"];
    const replicaIndexStr = tags["replicaIndex"];
    const jobName = tags["jobName"];
    if (!bundleId || replicaIndexStr === undefined || !jobName) return [];
    const spec = this.bundleMap
      .get(bundleId)
      ?.jobs.find((j) => j.jobName === jobName);
    return [
      {
        bundleId,
        replicaIndex: parseInt(replicaIndexStr, 10),
        jobName,
        ...(spec?.exclusive ? { exclusive: true } : {}),
      },
    ];
  }

  /** Sum the resource requirements of the given placed jobs into a node
   *  reservation. Capacity / cores / capabilities come from each job's
   *  bundle spec — the canonical typed requirements the placement engine
   *  scores against, looked up via the job's bundleId / jobName tags. Jobs
   *  whose bundle or spec we don't have yet contribute nothing. */
  private derivedReservation(placed: PlacedJobRef[]): ReturnType<typeof deriveReservation> {
    const claims: PlacedClaim[] = [];
    for (const { jobId, gpuIndex } of placed) {
      const requirements = this.requirementsForPlacedJob(jobId);
      if (requirements) claims.push({ requirements, ...(gpuIndex !== undefined ? { gpuIndex } : {}) });
    }
    return deriveReservation(claims);
  }

  /**
   * The GPU device a bound job holds, from its placement of record —
   * `PlacementResult.gpuIndex` on both `place` and `provision` decisions,
   * which is where placement records the device it chose. A migration
   * target's claim is in the parked target decision. A job hydrated from
   * history has no live decision (the synthesised record carries no
   * device) and so holds no device until it is next placed — the wire
   * carries none either (no worker populates `assignedJobs`).
   */
  private gpuIndexOfRecord(jobId: JobId, binding: "job" | "migrationTarget"): number | undefined {
    const info =
      binding === "job" ? this.bindings.placementOf(jobId) : this.bindings.migrationOf(jobId)?.targetPlacement;
    if (info && info.result.kind !== "failure") return info.result.gpuIndex;
    return undefined;
  }

  /** Invert the bindings once: nodeId → the jobs AutoManager placed
   *  there. Reads BOTH the jobs' own bindings and the migration targets,
   *  so a migrating job is counted on its source AND its target for as
   *  long as the overlap lasts — the honest reservation (a full cluster
   *  cannot migrate; hot spares are the answer, not a looser check). */
  private placedJobsByNode(): Map<NodeId, PlacedJobRef[]> {
    const out = new Map<NodeId, PlacedJobRef[]>();
    const add = (ref: PlacedJobRef, nodeId: NodeId): void => {
      const list = out.get(nodeId);
      if (list) list.push(ref);
      else out.set(nodeId, [ref]);
    };
    for (const { jobId, nodeId } of this.bindings.bound())
      add({ jobId, gpuIndex: this.gpuIndexOfRecord(jobId, "job") }, nodeId);
    for (const { jobId, nodeId } of this.bindings.targets())
      add({ jobId, gpuIndex: this.gpuIndexOfRecord(jobId, "migrationTarget") }, nodeId);
    return out;
  }

  /**
   * Overlay the derived reservation (from the jobs AutoManager placed on
   * the node) onto a wire inventory, deferring to a non-zero wire value —
   * if a producer ever advertises real reservations they're authoritative.
   */
  private withReservationOverlay(
    inv: TypedNodeInventory,
    placedHere: PlacedJobRef[]
  ): TypedNodeInventory {
    return overlayReservation(inv, this.derivedReservation(placedHere));
  }

  /** The canonical typed bundle spec for a placed job, looked up via its
   *  bundleId / jobName tags. Undefined for un-managed jobs. */
  private specForPlacedJob(jobId: JobId): BundleJobSpec | undefined {
    const tags = this.tagsForJob(jobId);
    if (!tags) return undefined;
    const bundleId = tags["bundleId"];
    const jobName = tags["jobName"];
    if (!bundleId || !jobName) return undefined;
    return this.bundleMap.get(bundleId)?.jobs.find((j) => j.jobName === jobName);
  }

  /** The canonical typed requirements for a placed job, from its bundle
   *  spec (the same source `placeAndProvision` scores against). */
  private requirementsForPlacedJob(jobId: JobId): JobRequirements | undefined {
    return this.specForPlacedJob(jobId)?.requirements;
  }

  // ---- Phase C6 / Phase G — hot spares, warmth, reservation -----------
  // The reconciler and the spare bookkeeping are src/hotSpares.ts; these
  // delegate.

  /**
   * 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
   */
  public isHotSpareNode(nodeId: NodeId): boolean {
    return this.hotSpares.isHotSpareNode(nodeId);
  }

  /**
   * 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
   */
  public setTemplateImages(templateName: string, images: string[]): void {
    this.hotSpares.setTemplateImages(templateName, images);
  }

  /**
   * 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
   */
  public templateImages(templateName: string): string[] {
    return this.hotSpares.templateImages(templateName);
  }

  /**
   * 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
   */
  public spareSpec(nodeId: NodeId): SpareSpec | undefined {
    return this.hotSpares.spareSpec(nodeId);
  }

  /**
   * 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
   */
  public async updateHotSpareNode(
    nodeId: NodeId,
    edit: { warmFor?: string[]; reservedFor?: string[]; exclusive?: boolean }
  ): Promise<void> {
    await this.hotSpares.updateHotSpareNode(nodeId, edit);
  }

  /**
   * 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
   */
  public async updateHotSpareSpecs(specs: HotSpareConfig[]): Promise<HotSpareSpecsDiff> {
    return this.hotSpares.updateHotSpareSpecs(specs);
  }

  /**
   * Is this a LOCAL SERVER that is a hot spare (marked through
   * `setLocalSpare`, not provisioned)? A spare to placement; nobody's to
   * the reconciler. See `HotSpares.isLocalSpare`.
   * @public
   */
  public isLocalSpare(nodeId: NodeId): boolean {
    return this.hotSpares.isLocalSpare(nodeId);
  }

  /**
   * Mark a local server as a hot spare (`spare: true`, optionally with its
   * `warmFor` / `reservedFor` / `exclusive`), or clear it. The daemon owns
   * the tag from there and drops it when a job binds, as for an elastic
   * spare. Rejects for a node that is not a local server or that holds a
   * job. See `HotSpares.setLocalSpare`.
   * @public
   */
  public async setLocalSpare(
    nodeId: NodeId,
    edit: { spare: boolean; warmFor?: string[]; reservedFor?: string[]; exclusive?: boolean }
  ): Promise<void> {
    await this.hotSpares.setLocalSpare(nodeId, edit);
  }

  /**
   * 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
   */
  public async setNodeIdleExpiry(nodeId: NodeId, ms: number): Promise<void> {
    await this.hotSpares.setNodeIdleExpiry(nodeId, ms);
  }

  /**
   * 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
   */
  public async prefetchTemplates(nodeId: NodeId, templateNames: string[]): Promise<string[]> {
    return this.hotSpares.prefetchTemplates(nodeId, templateNames);
  }

  private handleImagesPrefetched(raw: ManagerPB.ImagesPrefetchedEvent): void {
    this.hotSpares.onImagesPrefetched(raw);
  }

  private reconcileHotSpares(): void {
    this.hotSpares.reconcile();
  }

  // ---- Phase F — NodeService reconciliation ------------------------
  // The reconciler itself is src/nodeServices.ts; these delegate.

  /**
   * @public
   * The configured NodeServices.
   */
  public nodeServices(): NodeService[] {
    return this.nodeServiceReconciler.services();
  }

  /**
   * @public
   * Nodes on which the named NodeService is currently placed,
   * keyed by the runtime job's jobId.
   */
  public nodeServicePlacementsFor(serviceId: ServiceId): Map<NodeId, JobId> {
    return this.nodeServiceReconciler.placementsFor(serviceId);
  }

  private reconcileNodeServices(): void {
    this.nodeServiceReconciler.reconcile();
  }

  private tagsForJob(jobId: JobId): { [k: string]: string } | undefined {
    return this.jobMap.get(jobId)?.job.tags;
  }

  private poolNameForNode(nodeId: NodeId): string {
    const node = this.nodeMap.get(nodeId);
    return node?.nodeMetadata?.tags?.["pool"] ?? "";
  }

  // Tier within the pool, from the node's `tier` tag. Undefined for
  // pre-registered cluster nodes (no tier tag) — nodeMatchesTier then
  // treats them as the pool's fixed tier.
  private tierNameForNode(nodeId: NodeId): string | undefined {
    const node = this.nodeMap.get(nodeId);
    return node?.nodeMetadata?.tags?.["tier"];
  }

  private azForNode(nodeId: NodeId): string | undefined {
    const node = this.nodeMap.get(nodeId);
    return node?.nodeMetadata?.tags?.["az"];
  }

  // Cloud / failure-domain key for the resilience sameCloud rule: the node's
  // tier kind (aws / oci / local-servers), resolved via its pool+tier tags. Matches
  // how provisionDegradation and validation derive a pool's clouds.
  private cloudForNode(nodeId: NodeId): string | undefined {
    const poolName = this.poolNameForNode(nodeId);
    const tierName = this.tierNameForNode(nodeId);
    const pool = (this.settings.placementPools ?? []).find((p) => p.name === poolName);
    if (!pool) return undefined;
    const tier =
      tierName !== undefined
        ? pool.tiers.find((t) => t.name === tierName)
        : pool.tiers.find((t) => t.scaleOut === "fixed");
    return tier?.kind;
  }

  private handleBundleUpdated(raw: ManagerPB.BundleUpdated): void {
    if (!raw.bundle) return;
    const b = fromPbBundle(raw.bundle);
    this.bundleMap.set(b.bundleId, b);
    // Phase C2: operator pushed a fresh spec — clear any prior broken
    // state for this bundle and reset its replicas' restart counters
    // so the new spec gets a clean start.
    //
    // KNOWN GAP, measured 2026-09-04 (norsk-mgr-deploy TASKS.md): the daemon
    // also re-emits bundleUpdated for things that are NOT an operator edit —
    // notably the `markJobFailed` that shouldBreakOnRestart issues to persist
    // the broken verdict, whose echo arrives ~19ms later and clears the flag
    // that was just set. So `isBundleBroken` is effectively never observable
    // for a ceiling trip, and mgr's production reads `broken: false`.
    //
    // Not fixed here because the obvious fix is wrong: comparing the spec and
    // clearing only on a change breaks relaunching an UNCHANGED spec, which
    // is a real operator action (fix the image, retry the same production) and
    // is what the edit-and-relaunch test replays. The clearing signal probably
    // wants to be the arrival of a fresh non-terminal job for the bundle —
    // the daemon's actual relaunch protocol — rather than the bundle echo.
    if (this.brokenBundles.delete(b.bundleId)) {
      debuglog("Cleared broken flag on bundleUpdated", { bundleId: b.bundleId });
    }
    this.clearRestartHistoryForBundle(b.bundleId);
    this.settings.onBundleUpdated?.(b);
  }

  private handleBundleDeleted(raw: ManagerPB.BundleDeleted): void {
    if (!raw.bundleId) return;
    const id = raw.bundleId.bundleId;
    this.bundleMap.delete(id);
    this.brokenBundles.delete(id);
    this.clearRestartHistoryForBundle(id);
    const stop = this.pendingStopTimers.get(id);
    if (stop) {
      stop.timer.cancel();
      this.pendingStopTimers.delete(id);
    }
    this.settings.onBundleDeleted?.(id);
  }

  private clearRestartHistoryForBundle(bundleId: BundleId): void {
    for (const jobId of this.jobMap.keys()) {
      const tags = this.jobMap.get(jobId)?.job.tags ?? {};
      if (tags["bundleId"] === bundleId) {
        this.restartHistory.delete(jobId);
        this.terminalVerdictsIssued.delete(jobId);
      }
    }
  }

  private nodesByJob(jobId: JobId): Set<NodeSummary> {
    // Uses the job's binding so the "is this job placed yet?" check
    // works for multi-job nodes too — the older `tags["jobId"]` scan
    // always returned empty for cluster workers (the tag is per-node
    // and can only carry one value), which caused AutoManager to
    // repeatedly re-attempt placement and the daemon to reject with
    // `INVALID_ARGUMENT: Job already running on node`.
    const nodes = new Set<NodeSummary>();
    const nodeId = this.bindings.nodeOf(jobId);
    if (nodeId !== undefined) {
      const node = this.nodeMap.get(nodeId);
      if (node) nodes.add(node);
    }
    return nodes;
  }

  /** @internal */
  constructor(norsk: NorskManager, settings: AutoSettings) {
    this.norsk = norsk;
    this.settings = settings;
    this.clock = settings.clock ?? new RealClock();
    if (!this.settings.onError) {
      this.settings.onError = (err) => {
        debuglog("Unhandled error: %o", err);
      };
    }
    this.hotSpares = new HotSpares({
      settings: this.settings,
      clock: this.clock,
      norsk,
      bindings: this.bindings,
      active: () => !this.closed && this.hydrated,
      nodes: () => this.nodeMap.values(),
      nodeOf: (nodeId) => this.nodeMap.get(nodeId),
      inventoryOf: (nodeId) => this.inventoryMap.get(nodeId),
      poolNameForNode: (nodeId) => this.poolNameForNode(nodeId),
      nodeStarting: (nodeId) => {
        this.updateNodeMap(nodeId, "starting", undefined, undefined);
        this.settings.onNodeStarting?.(nodeId);
      },
      deleteNode: (nodeId) => {
        this.nodeMap.delete(nodeId);
      },
      mergeNodeTags: (nodeId, tags) => {
        const existing = this.nodeMap.get(nodeId);
        if (existing?.nodeMetadata) {
          this.updateNodeMap(
            nodeId,
            existing.nodeState,
            { ...existing.nodeMetadata, tags: { ...existing.nodeMetadata.tags, ...tags } },
            undefined
          );
        }
      },
      issueTerminate: async (nodeId) => this.issueTerminate(nodeId),
    });
    this.migrationSaga = new MigrationSaga({
      settings: this.settings,
      clock: this.clock,
      norsk,
      bindings: this.bindings,
      jobOf: (jobId) => this.jobMap.get(jobId),
      placeAndProvision: async (jobId, tags, recovery, migration) =>
        this.placeAndProvision(jobId, tags, recovery, migration),
      currentExcludeSet: () => this.currentExcludeSet(),
      anyOtherNodeAdvertises: (name, except) => this.anyOtherNodeAdvertises(name, except),
      onDispatchFailed: () => {
        this.metricsCounters.placementsDispatchFailed++;
      },
      onMigrationEnded: (state) => {
        if (this.drains.size > 0) this.noteDrainMigrationEnded(state);
      },
    });
    this.nodeServiceReconciler = new NodeServiceReconciler({
      services: () => this.settings.nodeServices ?? [],
      active: () => !this.closed && this.hydrated,
      nodes: () => this.nodeMap.values(),
      inventoryOf: (nodeId) => this.inventoryMap.get(nodeId),
      poolNameForNode: (nodeId) => this.poolNameForNode(nodeId),
      isCordoned: (nodeId) => this.cordonedNodes.has(nodeId),
      norsk,
      bindings: this.bindings,
      onError: (err) => this.settings.onError?.(err),
    });
  }

  /** @internal */
  private async runInternal(): Promise<void> {
    this.startedAt = this.clock.now();
    const [initialState, iter] = await eventStream(this.norsk, {
      pendingWindow: this.settings.pendingWindow,
    });
    // The initial state arrives only once the daemon's own initialisation
    // has settled (several seconds after connect). An embedder that gave
    // up on this instance in the meantime — norsk-mgr reconnects on every
    // settings change, and two edits a few seconds apart supersede the
    // first connection before its state lands — has closed it, and a
    // closed AutoManager must not hydrate and act: on norsk-e2e
    // 2026-09-15 the superseded instance reconciled hot spares from its
    // late initial state and provisioned a node the live instance then
    // provisioned again (targetCount 1, two spares).
    if (this.closed) {
      debuglog("Initial state arrived after close(); not hydrating");
      return;
    }

    // R7: Hydrate state from initial-state. AutoManager owns nothing
    // durable; clear-then-populate so the fleet view is exactly what the
    // server just sent. (An instance is entered here once — norsk-mgr's
    // "reconnect" is `close()` plus a fresh instance — so the clears are
    // a statement of intent rather than a live path.)
    this.jobMap.clear();
    this.nodeMap.clear();
    this.bundleMap.clear();
    this.inventoryMap.clear();
    this.unhealthyNodes.clear();
    this.recentFailures.clear();
    this.rejectionsPerNode.clear();
    this.restartHistory.clear();
    this.lastCountedDispatchRefusal.clear();
    this.brokenBundles.clear();
    this.terminalVerdictsIssued.clear();
    this.hotSpares.clear();
    this.bindings.clear();
    // Pending provisions are rebuilt below from the initial state's
    // nodesStarting — entries for nodes the server has since forgotten
    // (arrived-and-died while we were away) must not survive hydration.
    for (const t of this.pendingProvisionTimers.values()) t.cancel();
    this.pendingProvisionTimers.clear();
    this.pendingProvisions.clear();
    // Pending placement timers are session-local; the recovery loop
    // below re-evaluates each hydrated pre/active job and schedules
    // fresh timers (or places immediately) as appropriate.
    for (const { timer } of this.pendingPlacementTimers.values()) timer.cancel();
    this.pendingPlacementTimers.clear();
    // Same for stop timers — the recovery loop calls reconcileStopTimer
    // for each hydrated job so the daemon-persisted __stopDateTime tag
    // re-arms the timer.
    for (const { timer } of this.pendingStopTimers.values()) timer.cancel();
    this.pendingStopTimers.clear();
    for (const t of this.startupGraceTimers.values()) t.cancel();
    for (const t of this.unseenBoundNodes.values()) t.cancel();
    this.unseenBoundNodes.clear();
    this.startupGraceTimers.clear();
    // Migrations are in-flight client-side state that ends with the
    // instance (the v1 rule — design doc, Risks): `close()` aborts them
    // on the wire, so none can exist here. The bindings rebuilt from
    // history below say where a job whose migration was lost actually
    // is (a target that got as far as `running` carries it; one still
    // `prepared` is started below).
    this.nodeServiceReconciler.clear();

    for (const jobWithHistory of initialState.activeJobs) {
      this.jobMap.set(jobWithHistory.job.jobId, jobWithHistory);
      // Rebuild the binding from history — at hydration we don't get a
      // fresh placement-decision stream for jobs that were already
      // placed; the history is the source of truth. The same walk
      // yields the instance KEY the current instance wears: after a
      // migration that is `primary@xxxxxxxx`, and every RPC that
      // addresses the running instance must use it (`roleForJob`).
      const current = currentInstanceForJob(jobWithHistory);
      const nodeId = current?.nodeId;
      if (current !== undefined && nodeId !== undefined) {
        const jobId = jobWithHistory.job.jobId;
        const slot = this.slotForJob(jobId) ?? roleFromReplicaIndex(0);
        // A `prepared` instance with nothing else live is a stopThenStart
        // target whose saga was lost after the source stopped: nobody
        // else will ever send its StartPreparedJob. Start it.
        if (current.event === "prepared") {
          debuglog("Hydrated a prepared instance with no live sibling — starting it", {
            jobId,
            role: current.role,
            nodeId,
          });
          void this.norsk
            .startPreparedJob(jobId, current.role, nodeId)
            .catch((err) => this.settings.onError?.(err));
        }
        // Same reasoning for the placement of record: a job that was
        // placed in a prior generation has no live `onPlacementDecision`
        // event we can replay, so we synthesise a record from history. The
        // trace is empty (we never had it) but the result + at +
        // job-identity fields are all derivable.
        const synth = synthesisePlacementFromHistory(jobWithHistory, nodeId);
        this.bindings.hydrate(jobId, nodeId, current.role !== slot ? current.role : undefined, synth);
      }
      // Fire onJobUpdated for each hydrated job so consumers can
      // treat hydration identically to a stream of fresh
      // job-updated events. Without this, a consumer subscribed
      // through onJobUpdated wouldn't see existing jobs until the
      // next genuine update — proxies that build routing tables
      // from the event stream would stay empty until the operator
      // touched something.
      this.settings.onJobUpdated?.(jobWithHistory);
    }

    for (const runningNode of initialState.nodesRunning) {
      const nodeId = runningNode.nodeMetadata.nodeId;
      this.updateNodeMap(
        nodeId,
        "running",
        runningNode.nodeMetadata,
        runningNode.runningNodeMetadata
      );
      // Surface hydrated nodes to consumers as started — symmetric with
      // the job-hydration loop above replaying each job via onJobUpdated.
      // A worker already up before AutoManager (re)connected is delivered
      // as a `NodeRunning` snapshot, never a live `NodeStarted`; without
      // this, downstream listeners (e.g. the norsk-mgr Cluster page, which
      // only refetches on a node-started signal) never learn it joined.
      const node = this.nodeMap.get(nodeId);
      if (node) this.settings.onNodeStarted?.(node);
      // A provisioned node that came up while we were away still needs its
      // job bound — same hole as handleNodeStarted's live path (the binding
      // was rebuilt from history above, so the placement recovery loop
      // below will skip the job).
      void this.bindProvisionedJob(runningNode.nodeMetadata).catch((err) =>
        this.settings.onError?.(err)
      );
    }

    // In-flight provisions from a prior generation: a node the daemon
    // reports as STARTING was created but has not arrived. Replay the
    // live nodeStarting handling (nodeMap + startup grace) and rebuild
    // the pending-provision entry so placement sees the incoming node.
    // Must run BEFORE the recovery loop below: the triggering job has no
    // placement history (its dispatch was deferred), so recovery
    // re-places it — and must JOIN the incoming node, not spawn another.
    for (const startingNode of initialState.nodesStarting) {
      // Belt and braces with the SDK, which drops a node from nodesStarting
      // once its NodeRunning arrives in the same snapshot: a node already
      // hydrated as running must not be demoted to starting here (norsk-e2e
      // 2026-09-15: a spare re-discovered by a restarted daemon read as
      // "starting" in this view for good, and was provisioned again).
      if (this.nodeMap.get(startingNode.nodeMetadata.nodeId)?.nodeState === "running") continue;
      this.handleNodeStarting(startingNode.nodeMetadata);
      this.hydratePendingProvision(startingNode.nodeMetadata);
    }

    for (const ba of initialState.bundles) {
      if (ba.bundle) {
        const b = fromPbBundle(ba.bundle);
        this.bundleMap.set(b.bundleId, b);
        // Replay through onBundleUpdated, as jobs and nodes are above:
        // consumers treat hydration as a stream of fresh events, and
        // norsk-mgr clears its per-bundle last-failure record on this
        // callback — without the replay a stale failure outlived every
        // reconnect. Straight to the callback, not handleBundleUpdated:
        // the broken-bundle / restart-history state it clears is empty
        // on a fresh instance.
        this.settings.onBundleUpdated?.(b);
      }
    }
    for (const snap of initialState.inventories) {
      if (snap.inventory && snap.nodeId) {
        this.inventoryMap.set(
          snap.nodeId.nodeId,
          this.typedInventoryFrom(snap.nodeId.nodeId, snap.inventory)
        );
      }
    }

    // Recover any active jobs that we don't have a node for — these
    // exist when AutoManager has just (re)connected and the server
    // tells us about them via initial-state. Route through
    // `placeOrDefer` so scheduled jobs whose `startDateTime - lead`
    // is still in the future get a fresh timer; ones whose window
    // has already arrived (or was missed while mgr was down) place
    // immediately.
    // Nodes the histories name but the initial state does not: still
    // rediscovering, most likely. Hold their jobs rather than re-place them.
    this.grantHydrationGrace();
    for (const entry of this.jobMap.values()) {
      // Rebuild stop timers from the daemon-persisted __stopDateTime
      // tag on every hydrated job. Dedupes per bundleId — first job
      // wins, subsequent jobs in the same bundle are no-ops.
      this.reconcileStopTimer(entry);
      if (this.wantsPlacementNow(entry)) {
        await this.placeOrDefer(entry);
      }
    }

    // A close() that landed while the recovery loop above was awaiting a
    // placement has already run its cancels: creating the intervals now
    // would leak them (with RealClock they hold the event loop forever).
    if (this.closed) {
      debuglog("Closed during hydration; not starting periodic work");
      return;
    }

    // Periodic cleanup: terminate stopped nodes older than the threshold.
    // R19: cadence derived from removeStoppedNodesAfter (1/4 of the
    // threshold, capped to [10s, 5min]) so the sweep is proportionate
    // to the retention window — tests that lower the threshold to
    // seconds see the sweep fire quickly without rewiring the cadence.
    const cleanupCadenceMs = Math.max(
      10_000,
      Math.min(
        300_000,
        Math.floor((this.settings.removeStoppedNodesAfter * 1000) / 4)
      )
    );
    this.cleanupInterval = this.clock.setInterval(async () => {
      if (this.closed) return; // R14: don't run cleanup after close()
      // R13: snapshot entries up front so concurrent mutations during
      // the await terminateNode don't reshuffle traversal.
      const snapshot = Array.from(this.nodeMap.entries());
      for (const [nodeId, node] of snapshot) {
        if (this.closed) return;
        if (
          node.nodeState === "stopped" &&
          this.clock.now().getTime() - node.lastActivity.getTime() >
            this.settings.removeStoppedNodesAfter * 1000
        ) {
          await this.issueTerminate(nodeId).catch((err) => {
            this.settings.onError?.(err);
          });
          this.nodeMap.delete(nodeId);
          // Gone from our world; the once-only mark can go with it.
          this.terminateSignalled.delete(nodeId);
        }
      }
    }, cleanupCadenceMs);

    // Phase C6: periodic hot-spare reconciliation tick. Event-driven
    // reconciles handle most cases; this is the safety net for
    // missed signals (e.g. a provider that quietly terminated a
    // spare without Manager emitting nodeStopping).
    const reconcileInterval =
      this.settings.hotSpareReconcileIntervalMs ??
      DEFAULT_HOT_SPARE_RECONCILE_INTERVAL_MS;
    this.hotSpareReconcileInterval = this.clock.setInterval(
      () => this.reconcileHotSpares(),
      reconcileInterval
    );
    // Kick once at startup so a fresh AutoManager with hotSpares
    // configured begins provisioning immediately rather than waiting
    // for the first event or interval. This is the first reconcile that
    // can run: everything before it saw an unhydrated view.
    this.hydrated = true;
    this.reconcileHotSpares();

    // Phase F: same shape as hot-spare reconcile — periodic tick
    // for NodeServices, with a startup kick.
    const nsReconcileInterval =
      this.settings.nodeServiceReconcileIntervalMs ??
      DEFAULT_NODE_SERVICE_RECONCILE_INTERVAL_MS;
    this.nodeServiceReconcileInterval = this.clock.setInterval(
      () => this.reconcileNodeServices(),
      nsReconcileInterval
    );
    this.reconcileNodeServices();

    for await (const event of iter) {
      // close() holds the channel open while its migration aborts go
      // out; nothing that arrives in that window is ours to act on.
      if (this.closed) break;
      try {
        const eventType = event.event;
        switch (eventType) {
          case "jobUpdated":
          case "jobPending":
            await this.handleJobChanged(event.jobWithHistory);
            break;
          case "jobDeleted":
            this.handleJobDeleted(event.jobId);
            break;
          case "jobOutOfWindow":
            this.handleJobDeleted(event.jobWithHistory.job.jobId);
            break;
          case "jobInfo":
            this.settings.onJobInfo?.(event);
            break;
          case "nodeStarting":
            this.handleNodeStarting(event.nodeMetadata);
            break;
          case "nodeStarted":
            this.handleNodeStarted(event.nodeMetadata, event.runningNodeMetadata);
            break;
          case "nodeStopping":
            await this.handleNodeStopping(event.nodeId, event.reason);
            break;
          case "nodeStopped":
            await this.handleNodeStopped(event.nodeId);
            break;
          case "nodeTagsUpdated":
            this.handleNodeTagsUpdated(event.nodeMetadata);
            break;
          case "nodeInterruptionNotice":
            // The worker heard the provider's notice (IMDS) and the daemon
            // relayed it. The SDK's kinds become ours; the deadline rides
            // along (absent for a rebalance recommendation).
            await this.handleNodeInterruption(event.nodeId, {
              kind: event.kind === "spotInterruption" ? "spot-interruption" : "rebalance",
              ...(event.deadline !== undefined ? { deadline: event.deadline } : {}),
            });
            break;
          case "providerHealthChange":
            debuglog("Provider health change %o", event.health);
            break;
          case "eventStreamClosed":
            break;
          case "nodeInventoryUpdated":
            this.handleNodeInventoryUpdated(event.raw);
            break;
          case "jobRejected":
            this.handleJobRejected(event.raw);
            break;
          case "bundleUpdated":
            this.handleBundleUpdated(event.raw);
            break;
          case "bundleDeleted":
            this.handleBundleDeleted(event.raw);
            break;
          case "migrationState":
            this.migrationSaga.handleMigrationState(event.raw);
            break;
          case "migrationTargetReady":
            await this.migrationSaga.handleMigrationTargetReady(event.raw);
            break;
          case "migrationSourceStoppedOutput":
            await this.migrationSaga.handleMigrationSourceStoppedOutput(event.raw);
            break;
          case "migrationTargetApplied":
            await this.migrationSaga.handleMigrationTargetApplied(event.raw);
            break;
          case "migrationInitialStateDelivered":
            this.migrationSaga.handleMigrationInitialStateDelivered(event.raw);
            break;
          case "migrationAborted":
            await this.migrationSaga.handleMigrationAborted(event.raw);
            break;
          case "auditAppended":
            this.settings.onAuditAppended?.(event.entry);
            break;
          case "imagesPrefetched":
            // Phase G — a prefetch batch settled. Nothing to drive off it:
            // warmth is read from the node's re-advertised `heldImages`,
            // which arrives right behind this, so the event is here for
            // the operator's benefit (which refs landed, and why not) and
            // for the log.
            this.handleImagesPrefetched(event.raw);
            break;
          default: {
            const exhaustiveCheck: never = eventType;
            throw new Error(`Unhandled case: ${exhaustiveCheck}`);
          }
        }
        // Any of the above may have moved a job off (or onto) a draining
        // node: a migration's done, a job's deletion, a recovery re-place.
        if (this.drains.size > 0) this.reconcileDrains();
      } catch (err) {
        this.settings.onError?.(err);
      }
    }
  }

  // ---- job lifecycle ----

  /** jobId → timestamp (ms) of the runtime failure we have already recovered
   *  from, so redelivered `jobUpdated` events for the same failure are not
   *  counted again. Cleared when the job is deleted. */
  private readonly handledRuntimeFailures = new Map<JobId, number>();

  /** Jobs AutoManager has ruled terminally failed (a policy drop, a ceiling
   *  trip) and told the daemon so with `markJobFailed`. The verdict is on
   *  the wire, not yet echoed: until the job comes back with
   *  `terminalState`, it still reads `active` with no binding, which is
   *  exactly what `wantsPlacementNow` places. The daemon's own
   *  `processJobFailed` records `stopping` AND `stopped`, one `jobUpdated`
   *  each — so the second event of the very failure we just dropped on
   *  used to re-place the job as a fresh initial placement (a new node for
   *  a job the operator has been told is gone). Cleared when the echo lands
   *  (the durable flag takes over), when the job is deleted, and when the
   *  operator relaunches the bundle. */
  private readonly terminalVerdictsIssued = new Set<JobId>();

  private issueTerminalFailure(jobId: JobId, reason: string): void {
    this.terminalVerdictsIssued.add(jobId);
    void this.norsk.markJobFailed(jobId, reason).catch((err) => this.settings.onError?.(err));
  }

  private async handleJobChanged(jobWithHistory: JobWithHistory): Promise<void> {
    const job = jobWithHistory.job;
    const prevState = this.jobMap.get(job.jobId)?.job.state;
    this.jobMap.set(job.jobId, jobWithHistory);
    // The verdict has been persisted; the durable flag guards from here.
    if (job.terminalState) this.terminalVerdictsIssued.delete(job.jobId);
    this.settings.onJobUpdated?.(jobWithHistory);
    // Single-use node: the job ending releases its node for immediate
    // termination (data-leak protection — never lingers for reuse). The
    // binding still names the node at this point.
    if (job.state === "post" && prevState !== "post") {
      const spec = this.specForPlacedJob(job.jobId);
      const nodeId = this.bindings.nodeOf(job.jobId);
      if (spec?.singleUse && nodeId !== undefined) {
        this.maybeTerminateSingleUseNode(nodeId, job.jobId);
      }
    }
    // A job reaching `running` is the authoritative "this node is healthy"
    // signal — clear any placement-exclusion mark on its node. This is the
    // recovery signal that inventory heartbeats used to (wrongly) provide.
    // History is NEWEST-FIRST: the daemon prepends (`NEL.cons` in
    // JobStore.purs), so entry 0 is the latest transition. This used to read
    // `history[history.length - 1]`, which is the OLDEST entry — always
    // `created` in practice — so the health-mark clearing below never fired.
    const newest = jobWithHistory.history[0];
    if (newest?.event === "running") {
      this.clearNodeHealthMarks(newest.nodeMetadata.nodeId);
      // …and it ends any run of identical dispatch refusals for the job:
      // whatever the daemon was insisting on, it is no longer in force.
      this.lastCountedDispatchRefusal.delete(job.jobId);
    }
    // A migrating job's history entries are the saga's signals — keyed
    // on the instance key, so the target's `running` and the source's
    // `stopped` are told apart. Runs BEFORE the failure recovery below:
    // a target's failure is the saga's to handle, and a source failure
    // that completes into a running target needs no recovery at all.
    const handledByMigration = await this.migrationSaga.advanceMigrationOnJobHistory(jobWithHistory);
    // The middle rung of the escalation ladder. A worker that watched a
    // runtime die reports it, and the daemon records `stopping` with
    // reason "jobFailed" — but until this, nobody acted on it: the job kept
    // its node binding, so `wantsPlacementNow` saw it as placed and no
    // recovery ever ran. The runtime stayed `active` with dead containers
    // and the production still read healthy, which is exactly the
    // "recovery only ever fires on node/worker loss" hole.
    //
    // Keyed strictly on "jobFailed". The other stopping reasons already
    // have owners: nodeStopped/nodeTerminated go through the node paths
    // (re-placing here too would double-count against the restart ceiling),
    // and userRequested is an operator stop that must NOT come back.
    if (newest?.event === "stopping" && newest.reason === "jobFailed" && !handledByMigration) {
      await this.recoverFailedRuntime(job.jobId, newest.role, newest.timestamp, newest.additionalReason);
    }
    // Reconcile the per-bundle stop timer from the well-known tag on
    // this job. Done before the placement decisions so an already-past
    // stop time triggers deleteBundle without waiting for placement.
    this.reconcileStopTimer(jobWithHistory);
    if (this.wantsPlacementNow(jobWithHistory)) {
      await this.placeOrDefer(jobWithHistory);
    }
  }

  /**
   * Re-place a runtime the worker reported as failed, honouring the same
   * policy as any other recovery (drop / preferSpare / avoidSpare) and
   * counting against the same restart-rate ceiling — `placeAndProvision`
   * owns all of that; this only has to unbind and hand over.
   *
   * Idempotent by construction: the unbind is the trigger, so a repeated
   * `stopping/jobFailed` for a job we have already released finds no
   * binding and does nothing. Without that guard a redelivered event would
   * count twice against the ceiling and trip it early.
   */
  private async recoverFailedRuntime(
    jobId: JobId,
    role: Role,
    failedAt: Date,
    reason?: string
  ): Promise<void> {
    // A migration owns its own failure handling — and an event about
    // its TARGET key must never be mistaken for one about the source,
    // which still runs on the bound node. This used to unbind first and
    // ask afterwards, which released the source's binding on the
    // target's failure.
    const migration = this.migrationForJob(jobId);
    if (migration) {
      debuglog("Runtime failure during a migration — left to the saga", {
        jobId,
        role,
        migrationId: migration.migrationId,
        target: role === migration.targetRole,
      });
      return;
    }
    // The failure must be about the instance the job currently lives as.
    // After a handover the job's running instance wears its migrated key
    // (`primary@…`); a late `jobFailed` under the bare slot key is the old
    // source's container going away, and acting on it would unbind and
    // re-place the healthy target — the same restart-churn the hydration
    // replay used to cause (task 90, 2026-09-09).
    const currentRole = this.roleForJob(jobId);
    if (role !== currentRole) {
      debuglog("Runtime failure reported for a key that is not the job's current instance — ignored", {
        jobId,
        role,
        currentRole,
      });
      return;
    }
    // Idempotent per FAILURE, not per binding. The daemon redelivers
    // `jobUpdated` several times while a job settles, each carrying the same
    // newest `stopping/jobFailed` entry — and every recovery trigger counts
    // against the restart-rate ceiling by design ("the ceiling tracks failure
    // rate, which is real-time"). Keying on the binding alone let six
    // redeliveries of ONE failure trip the ceiling in 15 seconds and mark the
    // bundle broken. The history entry's timestamp identifies the failure
    // itself, so each one is acted on exactly once.
    const key = failedAt.getTime();
    if (this.handledRuntimeFailures.get(jobId) === key) return;
    this.handledRuntimeFailures.set(jobId, key);
    const nodeId = this.bindings.nodeOf(jobId);
    if (nodeId === undefined) return;
    debuglog("Runtime reported failed by its worker — recovering", { jobId, nodeId, reason });
    // Unbind first so the stale binding can't keep the job looking placed
    // (and clear the in-flight suppression so the retry isn't delayed) —
    // the same order the node-loss path uses.
    this.bindings.unbind(jobId);
    const jh = this.jobMap.get(jobId);
    const req = jh?.job.autoPlacement?.requirements;
    if (!jh || !req || jh.job.terminalState) return;
    if (jh.job.state !== "pre" && jh.job.state !== "active") return;
    // Do NOT exclude the node it died on: the runtime failed, which is not
    // evidence the node is bad, and excluding the only candidate in a
    // single-node pool would leave nothing to place onto — the failure
    // would never re-count and the ceiling would never trip. Same reasoning
    // as the dispatch-failure path.
    await this.placeAndProvision(jobId, jh.job.tags ?? {}, {
      excludeNodes: this.currentExcludeSet(),
    });
  }

  /** Read the `__stopDateTime` tag off the job, and (re)schedule the
   *  bundle's stop timer accordingly. Tag is set per-job but applies
   *  to the whole bundle — first-job-wins for read; all jobs in a
   *  bundle share the same value at submit. */
  private reconcileStopTimer(jh: JobWithHistory): void {
    const tags = jh.job.tags ?? {};
    const bundleId = tags["bundleId"];
    if (!bundleId) return;
    const stopStr = tags[STOP_DATE_TIME_TAG];
    const stopAt = stopStr ? new Date(stopStr) : undefined;
    const prev = this.pendingStopTimers.get(bundleId);
    // If the tag is gone OR unchanged, just keep / drop the existing
    // timer. Equality on millis is fine — daemon round-trips ISO
    // strings, no float fuzz.
    if (!stopAt || Number.isNaN(stopAt.getTime())) {
      if (prev) {
        prev.timer.cancel();
        this.pendingStopTimers.delete(bundleId);
      }
      return;
    }
    if (prev && prev.stopAt.getTime() === stopAt.getTime()) return;
    if (prev) prev.timer.cancel();
    this.scheduleStopTimer(bundleId, stopAt);
  }

  private scheduleStopTimer(bundleId: BundleId, stopAt: Date): void {
    const delayMs = stopAt.getTime() - this.clock.now().getTime();
    if (delayMs <= 0) {
      // Already past — fire immediately and don't keep a timer entry.
      debuglog("Stop time already past; deleting bundle now", { bundleId, stopAt: stopAt.toISOString() });
      void this.deleteBundle(bundleId).catch((e) =>
        this.settings.onError?.(e instanceof Error ? e : new Error(String(e))),
      );
      this.pendingStopTimers.delete(bundleId);
      return;
    }
    debuglog("Scheduling bundle stop", { bundleId, stopAt: stopAt.toISOString(), inMs: delayMs });
    const timer = this.clock.setTimeout(() => {
      this.pendingStopTimers.delete(bundleId);
      void this.deleteBundle(bundleId).catch((e) =>
        this.settings.onError?.(e instanceof Error ? e : new Error(String(e))),
      );
    }, delayMs);
    this.pendingStopTimers.set(bundleId, { timer, stopAt });
  }

  /**
   * Update (or clear) a bundle's auto-stop time. Calls `updateBundle`
   * so the new value persists via the daemon's tag store and any other
   * clients see it through their event streams. Pass `undefined` to
   * cancel a previously-set stop time.
   *
   * Throws if the bundle isn't known to this AutoManager (no live
   * bundleMap entry); operators wanting to set stop on a not-yet-
   * streamed bundle should retry after JobPending fires.
   * @public
   */
  public async setBundleStopTime(bundleId: BundleId, stopDateTime?: Date): Promise<void> {
    const bundle = this.bundleById(bundleId);
    if (!bundle) {
      throw new Error(`No bundle ${bundleId} in AutoManager state`);
    }
    const updated: TypedBundle = {
      ...bundle,
      jobs: bundle.jobs.map((j) =>
        j.launch?.kind === "productTemplate"
          ? { ...j, launch: { ...j.launch, stopDateTime } }
          : j,
      ),
    };
    await this.updateBundle(updated);
  }

  /**
   * Snapshot of all currently-scheduled bundle stop timers. UIs render
   * this alongside `pendingPlacements()` for an "upcoming events"
   * view.
   * @public
   */
  public bundleStopTimes(): Array<{ bundleId: BundleId; stopAt: Date }> {
    return [...this.pendingStopTimers.entries()]
      .map(([bundleId, { stopAt }]) => ({ bundleId, stopAt }))
      .sort((a, b) => a.stopAt.getTime() - b.stopAt.getTime());
  }

  /** A job that should be running but currently has no node bound to
   *  it: it carries placement `requirements`, isn't manual-start, is in
   *  a placeable state (`pre` / `active`), and `nodesByJob` is empty.
   *  This is the single source of truth for "this job wants a placement
   *  and doesn't have one" — shared by initial hydration, the per-job
   *  `handleJobChanged` path, and the capacity-driven reconcile sweep,
   *  so the decision lives in exactly one place. */
  private wantsPlacementNow(jh: JobWithHistory): boolean {
    const job = jh.job;
    if (!job.autoPlacement?.requirements) return false;
    if ((job.tags ?? {})["startMode"] === "manual") return false;
    // Durable terminal verdict (poison / policy drop) persisted by a prior
    // session: never re-place it. This is what makes "fully failed" survive a
    // daemon restart — on reconnect the job hydrates with terminalState set.
    if (job.terminalState) return false;
    if (this.terminalVerdictsIssued.has(job.jobId)) return false;
    if (job.state !== "pre" && job.state !== "active") return false;
    if (this.boundToUnseenNodeInGrace(job.jobId)) return false;
    return this.nodesByJob(job.jobId).size === 0;
  }

  /**
   * Is this job bound to a node the daemon has yet to report, inside the
   * grace a (re)connect grants such nodes? Such a job is placed, as far as
   * anyone can tell, and must not be placed again.
   */
  private boundToUnseenNodeInGrace(jobId: JobId): boolean {
    const nodeId = this.bindings.nodeOf(jobId);
    if (nodeId === undefined) return false;
    if (this.nodeMap.has(nodeId)) return false;
    return this.unseenBoundNodes.has(nodeId);
  }

  /**
   * After hydration: every node a job is bound to (from its history) that
   * the initial state did not carry gets a grace. When the node reports in
   * (`nodeStarted`) the grace is cancelled; when it expires the bindings to
   * it are released and the jobs re-placed, exactly as if the node had
   * just been reported stopped.
   */
  private grantHydrationGrace(): void {
    for (const t of this.unseenBoundNodes.values()) t.cancel();
    this.unseenBoundNodes.clear();
    const unseen = new Set<NodeId>();
    for (const nodeId of this.bindings.boundNodes()) {
      if (!this.nodeMap.has(nodeId)) unseen.add(nodeId);
    }
    for (const nodeId of unseen) {
      const boundJobs = this.bindings.jobsOn(nodeId);
      debuglog("Hydrated jobs are bound to a node the daemon has not reported yet — holding placement", {
        nodeId,
        jobs: boundJobs,
        graceMs: HYDRATION_NODE_GRACE_MS,
      });
      const timer = this.clock.setTimeout(() => {
        this.unseenBoundNodes.delete(nodeId);
        if (this.nodeMap.has(nodeId)) return;
        const stillBound = this.bindings.jobsOn(nodeId);
        if (stillBound.length === 0) return;
        this.settings.onError?.(
          new Error(
            `node ${nodeId} was never reported after the reconnect; releasing ${stillBound.length} job(s) bound to it for re-placement`
          )
        );
        for (const jobId of stillBound) this.bindings.unbind(jobId);
        this.reconcilePendingPlacements();
      }, HYDRATION_NODE_GRACE_MS);
      this.unseenBoundNodes.set(nodeId, timer);
    }
  }

  /**
   * Re-run placement for every job that wants a node but hasn't got
   * one. Called whenever a *capacity-increasing* event arrives — a node
   * comes online (`nodeStarted`), a worker reports fresh inventory
   * (`nodeInventoryUpdated`, which also covers an un-cordon or a node
   * that has just freed reserved capacity), or a job finishes and
   * releases its node (`jobDeleted`). Those are the only transitions
   * that can turn a previously-unplaceable job placeable, so they are
   * exactly when re-running the engine is worthwhile — there is no
   * polling retry, the trigger is always a concrete event.
   *
   * Placement stays implemented once: each candidate routes through
   * `placeOrDefer` (the same entry point hydration and `handleJobChanged`
   * use), inheriting its scheduling, the in-flight gate, and the
   * failure path. Jobs already waiting on a schedule timer or a recovery
   * backoff are skipped — their own timers own them — so a capacity
   * event never disturbs a future-scheduled or backing-off job.
   *
   * Cheap and idempotent (mirrors `reconcileHotSpares` /
   * `reconcileNodeServices`); safe to call on every relevant event.
   */
  private reconcilePendingPlacements(): void {
    for (const jh of this.jobMap.values()) {
      if (!this.wantsPlacementNow(jh)) continue;
      const jobId = jh.job.jobId;
      if (this.pendingPlacementTimers.has(jobId)) continue;
      if (this.pendingPlacementRetries.has(jobId)) continue;
      void this.placeOrDefer(jh).catch((err) => this.settings.onError?.(err));
    }
  }

  /** Decide whether to place immediately or defer until the job's
   *  scheduled `startDateTime` (minus pool lead time) arrives.
   *  Re-entry from job-update events: cancels and re-evaluates so a
   *  schedule change moves the timer. Hydrated jobs with `startTime -
   *  lead` in the past place immediately (operator intent was to run;
   *  schedule was a "when", not a "drop if missed"). */
  private async placeOrDefer(jh: JobWithHistory): Promise<void> {
    const job = jh.job;
    const requirements = job.autoPlacement?.requirements;
    if (!requirements) return;
    const tags = job.tags ?? {};

    // Any previous pending timer for this jobId is now stale — either
    // we'll re-schedule below or we'll place immediately. Either way,
    // drop the prior timer.
    const prev = this.pendingPlacementTimers.get(job.jobId);
    if (prev) {
      prev.timer.cancel();
      this.pendingPlacementTimers.delete(job.jobId);
    }

    const startDateTime = job.startDateTime;
    if (!startDateTime) {
      // No schedule — fire now, current behaviour.
      await this.placeAndProvision(job.jobId, tags);
      return;
    }

    const leadMs = this.placementLeadMsForJob(jh);
    const fireAt = startDateTime.getTime() - leadMs;
    const delayMs = fireAt - this.clock.now().getTime();
    if (delayMs <= 0) {
      // Schedule has arrived (or passed — hydration after mgr was
      // down through the launch window). Operator's intent was for
      // the job to be running; honour it.
      await this.placeAndProvision(job.jobId, tags);
      return;
    }

    debuglog("Deferring placement until scheduled startDateTime", {
      jobId: job.jobId,
      startDateTime: startDateTime.toISOString(),
      leadMs,
      fireInMs: delayMs,
    });
    const timer = this.clock.setTimeout(() => {
      this.pendingPlacementTimers.delete(job.jobId);
      // Re-check that we still want to place — the job may have been
      // deleted or another instance placed it during the wait. Cheap
      // re-test using nodesByJob (binding-backed and O(1)).
      const live = this.jobMap.get(job.jobId);
      if (!live) return;
      if (this.nodesByJob(job.jobId).size > 0) return;
      if (this.boundToUnseenNodeInGrace(job.jobId)) return;
      void this.placeAndProvision(job.jobId, tags);
    }, delayMs);
    this.pendingPlacementTimers.set(job.jobId, { timer, scheduledFor: new Date(fireAt) });
  }

  /** Maximum `placementLeadMs` across the bundle pool's tiers (worst case
   *  — guarantees the job is ready by `startDateTime` regardless of which
   *  tier the placement engine ends up picking). Falls back to the per-kind
   *  default when a tier doesn't specify one. Returns 0 if the bundle or
   *  pool isn't found — we'd rather race the schedule than indefinitely
   *  delay placement. */
  private placementLeadMsForJob(jh: JobWithHistory): number {
    const bundleId = (jh.job.tags ?? {})["bundleId"];
    if (!bundleId) return 0;
    const bundle = this.bundleMap.get(bundleId);
    if (!bundle) return 0;
    const pools = this.settings.placementPools ?? [];
    const pool = pools.find((p) => p.name === bundle.pool);
    if (!pool) return 0;
    let max = 0;
    for (const tier of pool.tiers) {
      const lead = tier.placementLeadMs ?? defaultPlacementLeadMs(tier.kind);
      if (lead > max) max = lead;
    }
    return max;
  }

  /**
   * Currently-scheduled placements (jobs AutoManager has deferred
   * because their `startDateTime - leadTime` is still in the future).
   * Returns one entry per pending timer; the UI can render an
   * "upcoming jobs" list and show when each will go live.
   * @public
   */
  public pendingPlacements(): Array<{ jobId: JobId; scheduledFor: Date }> {
    return [...this.pendingPlacementTimers.entries()]
      .map(([jobId, { scheduledFor }]) => ({ jobId, scheduledFor }))
      .sort((a, b) => a.scheduledFor.getTime() - b.scheduledFor.getTime());
  }

  private handleJobDeleted(jobId: JobId): void {
    // Pure local-state cleanup. The daemon's deleteBundle saga is now
    // responsible for stopping the worker-side compose stacks before
    // emitting JobDeleted; node-level lifecycle (terminate / drain) is
    // AutoManager policy (drainNode, hot-spare reconciliation) and
    // intentionally not coupled to job deletion. Prior versions called
    // `terminateNode` here, which was provider-incorrect for cluster
    // workers (terminate is a no-op on long-lived operator-attached
    // nodes; the worker's compose stack would leak as a result).
    //
    // The ONE deliberate exception: a `singleUse` job's node is terminated
    // as it ends (opt-in, cloud-tier-gated — see
    // maybeTerminateSingleUseNode). Capture the binding + spec before the
    // cleanup below erases them.
    const singleUseNode = this.specForPlacedJob(jobId)?.singleUse
      ? this.bindings.nodeOf(jobId)
      : undefined;
    const existed = this.jobMap.delete(jobId);
    this.restartHistory.delete(jobId);
    this.lastCountedDispatchRefusal.delete(jobId);
    // A migration in flight for a deleted job has nothing left to
    // migrate: the daemon's delete saga stops every instance, both keys
    // included, so no abort goes on the wire — only the local state.
    const migration = this.migrationForJob(jobId);
    if (migration) this.migrationSaga.markMigrationAborted(migration, "job deleted");
    this.bindings.forget(jobId);
    this.handledRuntimeFailures.delete(jobId);
    this.terminalVerdictsIssued.delete(jobId);
    const pending = this.pendingPlacementTimers.get(jobId);
    if (pending) {
      pending.timer.cancel();
      this.pendingPlacementTimers.delete(jobId);
    }
    const pendingRetry = this.pendingPlacementRetries.get(jobId);
    if (pendingRetry) {
      pendingRetry.cancel();
      this.pendingPlacementRetries.delete(jobId);
    }
    if (existed) this.settings.onJobDeleted?.(jobId);
    if (singleUseNode !== undefined) this.maybeTerminateSingleUseNode(singleUseNode, jobId);
    // A job leaving frees the capacity it reserved, which may let a
    // previously-unplaceable job land. The freed capacity becomes
    // visible to the placement engine once the worker re-reports
    // inventory (`nodeInventoryUpdated` will reconcile again then); this
    // trigger covers the case where AutoManager's cached inventory
    // already reflects the release.
    this.reconcilePendingPlacements();
  }

  /** Nodes already dispatched a single-use termination — dedupes the
   *  per-member triggers when a co-located group ends together. */
  private singleUseTerminated = new Set<NodeId>();

  /**
   * A single-use job on `nodeId` has ended. Terminate the node NOW —
   * never linger for reuse (the flag exists so one event's data can't
   * leak onto a later event's runtime) — provided:
   *  - the node is provisioned cloud capacity (aws/oci tier). Registered
   *    cluster nodes are operator hardware; terminating them is the exact
   *    provider-mistake the old terminate-on-jobDeleted was removed for.
   *  - no OTHER live job is still bound to it (last-man-out for a
   *    co-located group; validation keeps singleUse uniform across a
   *    group so the last member out is always single-use).
   */
  private maybeTerminateSingleUseNode(nodeId: NodeId, endedJobId: JobId): void {
    if (this.singleUseTerminated.has(nodeId)) return;
    const kind = this.cloudForNode(nodeId);
    if (kind !== "aws" && kind !== "oci") {
      debuglog("Single-use job ended on a non-cloud node — not terminating", {
        nodeId,
        jobId: endedJobId,
        kind,
      });
      return;
    }
    for (const otherId of this.bindings.jobsOn(nodeId)) {
      if (otherId === endedJobId) continue;
      const other = this.jobMap.get(otherId);
      if (other && other.job.state !== "post") return; // sibling still live
    }
    this.singleUseTerminated.add(nodeId);
    debuglog("Single-use node released — terminating", { nodeId, jobId: endedJobId });
    // The node is now a corpse-in-waiting: make it unplaceable IMMEDIATELY.
    // The NodeStopping event that normally drops its inventory (Fix B)
    // arrives only after the daemon processes the terminate — until then
    // the stale inventory reads as free capacity, and a replacement job
    // (e.g. a Retry's re-expansion) places straight onto it, dispatch-fails
    // with "Node is not in running state", and re-places onto the same
    // corpse until the restart ceiling fails the production (observed live
    // 2026-08-27). The recentFailures mark covers the window where a
    // still-alive worker heartbeats the inventory back in.
    this.inventoryMap.delete(nodeId);
    this.recentFailures.set(
      nodeId,
      new Date(this.clock.now().getTime() + this.cfgFailureBackoffMs())
    );
    void this.issueTerminate(nodeId).catch((err) => {
      const msg = err instanceof Error ? err.message : String(err);
      // NOT_FOUND = the daemon already reaped it — the outcome we wanted.
      if (/NOT_FOUND/i.test(msg)) {
        debuglog("Single-use terminate: node already gone", { nodeId });
        return;
      }
      this.settings.onError?.(err instanceof Error ? err : new Error(msg));
    });
  }

  // ---- node lifecycle ----

  private handleNodeStarting(nodeMetadata: NodeMetadata): void {
    this.updateNodeMap(nodeMetadata.nodeId, "starting", nodeMetadata, undefined);
    this.settings.onNodeStarting?.(nodeMetadata.nodeId);
    this.armStartupGrace(nodeMetadata.nodeId);
  }

  /** Phase C3. Schedule the grace timer; replace any prior timer. */
  private armStartupGrace(nodeId: NodeId): void {
    this.startupGraceTimers.get(nodeId)?.cancel();
    const grace = this.cfgJobStartupGraceMs();
    const timer = this.clock.setTimeout(() => {
      this.startupGraceTimers.delete(nodeId);
      debuglog("Node startup grace exceeded — terminating", { nodeId, grace });
      void this.issueTerminate(nodeId).catch((err) => this.settings.onError?.(err));
    }, grace);
    this.startupGraceTimers.set(nodeId, timer);
  }

  private cancelStartupGrace(nodeId: NodeId): void {
    this.startupGraceTimers.get(nodeId)?.cancel();
    this.startupGraceTimers.delete(nodeId);
  }

  private handleNodeStarted(
    nodeMetadata: NodeMetadata,
    runningNodeMetadata: RunningNodeMetadata
  ): void {
    this.updateNodeMap(
      nodeMetadata.nodeId,
      "running",
      nodeMetadata,
      runningNodeMetadata
    );
    this.cancelStartupGrace(nodeMetadata.nodeId);
    // A node hydrated jobs were waiting for has reported in.
    const unseen = this.unseenBoundNodes.get(nodeMetadata.nodeId);
    if (unseen) {
      unseen.cancel();
      this.unseenBoundNodes.delete(nodeMetadata.nodeId);
    }
    // Phase C2: a successful start clears restart history for this
    // job — the "fundamentally broken" diagnosis only fires when no
    // amount of replacement is helping, so a job that has ever started
    // is by definition not fundamentally broken.
    const jobId = getJobIdForNode(nodeMetadata);
    if (jobId) {
      this.restartHistory.delete(jobId);
      this.lastCountedDispatchRefusal.delete(jobId);
    }
    // Phase C6: a pending spare just landed.
    this.hotSpares.spareLanded(nodeMetadata.nodeId);
    const node = this.nodeMap.get(nodeMetadata.nodeId);
    if (node) this.settings.onNodeStarted?.(node);
    // The node we provisioned for a job has come up — bind the job to it.
    // Nothing else does this: createAwsNode/createOciNode carry the job
    // only as tags the daemon doesn't act on, and the placement engine
    // never re-runs for the job (the binding was set at provision time, so
    // `wantsPlacementNow` is false). Without this call the job sits in
    // "pre" until the daemon's idle reaper terminates the node — every
    // historical case where dispatch appeared to work was churn: some
    // other node's death released the binding and recovery re-placed onto
    // this one. Observed live, 2026-08-13 worker e2e.
    void this.bindProvisionedJob(nodeMetadata).catch((err) => this.settings.onError?.(err));
    this.reconcileHotSpares();
    this.reconcileNodeServices();
    // A node coming online is new capacity — a cluster worker joining or
    // a hot spare landing can host a job that previously had nowhere to
    // go. Re-attempt pending placements.
    this.reconcilePendingPlacements();
    // A migration target's node coming up is NOT a migration signal:
    // the saga advances on the target instance's own history events
    // (running / prepared), which bindProvisionedJob's dispatch above
    // sets in motion.
  }

  /** Dispatch startJob for the job a freshly-running node was provisioned
   *  for. Guards make this idempotent and narrow:
   *  - only nodes the AutoManager provisioned (`auto-manager` tag) with a
   *    `jobId` tag (spares carry no jobId);
   *  - only when the job is bound to this node, or to nothing (hydration
   *    after a restart mid-boot: history had no node entry, so the binding is
   *    empty — the tagged node claims the job rather than letting the
   *    recovery loop provision a duplicate). A double-provision loser —
   *    job bound to a DIFFERENT node — stays idle for the reaper;
   *  - only jobs in `pre` or `active` (matching wantsPlacementNow and the
   *    daemon's own provisionJob validation — an `active` job whose node
   *    died gets re-provisioned and needs this bind exactly like a fresh
   *    one; observed live 2026-08-14: a launch-failed job stayed `active`,
   *    a `pre`-only guard skipped the replacement node, and the job hung).
   *    Genuinely-running actives are covered by the binding guard above and
   *    the daemon's ProvisionJobDuplicate check;
   *  - never while a migration owns the job's lifecycle.
   *  the binding is (re)confirmed synchronously before the await so the
   *  hydration recovery loop can't re-place the job under the dispatch.
   *  The daemon rejects duplicates (`ProvisionJobDuplicate`), so a re-fire
   *  is an error surfaced via onError, not a double-start. */
  private async bindProvisionedJob(nodeMetadata: NodeMetadata): Promise<void> {
    const tags = nodeMetadata.tags ?? {};
    if (tags["auto-manager"] !== "true") return;
    const nodeId = nodeMetadata.nodeId;
    const ownerJobId = tags["jobId"];
    // Dispatch the node's provisioning owner IF it is still bindable. Each
    // guard skips ONLY the owner — never the join-er loop below, which MUST
    // run even when the owner was deleted / went terminal before the node
    // arrived. Otherwise a job that joined this node during its boot
    // (prospective placement) is never dispatched and hangs in `pre` until
    // the node is reaped. Observed live 2026-08-25: node provisioned for
    // repro-e, repro-e deleted mid-boot, repro-f prospectively joined it;
    // bindProvisionedJob bailed on the missing owner and abandoned repro-f.
    // A node provisioned for a migration TARGET carries the target's key
    // in its role tag. Its dispatch is the migration's (below), never
    // the owner path's: with no migration in flight for it — aborted,
    // or lost to a reconnect — it is a leftover, and starting a fresh
    // instance under a stale migration key would only be a duplicate
    // the daemon refuses. Leave it idle for the reaper.
    const ownerIsStaleMigrationTarget =
      ownerJobId !== undefined &&
      (tags["role"] ?? "").includes("@") &&
      !this.bindings.isTargetOn(ownerJobId, nodeId);
    if (ownerIsStaleMigrationTarget) {
      debuglog("Arrived node was provisioned for a migration target no longer in flight — not binding", {
        jobId: ownerJobId,
        role: tags["role"],
        nodeId,
      });
    }
    if (ownerJobId && !ownerIsStaleMigrationTarget) {
      const bound = this.bindings.nodeOf(ownerJobId);
      const jh = this.jobMap.get(ownerJobId);
      const ownerBindable =
        (bound === undefined || bound === nodeId) &&
        jh !== undefined &&
        !jh.job.terminalState &&
        (jh.job.state === "pre" || jh.job.state === "active") &&
        this.bindings.migrationOf(ownerJobId) === undefined;
      if (ownerBindable) {
        const role =
          tags["role"] ?? roleFromReplicaIndex(parseInt(tags["replicaIndex"] ?? "0", 10));
        this.bindings.adopt(ownerJobId, nodeId);
        // A restart/reconnect hydration replays running nodes through this
        // path — but if the job's own history shows the daemon already
        // accepted a dispatch on this node (provisioned/running with no
        // later teardown), the job is not waiting for its deferred
        // dispatch: it is ALREADY live. Re-firing startJob would only draw
        // ProvisionJobDuplicate, and the dispatch-failure recovery would
        // then unbind → re-place → same node → duplicate again, burning
        // the restart ceiling and stopping a healthy production within a
        // second (observed live on every norsk-mgr restart, 2026-08-26).
        // Adopt the binding instead.
        if (daemonOwnsJobOnNode(jh, nodeId)) {
          debuglog("Provisioned node already runs its job — adopting binding", {
            jobId: ownerJobId,
            nodeId,
          });
        } else {
          debuglog("Binding provisioned node to its job", { jobId: ownerJobId, nodeId, role });
          try {
            await this.norsk.startJob(ownerJobId, role, nodeId);
          } catch (err) {
            this.handleDispatchFailure(ownerJobId, nodeId, err, "bindProvisionedJob");
          }
        }
      }
    }
    // Join-ers: jobs placed onto this node while it was still launching
    // (prospective placement) bound with dispatch deferred
    // to exactly this moment. The tag-named owner above is skipped.
    for (const joinedJobId of this.bindings.jobsOn(nodeId)) {
      if (joinedJobId === ownerJobId) continue;
      const joined = this.jobMap.get(joinedJobId);
      if (!joined || joined.job.terminalState) continue;
      if (joined.job.state !== "pre" && joined.job.state !== "active") continue;
      if (this.bindings.migrationOf(joinedJobId) !== undefined) continue;
      // Same adopt-don't-redispatch rule as the owner path above.
      if (daemonOwnsJobOnNode(joined, nodeId)) {
        debuglog("Arrived node already runs joined job — adopting binding", {
          jobId: joinedJobId,
          nodeId,
        });
        continue;
      }
      const joinedRole = roleFromReplicaIndex(
        parseInt(joined.job.tags?.["replicaIndex"] ?? "0", 10)
      );
      debuglog("Binding joined job to arrived node", { jobId: joinedJobId, nodeId, role: joinedRole });
      try {
        await this.norsk.startJob(joinedJobId, joinedRole, nodeId);
      } catch (err) {
        this.handleDispatchFailure(joinedJobId, nodeId, err, "bindProvisionedJob:joiner");
      }
    }
    // Migration targets bound to this node — the node was provisioned
    // for one, or one joined it while it was booting. Dispatched with
    // the migration fields, and looked up NOW rather than at bind time:
    // a migration aborted meanwhile has released its binding, and its
    // target must not be started for nobody.
    for (const targetJobId of this.bindings.targetsOn(nodeId)) {
      const state = this.migrationForJob(targetJobId);
      if (!state || state.phase !== "launching") continue;
      debuglog("Binding migration target to arrived node", {
        jobId: targetJobId,
        role: state.targetRole,
        migrationId: state.migrationId,
        nodeId,
      });
      await this.migrationSaga.dispatchMigrationTarget(state, nodeId);
    }
  }

  /** A startJob dispatch to a bound node failed. Unlike a launch rejection
   *  (the worker refused after starting to launch), this is the daemon or
   *  transport refusing the dispatch itself — historically it only bumped a
   *  counter and fired a context-free onError, leaving the job bound to a
   *  node that never ran it: the node sat idle until the elastic sweeper
   *  reaped it (5-min dead zone), and the failure never counted toward the
   *  restart ceiling, so a persistent dispatch failure looped forever
   *  instead of terminating in a Failed bundle (observed live 2026-08-25).
   *
   *  Now: release the binding, surface the error WITH job+node context, and
   *  re-place through the recovery path — which runs the restart-rate
   *  ceiling, so a job that can never dispatch trips `shouldBreakOnRestart`
   *  → markJobFailed → BundleFailed rather than spinning. Excludes the node
   *  we couldn't reach from the immediate re-place. */
  private handleDispatchFailure(
    jobId: JobId,
    nodeId: NodeId,
    err: unknown,
    phase: string
  ): void {
    const msg = err instanceof Error ? err.message : String(err);
    // ProvisionJobDuplicate ("Job already running on node") — but ONLY when
    // the job's own history corroborates it (provisioned/running on this
    // node, no later teardown): the dispatch was redundant (a hydration
    // replay), not failed. Adopt: keep the binding, no ceiling count, no
    // re-place. Routed through the generic failure path, this rejection
    // used to loop unbind → re-place → same node → duplicate → … seven
    // times in under a second, trip the restart ceiling, and stop a
    // healthy production on every norsk-mgr restart (2026-08-26).
    //
    // WITHOUT that corroboration the duplicate is stale — a prior
    // incarnation of the same jobId still tearing down on the node (e.g. a
    // production Retry cascade-replacing its jobs; the daemon releases the
    // binding when the old instance finishes stopping). Adopting there
    // wedges the new job in `pre` forever (observed live 2026-08-26, first
    // deploy of this fix) — so it falls through to the normal
    // dispatch-failure recovery, whose re-place converges once the old
    // binding releases.
    if (/already running on node/i.test(msg)) {
      const dupJh = this.jobMap.get(jobId);
      if (dupJh && daemonOwnsJobOnNode(dupJh, nodeId)) {
        debuglog("startJob duplicate — daemon already runs the job; adopting binding", {
          jobId,
          nodeId,
          phase,
        });
        this.lastCountedDispatchRefusal.delete(jobId);
        return;
      }
    }
    // How this refusal is to be treated (item 5, 2026-09-10): whether the
    // node that gave it stays out of the immediate re-place, and whether a
    // consecutive repeat of it is a new failure for the ceiling.
    const refusal = classifyDispatchFailure(msg);
    const signature = `${nodeId}|${refusal.reason}`;
    const repeatOfCountedCondition =
      refusal.standing && this.lastCountedDispatchRefusal.get(jobId) === signature;
    if (refusal.standing) this.lastCountedDispatchRefusal.set(jobId, signature);
    else this.lastCountedDispatchRefusal.delete(jobId);
    // "Node is not in running state": the daemon says this node is
    // stopping/terminated — it is NOT viable capacity, whatever our cached
    // inventory claims (its NodeStopping event may still be in flight, and
    // a still-alive worker can even heartbeat the inventory back in). Drop
    // the inventory and mark the node recently-failed so the recovery
    // re-place below lands ELSEWHERE — without this, placement re-picked
    // the same corpse until the restart ceiling failed the production
    // (observed live 2026-08-27: single-use terminate + Retry re-expansion).
    if (/not in (a )?running state/i.test(msg)) {
      this.inventoryMap.delete(nodeId);
      this.recentFailures.set(
        nodeId,
        new Date(this.clock.now().getTime() + this.cfgFailureBackoffMs())
      );
    }
    this.metricsCounters.placementsDispatchFailed++;
    // The attempt conclusively failed, so the in-flight suppression window
    // must not hold the next placement back — the same reasoning (and the
    // same pairing with the unbind) as handleJobRejected. It matters more
    // now that a standing refusal steers the immediate re-place away from
    // this node: what converges the job is the next capacity event, and
    // the gate would silently swallow the first of them.
    if (!this.bindings.releaseFrom(jobId, nodeId)) this.bindings.clearAttempt(jobId);
    this.settings.onError?.(
      new Error(`startJob dispatch failed [${phase}] for job ${jobId} on node ${nodeId}: ${msg}`)
    );
    const jh = this.jobMap.get(jobId);
    const req = jh?.job.autoPlacement?.requirements;
    if (!jh || !req || jh.job.terminalState) return;
    // Recovery re-place: passing excludeNodes marks this a recovery attempt,
    // so placeAndProvision runs the restart-rate ceiling + backoff (that is
    // where the failure is counted — one count per re-place, no double).
    //
    // An UNCLASSIFIED dispatch error does not exclude the failed node: a
    // dispatch glitch is not evidence the node is bad, and excluding the
    // only candidate would leave nothing to dispatch to, so the failure
    // would never re-count and the ceiling would never trip. Retrying (same
    // node included) converges: either it starts, or the repeated failure
    // trips the ceiling → markJobFailed → BundleFailed.
    //
    // A STANDING refusal is the opposite case and gets the opposite
    // treatment (2026-09-10). `Job already running on node` from a daemon
    // holding a leaked binding cannot be cleared by asking again, so the
    // node comes out of the immediate re-place and placement converges
    // somewhere else. If there IS nowhere else, the job stays unplaced
    // rather than storming: the next capacity event (an inventory
    // heartbeat, a node arriving) re-places it through the ordinary
    // non-recovery path, with the refusing node eligible again — which is
    // also how it recovers once the stale binding finally releases.
    // (A rejection that BLAMES the node — no capacity, missing hardware —
    // excludes it too; see handleJobRejected.)
    const exclude = this.currentExcludeSet();
    if (refusal.excludeNode) exclude.add(nodeId);
    void this.placeAndProvision(jobId, jh.job.tags ?? {}, {
      excludeNodes: exclude,
      ...(repeatOfCountedCondition ? { repeatOfCountedCondition: true } : {}),
    }).catch((e) => this.settings.onError?.(e));
  }

  private async handleNodeStopping(
    nodeId: NodeId,
    _reason: NodeStopping["reason"]
  ): Promise<void> {
    this.updateNodeMap(nodeId, "stopping", undefined, undefined);
    this.cancelStartupGrace(nodeId);
    this.singleUseTerminated.delete(nodeId); // node going away — drop the dedupe mark
    // Surface to consumers so cluster-membership views (mgr's Cluster
    // page) refresh promptly. Without this callback, a Ctrl-C'd
    // cluster worker only surfaces in mgr UI when the eventual
    // NodeStopped/NodeTerminated arrives — which can be many seconds
    // later via the daemon's keepalive timeout + cleanup loop.
    this.settings.onNodeStopping?.(nodeId);
    await this.recoverLostNode(nodeId);
  }

  /**
   * The recovery a lost node needs, whichever event reported the loss:
   * `NodeStopping` normally, or a `NodeStopped` with no `NodeStopping`
   * before it (the daemon emits the two from separate sites). Drops the
   * node as placeable capacity, resolves migrations touching it, and
   * re-places every job still bound to it.
   */
  private async recoverLostNode(nodeId: NodeId): Promise<void> {
    // Mark the node as recently-failed so subsequent placements skip it
    // for `failureBackoffMs`. Cleared when the node next reports a
    // healthy `nodeInventoryUpdated`.
    const window = this.cfgFailureBackoffMs();
    this.recentFailures.set(
      nodeId,
      new Date(this.clock.now().getTime() + window)
    );
    // Fix B (2026-08-25): a dying node must stop being placeable capacity.
    // inventoryMap is otherwise only cleared on a full resync, so a reaped
    // node lingers as `reachable` and placement keeps selecting it — the job
    // then dispatches to a NodeId the daemon no longer has (NOT_FOUND). Drop
    // it now, before the recovery re-place below runs.
    this.inventoryMap.delete(nodeId);

    // Migrations touching this node are resolved FIRST, so the ordinary
    // recovery below sees the post-migration bindings:
    //  - the node holds a migration TARGET: the target is lost — abort
    //    (the source, elsewhere, is untouched) or, past the point of no
    //    return, fail forward into recovery;
    //  - the node holds a migration SOURCE: if the target is already
    //    running, the job completes into it — that is what a migration
    //    is for, and R8's "abort and leave the job unplaced" is gone;
    //    otherwise abort, and the ordinary re-place below recovers the
    //    job like any other node loss (its binding still names this node).
    for (const state of this.inFlightMigrations()) {
      if (state.targetNodeId === nodeId) {
        await this.migrationSaga.handleTargetLost(state, `target node stopping during phase ${state.phase}`);
      } else if (state.sourceNodeId === nodeId) {
        await this.migrationSaga.handleSourceLost(state, `source node stopping during phase ${state.phase}`);
      }
    }

    // A node its provider reclaimed (spot interruption / rebalance —
    // `handleNodeInterruption`): the jobs it takes with it are re-placed
    // like any other loss, but the loss is the provider's, not theirs, so
    // it does not count against their restart ceiling.
    const reclaimed = this.reclaimedNodes.has(nodeId);
    if (reclaimed) debuglog("Lost node was reclaimed by its provider — recovery will not count", { nodeId });
    const recovery = (): { excludeNodes: Set<NodeId>; reclaimed?: boolean } => ({
      excludeNodes: this.currentExcludeSet(),
      ...(reclaimed ? { reclaimed: true } : {}),
    });
    const node = this.nodeMap.get(nodeId);
    const jobId = node?.provisional?.jobId ?? getJobIdForNode(node?.nodeMetadata);
    if (jobId) {
      // Only when the owner still lives here (or is unbound): a job that
      // has since been re-placed — or has just completed a migration
      // onto another node — is running elsewhere, and re-placing it
      // would make a third instance.
      const bound = this.bindings.nodeOf(jobId);
      if (bound !== undefined && bound !== nodeId) {
        debuglog("Stopping node's owner job now lives elsewhere — no recovery", {
          nodeId,
          jobId,
          boundTo: bound,
        });
      } else {
        // Unbind first, exactly as the boundHere loop below and
        // recoverFailedRuntime do. Left bound to the corpse, a job whose
        // recovery placement fails (no capacity yet) reads as placed to
        // `wantsPlacementNow`, and the reconcile that runs when capacity
        // arrives skips it — for good, since nodesByJob ignores nodeState.
        this.bindings.unbind(jobId);
        const entry = this.jobMap.get(jobId);
        const job = entry?.job;
        const req = job?.autoPlacement?.requirements;
        if (job && req && !job.terminalState && (job.state === "pre" || job.state === "active")) {
          await this.placeAndProvision(job.jobId, job.tags ?? {}, recovery());
        }
      }
    }
    // Fix A (2026-08-25): recover EVERY job bound to this dying node, not just
    // the node's provisioning owner (handled above via its metadata tag). A
    // job placed onto an already-existing node — e.g. one binpacked onto an
    // empty, still-lingering elastic node that is then reaped — lives only in
    // its binding, never in the node's daemon-side runningJobs, so the daemon
    // emits no JobStopped for it and the owner-only recovery misses it. Left
    // bound to the dead node, `wantsPlacementNow` reads it as already-placed
    // and the periodic reconcile skips it — it hangs in `pre` forever
    // (observed live 2026-08-25: repro-f on a reaped repro-e node). Generalises
    // the orphan-scan abandonPendingProvision already does for pending nodes.
    const boundHere = this.bindings.jobsOn(nodeId).filter((j) => j !== jobId);
    for (const boundJobId of boundHere) {
      // Unbind first so the stale binding can't keep the job looking placed
      // (and clear the in-flight suppression so the retry isn't delayed).
      // No migration can still be in flight for these: every one whose
      // source was this node was resolved above.
      this.bindings.unbind(boundJobId);
      const jh = this.jobMap.get(boundJobId);
      const jreq = jh?.job.autoPlacement?.requirements;
      if (jh && jreq && !jh.job.terminalState && (jh.job.state === "pre" || jh.job.state === "active")) {
        await this.placeAndProvision(boundJobId, jh.job.tags ?? {}, recovery());
      }
    }
    // Phase C6: if the stopping node was a spare, reconcile to
    // provision a replacement promptly (don't wait for the tick).
    this.reconcileHotSpares();
    // Phase F: lost capacity → reconcile NodeService placements.
    // Also drop any tracked placement on this node (the runtime job
    // is gone with the node).
    this.dropNodeServicePlacementsOnNode(nodeId);
    this.reconcileNodeServices();
  }

  /**
   * The daemon changed a node's tags. Mostly one thing: a hot spare
   * was given a job and is a spare no longer — the daemon drops `spare`
   * the moment a job is bound to the node (task 90, the spare-pool churn:
   * the daemon's empty-node sweep used to reap every spare five minutes
   * after it came up, because a spare IS an empty elastic node, and the
   * reconciler replaced it, cold, for ever; now a node tagged `spare` is
   * exempt from the sweep, and ordinary once consumed). The metadata is
   * adopted whole, and a spare that stopped being one is a shortfall the
   * reconciler fills straight away — unless it was a LOCAL spare, which the
   * reconciler never counted, so the reconcile it runs finds nothing short.
   * The other thing it is: an operator marking or clearing a local server
   * as a spare (`setLocalSpare`), whose tags land the same way.
   */
  private handleNodeTagsUpdated(nodeMetadata: NodeMetadata): void {
    const nodeId = nodeMetadata.nodeId;
    const existing = this.nodeMap.get(nodeId);
    if (!existing) return;
    const wasSpare = this.isHotSpareNode(nodeId);
    this.updateNodeMap(nodeId, existing.nodeState, nodeMetadata, undefined);
    const isSpare = this.isHotSpareNode(nodeId);
    debuglog("Node tags updated", { nodeId, tags: nodeMetadata.tags, wasSpare, isSpare });
    if (wasSpare && !isSpare) this.reconcileHotSpares();
  }

  private async handleNodeStopped(nodeId: NodeId): Promise<void> {
    // A NodeStopped with no NodeStopping before it is the only report of
    // this node's loss: it gets the recovery the stopping path runs. Only
    // for a node we knew as running — a pending provision's jobs are
    // re-placed by abandonPendingProvision just below (running the loss
    // recovery too would place them twice), and a node already `stopping`
    // has had its recovery.
    const lostWithoutWarning = this.nodeMap.get(nodeId)?.nodeState === "running";
    this.abandonPendingProvision(nodeId, "node stopped");
    // Idempotent with handleNodeStopping (Fix B).
    this.inventoryMap.delete(nodeId);
    this.updateNodeMap(nodeId, "stopped", undefined, undefined);
    // Phase C6: a spare going away should trigger a replacement. A spare
    // that stops while still pending never ran: its create failed, and the
    // replacement waits out the backoff rather than repeating the failure.
    this.hotSpares.spareStopped(nodeId);
    this.settings.onNodeStopped?.(nodeId);
    if (lostWithoutWarning) {
      this.cancelStartupGrace(nodeId);
      this.singleUseTerminated.delete(nodeId);
      await this.recoverLostNode(nodeId); // also reconciles spares + NodeServices
      this.reclaimedNodes.delete(nodeId); // the reclaim is over with the node
      return;
    }
    this.reclaimedNodes.delete(nodeId);
    this.reconcileHotSpares();
    // Phase F: clear NodeService placements on stopped nodes.
    this.dropNodeServicePlacementsOnNode(nodeId);
    this.reconcileNodeServices();
  }

  private dropNodeServicePlacementsOnNode(nodeId: NodeId): void {
    this.nodeServiceReconciler.dropNode(nodeId);
  }

  // ---- maps ----

  private updateNodeMap(
    nodeId: NodeId,
    nodeState: NodeState,
    nodeMetadata: NodeMetadata | undefined,
    runningNodeMetadata: RunningNodeMetadata | undefined,
    provisional?: { jobId: JobId; role: Role }
  ): void {
    const existing = this.nodeMap.get(nodeId);
    this.nodeMap.set(nodeId, {
      nodeId,
      nodeState,
      lastActivity: this.clock.now(),
      nodeMetadata: nodeMetadata ?? existing?.nodeMetadata,
      runningNodeMetadata: runningNodeMetadata ?? existing?.runningNodeMetadata,
      provisional: provisional ?? existing?.provisional,
    });
  }
}

function summariseTrace(trace: PlacementTrace): string {
  // One-line per attempted pool, listing how nodes were filtered
  // (grouped by reason) and what happened. Compact enough that it
  // doesn't drown a debuglog stream when many placements happen in a
  // burst, but informative enough that "why did my job land there?"
  // is immediately answerable.
  const lines = trace.attempts.map((a) => {
    const groups: Record<string, number> = {};
    for (const f of a.filtered) groups[f.reason] = (groups[f.reason] ?? 0) + 1;
    const filterSummary = Object.entries(groups)
      .map(([reason, count]) => `${reason}=${count}`)
      .join(", ");
    const winner =
      a.outcome.kind === "placed"
        ? ` → placed on ${a.outcome.nodeId}${a.outcome.gpuIndex !== undefined ? ` (gpu ${a.outcome.gpuIndex})` : ""}`
        : a.outcome.kind === "provisioned"
          ? ` → provisioned ${a.outcome.instanceType}`
          : a.outcome.kind === "exhausted"
            ? " → fixed-pool exhausted, fall through"
            : a.outcome.kind === "no-instance-type"
              ? " → no allowed instance type fits, fall through"
              : " → pool not configured";
    const matchSummary = `${a.scored.length}/${a.initialCandidates} matched`;
    const filterPart = filterSummary ? `, filtered: ${filterSummary}` : "";
    return `[${a.pool}] ${matchSummary}${filterPart}${winner}`;
  });
  return `${trace.decision}: ${lines.join(" || ")}`;
}

export function getJobIdForNode(nodeMetadata: NodeMetadata | undefined): JobId | undefined {
  return nodeMetadata?.tags["jobId"];
}

/**
 * Walk a job's history backward and return the nodeId where it is
 * currently placed (or was most recently placed). Prefers `running`
 * entries; falls back to `provisioned` for jobs still booting their
 * node. Returns undefined for jobs that have never been placed or
 * whose history is empty (defensive — runtime data always has at
 * least a `created` entry, which carries no nodeMetadata).
 *
 * This is the canonical "where does this job live" lookup and
 * underpins `AutoManager.nodeForJob`. It works regardless of whether
 * a node hosts one job (cloud, today) or many (cluster, or future
 * cloud N:1) because the nodeId is recorded against the JOB rather
 * than scanned out of NODE tags.
 */
/**
 * Does the daemon already hold a live dispatch for this job on `nodeId`?
 * True when the newest decisive history event is `provisioned` or
 * `running` on that node — i.e. startJob was already accepted and no
 * teardown (`stopping`/`stopped`/`completed`/`rejected`) has landed
 * since. Used by the bind/dispatch paths to ADOPT such a binding rather
 * than re-fire startJob, which the daemon would refuse as
 * ProvisionJobDuplicate ("Job already running on node") — and which the
 * dispatch-failure recovery would then escalate into a restart-ceiling
 * trip that kills the healthy job (the 2026-08-26 restart incident).
 *
 * A job that previously ran on the node but has since stopped returns
 * false: the daemon released it, so a fresh dispatch is required (and
 * valid).
 */
function daemonOwnsJobOnNode(jh: JobWithHistory, nodeId: NodeId): boolean {
  // Wire history is NEWEST-first (JobStore prepends via NEL.cons and the
  // ServiceHost serialises with NEL.toList) — index 0 is the latest
  // event.
  //
  // The question is asked per INSTANCE KEY, not per job. After a handover
  // the job's newest events are the SOURCE's `stopping` / `stopped` under
  // the slot key, written after the target's `running` under its migrated
  // key — a job-wide "first decisive event wins" scan therefore answered
  // "not owned" for every migrated job, and the hydration replay of its
  // node through bindProvisionedJob re-fired startJob under the bare slot
  // key onto the node the instance was running on. The worker refused
  // ("already held under role primary@…"), the dispatch-failure recovery
  // unbound and re-placed, and the job churned for 90 s before landing on
  // a fresh node (observed live 2026-09-09 13:26Z, task 90). So: find the
  // instance the job currently lives as (its newest `running`, else its
  // newest placement) and decide on the newest event ABOUT THAT KEY.
  // Events under ANOTHER key (the source a handover left behind) are
  // skipped; `completed` / `rejected` carry no key and are job-level
  // teardowns, so they stay decisive for whatever key is current — a
  // `rejected` after a `provisioned` still means "dispatch again".
  const current = currentInstanceForJob(jh);
  if (current === undefined || current.nodeId !== nodeId) return false;
  for (const e of jh.history) {
    switch (e.event) {
      case "completed":
      case "rejected":
        return false;
      case "provisioned":
      case "prepared":
      case "running":
        if (e.role !== current.role) break;
        return e.nodeMetadata.nodeId === nodeId;
      case "stopping":
      case "stopped":
        if (e.role !== current.role) break;
        return false;
      default:
        break; // created / updated — not decisive, keep scanning
    }
  }
  return false;
}

export function currentNodeIdForJob(jh: JobWithHistory | undefined): NodeId | undefined {
  return currentInstanceForJob(jh)?.nodeId;
}

/**
 * The job's current instance, from history: where it is (or was most
 * recently placed) and the instance KEY it wears there. Prefers the
 * most recent `running` over the most recent `provisioned` / `prepared`
 * — a job provisioned, run, stopped and re-provisioned but not yet run
 * again reports the newest provisioned node, but a running instance
 * always wins. `prepared` ranks with `provisioned`: containers exist,
 * none run.
 *
 * Wire history is NEWEST-first (JobStore prepends), so the scan starts
 * at index 0 and the first `running` met IS the most recent one. (The
 * earlier oldest-first walk returned the OLDEST running node — the same
 * answer for a job that has only ever lived on one node, and the wrong
 * one for a job that has moved: recovery, or a completed migration.)
 *
 * The role matters because after a migration the running instance is
 * `primary\@xxxxxxxx`, not `primary`, and that is the key the Manager
 * routes StopJob and the handshake by.
 */
export function currentInstanceForJob(
  jh: JobWithHistory | undefined
): { nodeId: NodeId; role: Role; event: "running" | "provisioned" | "prepared" } | undefined {
  if (!jh) return undefined;
  let placedButNotRunning: { nodeId: NodeId; role: Role; event: "provisioned" | "prepared" } | undefined;
  for (const e of jh.history) {
    if (e.event === "running") {
      return { nodeId: e.nodeMetadata.nodeId, role: e.role, event: "running" };
    }
    if ((e.event === "provisioned" || e.event === "prepared") && placedButNotRunning === undefined) {
      placedButNotRunning = { nodeId: e.nodeMetadata.nodeId, role: e.role, event: e.event };
    }
  }
  return placedButNotRunning;
}

/**
 * Build a `PlacementDecisionInfo` from a hydrated job's history. Used
 * on R7 reconnect to seed the placement of record for jobs that were
 * placed in a previous generation — we have no live trace, so this
 * synthesises a minimal record (empty `trace.attempts`) that's still
 * shaped like a real decision so consumers can render it uniformly.
 *
 * `nodeId` is passed in by the caller (already resolved via
 * `currentNodeIdForJob`), to avoid walking the history twice.
 *
 * Returns undefined if the job's tags are too sparse to derive
 * `bundleId` / `replicaIndex` / `jobName` (un-managed jobs the
 * AutoManager-driven UI couldn't usefully display anyway).
 */
function synthesisePlacementFromHistory(jh: JobWithHistory, nodeId: NodeId): PlacementDecisionInfo | undefined {
  const tags = jh.job.tags ?? {};
  const bundleId = tags["bundleId"];
  const replicaIndexStr = tags["replicaIndex"];
  const jobName = tags["jobName"];
  if (!bundleId || replicaIndexStr === undefined || !jobName) return undefined;
  const replicaIndex = Number(replicaIndexStr);
  if (!Number.isFinite(replicaIndex)) return undefined;

  // Find the most recent placement event for `at` + decision kind.
  // running > provisioned (same precedence as currentNodeIdForJob).
  // If the only event we found is `provisioned`, decision="provision"
  // (the node was started for this job); `running` could be either —
  // we treat it as "place" because by the time we hydrate the node is
  // already running, so functionally the job was bound to an existing
  // running node. The trace stays empty either way.
  let at: Date | undefined;
  let decision: "place" | "provision" = "place";
  // NEWEST-first, like currentInstanceForJob.
  for (const e of jh.history) {
    if (e.event === "running") {
      at = e.timestamp instanceof Date ? e.timestamp : new Date(e.timestamp);
      decision = "place";
      break;
    }
    if (e.event === "provisioned" && at === undefined) {
      at = e.timestamp instanceof Date ? e.timestamp : new Date(e.timestamp);
      decision = "provision";
    }
  }
  if (!at) return undefined;

  // `pool`, `tier` and `instanceType` aren't on the history entries —
  // leave them as best-guess placeholders. Consumers that care about
  // accurate pool/tier attribution should walk the node's tags via
  // nodeMap.get(nodeId).nodeMetadata.tags.{pool,tier}.
  const result: PlacementResult =
    decision === "place"
      ? { kind: "place", nodeId, pool: "", tier: "" }
      : { kind: "provision", pool: "", tier: "", instanceType: "", launchMode: "on-demand" };
  const trace: PlacementTrace = { attempts: [], decision };
  return { jobId: jh.job.jobId, bundleId, replicaIndex, jobName, result, trace, at };
}