import { readFileSync } from "node:fs";
import { homedir } from "node:os";
import { z } from "zod";
import type { Adapter, FetchLike } from "../adapter";
import { CollectorFatalError } from "../errors";
import type { ActionRef, AdapterResult, Item, Panel, Severity } from "../schema";
import {
  loadBuildboxRegistry,
  RegistryUnavailableError,
  type BuildboxRegistry,
} from "../buildbox-registry";

export type FetchFn = FetchLike;

/** Per-host controller state (R2). */
export const HostStateSchema = z.enum([
  "available",
  "draining",
  "maintenance",
  "restoring",
  "degraded",
]);
export type HostState = z.infer<typeof HostStateSchema>;

const CapabilityProbeSchema = z
  .object({
    name: z.string(),
    ok: z.boolean(),
    version: z.string().optional(),
    detail: z.string().optional(),
    checkedAt: z.string().optional(),
  })
  .strict();

const HostCapabilitySchema = z
  .object({
    probes: z.array(CapabilityProbeSchema),
    circuitOpen: z.boolean().optional(),
    missingCommand: z.string().optional(),
  })
  .strict();

const HostStatusSchema = z
  .object({
    state: HostStateSchema,
    role: z.enum(["builder", "workstation"]),
    primary: z.boolean().optional(),
    enrolling: z.boolean().optional(),
    capability: HostCapabilitySchema.optional(),
  })
  .strict();

const QueueOwnerSchema = z
  .object({
    pid: z.number().int(),
    starttime: z.number().int(),
    label: z.string().optional(),
  })
  .strict();

const QueueTicketSchema = z
  .object({
    position: z.number().int(),
    key: z.string(),
    repo: z.string(),
    owner: QueueOwnerSchema,
    enqueueAgeSeconds: z.number(),
    state: z.string(),
    dispatchTarget: z.string().optional(),
  })
  .strict();

const QueueStatusSchema = z
  .object({
    depth: z.number().int(),
    oldestAgeSeconds: z.number(),
    p95AgeSeconds: z.number(),
    sloSeconds: z.number(),
    dispatchTarget: z.string().optional(),
    spill: z
      .object({
        active: z.boolean(),
        stale: z.boolean().optional(),
      })
      .strict()
      .optional(),
    tickets: z.array(QueueTicketSchema),
    wedge: z
      .object({
        host: z.string(),
        detail: z.string(),
      })
      .strict()
      .optional(),
  })
  .strict();

const PublicationStateSchema = z
  .object({
    state: z.enum(["none", "staged", "promoted", "blocked", "discarded"]),
    reason: z.string().optional(),
  })
  .strict();

const RemoteJobSchema = z
  .object({
    id: z.string(),
    repo: z.string(),
    snapshot: z.string(),
    stage: z.string(),
    host: z.string(),
    rc: z.number().int().nullable().optional(),
    pullBytes: z.number().int().nullable().optional(),
    pullDurationSeconds: z.number().nullable().optional(),
    publication: PublicationStateSchema.optional(),
  })
  .strict();

const OffloadKpiSamplesSchema = z
  .object({
    remote24h: z.number().int().nonnegative(),
    remoteSuccessPct: z.number().min(0).max(100).optional(),
    exit127: z
      .object({
        count: z.number().int().nonnegative(),
        hosts: z.array(z.string()),
      })
      .strict(),
  })
  .strict();

/** Closed schema matching `controller/src/status.ts`. */
export const ControllerStatusSchema = z
  .object({
    desired: HostStateSchema,
    observed: HostStateSchema,
    revision: z.number().int(),
    lease: z
      .object({
        active: z.boolean(),
        expired: z.boolean().optional(),
        host: z.string().nullable().optional(),
        expiresAt: z.string().nullable().optional(),
        reason: z.string().nullable().optional(),
      })
      .strict(),
    capacity: z
      .object({
        builders: z.number().int(),
        idleSlots: z.number().int(),
        overloaded: z.boolean().optional(),
      })
      .strict(),
    reconciler: z
      .object({
        healthy: z.boolean(),
        lastAt: z.string().optional(),
      })
      .strict(),
    dispatch: z
      .object({
        state: z.enum(["healthy", "wedged", "paused"]),
        detail: z.string().optional(),
        host: z.string().optional(),
      })
      .strict(),
    hosts: z.record(HostStatusSchema),
    landConduct: z
      .record(
        z
          .object({
            lastPassAt: z.string().nullable(),
            lastOk: z.boolean(),
            lastDetail: z.string(),
            consecutiveFailures: z.number().int().nonnegative(),
          })
          .strict(),
      )
      .optional(),
    deployWatcher: z
      .object({
        targetSha: z.string(),
        attempts: z.number().int().nonnegative(),
        lastStatus: z.string(),
        lastDetail: z.string(),
        lastAt: z.string(),
        lastOk: z.boolean(),
        failureClass: z.enum(["none", "transient", "permanent"]),
        nextRetryAt: z.string().datetime().nullable(),
      })
      .strict()
      .nullable()
      .optional(),
    queue: QueueStatusSchema,
    jobs: z.array(RemoteJobSchema),
    kpis: OffloadKpiSamplesSchema.optional(),
  })
  // Every field the deck reads is still validated exactly, nested objects included. Only
  // the top level tolerates keys it does not know: the controller gaining a field is an
  // additive change, and rejecting the whole response for one made the entire fleet
  // disappear from the deck until someone noticed (landConduct, deployWatcher).
  .passthrough();
export type ControllerStatus = z.infer<typeof ControllerStatusSchema>;

const PromVectorSampleSchema = z
  .object({
    metric: z.record(z.string()),
    value: z.tuple([z.number(), z.string()]),
  })
  .strict();

const PromVectorDataSchema = z
  .object({
    resultType: z.literal("vector"),
    result: z.array(PromVectorSampleSchema),
  })
  .strict();

