// Pure-function placement engine. Given a job-to-place, the bundle it
// belongs to, the current inventory snapshot, and the pool configuration,
// returns a placement decision: place on a specific node, provision a new
// node in an elastic pool, or fail.
//
// Per §6 of the design doc: filter by hard constraints, then rank the
// survivors. A backup (replicaIndex > 0) is placed from its
// `resiliencePolicy.pool` when one is set; everything else from the
// bundle's pool.
//
// Ranking is a lexicographic KEY TUPLE (`placementKey`), compared component
// by component, most significant first — see PlacementKey for the order.
// There are no weights to tune: "avoid-spare outranks warmth" is true by
// construction, not by the size of two constants.

import {
  Bundle,
  BundleId,
  BundleJobSpec,
  Capability,
  CapabilityRequirement,
  GpuResource,
  JobRequirements,
  NodeId,
  NodeInventory,
  OnRuntimeFailure,
  ResiliencePolicy,
} from "./types";
import {
  freeCapabilityFor,
  freeCapacity,
  freeCores,
  reservedHostPortSet,
  usedCapacityFraction,
} from "./inventory";

/** @public */
export interface NodeView {
  nodeId: NodeId;
  poolName: string;
  /**
   * Which tier within the pool this node belongs to. Set from
   * `tags["tier"]` at provision time for elastic tiers. Pre-registered
   * cluster nodes carry no tier tag (undefined) and are treated as
   * belonging to their pool's `fixed` tier — see `nodeMatchesTier`.
   */
  tierName?: string;
  inventory: NodeInventory;
  /** Jobs currently running on this node (or pending placement onto it). */
  runningJobs: RunningJob[];
  /** Topology key for the resilience sameAz rule. Cluster pools can leave undefined. */
  az?: string;
  /**
   * Cloud / failure-domain key for the resilience sameCloud rule — the
   * node's provider (e.g. "aws" / "oci" / "cluster"). Undefined when
   * unknown; sameCloud then can't be evaluated against this node.
   */
  cloud?: string;
  /** Hot spare flag (Phase C). Currently unused by hard-constraint filters. */
  isHotSpare: boolean;
  /**
   * Phase G — the container images this node advertises holding, as
   * observed (the worker reports its local image store in its inventory).
   * Warmth is measured against this, never declared: a node that just ran
   * `abc` is warm for `abc` without anyone saying so. Absent/empty ⇒
   * unknown, which scores as cold rather than as warm.
   */
  heldImages?: HeldImage[];
  /**
   * Phase G — product-template names this node is RESERVED for. Set on a
   * hot spare from its `HotSpareConfig`. Reservation is who may take the
   * node; warmth is what is on its disk. Two properties, never one flag.
   */
  reservedFor?: string[];
  /**
   * Phase G — the reservation is exclusive: while this node is still
   * unconsumed reserve, only a job whose template is in `reservedFor` may
   * place onto it (unless the placement carries `overrideReservation`).
   * Default (absent/false) is non-exclusive: anyone may take it, and the
   * reserved templates merely score it first.
   */
  exclusiveReservation?: boolean;
  /**
   * A node still launching (machine-types.md Part B): its inventory is the
   * machine type's DECLARED shape, synthesized by AutoManager from the
   * in-flight provision. Passes the normal filter chain, carries a scoring
   * penalty so real capacity wins when both fit, and counts against tier
   * caps. Placement onto it defers job dispatch until the node arrives.
   */
  prospective?: boolean;
}

/**
 * Phase G — one container image, either as a product template needs it or
 * as a node holds it. `digest` is the real identity; `repoTag` is the
 * fallback the Manager has to match on when the template's images have not
 * been resolved to digests (the tar is opaque to Manager, so the image set
 * is supplied by the embedder and may name only `repository:tag`).
 *
 * @public
 */
export interface ImageRef {
  /** `repository:tag` exactly as written. */
  repoTag?: string;
  /** Content digest, `sha256:...`, when known. */
  digest?: string;
}

/**
 * Phase G — an image a node advertises holding. One digest may carry
 * several `repository:tag` names on the same node.
 *
 * @public
 */
export interface HeldImage {
  digest?: string;
  repoTags?: string[];
}

/**
 * Phase G — what fraction of the images this job needs the node already
 * holds, in [0,1]. Digest match first (the honest identity); when the
 * template's image set was not resolved to digests, `repository:tag` is
 * the fallback — a tag can be re-pushed under the same name, so a tag
 * match is a good guess about the pull cost, not a guarantee of identity.
 *
 * An unknown image set (empty `images`), or a node that advertises nothing,
 * scores 0: no warmth signal, no bias. That is what every caller sees until
 * the worker actually advertises its image store.
 *
 * @public
 */
export function warmth(node: NodeView, images: ImageRef[]): number {
  if (images.length === 0) return 0;
  const held = node.heldImages ?? [];
  if (held.length === 0) return 0;
  const digests = new Set<string>();
  const repoTags = new Set<string>();
  for (const h of held) {
    if (h.digest !== undefined) digests.add(h.digest);
    for (const t of h.repoTags ?? []) repoTags.add(t);
  }
  let hits = 0;
  for (const img of images) {
    if (img.digest !== undefined && digests.has(img.digest)) {
      hits++;
      continue;
    }
    if (img.repoTag !== undefined && repoTags.has(img.repoTag)) hits++;
  }
  return hits / images.length;
}

/** @public */
export interface RunningJob {
  bundleId: BundleId;
  replicaIndex: number;
  jobName: string;
  /** The occupant's spec declares exclusive node allocation — no other
   *  job may place onto this node (its own co-located peers excepted). */
  exclusive?: boolean;
}

/**
 * A pool is an ordered list of capacity tiers. Placement walks the tiers
 * in order (cheapest first) and takes the first that can satisfy the job
 * — the tier order *is* the cost cascade (no cost weight in v1). The
 * homogeneous single-cloud pool of before is just a one-tier pool.
 *
 * @public
 */
export interface PlacementPool {
  name: string;
  /** Ordered cheapest → most expensive. Placement tries them in order. */
  tiers: PlacementTier[];
  /**
   * Default true: nodes' security groups are reconciled at runtime as
   * jobs with extra SGs come and go (worker-security-groups.md). false
   * fixes each node's SG set at provision time — the opt-out is stamped
   * onto provisioned nodes as `tags["dynamicSecurityGroups"] = "false"`,
   * which the Manager's reconcile paths honour.
   */
  dynamicSecurityGroups?: boolean;
}

/**
 * One homogeneous capacity segment within a pool: a single cloud/cluster
 * kind + region with its own scale-out behaviour and candidate shapes.
 *
 * @public
 */
export interface PlacementTier {
  /** Unique within the pool. Tagged onto provisioned nodes as `tags["tier"]`. */
  name: string;
  /**
   * Cloud / local-servers kind. Used by AutoManager to decide which gRPC
   * call to make when provisioning (createAwsNode vs createOciNode vs
   * local-servers-tier placement, which uses startJob against an existing
   * registered server).
   */
  kind: "aws" | "oci" | "local-servers";
  /**
   * For aws/oci elastic tiers: which region/availability-domain to
   * provision into. Cluster tiers leave undefined.
   */
  region?: string;
  packingStrategy: "binpack" | "spread";
  scaleOut: "elastic" | "fixed";
  /**
   * For elastic tiers: shapes that can be provisioned on demand. For
   * fixed tiers: typically empty (the tier consists of registered
   * cluster nodes already in `inventory`).
   */
  candidateInstanceTypes: InstanceTypeOption[];
  /**
   * How long ahead of a scheduled job's `startDateTime` AutoManager
   * should kick off placement, so the worker is ready to run by the
   * requested time. Cluster tiers place against an existing running
   * node (~instant) so the default is 0. Cloud tiers have to boot a
   * fresh instance — defaults to 5 minutes; operators set per-tier
   * when their boot time differs. AutoManager picks the maximum across
   * the bundle pool's tiers (worst case — guarantees readiness
   * regardless of which tier the placement engine settles on).
   */
  placementLeadMs?: number;
  /**
   * How nodes in this tier are purchased. `"spot"` requests interruptible
   * spot-market capacity; `"reserved"` is an on-demand launch the operator
   * knows is covered by a standing reservation (same AWS call as on-demand,
   * but capped — see `maxNodes`). Defaults to `"on-demand"`.
   */
  launchMode?: LaunchMode;
  /**
   * The root EBS volume of nodes this tier provisions (aws tiers only).
   * Unset ⇒ the Manager's default, 100 GiB gp2. The size is where this
   * matters: it is the node's image store, and a tier warmed for many
   * templates (`warmFor`) holds more images than a bare one.
   */
  rootVolume?: RootVolume;
  /**
   * Failure-domain reliability of this tier's capacity. Defaults to
   * `"interruptible"` for spot, `"durable"` otherwise. Read by the
   * resilience engine — an interruptible-tier primary with no durable
   * backup is flagged degraded.
   */
  reliability?: Reliability;
  /**
   * Cap on the number of nodes this tier may hold (tier-wide, across
   * bundles). The cost cascade spills to the next tier once the cap is
   * reached. Used mainly for reserved tiers — exceeding the reservation
   * count just bills as on-demand, defeating the point. Unset ⇒ no cap.
   */
  maxNodes?: number;
  /**
   * How long an in-flight provision may stay prospective before it is
   * abandoned and its jobs re-placed (machine-types.md Part B). Default:
   * max(placementLeadMs, 5 min) × 2.
   */
  provisionTimeoutMs?: number;
}

/** @public */
export type LaunchMode = "on-demand" | "spot" | "reserved";

/** An AWS node's root volume: the size and EBS type it is launched with. @public */
export interface RootVolume {
  sizeGib: number;
  /** Defaults to "gp2". */
  volumeType?: "gp2" | "gp3" | "io1" | "io2";
  /** Provisioned IOPS, for the types that take one (gp3, io1, io2). */
  iops?: number;
}

/** @public */
export type Reliability = "durable" | "interruptible";

/** The tier's launch mode, defaulting to on-demand. @public */
export function tierLaunchMode(tier: PlacementTier): LaunchMode {
  return tier.launchMode ?? "on-demand";
}

/** The tier's reliability — explicit, else interruptible for spot. @public */
export function tierReliability(tier: PlacementTier): Reliability {
  return tier.reliability ?? (tierLaunchMode(tier) === "spot" ? "interruptible" : "durable");
}

/** How long an in-flight provision may stay prospective before it is
 *  abandoned: generous relative to expected boot time so a slow image pull
 *  doesn't orphan jobs, bounded so jobs never wait forever on a node that
 *  is not coming. @public */
export function provisionTimeoutMs(tier: PlacementTier): number {
  if (tier.provisionTimeoutMs !== undefined) return tier.provisionTimeoutMs;
  return Math.max(tier.placementLeadMs ?? defaultPlacementLeadMs(tier.kind), 5 * 60 * 1000) * 2;
}

/** Default placement lead time by tier kind. Cluster tiers place
 *  against existing running nodes so we don't need lead. Cloud tiers
 *  default to 5 minutes — typical EC2/OCI instance boot + image pull
 *  budget; operators can override per-tier via PlacementTier. */
export function defaultPlacementLeadMs(kind: PlacementTier["kind"]): number {
  return kind === "local-servers" ? 0 : 5 * 60 * 1000;
}

/**
 * Whether a node belongs to (pool, tier). Elastic nodes carry an explicit
 * `tierName` (set from `tags["tier"]` at provision). Pre-registered cluster
 * nodes carry no tier tag and are taken to belong to the pool's `fixed`
 * tier — so a pool must have at most one fixed tier (enforced in
 * settingsValidation) for this to be unambiguous.
 *
 * @public
 */
export function nodeMatchesTier(
  n: NodeView,
  pool: PlacementPool,
  tier: PlacementTier
): boolean {
  if (n.poolName !== pool.name) return false;
  if (n.tierName !== undefined) return n.tierName === tier.name;
  return tier.scaleOut === "fixed";
}

/** @public */
export interface InstanceTypeOption {
  instanceType: string;
  totalCapacity: number;
  totalCores: number;
  capabilities: Capability[];
  /** GPUs we could provision a node with — for feasibility check. */
  gpus?: { totalCapacity: number; model?: string }[];
}