const PromVectorResultSchema = z
  .object({
    status: z.literal("success"),
    data: PromVectorDataSchema,
  })
  .strict();
type PromVectorData = z.infer<typeof PromVectorDataSchema>;

/**
 * Written by buildbox-parity.sh after every convergence pass. A host absent from it has
 * no recorded verdict — the deck says so rather than showing the fleet as converged.
 */
const ParityStateSchema = z.object({
  schemaVersion: z.literal(1),
  hosts: z.record(
    z.object({
      verdict: z.enum(["converged", "drifted", "unreachable", "error"]),
      detail: z.string(),
      probedAt: z.string(),
    }),
  ),
});
export type HostParity = z.infer<typeof ParityStateSchema>["hosts"][string];

export const DEFAULT_PARITY_STATE_PATH = "~/.claude/run/buildbox-parity/state.json";

/**
 * Null for a machine where the timer has not run yet or wrote something this collector
 * cannot read: unknown parity is never worth failing the whole offload poll over.
 */
export function readParityState(
  path: string,
  readFile: (path: string) => string,
): Record<string, HostParity> | null {
  try {
    const raw = JSON.parse(readFile(path.replace(/^~(?=\/)/, homedir())));
    const parsed = ParityStateSchema.safeParse(raw);
    return parsed.success ? parsed.data.hosts : null;
  } catch {
    return null;
  }
}

export interface OffloadAdapterOptions {
  id?: string;
  interval?: number;
  fetchImpl: FetchFn;
  /** Build controller base URL (GET /status). */
  controllerUrl?: string;
  /** Overridable only so tests can drive the unreadable-registry branch. */
  loadRegistry?: () => Promise<BuildboxRegistry>;
  /** Prometheus query API base URL (GET /api/v1/query). */
  metricsUrl?: string;
  token: string;
  now?: () => number;
  /** Percent of the last 5 min with every task stalled on memory that counts as a stalled box. */
  stallPercent?: number;
  /** Percent of a box's /tmp that counts as full — at 100 its writes are already failing. */
  tmpFullPercent?: number;
  /** buildbox-parity.sh state file; `~` is expanded. */
  parityStatePath?: string;
  readFileImpl?: (path: string) => string;
}

export function requireOffloadToken(token: string): string {
  const trimmed = token.trim();
  if (!trimmed) {
    throw new CollectorFatalError(
      "TOKEN_EMPTY",
      "offload adapter requires a non-empty collector token",
    );
  }
  return trimmed;
}

/** Live work units a box observes on itself, split by the path that placed them. */
export interface RemoteWorkCounts {
  agentSeats: number;
  remoteBuildJobs: number;
  offloadShells: number;
}

const REMOTE_WORK_FIELD_BY_KIND: Record<string, keyof RemoteWorkCounts> = {
  agent_seat: "agentSeats",
  remote_build_job: "remoteBuildJobs",
  offload_shell: "offloadShells",
};

/**
 * Every kind must be present: a partial series means the box reported only some paths, and a
 * total summed from an incomplete split would understate the work without saying so.
 */
function remoteWorkFrom(byKind: Map<string, number> | undefined): RemoteWorkCounts | undefined {
  if (!byKind) return undefined;
  const counts: Partial<RemoteWorkCounts> = {};
  for (const [kind, field] of Object.entries(REMOTE_WORK_FIELD_BY_KIND)) {
    const value = byKind.get(kind);
    if (value === undefined) return undefined;
    counts[field] = value;
  }
  return counts as RemoteWorkCounts;
}

interface HostFleetMetrics {
  sessions?: number;
  remoteWork?: RemoteWorkCounts;
  load?: number;
  cores?: number;
  coreLoads?: number[];
  running?: number;
  slotsFree?: number;
  slotsTotal?: number;
  builds24h?: number;
  dispatchAccept?: boolean;
  memUsedBytes?: number;
  memTotalBytes?: number;
  swapUsedBytes?: number;
  swapTotalBytes?: number;
  netRxBytesPerSecond?: number;
  netTxBytesPerSecond?: number;
  disk?: Array<{ mountpoint: string; freeBytes: number; sizeBytes: number }>;
  tempPkg?: number;
  tempMax?: number;
  tempCrit?: number;
  guard?: HostGuardMetrics;
}

// Box-side mirror of the laptop mem-guard and tmpjail signals. Kernel-measured facts only:
// a stall fraction, tmpfs fill, and per-slice memory/pids/OOM-kill counters.
interface HostGuardMetrics {
  stall: { some60: number | null; full60: number | null; full300: number | null };
  tmpUsedBytes: number | null;
  tmpSizeBytes: number | null;
  workSlices: Array<{
    slice: string;
    memoryBytes: number | null;
    pidsCurrent: number | null;
    oomKillTotal: number | null;
  }>;
}

interface CapabilityExitCount {
  host: string;
  command: string;
  code: string;
  count: number;
}

interface OffloadMetrics {
  queueDepth?: number;
  queueOldestAgeSeconds?: number;
  queueP95AgeSeconds?: number;
  fallbackLeaseExpired?: boolean;
  artifactCasMismatchTotal?: number;
  hostFleet: Record<string, HostFleetMetrics>;
  capabilityExits: CapabilityExitCount[];
}

const DEFAULT_INTERVAL_MS = 30_000;
const DEFAULT_CONTROLLER_URL = "http://127.0.0.1:8787";
const DEFAULT_METRICS_URL = "http://127.0.0.1:8787";
const DEFAULT_STALL_PERCENT = 20;
const DEFAULT_TMP_FULL_PERCENT = 100;

const METRIC_QUERIES = {
  queueDepth: "build_offload_queue_depth",
  queueOldestAge: "build_offload_queue_oldest_age_seconds",
  queueP95Age: "build_offload_queue_age_p95_seconds",
  fallbackLeaseExpired: "build_offload_fallback_lease_expired",
  artifactCasMismatch: "build_offload_artifact_cas_mismatch_total",
  hostRunning: 'build_offload_host_running_jobs{host!=""}',
  hostSlotsFree: 'build_offload_host_slots_free{host!=""}',
  hostSlotsTotal: 'build_offload_host_slots_total{host!=""}',
  hostDispatchAccept: 'build_offload_host_dispatch_accept{host!=""}',
  hostLoad: 'build_offload_host_load{host!=""}',
  hostCores: 'build_offload_host_cores{host!=""}',
  hostBuilds24h: 'build_offload_host_builds_24h{host!=""}',
  hostSessions: 'build_offload_host_sessions{host!=""}',
  hostRemoteWork: 'build_offload_host_remote_work{host!="",kind!=""}',
  hostMemUsed: 'build_offload_host_mem_used_bytes{host!=""}',
  hostMemTotal: 'build_offload_host_mem_total_bytes{host!=""}',
  hostSwapUsed: 'build_offload_host_swap_used_bytes{host!=""}',
  hostSwapTotal: 'build_offload_host_swap_total_bytes{host!=""}',
  hostNetRx: 'build_offload_host_net_rx_bytes_per_second{host!=""}',
  hostNetTx: 'build_offload_host_net_tx_bytes_per_second{host!=""}',
  hostDiskFree: 'build_offload_host_disk_free_bytes{host!=""}',
  hostDiskSize: 'build_offload_host_disk_size_bytes{host!=""}',
  hostCoreLoad: 'build_offload_host_core_load_percent{host!=""}',
  hostTempPkg: 'build_offload_host_cpu_temp_celsius{host!="",sensor="pkg"}',
  hostTempMax: 'build_offload_host_cpu_temp_celsius{host!="",sensor="max"}',
  hostTempCrit: 'build_offload_host_cpu_temp_celsius{host!="",sensor="crit"}',
  capabilityExit: 'build_offload_exit_total{host!="",command!="",code=~"126|127"}',
  hostMemoryStall: 'build_offload_host_memory_stall_percent{host!="",window!=""}',
  hostTmpUsed: 'build_offload_host_tmp_used_bytes{host!=""}',
  hostTmpSize: 'build_offload_host_tmp_size_bytes{host!=""}',
  hostSliceMemory: 'build_offload_host_work_slice_memory_bytes{host!="",slice!=""}',
  hostSliceOomKills: 'build_offload_host_work_slice_oom_kills_total{host!="",slice!=""}',
  hostSlicePids: 'build_offload_host_work_slice_pids{host!="",slice!=""}',
} as const;

/** Raised when the controller fetch fails at the transport layer (connection refused,
 * DNS failure, timeout) — the controller is unreachable. A resolved-but-error response
 * (non-2xx or malformed body) is reachable-but-bad and stays a plain Error so the poll
 * throws to retain prior state. Structural, so it holds across runtimes (Bun and Node). */
class OffloadControllerUnreachableError extends Error {
  constructor(url: string, cause: unknown) {
    super(`offload controller unreachable: ${url}`, { cause });
    this.name = "OffloadControllerUnreachableError";
  }
}

function isControllerUnreachable(error: unknown): boolean {
  return error instanceof OffloadControllerUnreachableError;
}

async function fetchOrUnreachable(fetchImpl: FetchFn, url: string, token: string) {
  try {
    return await fetchImpl(url, {
      headers: { authorization: `Bearer ${token}` },
    });
  } catch (cause) {
    throw new OffloadControllerUnreachableError(url, cause);
  }
}

async function requestControllerStatus(
  fetchImpl: FetchFn,
  url: string,
  token: string,
): Promise<ControllerStatus> {
  const res = await fetchOrUnreachable(fetchImpl, url, token);
  if (!res.ok) {
    throw new Error(`offload request failed (${res.status}): ${url}`);
  }
  let body: unknown;
  try {
    body = await res.json();
  } catch (cause) {
    throw new Error(`offload status response is not valid JSON: ${url}`, { cause });
  }
  return ControllerStatusSchema.parse(body);
}

async function runMetricQuery(
  fetchImpl: FetchFn,
  metricsUrl: string,
  expr: string,
  token: string,
): Promise<PromVectorData> {
  const url = `${metricsUrl}/api/v1/query?query=${encodeURIComponent(expr)}`;
  const res = await fetchImpl(url, {
    headers: { authorization: `Bearer ${token}` },
  });
  if (!res.ok) {
    throw new Error(`offload metrics query failed (${res.status}): ${expr}`);
  }
  let body: unknown;
  try {
    body = await res.json();
  } catch (cause) {
    throw new Error(`offload metrics query response is not valid JSON: ${expr}`, { cause });
  }
  return PromVectorResultSchema.parse(body).data;
}

/** A missing session series is a coverage gap, not evidence that a host has zero sessions. */
async function runOptionalMetricQuery(
  fetchImpl: FetchFn,
  metricsUrl: string,
  expr: string,
  token: string,
): Promise<PromVectorData | undefined> {
  try {
    return await runMetricQuery(fetchImpl, metricsUrl, expr, token);
  } catch {
    return undefined;
  }
}

function scalarFromResult(data: PromVectorData | undefined): number | undefined {
  const sample = data?.result[0];
  if (!sample) return undefined;
  return Number(sample.value[1]);
}

function boolFromScalar(value: number | undefined): boolean {
  return value !== undefined && value > 0;
}

function labeledSeries(data: PromVectorData | undefined, label: string): Map<string, number> {
  const out = new Map<string, number>();
  for (const sample of data?.result ?? []) {
    const key = sample.metric[label];
    if (!key) continue;
    out.set(key, Number(sample.value[1]));
  }
  return out;
}

function labeledSeriesWithExtra(
  data: PromVectorData | undefined,
  keys: string[],
): Array<{ labels: Record<string, string>; value: number }> {
  return (data?.result ?? []).map((sample) => {
    const row = { labels: {} as Record<string, string>, value: Number(sample.value[1]) };
    for (const key of keys) {
      const v = sample.metric[key];
      if (v !== undefined) row.labels[key] = v;
    }
    return row;
  });
}