/** @public */
export interface RecoveryContext {
  /** NodeIds the placement engine must skip (e.g., recently failed). */
  excludeNodes: Set<NodeId>;
  /**
   * The recovering job's failure policy, which is what decides how the
   * reserve is treated for this placement:
   *  - `"avoidSpare"`  — an unconsumed spare ranks below every ordinary
   *    node (PlacementKey.avoidedSpare), so a routine failure does not
   *    consume a reserve that exists for the workloads which asked for one;
   *  - `"preferSpare"` — among otherwise-equal candidates an unconsumed
   *    spare ranks first (PlacementKey.preferredSpare). Warmth outranks
   *    this: what `preferSpare` reaches for is a node the job starts on
   *    quickly, and spare-ness was only ever a proxy for that (fix-list 9);
   *  - `"drop"` / undefined — no opinion about spares (a planned migration
   *    passes undefined: the reserve is protected by reservation there).
   */
  onRuntimeFailure?: OnRuntimeFailure;
}

/** @public */
export interface PlacementInput {
  bundle: Bundle;
  job: BundleJobSpec;
  replicaIndex: number;
  inventory: NodeView[];
  pools: Map<string, PlacementPool>;
  recoveryContext?: RecoveryContext;
  /**
   * Phase G — the images this job's product template needs, so candidates
   * can be scored by `warmth`. Absent/empty ⇒ warmth is unknown and every
   * node scores 0 for it, which is exactly the behaviour before Phase G.
   */
  jobImages?: ImageRef[];
  /**
   * Phase G — the job's product-template name, for reservation matching
   * (`NodeView.reservedFor`). Absent ⇒ the job matches no reservation, so
   * every exclusive reserve is closed to it.
   */
  templateName?: string;
  /**
   * Phase G — ignore exclusive reservations for this one placement. The
   * operator's explicit "yes, take the reserved spare" on a launch or a
   * migrate. Does not affect the `avoidSpare` penalty, which is a
   * different question.
   */
  overrideReservation?: boolean;
  /**
   * Hard node pin (AutoPlacement.target_node) — "this node or nowhere",
   * the API-only escape hatch for cable-bound hardware. Every other node
   * filters out as wrong-target-node; the target itself still runs the
   * full constraint chain so the real blocker (port conflict, capacity)
   * surfaces in the trace. A pinned job never provisions a fresh node.
   * Label-matched capability requirements are the preferred targeting.
   */
  targetNode?: NodeId;
}

/** @public */
/** Which anti-affinity flag(s) a backup placement couldn't honour. */
export type ResilienceFlag = "sameNode" | "sameAz" | "sameCloud";

/**
 * Set on a backup placement that landed inside a failure domain its
 * resilience policy asked to avoid (best-effort: no compliant node had
 * capacity). The engine places it anyway and surfaces this so the
 * operator knows the bundle's DR posture is currently reduced.
 *
 * @public
 */
export interface ResilienceDegradation {
  violated: ResilienceFlag[];
  /**
   * Set when the primary landed on interruptible (spot) capacity with no
   * durable backup — a single interruption takes the workload down. Surfaced
   * so the operator can add a durable backup.
   */
  interruptiblePrimary?: boolean;
}

/** @public */
export type PlacementResult =
  | {
      kind: "place";
      nodeId: NodeId;
      gpuIndex?: number;
      pool: string;
      tier: string;
      degraded?: ResilienceDegradation;
    }
  | {
      kind: "provision";
      pool: string;
      tier: string;
      instanceType: string;
      launchMode: LaunchMode;
      /**
       * The device (index into the shape's `gpus`, which is how the
       * prospective node view numbers them) the triggering job will take
       * on the new node, when it needs one — so its claim reserves that
       * device from the moment the provision is bound, exactly as a
       * `place` decision's does.
       */
      gpuIndex?: number;
      degraded?: ResilienceDegradation;
    }
  | { kind: "failure"; reason: PlacementFailureReason };

/** @public */
export interface PlacementFailureReason {
  code: "noCapacityInAnyPool" | "noConfiguredPools";
  triedPools: string[];
}

/**
 * Structured trace of a placement decision — one entry per pool
 * attempted, with per-node filter outcomes and resulting scores. Used
 * for "why did my job land here?" debugging.
 *
 * @public
 */
export interface PlacementTrace {
  attempts: PoolAttempt[];
  /** Final decision kind, mirroring PlacementResult.kind. */
  decision: "place" | "provision" | "failure";
}

/** @public */
export interface PoolAttempt {
  pool: string;
  /** Which tier of the pool this attempt covers. */
  tier: string;
  /** Total nodes considered in this pool before any filtering. */
  initialCandidates: number;
  /** Why each non-matching node was filtered out. Indexed by nodeId. */
  filtered: { nodeId: NodeId; reason: FilterReason }[];
  /**
   * Nodes that survived all filters, winner first. `key` is what ranked
   * them (compare component by component, see PlacementKey); `score` is a
   * scalar rendering of the same key for displays that expect one number
   * — it is NOT what decides the order and cannot always separate two
   * keys that the tuple does.
   */
  scored: { nodeId: NodeId; score: number; key: PlacementKey }[];
  /** What this pool attempt produced. */
  outcome:
    | { kind: "placed"; nodeId: NodeId; gpuIndex?: number }
    | { kind: "provisioned"; instanceType: string }
    | { kind: "exhausted" }
    | { kind: "no-instance-type" };
}

/** @public */
export type FilterReason =
  | "wrong-pool"
  | "wrong-tier"
  | "excluded-recovery"
  | "wrong-target-node"
  | "unreachable"
  | "cordoned"
  | "missing-capability"
  | "insufficient-capacity"
  | "insufficient-cores"
  | "port-conflict"
  | "no-fitting-gpu"
  | "node-not-empty"
  | "node-hosts-exclusive"
  | "reserved-for-other-template"
  | "fails-intra-replica"
  | "fails-intra-az";

/**
 * Core placement function. Pure — no side effects, no I/O.
 *
 * @public
 */
export function place(input: PlacementInput): PlacementResult {
  return placeWithTrace(input).result;
}