/** host -> (inner label value -> sample), for the two-label host guard series. */
function nestedSeries(
  data: PromVectorData | undefined,
  innerLabel: string,
): Map<string, Map<string, number>> {
  const out = new Map<string, Map<string, number>>();
  for (const sample of data?.result ?? []) {
    const host = sample.metric.host;
    const inner = sample.metric[innerLabel];
    if (!host || !inner) continue;
    let byInner = out.get(host);
    if (!byInner) {
      byInner = new Map();
      out.set(host, byInner);
    }
    byInner.set(inner, Number(sample.value[1]));
  }
  return out;
}

function buildHostGuard(
  stall: Map<string, number> | undefined,
  tmpUsedBytes: number | undefined,
  tmpSizeBytes: number | undefined,
  sliceMemory: Map<string, number> | undefined,
  sliceOom: Map<string, number> | undefined,
  slicePids: Map<string, number> | undefined,
): HostGuardMetrics {
  const sliceNames = [
    ...new Set([
      ...(sliceMemory?.keys() ?? []),
      ...(sliceOom?.keys() ?? []),
      ...(slicePids?.keys() ?? []),
    ]),
  ].sort();
  return {
    stall: {
      some60: stall?.get("some60") ?? null,
      full60: stall?.get("full60") ?? null,
      full300: stall?.get("full300") ?? null,
    },
    tmpUsedBytes: tmpUsedBytes ?? null,
    tmpSizeBytes: tmpSizeBytes ?? null,
    workSlices: sliceNames.map((slice) => ({
      slice,
      memoryBytes: sliceMemory?.get(slice) ?? null,
      pidsCurrent: slicePids?.get(slice) ?? null,
      oomKillTotal: sliceOom?.get(slice) ?? null,
    })),
  };
}

async function fetchMetrics(
  fetchImpl: FetchFn,
  metricsUrl: string,
  token: string,
): Promise<OffloadMetrics> {
  const [
    queueDepthData,
    queueOldestData,
    queueP95Data,
    fallbackLeaseData,
    artifactCasData,
    hostRunningData,
    hostSlotsFreeData,
    hostSlotsTotalData,
    hostDispatchData,
    hostLoadData,
    hostCoresData,
    hostBuilds24hData,
    hostSessionsData,
    hostRemoteWorkData,
    hostMemUsedData,
    hostMemTotalData,
    hostSwapUsedData,
    hostSwapTotalData,
    hostNetRxData,
    hostNetTxData,
    hostDiskFreeData,
    hostDiskSizeData,
    hostCoreLoadData,
    hostTempPkgData,
    hostTempMaxData,
    hostTempCritData,
    capabilityExitData,
    hostMemoryStallData,
    hostTmpUsedData,
    hostTmpSizeData,
    hostSliceMemoryData,
    hostSliceOomKillsData,
    hostSlicePidsData,
  ] = await Promise.all([
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.queueDepth, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.queueOldestAge, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.queueP95Age, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.fallbackLeaseExpired, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.artifactCasMismatch, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostRunning, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSlotsFree, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSlotsTotal, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostDispatchAccept, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostLoad, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostCores, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostBuilds24h, token),
    runOptionalMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSessions, token),
    runOptionalMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostRemoteWork, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostMemUsed, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostMemTotal, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSwapUsed, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSwapTotal, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostNetRx, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostNetTx, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostDiskFree, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostDiskSize, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostCoreLoad, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostTempPkg, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostTempMax, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostTempCrit, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.capabilityExit, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostMemoryStall, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostTmpUsed, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostTmpSize, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSliceMemory, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSliceOomKills, token),
    runMetricQuery(fetchImpl, metricsUrl, METRIC_QUERIES.hostSlicePids, token),
  ]);

  const hostRunning = labeledSeries(hostRunningData, "host");
  const hostSlotsFree = labeledSeries(hostSlotsFreeData, "host");
  const hostSlotsTotal = labeledSeries(hostSlotsTotalData, "host");
  const hostDispatch = labeledSeries(hostDispatchData, "host");
  const hostLoad = labeledSeries(hostLoadData, "host");
  const hostCores = labeledSeries(hostCoresData, "host");
  const hostBuilds24h = labeledSeries(hostBuilds24hData, "host");
  const hostSessions = labeledSeries(hostSessionsData, "host");
  const hostMemUsed = labeledSeries(hostMemUsedData, "host");
  const hostMemTotal = labeledSeries(hostMemTotalData, "host");
  const hostSwapUsed = labeledSeries(hostSwapUsedData, "host");
  const hostSwapTotal = labeledSeries(hostSwapTotalData, "host");
  const hostNetRx = labeledSeries(hostNetRxData, "host");
  const hostNetTx = labeledSeries(hostNetTxData, "host");
  const hostTempPkg = labeledSeries(hostTempPkgData, "host");
  const hostTempMax = labeledSeries(hostTempMaxData, "host");
  const hostTempCrit = labeledSeries(hostTempCritData, "host");

  const diskFreeByHost = new Map<string, Map<string, number>>();
  for (const sample of hostDiskFreeData?.result ?? []) {
    const host = sample.metric.host;
    const mount = sample.metric.mountpoint;
    if (!host || !mount) continue;
    let mounts = diskFreeByHost.get(host);
    if (!mounts) {
      mounts = new Map();
      diskFreeByHost.set(host, mounts);
    }
    mounts.set(mount, Number(sample.value[1]));
  }

  const diskSizeByHost = new Map<string, Map<string, number>>();
  for (const sample of hostDiskSizeData?.result ?? []) {
    const host = sample.metric.host;
    const mount = sample.metric.mountpoint;
    if (!host || !mount) continue;
    let mounts = diskSizeByHost.get(host);
    if (!mounts) {
      mounts = new Map();
      diskSizeByHost.set(host, mounts);
    }
    mounts.set(mount, Number(sample.value[1]));
  }

  const coreLoadsByHost = new Map<string, number[]>();
  for (const sample of hostCoreLoadData?.result ?? []) {
    const host = sample.metric.host;
    const core = sample.metric.core;
    if (!host || core === undefined) continue;
    const loads = coreLoadsByHost.get(host) ?? [];
    const index = Number(core);
    loads[index] = Number(sample.value[1]);
    coreLoadsByHost.set(host, loads);
  }

  const hostTmpUsed = labeledSeries(hostTmpUsedData, "host");
  const hostTmpSize = labeledSeries(hostTmpSizeData, "host");
  const remoteWorkByHost = nestedSeries(hostRemoteWorkData, "kind");
  const stallByHost = nestedSeries(hostMemoryStallData, "window");
  const sliceMemoryByHost = nestedSeries(hostSliceMemoryData, "slice");
  const sliceOomByHost = nestedSeries(hostSliceOomKillsData, "slice");
  const slicePidsByHost = nestedSeries(hostSlicePidsData, "slice");

  const hostFleet: Record<string, HostFleetMetrics> = {};
  const hostNames = new Set<string>([
    ...hostRunning.keys(),
    ...remoteWorkByHost.keys(),
    ...hostLoad.keys(),
    ...hostTempPkg.keys(),
  ]);
  for (const host of hostNames) {
    const diskMounts = new Set([
      ...(diskFreeByHost.get(host)?.keys() ?? []),
      ...(diskSizeByHost.get(host)?.keys() ?? []),
    ]);
    hostFleet[host] = {
      load: hostLoad.get(host),
      cores: hostCores.get(host),
      coreLoads: coreLoadsByHost.get(host),
      running: hostRunning.get(host),
      slotsFree: hostSlotsFree.get(host),
      slotsTotal: hostSlotsTotal.get(host),
      builds24h: hostBuilds24h.get(host),
      sessions: hostSessions.get(host),
      remoteWork: remoteWorkFrom(remoteWorkByHost.get(host)),
      dispatchAccept: (hostDispatch.get(host) ?? 0) > 0,
      memUsedBytes: hostMemUsed.get(host),
      memTotalBytes: hostMemTotal.get(host),
      swapUsedBytes: hostSwapUsed.get(host),
      swapTotalBytes: hostSwapTotal.get(host),
      netRxBytesPerSecond: hostNetRx.get(host),
      netTxBytesPerSecond: hostNetTx.get(host),
      disk: [...diskMounts].map((mountpoint) => ({
        mountpoint,
        freeBytes: diskFreeByHost.get(host)?.get(mountpoint) ?? 0,
        sizeBytes: diskSizeByHost.get(host)?.get(mountpoint) ?? 0,
      })),
      tempPkg: hostTempPkg.get(host),
      tempMax: hostTempMax.get(host),
      tempCrit: hostTempCrit.get(host),
      guard: buildHostGuard(
        stallByHost.get(host),
        hostTmpUsed.get(host),
        hostTmpSize.get(host),
        sliceMemoryByHost.get(host),
        sliceOomByHost.get(host),
        slicePidsByHost.get(host),
      ),
    };
  }

  const capabilityExits: CapabilityExitCount[] = labeledSeriesWithExtra(
    capabilityExitData,
    ["host", "command", "code"],
  ).map((row) => ({
    host: row.labels.host!,
    command: row.labels.command!,
    code: row.labels.code!,
    count: row.value,
  }));

  return {
    queueDepth: scalarFromResult(queueDepthData),
    queueOldestAgeSeconds: scalarFromResult(queueOldestData),
    queueP95AgeSeconds: scalarFromResult(queueP95Data),
    fallbackLeaseExpired: boolFromScalar(scalarFromResult(fallbackLeaseData)),
    artifactCasMismatchTotal: scalarFromResult(artifactCasData),
    hostFleet,
    capabilityExits,
  };
}

function formatDuration(seconds: number): string {
  if (seconds < 60) return `${Math.round(seconds)}s`;
  if (seconds < 3600) return `${Math.floor(seconds / 60)}m`;
  const hours = Math.floor(seconds / 3600);
  const minutes = Math.floor((seconds % 3600) / 60);
  return minutes > 0 ? `${hours}h${minutes}m` : `${hours}h`;
}

function formatBytes(bytes: number): string {
  if (bytes >= 1_000_000_000) return `${(bytes / 1_000_000_000).toFixed(1)} GB`;
  if (bytes >= 1_000_000) return `${(bytes / 1_000_000).toFixed(0)} MB`;
  return `${bytes} B`;
}

function buildItem(
  source: string,
  ts: string,
  incidentKey: string,
  severity: Severity,
  title: string,
  detail: string,
  actions: ActionRef[],
): Item {
  return {
    id: `${source}:${incidentKey}`,
    source,
    severity,
    kind: "build",
    title,
    detail,
    ts,
    actions,
  };
}

function builderCapacityIdle(status: ControllerStatus, metrics: OffloadMetrics): boolean {
  for (const [host, hostStatus] of Object.entries(status.hosts)) {
    if (hostStatus.role !== "builder" || hostStatus.enrolling) continue;
    if (hostStatus.state !== "available") continue;
    const fleet = metrics.hostFleet[host];
    const running = fleet?.running ?? 0;
    const slotsFree = fleet?.slotsFree ?? status.capacity.idleSlots;
    if (running === 0 && slotsFree > 0) return true;
  }
  return status.capacity.idleSlots > 0;
}

function queueStalled(status: ControllerStatus, metrics: OffloadMetrics): boolean {
  const oldest =
    metrics.queueOldestAgeSeconds ??
    status.queue.oldestAgeSeconds;
  const slo = status.queue.sloSeconds;
  return oldest > slo && builderCapacityIdle(status, metrics);
}