/**
 * Placement with a structured trace of every pool attempted and every
 * node's filter outcome. AutoManager logs the trace via debuglog when
 * placing; tests use it to assert on the precise reason a candidate
 * was rejected.
 *
 * @public
 */
export function placeWithTrace(
  input: PlacementInput
): { result: PlacementResult; trace: PlacementTrace } {
  const { bundle, job } = input;
  const tried: string[] = [];
  const attempts: PoolAttempt[] = [];
  // Group-aware fit: the first mover of a co-located group must pick (or
  // provision) a node the WHOLE group fits on — see groupFitRequirements.
  const fitReq = groupFitRequirements(input);

  // A backup is placed from its resilience policy's pool when one is set
  // (README: "primary in AWS, backup in OCI"); the primary always from the
  // bundle's own pool.
  const poolName =
    input.replicaIndex > 0 && bundle.resiliencePolicy?.pool !== undefined
      ? bundle.resiliencePolicy.pool
      : bundle.pool;
  const pool = input.pools.get(poolName);
  if (!pool || pool.tiers.length === 0) {
    return {
      result: { kind: "failure", reason: { code: "noConfiguredPools", triedPools: [] } },
      trace: { attempts: [], decision: "failure" },
    };
  }

  // Walk the tiers cheapest-first; first tier that can place or provision
  // wins. This positional order is the cost cascade (no cost weight in v1).
  for (const tier of pool.tiers) {
    tried.push(tier.name);

    const { matched, filtered } = filterCandidatesTraced(input, pool, tier, fitReq);
    const initialCandidates = input.inventory.length;

    if (matched.length > 0) {
      const scored = matched.map((n) => ({
        node: n,
        key: placementKey(n, tier.packingStrategy, input),
      }));
      // Best key first; lexicographic nodeId is the final component.
      scored.sort((a, b) => {
        const c = comparePlacementKeys(a.key, b.key);
        if (c !== 0) return c;
        return a.node.nodeId.localeCompare(b.node.nodeId);
      });
      const winner = scored[0].node;
      const gpuIndex = pickGpuIndex(winner, job.requirements, tier.packingStrategy);
      const degraded = placeDegradation(winner, input) ?? primaryInterruptibleDegradation(tier, input);
      attempts.push({
        pool: pool.name,
        tier: tier.name,
        initialCandidates,
        filtered,
        scored: scored.map((s) => ({ nodeId: s.node.nodeId, score: keyScalar(s.key), key: s.key })),
        outcome: { kind: "placed", nodeId: winner.nodeId, gpuIndex },
      });
      return {
        result: {
          kind: "place",
          nodeId: winner.nodeId,
          gpuIndex,
          pool: pool.name,
          tier: tier.name,
          ...(degraded ? { degraded } : {}),
        },
        trace: { attempts, decision: "place" },
      };
    }

    // A tier at its node cap can't grow — the cost cascade spills to the
    // next tier. (Placing onto an existing node above doesn't add a node,
    // so the cap only gates provisioning.)
    const atCap = tier.maxNodes !== undefined && tierNodeCount(input.inventory, pool, tier) >= tier.maxNodes;

    // A pinned job runs on its target or nowhere — never a fresh node.
    if (tier.scaleOut === "elastic" && !atCap && input.targetNode === undefined) {
      const instanceType = findProvisionableInstanceType(tier, fitReq);
      if (instanceType) {
        const degraded = provisionDegradation(tier, input) ?? primaryInterruptibleDegradation(tier, input);
        const shape = tier.candidateInstanceTypes.find((it) => it.instanceType === instanceType);
        const gpuIndex = shape ? pickShapeGpuIndex(shape, job.requirements) : undefined;
        attempts.push({
          pool: pool.name,
          tier: tier.name,
          initialCandidates,
          filtered,
          scored: [],
          outcome: { kind: "provisioned", instanceType },
        });
        return {
          result: {
            kind: "provision",
            pool: pool.name,
            tier: tier.name,
            instanceType,
            launchMode: tierLaunchMode(tier),
            ...(gpuIndex !== undefined ? { gpuIndex } : {}),
            ...(degraded ? { degraded } : {}),
          },
          trace: { attempts, decision: "provision" },
        };
      }
      attempts.push({
        pool: pool.name,
        tier: tier.name,
        initialCandidates,
        filtered,
        scored: [],
        outcome: { kind: "no-instance-type" },
      });
      continue;
    }
    attempts.push({
      pool: pool.name,
      tier: tier.name,
      initialCandidates,
      filtered,
      scored: [],
      outcome: { kind: "exhausted" },
    });
  }

  return {
    result: { kind: "failure", reason: { code: "noCapacityInAnyPool", triedPools: tried } },
    trace: { attempts, decision: "failure" },
  };
}

// ---------- filter ----------

function filterCandidatesTraced(
  input: PlacementInput,
  pool: PlacementPool,
  tier: PlacementTier,
  fitReq: JobRequirements
): { matched: NodeView[]; filtered: { nodeId: NodeId; reason: FilterReason }[] } {
  const { inventory, recoveryContext } = input;
  const matched: NodeView[] = [];
  const filtered: { nodeId: NodeId; reason: FilterReason }[] = [];

  for (const n of inventory) {
    const reason = firstFilterMiss(n, input, pool, tier, recoveryContext, fitReq);
    if (reason === undefined) {
      matched.push(n);
    } else {
      // Skip "wrong-pool"/"wrong-tier" entries from the trace by default —
      // for a large cross-pool inventory the "this isn't in my pool/tier"
      // entries dominate the trace and aren't useful. Keep all other reasons.
      if (reason !== "wrong-pool" && reason !== "wrong-tier")
        filtered.push({ nodeId: n.nodeId, reason });
    }
  }
  return { matched, filtered };
}