function deriveItems(
  source: string,
  ts: string,
  status: ControllerStatus,
  metrics: OffloadMetrics,
): Item[] {
  const items: Item[] = [];
  const revision = String(status.revision);

  if (queueStalled(status, metrics)) {
    const oldest =
      metrics.queueOldestAgeSeconds ??
      status.queue.oldestAgeSeconds;
    const idleSlots = status.capacity.idleSlots;
    const queueJobs = status.queue.tickets
      .slice(0, 5)
      .map((t) => `${t.key}@${t.repo}`)
      .join(", ");
    items.push(
      buildItem(
        source,
        ts,
        "remote-idle-queue-stalled",
        "act",
        "Remote capacity idle while queue is stalled",
        `${status.queue.depth} builds waiting (oldest ${formatDuration(oldest)}, SLO ${formatDuration(status.queue.sloSeconds)}); ${idleSlots} builder slots idle. Queue: ${queueJobs}`,
        [
          {
            verb: "admission-reconcile",
            args: { expectedRevision: revision },
            label: "Reconcile dispatch",
            recommended: true,
          },
          {
            verb: "recall-spill",
            args: { expectedRevision: revision },
            label: "Recall laptop spill",
          },
        ],
      ),
    );
  }

  // The exit counter is monotonic for the controller's lifetime, so it can say "happened"
  // but never "still broken". Only the persisted open breaker decides the item exists.
  const capabilityGroups = new Map<
    string,
    { host: string; command: string; count: number; code: string | null }
  >();
  for (const [host, hostStatus] of Object.entries(status.hosts)) {
    const command = hostStatus.capability?.circuitOpen
      ? hostStatus.capability.missingCommand
      : undefined;
    if (!command) continue;
    let observed: CapabilityExitCount | undefined;
    for (const exit of metrics.capabilityExits) {
      if (exit.host !== host || exit.command !== command) continue;
      if (!observed || exit.count > observed.count) observed = exit;
    }
    capabilityGroups.set(`${host}:${command}`, {
      host,
      command,
      count: observed?.count ?? 0,
      code: observed?.code ?? null,
    });
  }

  for (const [, group] of capabilityGroups) {
    const linkedJobs = status.jobs
      .filter(
        (job) =>
          job.host === group.host &&
          (job.rc === 126 || job.rc === 127),
      )
      .map((job) => job.id);
    const jobLinks = linkedJobs.join(", ");
    items.push(
      buildItem(
        source,
        ts,
        `capability-missing:${group.host}:${group.command}`,
        "act",
        group.code === null
          ? `Capability quarantined — ${group.command} on ${group.host}`
          : `Capability quarantined — ${group.command} on ${group.host}, ${group.count}× exit ${group.code}`,
        group.code === null
          ? `host ${group.host} · command ${group.command} · circuit open, no command-not-found exit recorded this controller run · jobs: ${jobLinks}`
          : `host ${group.host} · command ${group.command} · ${group.count}× exit ${group.code} · jobs: ${jobLinks}`,
        [
          {
            verb: "host-quarantine",
            args: { host: group.host, command: group.command, expectedRevision: revision },
            label: "Quarantine host",
            recommended: true,
          },
        ],
      ),
    );
  }

  if (status.lease.expired || metrics.fallbackLeaseExpired) {
    items.push(
      buildItem(
        source,
        ts,
        "fallback-lease-expired",
        "act",
        "Local fallback lease expired",
        status.lease.reason ??
          `Fallback lease on ${status.lease.host ?? "workstation"} expired; remote-only policy may be violated.`,
        [
          {
            verb: "ci-reconcile",
            args: { expectedRevision: revision },
            label: "Reconcile CI eligibility",
            recommended: true,
          },
        ],
      ),
    );
  }

  const blockedJobs = status.jobs.filter((job) => job.publication?.state === "blocked");
  if (blockedJobs.length > 0 || (metrics.artifactCasMismatchTotal ?? 0) > 0) {
    for (const job of blockedJobs) {
      items.push(
        buildItem(
          source,
          ts,
          `artifact-publication-blocked:${job.id}`,
          "act",
          `Artifact publication blocked — ${job.id}`,
          job.publication?.reason ??
            `Snapshot/CAS mismatch for ${job.snapshot}; publication halted (never auto-promote).`,
          [
            {
              verb: "job-retry",
              args: { jobId: job.id, expectedRevision: revision },
              label: "Retry job",
            },
          ],
        ),
      );
    }
  }

  for (const [host, fleet] of Object.entries(metrics.hostFleet)) {
    if (
      fleet.tempPkg !== undefined &&
      fleet.tempCrit !== undefined &&
      fleet.tempPkg >= fleet.tempCrit
    ) {
      items.push(
        buildItem(
          source,
          ts,
          `cpu-temp-critical:${host}`,
          "act",
          `CPU temperature critical — ${host}`,
          `Package temp ${fleet.tempPkg}°C ≥ critical ${fleet.tempCrit}°C; new dispatch paused until it clears.`,
          [
            {
              verb: "box-drain",
              args: { host, expectedRevision: revision },
              label: "Drain box",
              recommended: true,
            },
          ],
        ),
      );
    }
  }

  return items;
}

export interface GuardWatchState {
  /** host/slice -> OOM-kill total last seen. Seeded silently so restart never renotifies. */
  oomBaseline: Map<string, number>;
}

export function createGuardWatchState(): GuardWatchState {
  return { oomBaseline: new Map() };
}