function firstFilterMiss(
  n: NodeView,
  input: PlacementInput,
  pool: PlacementPool,
  tier: PlacementTier,
  recoveryContext: RecoveryContext | undefined,
  fitReq: JobRequirements
): FilterReason | undefined {
  const { bundle, job, replicaIndex, inventory } = input;
  if (input.targetNode !== undefined && n.nodeId !== input.targetNode)
    return "wrong-target-node";
  if (n.poolName !== pool.name) return "wrong-pool";
  if (!nodeMatchesTier(n, pool, tier)) return "wrong-tier";
  if (recoveryContext?.excludeNodes.has(n.nodeId)) return "excluded-recovery";
  if (!n.inventory.reachable) return "unreachable";
  if (n.inventory.cordoned) return "cordoned";
  if (reservedAgainst(n, input)) return "reserved-for-other-template";
  // Sizing checks run against the group-aware fitReq (== job.requirements
  // when the job has no unplaced co-located peers).
  if (!matchesAllCapabilities(n.inventory, fitReq.requiredCapabilities))
    return "missing-capability";
  if (freeCapacity(n.inventory) < fitReq.requiredCapacity)
    return "insufficient-capacity";
  if (freeCores(n.inventory) < (fitReq.requiredCores ?? 0))
    return "insufficient-cores";
  if (hasPortConflict(n.inventory, fitReq.requiredHostPorts))
    return "port-conflict";
  if (!matchesGpu(n.inventory.gpus, job.requirements)) return "no-fitting-gpu";
  {
    const excl = exclusivityMiss(n, bundle, job, replicaIndex);
    if (excl !== undefined) return excl;
  }
  if (!satisfiesIntraReplica(n, bundle, job, replicaIndex, inventory))
    return "fails-intra-replica";
  if (!satisfiesIntraReplicaAz(n, bundle, replicaIndex, inventory))
    return "fails-intra-az";
  // Resilience anti-affinity (backup vs primary) is NOT a hard filter — it
  // is best-effort, applied as a scoring penalty so the backup still
  // places when no compliant node has capacity (then flagged degraded).
  return undefined;
}

/**
 * Phase G — is this node exclusively reserved AGAINST this job?
 *
 * Only an UNCONSUMED reserve is a reserve. A spare that already carries a
 * job has been spent; keeping its reservation would strand the rest of the
 * node's capacity for ever, and the guarantee the operator asked for was
 * "one warm node ready for xyz", not "this machine belongs to xyz until it
 * dies". The same `runningJobs.length === 0` test that decides the
 * avoid-spare penalty decides this, and for the same reason.
 *
 * A job with no template name matches no reservation — an exclusive reserve
 * is closed to it, which is the conservative answer.
 */
/**
 * A spare that is still a reserve: tagged `spare` AND carrying nothing. The
 * daemon drops the tag the moment a job is bound to the node and announces
 * it (`nodeTagsUpdated`), so the tag alone is the truth once that lands;
 * the `runningJobs` check covers the window before it does, and also
 * counts jobs AutoManager has placed but that have yet to appear in any
 * report, so two placements in quick succession cannot both be told the
 * same spare is still pristine. Every consumer used to re-derive "is it
 * still one" for itself, and one of them — the prefer bonus — forgot, so
 * `preferSpare` biased toward a spare already carrying a job as strongly
 * as toward a pristine one (task 90, fix-list 14).
 */
export function isUnconsumedSpare(n: NodeView): boolean {
  return n.isHotSpare && n.runningJobs.length === 0;
}

function reservedAgainst(n: NodeView, input: PlacementInput): boolean {
  if (n.exclusiveReservation !== true) return false;
  if (input.overrideReservation === true) return false;
  if (n.runningJobs.length > 0) return false;
  const t = input.templateName;
  return t === undefined || !(n.reservedFor ?? []).includes(t);
}

/** A fixed host socket is exclusive: the job is refused any node where
 *  one of its declared ports is already claimed. Declared ports only —
 *  the node-side bind check stays authoritative for everything placement
 *  cannot see (workflow-content listeners, host processes). */
function hasPortConflict(
  inv: NodeInventory,
  requiredHostPorts: string[] | undefined
): boolean {
  if (!requiredHostPorts?.length) return false;
  const claimed = reservedHostPortSet(inv);
  return requiredHostPorts.some((p) => claimed.has(p));
}

function matchesAllCapabilities(
  inv: NodeInventory,
  reqs: CapabilityRequirement[]
): boolean {
  // Attribute-aware and multi-entry: a node may advertise several
  // same-named capabilities (one per physical card, distinguished by
  // attributes), and a requirement's free count is computed over exactly
  // the subset its attributes name — see freeCapabilityFor.
  for (const req of reqs) {
    if (freeCapabilityFor(inv, req) < req.count) return false;
  }
  return true;
}

function matchesGpu(gpus: GpuResource[], req: JobRequirements): boolean {
  if (req.requiredGpuCapacity === undefined || req.requiredGpuCapacity <= 0) {
    return true;
  }
  return gpus.some((g) => gpuFits(g, req));
}

function gpuFits(g: GpuResource, req: JobRequirements): boolean {
  if (g.exclusivelyReserved) return false;
  if (req.requiredGpuModel !== undefined && g.model !== req.requiredGpuModel)
    return false;
  const free = g.totalCapacity - g.reservedCapacity;
  return free >= (req.requiredGpuCapacity ?? 0);
}

/**
 * Effective sizing requirements for the placing job. When the job names
 * coLocateWith peers that are NOT yet placed, their capacity / cores /
 * host-ports / capabilities are summed in: the group's first mover must
 * pick (or provision) a node the WHOLE group fits on — sized to itself
 * alone, it could land the group on a node too small for the followers,
 * wedging them (coLocateWith pins them to that node). Already-placed
 * peers contribute nothing; their reservations are in the node view, and
 * the followers chasing them fit-check only themselves.
 *
 * v1 limitation: per-job GPU model/capacity is NOT group-summed — a
 * group whose followers need GPUs the leader doesn't declare can still
 * mis-lead. Declare the GPU on every member (or lead with the GPU job).
 */
function groupFitRequirements(input: PlacementInput): JobRequirements {
  const { bundle, job, replicaIndex, inventory } = input;
  const coLocateWith = effectiveCoLocateWith(bundle, job);
  if (coLocateWith.length === 0) return job.requirements;
  let capacity = job.requirements.requiredCapacity;
  let cores = job.requirements.requiredCores ?? 0;
  const ports = [...(job.requirements.requiredHostPorts ?? [])];
  const capabilities = [...job.requirements.requiredCapabilities];
  let summedAny = false;
  for (const peer of coLocateWith) {
    if (findRunningJob(inventory, bundle.bundleId, replicaIndex, peer)) continue;
    const spec = bundle.jobs.find((j) => j.jobName === peer);
    if (!spec) continue;
    summedAny = true;
    capacity += spec.requirements.requiredCapacity;
    cores += spec.requirements.requiredCores ?? 0;
    ports.push(...(spec.requirements.requiredHostPorts ?? []));
    capabilities.push(...spec.requirements.requiredCapabilities);
  }
  if (!summedAny) return job.requirements;
  return {
    ...job.requirements,
    requiredCapacity: capacity,
    ...(cores > 0 ? { requiredCores: cores } : {}),
    ...(ports.length > 0 ? { requiredHostPorts: ports } : {}),
    requiredCapabilities: mergeCapabilityRequirements(capabilities),
  };
}

/** Merge same-shaped capability requirements by summing counts — two
 *  members each needing one `norsk.io/gpu` fold to a count of 2 so the
 *  node (or machine type) must genuinely carry both. */
function mergeCapabilityRequirements(reqs: CapabilityRequirement[]): CapabilityRequirement[] {
  const out = new Map<string, CapabilityRequirement>();
  for (const r of reqs) {
    const key = `${r.name}\0${JSON.stringify(r.attributeMatches ?? {})}`;
    const existing = out.get(key);
    if (existing) out.set(key, { ...existing, count: existing.count + r.count });
    else out.set(key, { ...r });
  }
  return [...out.values()];
}

/**
 * Exclusive-node allocation, both directions:
 *  - an `exclusive` job only takes a node with no other occupants
 *    ("node-not-empty");
 *  - no job takes a node occupied by an exclusive job
 *    ("node-hosts-exclusive").
 * Same-replica peers named in the placing job's `coLocateWith` are exempt
 * (an exclusive co-located group shares one node), as is the job's own
 * lingering entry (re-placement onto its current node). Prospective
 * (still-booting) nodes flow through the same check, so join-ers are
 * blocked from a node an exclusive job is provisioning.
 */
function exclusivityMiss(
  n: NodeView,
  bundle: Bundle,
  job: BundleJobSpec,
  replicaIndex: number
): FilterReason | undefined {
  const isGroupPeer = (rj: RunningJob): boolean =>
    rj.bundleId === bundle.bundleId &&
    rj.replicaIndex === replicaIndex &&
    (rj.jobName === job.jobName || effectiveCoLocateWith(bundle, job).includes(rj.jobName));
  const occupants = n.runningJobs.filter((rj) => !isGroupPeer(rj));
  if (occupants.length === 0) return undefined;
  if (job.exclusive) return "node-not-empty";
  if (occupants.some((rj) => rj.exclusive)) return "node-hosts-exclusive";
  return undefined;
}

/** Whether the job carries an explicit intra-replica rule of its own. */
function hasExplicitIntraReplicaRule(job: BundleJobSpec): boolean {
  return (job.coLocateWith?.length ?? 0) > 0 || (job.separateFrom?.length ?? 0) > 0;
}

/**
 * The jobs this one must share a node with, in its replica. Explicit
 * `coLocateWith` when the job declares any rule of its own; otherwise, when
 * the bundle's `intraReplicaPlacement.sameNodeDefault` is true, every other
 * job of the bundle that declares no rule either — the "default group".
 * The rule is as hard as an explicit coLocateWith (filter, group-fit
 * sizing, exclusivity exemption all read this). Absent/false ⇒ nothing
 * implicit, which is how every bundle behaved before the field was read.
 */
function effectiveCoLocateWith(bundle: Bundle, job: BundleJobSpec): string[] {
  if (hasExplicitIntraReplicaRule(job)) return job.coLocateWith ?? [];
  if (bundle.intraReplicaPlacement?.sameNodeDefault !== true) return [];
  return bundle.jobs
    .filter((j) => j.jobName !== job.jobName && !hasExplicitIntraReplicaRule(j))
    .map((j) => j.jobName);
}

function satisfiesIntraReplica(
  n: NodeView,
  bundle: Bundle,
  job: BundleJobSpec,
  replicaIndex: number,
  inventory: NodeView[]
): boolean {
  // coLocateWith (explicit, or the sameNodeDefault group): every named job
  // in this same replica must be running on this node.
  for (const peer of effectiveCoLocateWith(bundle, job)) {
    const peerNode = findRunningJob(inventory, bundle.bundleId, replicaIndex, peer);
    // If peer not yet placed, no constraint to check yet.
    if (peerNode && peerNode.nodeId !== n.nodeId) return false;
  }
  // separateFrom: every named peer must be on a different node.
  if (job.separateFrom && job.separateFrom.length > 0) {
    for (const peer of job.separateFrom) {
      const peerNode = findRunningJob(inventory, bundle.bundleId, replicaIndex, peer);
      if (peerNode && peerNode.nodeId === n.nodeId) return false;
    }
  }
  return true;
}

function satisfiesIntraReplicaAz(
  n: NodeView,
  bundle: Bundle,
  replicaIndex: number,
  inventory: NodeView[]
): boolean {
  const sameAz = bundle.intraReplicaPlacement?.sameAz ?? "soft";
  if (sameAz !== "hard") return true;
  const peerAzs = peerAzsInSameReplica(inventory, bundle.bundleId, replicaIndex);
  if (peerAzs.length === 0) return true;
  // Hard sameAz with peers in known AZs: this node's AZ must match one of theirs.
  return n.az !== undefined && peerAzs.includes(n.az);
}

function peerAzsInSameReplica(
  inventory: NodeView[],
  bundleId: BundleId,
  replicaIndex: number
): string[] {
  const azs = new Set<string>();
  for (const n of inventory) {
    if (
      n.az !== undefined &&
      n.runningJobs.some(
        (rj) => rj.bundleId === bundleId && rj.replicaIndex === replicaIndex
      )
    ) {
      azs.add(n.az);
    }
  }
  return [...azs];
}

// ---------- resilience anti-affinity (backup vs primary) ----------

/** True when this placement is a backup whose policy carries anti-affinity. */
function backupPolicy(input: PlacementInput): ResiliencePolicy | undefined {
  return input.replicaIndex > 0 ? input.bundle.resiliencePolicy : undefined;
}