// Box-side guard alerts. Each condition is a failure that already happened on a box — a
// kernel kill, a filesystem that can no longer be written, a box making no forward
// progress. None of them is a resource level, and none of them kills anything.
function deriveGuardItems(
  source: string,
  ts: string,
  metrics: OffloadMetrics,
  state: GuardWatchState,
  stallPercent: number,
  tmpFullPercent: number,
  revision: string,
): Item[] {
  const items: Item[] = [];
  for (const [host, fleet] of Object.entries(metrics.hostFleet)) {
    const guard = fleet.guard;
    if (!guard) continue;

    for (const slice of guard.workSlices) {
      if (slice.oomKillTotal === null) continue;
      const key = `${host}/${slice.slice}`;
      const previous = state.oomBaseline.get(key);
      state.oomBaseline.set(key, slice.oomKillTotal);
      if (previous === undefined || slice.oomKillTotal <= previous) continue;
      items.push(
        buildItem(
          source,
          ts,
          `oom-kill:${host}:${slice.slice}:${slice.oomKillTotal}`,
          "act",
          `Work killed by the kernel on ${host}`,
          `${slice.slice} exceeded its memory ceiling — ${slice.oomKillTotal - previous} kill(s) since the last check, ${slice.oomKillTotal} since boot. Whatever ran there failed on the box; this laptop was never involved.`,
          [],
        ),
      );
    }

    if (
      guard.tmpSizeBytes !== null &&
      guard.tmpUsedBytes !== null &&
      guard.tmpSizeBytes > 0 &&
      (guard.tmpUsedBytes / guard.tmpSizeBytes) * 100 >= tmpFullPercent
    ) {
      items.push(
        buildItem(
          source,
          ts,
          `tmp-full:${host}`,
          "act",
          `Scratch tmpfs full on ${host}`,
          `/tmp holds ${formatBytes(guard.tmpUsedBytes)} of ${formatBytes(guard.tmpSizeBytes)} — writes on that box are failing, so runs placed there will fail until it drains.`,
          [],
        ),
      );
    }

    if (guard.stall.full300 !== null && guard.stall.full300 >= stallPercent) {
      items.push(
        buildItem(
          source,
          ts,
          `memory-stall:${host}`,
          "act",
          `${host} is making no forward progress`,
          `Every task on the box was blocked on memory for ${guard.stall.full300.toFixed(1)}% of the last five minutes. Nothing was killed — drain the box to let it recover, or leave it if the work is expected to be this heavy.`,
          [
            {
              verb: "box-drain",
              args: { host, expectedRevision: revision },
              label: "Drain box",
            },
          ],
        ),
      );
    }
  }
  return items;
}

function buildOffloadControlPanel(
  status: ControllerStatus,
  ts: string,
  stale: boolean,
  registry: BuildboxRegistry,
): Panel {
  // Capability is an observation about a host the registry declares reachable. A controller
  // host absent from the registry would otherwise put probes for an unauthorized target on
  // the page, which is the same identity leak the fleet panel refuses.
  const capabilityHosts = Object.fromEntries(
    registry.hosts
      .filter((declared) => declared.state === "reachable" && status.hosts[declared.name])
      .map((declared) => [
        declared.name,
        status.hosts[declared.name]!.capability ?? { probes: [] },
      ]),
  );
  return {
    id: "offload-control",
    ts,
    data: {
      stale,
      desired: status.desired,
      observed: status.observed,
      revision: status.revision,
      lease: status.lease,
      reconciler: status.reconciler,
      dispatch: status.dispatch,
      spill: status.queue.spill,
      capabilityByHost: capabilityHosts,
    },
  };
}

function buildClusterQueuePanel(
  status: ControllerStatus,
  metrics: OffloadMetrics,
  ts: string,
  stale: boolean,
): Panel {
  return {
    id: "cluster-queue",
    ts,
    data: {
      stale,
      depth: metrics.queueDepth ?? status.queue.depth,
      oldestAgeSeconds: metrics.queueOldestAgeSeconds ?? status.queue.oldestAgeSeconds,
      p95AgeSeconds: metrics.queueP95AgeSeconds ?? status.queue.p95AgeSeconds,
      sloSeconds: status.queue.sloSeconds,
      dispatchTarget: status.queue.dispatchTarget,
      spill: status.queue.spill,
      wedge: status.queue.wedge,
      tickets: status.queue.tickets,
      kpis: {
        ...status.kpis,
        ...(() => {
          const pullDurations = status.jobs
            .map((job) => job.pullDurationSeconds)
            .filter((seconds): seconds is number => seconds !== null && seconds !== undefined);
          return pullDurations.length > 0
            ? { longestPullMs: Math.max(...pullDurations) * 1000 }
            : {};
        })(),
      },
    },
  };
}

function buildRemoteJobsPanel(status: ControllerStatus, ts: string, stale: boolean): Panel {
  return {
    id: "remote-jobs",
    ts,
    data: {
      stale,
      jobs: status.jobs.map((job) => ({
        id: job.id,
        repo: job.repo,
        snapshot: job.snapshot,
        stage: job.stage,
        host: job.host,
        rc: job.rc ?? null,
        pullBytes: job.pullBytes ?? null,
        pullDurationSeconds: job.pullDurationSeconds ?? null,
        publication: job.publication ?? { state: "none" },
      })),
    },
  };
}

/**
 * The registry declares every CI host. Controller data is observation only: it cannot add
 * host identity, because that would let a stale or malformed controller response invent a
 * target the registry has not authorized.
 */