interface PrimaryDomains {
  nodeIds: Set<NodeId>;
  azs: Set<string>;
  clouds: Set<string>;
}

/** Failure domains the primary (replica 0) currently occupies. */
function primaryDomains(inventory: NodeView[], bundleId: BundleId): PrimaryDomains {
  const nodeIds = new Set<NodeId>();
  const azs = new Set<string>();
  const clouds = new Set<string>();
  for (const n of inventory) {
    if (n.runningJobs.some((rj) => rj.bundleId === bundleId && rj.replicaIndex === 0)) {
      nodeIds.add(n.nodeId);
      if (n.az !== undefined) azs.add(n.az);
      if (n.cloud !== undefined) clouds.add(n.cloud);
    }
  }
  return { nodeIds, azs, clouds };
}

/** Which `forbid` flags placing the backup on `n` would violate. */
function resilienceViolations(
  n: NodeView,
  policy: ResiliencePolicy,
  primary: PrimaryDomains
): ResilienceFlag[] {
  const violated: ResilienceFlag[] = [];
  if (policy.sameNode === "forbid" && primary.nodeIds.has(n.nodeId)) violated.push("sameNode");
  if (policy.sameAz === "forbid" && n.az !== undefined && primary.azs.has(n.az)) violated.push("sameAz");
  if (policy.sameCloud === "forbid" && n.cloud !== undefined && primary.clouds.has(n.cloud))
    violated.push("sameCloud");
  return violated;
}

/** Degradation for placing a backup on an existing node `n`, if any. */
function placeDegradation(n: NodeView, input: PlacementInput): ResilienceDegradation | undefined {
  const policy = backupPolicy(input);
  if (!policy) return undefined;
  const violated = resilienceViolations(n, policy, primaryDomains(input.inventory, input.bundle.bundleId));
  return violated.length > 0 ? { violated } : undefined;
}

/**
 * Degradation for provisioning a backup into `tier`. Only the cloud domain
 * is knowable pre-boot (it is the tier's provider); node/AZ are decided
 * once the instance starts, so they aren't evaluated here.
 */
function provisionDegradation(tier: PlacementTier, input: PlacementInput): ResilienceDegradation | undefined {
  const policy = backupPolicy(input);
  if (!policy || policy.sameCloud !== "forbid") return undefined;
  const primary = primaryDomains(input.inventory, input.bundle.bundleId);
  return primary.clouds.has(tier.kind) ? { violated: ["sameCloud"] } : undefined;
}

/** Degradation for placing the *primary* (replica 0) onto interruptible
 *  capacity with no durable backup configured — one interruption and the
 *  workload is gone. (Backups have their own anti-affinity degradation.) */
function primaryInterruptibleDegradation(
  tier: PlacementTier,
  input: PlacementInput
): ResilienceDegradation | undefined {
  if (input.replicaIndex !== 0 || input.bundle.resiliencePolicy) return undefined;
  if (tierReliability(tier) !== "interruptible") return undefined;
  return { violated: [], interruptiblePrimary: true };
}

/** Count of nodes currently in (pool, tier) — used for the tier node cap. */
function tierNodeCount(inventory: NodeView[], pool: PlacementPool, tier: PlacementTier): number {
  let n = 0;
  for (const node of inventory) if (nodeMatchesTier(node, pool, tier)) n++;
  return n;
}

function findRunningJob(
  inventory: NodeView[],
  bundleId: BundleId,
  replicaIndex: number,
  jobName: string
): NodeView | undefined {
  return inventory.find((n) =>
    n.runningJobs.some(
      (rj) =>
        rj.bundleId === bundleId &&
        rj.replicaIndex === replicaIndex &&
        rj.jobName === jobName
    )
  );
}

// ---------- rank ----------

/**
 * The ranking key of a candidate node. Compared lexicographically by
 * `comparePlacementKeys`, first component most significant:
 *
 *  1. `resilienceViolations` — fewer first. A backup lands outside the
 *     primary's `forbid` domains whenever any node there fits; best-effort,
 *     so a violating node still places (flagged degraded) when none does.
 *  2. `prospective` — a node that has booted before one that has not. A
 *     still-launching node is preferable only to provisioning another.
 *  3. `avoidedSpare` — set when the policy in force says avoidSpare and the
 *     node is an unconsumed spare: the reserve exists for the workloads that
 *     asked for it, and "who the node is for" outranks how fast it starts.
 *     Recovery reads the job's `onRuntimeFailure`; a first launch reads the
 *     bundle's `onLaunch` (default avoidSpare; `allowSpare` merely lifts
 *     this, it adds no preference). Soft by construction: ranking only
 *     orders nodes that MATCHED, and provisioning runs only when nothing
 *     did, so avoiding a spare can never cause a provision — deliberately,
 *     because a job waiting minutes for a fresh node beside an idle spare is
 *     the worse failure of the two.
 *  4. `warmth` — higher first (Phase G: a job launched in 7 s on a node that
 *     held its images and 64–118 s on one that did not).
 *  5. `reservationMatch` — a node reserved for this job's template, while
 *     it is still a reserve (carrying nothing). A spent reserve is an
 *     ordinary node here, exactly as it is for the exclusive filter.
 *  6. `preferredSpare` — set when the policy says preferSpare and the node
 *     is an unconsumed spare: what is left of "prefer the spare" once
 *     warmth has spoken (fix-list 9, 14).
 *  7. `packing` — higher first: the used fraction under binpack, its
 *     negation under spread.
 *  8. `sameAz` — the node shares an AZ with a same-replica peer (soft
 *     intraReplica sameAz; `hard` is a filter, `off` never sets this).
 *  9. nodeId — applied by the caller for determinism.
 *
 * @public
 */
export interface PlacementKey {
  resilienceViolations: number;
  prospective: boolean;
  avoidedSpare: boolean;
  warmth: number;
  reservationMatch: boolean;
  preferredSpare: boolean;
  packing: number;
  sameAz: boolean;
}

/** Negative when `a` ranks before `b`; 0 when the tuples are equal. @public */
export function comparePlacementKeys(a: PlacementKey, b: PlacementKey): number {
  if (a.resilienceViolations !== b.resilienceViolations)
    return a.resilienceViolations - b.resilienceViolations;
  if (a.prospective !== b.prospective) return a.prospective ? 1 : -1;
  if (a.avoidedSpare !== b.avoidedSpare) return a.avoidedSpare ? 1 : -1;
  if (a.warmth !== b.warmth) return b.warmth - a.warmth;
  if (a.reservationMatch !== b.reservationMatch) return a.reservationMatch ? -1 : 1;
  if (a.preferredSpare !== b.preferredSpare) return a.preferredSpare ? -1 : 1;
  if (a.packing !== b.packing) return b.packing - a.packing;
  if (a.sameAz !== b.sameAz) return a.sameAz ? -1 : 1;
  return 0;
}

function placementKey(
  n: NodeView,
  strategy: "binpack" | "spread",
  input: PlacementInput
): PlacementKey {
  const { bundle, replicaIndex, inventory, recoveryContext } = input;

  const policy = backupPolicy(input);
  const violations = policy
    ? resilienceViolations(n, policy, primaryDomains(inventory, bundle.bundleId)).length
    : 0;

  // Two different moments answer the spare question separately:
  //   - recovering  → the job's own onRuntimeFailure, via recoveryContext
  //   - first launch → the bundle's onLaunch, which defaults to avoiding
  // Only an UNCONSUMED spare is reserve (see isUnconsumedSpare).
  const unconsumedSpare = isUnconsumedSpare(n);
  const recovering = recoveryContext !== undefined;
  const failurePolicy = recoveryContext?.onRuntimeFailure;
  const avoiding = recovering
    ? failurePolicy === "avoidSpare"
    : (bundle.onLaunch ?? "avoidSpare") === "avoidSpare";
  const preferring = recovering && failurePolicy === "preferSpare";

  const reservationMatch =
    n.runningJobs.length === 0 &&
    input.templateName !== undefined &&
    (n.reservedFor ?? []).includes(input.templateName);

  const used = usedCapacityFraction(n.inventory);

  const intraAz = bundle.intraReplicaPlacement?.sameAz ?? "soft";
  const sameAz =
    intraAz === "soft" &&
    n.az !== undefined &&
    peerAzsInSameReplica(inventory, bundle.bundleId, replicaIndex).includes(n.az);

  return {
    resilienceViolations: violations,
    prospective: n.prospective === true,
    avoidedSpare: avoiding && unconsumedSpare,
    warmth: warmth(n, input.jobImages ?? []),
    reservationMatch,
    preferredSpare: preferring && unconsumedSpare,
    packing: strategy === "binpack" ? used : -used,
    sameAz,
  };
}

/**
 * A scalar rendering of the key for `PoolAttempt.scored[].score`, for
 * displays that show one number per node. Component weights are spaced so
 * the scalar agrees with the tuple order except where two keys differ only
 * by a sliver of warmth — the tuple, not this, decides.
 */
function keyScalar(k: PlacementKey): number {
  return (
    -1000 * k.resilienceViolations -
    100 * (k.prospective ? 1 : 0) -
    50 * (k.avoidedSpare ? 1 : 0) +
    10 * k.warmth +
    4 * (k.reservationMatch ? 1 : 0) +
    2 * (k.preferredSpare ? 1 : 0) +
    k.packing +
    0.1 * (k.sameAz ? 1 : 0)
  );
}

// ---------- GPU pick ----------

function pickGpuIndex(
  n: NodeView,
  req: JobRequirements,
  strategy: "binpack" | "spread"
): number | undefined {
  if (req.requiredGpuCapacity === undefined || req.requiredGpuCapacity <= 0) {
    return undefined;
  }
  const eligible = n.inventory.gpus.filter((g) => gpuFits(g, req));
  if (eligible.length === 0) return undefined;
  // Pack: pick the GPU with least free capacity that still fits. Spread:
  // pick the most free. Same packing intent as node-level scoring.
  const sorted = eligible.slice().sort((a, b) => {
    const freeA = a.totalCapacity - a.reservedCapacity;
    const freeB = b.totalCapacity - b.reservedCapacity;
    if (freeA !== freeB) {
      return strategy === "binpack" ? freeA - freeB : freeB - freeA;
    }
    return a.index - b.index;
  });
  return sorted[0].index;
}

/** The device a job takes on a node provisioned from `it`: the first of
 *  the shape's GPUs that fits, numbered as the prospective node view
 *  numbers them (array position). Every device is empty on a new node, so
 *  packing strategy does not enter into it. */
function pickShapeGpuIndex(it: InstanceTypeOption, req: JobRequirements): number | undefined {
  if (req.requiredGpuCapacity === undefined || req.requiredGpuCapacity <= 0) return undefined;
  const index = (it.gpus ?? []).findIndex(
    (g) =>
      (req.requiredGpuModel === undefined || g.model === req.requiredGpuModel) &&
      g.totalCapacity >= (req.requiredGpuCapacity ?? 0)
  );
  return index >= 0 ? index : undefined;
}

// ---------- elastic provision ----------

function findProvisionableInstanceType(
  tier: PlacementTier,
  req: JobRequirements
): string | undefined {
  for (const it of tier.candidateInstanceTypes) {
    if (instanceTypeSatisfies(it, req)) return it.instanceType;
  }
  return undefined;
}

function instanceTypeSatisfies(
  it: InstanceTypeOption,
  req: JobRequirements
): boolean {
  if (it.totalCapacity < req.requiredCapacity) return false;
  if (it.totalCores < (req.requiredCores ?? 0)) return false;
  for (const cap of req.requiredCapabilities) {
    const found = it.capabilities.find((c) => c.name === cap.name);
    if (!found || found.count < cap.count) return false;
    if (cap.attributeMatches) {
      for (const [k, v] of Object.entries(cap.attributeMatches)) {
        if (found.attributes?.[k] !== v) return false;
      }
    }
  }
  if (req.requiredGpuCapacity !== undefined && req.requiredGpuCapacity > 0) {
    const gpus = it.gpus ?? [];
    const fits = gpus.some(
      (g) =>
        (req.requiredGpuModel === undefined || g.model === req.requiredGpuModel) &&
        g.totalCapacity >= (req.requiredGpuCapacity ?? 0)
    );
    if (!fits) return false;
  }
  return true;
}