function buildFleetPanel(
  status: ControllerStatus,
  metrics: OffloadMetrics,
  ts: string,
  stale: boolean,
  registry: BuildboxRegistry,
  parity: Record<string, HostParity> | null,
): Panel {
  const hosts = registry.hosts.map((declared) => {
    const name = declared.name;
    const hostStatus = status.hosts[name];
    const fleet = metrics.hostFleet[name] ?? {};
    return {
      host: name,
      role: hostStatus?.role ?? (declared.roles.includes("builder") ? "builder" : "workstation"),
      state: hostStatus?.state ?? null,
      registryState: declared.state,
      registryRoles: [...declared.roles],
      rustdesk: declared.rustdesk,
      // A controller outage is not evidence about enrollment: null means unobserved,
      // false means the controller answered and does not know this host.
      enrolled: stale ? null : hostStatus !== undefined,
      primary: hostStatus?.primary ?? false,
      enrolling: hostStatus?.enrolling ?? false,
      load: fleet.load ?? null,
      cores: fleet.cores ?? null,
      coreLoads: fleet.coreLoads ?? [],
      running: fleet.running ?? null,
      slotsFree: fleet.slotsFree ?? null,
      slotsTotal: fleet.slotsTotal ?? null,
      builds24h: fleet.builds24h ?? null,
      sessions: fleet.sessions ?? null,
      remoteWork: fleet.remoteWork ?? null,
      dispatchAccept: fleet.dispatchAccept ?? null,
      capability: hostStatus?.capability ?? { probes: [] },
      memUsedBytes: fleet.memUsedBytes ?? null,
      memTotalBytes: fleet.memTotalBytes ?? null,
      swapUsedBytes: fleet.swapUsedBytes ?? null,
      swapTotalBytes: fleet.swapTotalBytes ?? null,
      netRxBytesPerSecond: fleet.netRxBytesPerSecond ?? null,
      netTxBytesPerSecond: fleet.netTxBytesPerSecond ?? null,
      disk: fleet.disk ?? [],
      temp: {
        pkg: fleet.tempPkg ?? null,
        max: fleet.tempMax ?? null,
        crit: fleet.tempCrit ?? null,
      },
      guard: fleet.guard ?? null,
      parity: parity?.[name] ?? null,
    };
  });
  return {
    id: "fleet",
    ts,
    data: { stale, hosts, registryError: null },
  };
}

function controllerDownSnapshot(
  source: string,
  ts: string,
  registry: BuildboxRegistry,
  parity: Record<string, HostParity> | null,
): AdapterResult {
  // Host identity does not come from the controller, so a controller outage must not empty
  // the fleet: the declared hosts stay listed, with every observation marked stale.
  const emptyStatus = { hosts: {} } as ControllerStatus;
  const stalePanels: Panel[] = [
    { id: "offload-control", ts, data: { stale: true } },
    { id: "cluster-queue", ts, data: { stale: true } },
    { id: "remote-jobs", ts, data: { stale: true } },
    buildFleetPanel(
      emptyStatus,
      { hostFleet: {}, capabilityExits: [] },
      ts,
      true,
      registry,
      parity,
    ),
  ];
  return {
    panels: stalePanels,
    items: [
      buildItem(
        source,
        ts,
        "controller-down",
        "act",
        "Build controller unreachable",
        "GET /status failed; offload panels are stale until the controller recovers.",
        [],
      ),
    ],
  };
}

function registryUnavailableSnapshot(
  source: string,
  ts: string,
  registryError: string,
): AdapterResult {
  return {
    panels: [
      { id: "offload-control", ts, data: { stale: true } },
      { id: "cluster-queue", ts, data: { stale: true } },
      { id: "remote-jobs", ts, data: { stale: true } },
      { id: "fleet", ts, data: { stale: true, hosts: [], registryError } },
    ],
    items: [
      buildItem(
        source,
        ts,
        "buildbox-registry-unavailable",
        "act",
        "Buildbox registry unavailable",
        "Host identity is unavailable, so the collector did not query the build controller or expose host actions.",
        [],
      ),
    ],
  };
}

export function createOffloadAdapter(opts: OffloadAdapterOptions): Adapter {
  if (!opts.fetchImpl) {
    throw new Error("fetchImpl is required");
  }

  const id = opts.id ?? "offload";
  const interval = opts.interval ?? DEFAULT_INTERVAL_MS;
  const fetchImpl = opts.fetchImpl;
  const controllerUrl = opts.controllerUrl ?? DEFAULT_CONTROLLER_URL;
  const metricsUrl = opts.metricsUrl ?? DEFAULT_METRICS_URL;
  const token = requireOffloadToken(opts.token);
  const readRegistry = opts.loadRegistry ?? (() => loadBuildboxRegistry());
  const now = opts.now ?? Date.now;
  const stallPercent = opts.stallPercent ?? DEFAULT_STALL_PERCENT;
  const tmpFullPercent = opts.tmpFullPercent ?? DEFAULT_TMP_FULL_PERCENT;
  const parityStatePath = opts.parityStatePath ?? DEFAULT_PARITY_STATE_PATH;
  const readFileImpl = opts.readFileImpl ?? ((path: string) => readFileSync(path, "utf8"));
  const guardState = createGuardWatchState();

  async function poll(): Promise<AdapterResult> {
    const ts = new Date(now()).toISOString();
    const parity = readParityState(parityStatePath, readFileImpl);

    // The registry gates every CI host operation. Without it, controller observations
    // are an untrusted subset and must not trigger a controller request or host action.
    let registry: BuildboxRegistry;
    try {
      registry = await readRegistry();
    } catch (error) {
      if (!(error instanceof RegistryUnavailableError)) throw error;
      return registryUnavailableSnapshot(id, ts, error.message);
    }

    let status: ControllerStatus;
    try {
      status = await requestControllerStatus(
        fetchImpl,
        `${controllerUrl}/status`,
        token,
      );
    } catch (error) {
      if (isControllerUnreachable(error)) {
        return controllerDownSnapshot(id, ts, registry, parity);
      }
      throw error;
    }

    const metrics = await fetchMetrics(fetchImpl, metricsUrl, token);
    const items = [
      ...deriveItems(id, ts, status, metrics),
      ...deriveGuardItems(
        id,
        ts,
        metrics,
        guardState,
        stallPercent,
        tmpFullPercent,
        String(status.revision),
      ),
    ];
    const panels: Panel[] = [
      buildOffloadControlPanel(status, ts, false, registry),
      buildClusterQueuePanel(status, metrics, ts, false),
      buildRemoteJobsPanel(status, ts, false),
      buildFleetPanel(status, metrics, ts, false, registry, parity),
    ];
    return { items, panels };
  }

  return { id, interval, poll };
}