/**
 * Fleet runtime drift detection — the predicates behind `celilo system
 * doctor`'s fleet section (designs/CELILO_DOCTOR_FLEET_DRIFT.md, ISS-0113).
 *
 * Every facet of post-migration drift that workstream B caught by hand —
 * a dead/stale dispatcher (ISS-0086), an empty `subscribers` table
 * (ISS-0088), a capability chain that never re-derived (ISS-0095 /
 * idp_dmz_ip) — becomes one check here. Each check asserts the *outcome*,
 * not a proxy for it (design D5): "a dispatcher process exists" is not the
 * same as "it's the supervised, current one that's actually emitting timer
 * ticks", and the difference is exactly the bug B hit.
 *
 * Detection is read-only and cheap (design D3). These predicates are also
 * the building blocks the defensive wiring (workstream D) and the celilo
 * MCP (ISS-0112) call — so they take their bus + DB handles as arguments
 * and return structured findings rather than rendering or exiting. The
 * rendering + `--fix` orchestration lives in the doctor command.
 */

import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import { type Bus, describeError } from '@celilo/event-bus';
import { desc, eq, inArray } from 'drizzle-orm';
import { parse as parseYaml } from 'yaml';
import { ProxmoxClient, type ProxmoxCredentials } from '../api-clients/proxmox';
import { getModuleStoragePath } from '../config/paths';
import { type DbClient, findMigrationsFolder } from '../db/client';
import { getMigrationStatus } from '../db/migration-status';
import { buildBusHookRuns, capabilities as capabilitiesTable, modules } from '../db/schema';
import { findSchemaDrift } from '../db/schema-introspection';
import { loadControlPlaneSubnet, resolveFirewallNatIp } from '../hooks/capability-loader';
import type { ModuleManifest } from '../manifest/schema';

import { getServiceCredentials, listContainerServices } from './container-service';
import {
  CONTROL_PLANE_MODULE_ID,
  getModuleSystems,
  listAllModuleSystems,
} from './deployed-systems';
import { listDnsInternalRecords } from './dns-internal-records';
import {
  type LaunchdUnitStatus,
  SUPERVISOR_SCOPES,
  type SupervisorPlatform,
  type SupervisorScope,
  launchdUnitStatus,
  readInstalledUnit,
  unitMainPid,
} from './events-daemon';
import { probeMachines } from './machine-probe';
import { describePausedModule, listPausedModules } from './module-pause';
import { resolveSubscription } from './module-subscriptions';

/** The module that IS celilo's control plane. */
const CONTROL_PLANE_MODULE = CONTROL_PLANE_MODULE_ID;

/**
 * Zones reachable from the operator's LAN. A celilo placement zone other
 * than `internal` is firewall-segmented — an unmanaged LAN device has no
 * route into it, so an internal-DNS record pointing at a container IP there
 * is unreachable (the bug). The `internal` zone IS the LAN, so a record at
 * one of its systems' IPs is fine. (Never pin literal subnets to zones —
 * compare by the zone role the system carries.)
 */
const LAN_REACHABLE_ZONE = 'internal';

/**
 * Records under the dedicated `.infra.<zone>` label are system-IDENTITY names,
 * registered by modules/technitium/scripts/on-system-event.ts at the system's
 * own container IP ON PURPOSE — the zone-side name for in-zone / VPN access,
 * deliberately kept separate from the natIp records LAN devices use (so a
 * container-IP identity record doesn't clobber the natIp record public_web
 * needs). They are NOT LAN-reachability records, so the natIp rule doesn't
 * apply to them — the service-DNS check skips them.
 */
const SYSTEM_IDENTITY_HOST = /\.infra\./;

export type FleetFindingStatus = 'ok' | 'warn' | 'fail';

/**
 * One drift facet's verdict. `autoFixable` marks the checks `--fix` may
 * run unattended (today: subscribers resync only) — everything else is
 * report + a named manual remediation, never a surprise prod redeploy.
 */
export interface FleetFinding {
  /** Stable id for the check (e.g. 'dispatcher'); not user-facing prose. */
  id: string;
  title: string;
  status: FleetFindingStatus;
  summary: string;
  /** Extra context lines, rendered indented under the summary. */
  detail: string[];
  /** A concrete next step, or null when status is ok. */
  remediation: string | null;
  autoFixable: boolean;
}

/**
 * A timer subscriber should see a fresh tick within its interval plus
 * slack. 15m is the shortest DDNS-refresh cadence (namecheap); a tick
 * older than this means the dispatcher isn't emitting (the workstream-B
 * stale-orphan symptom). One window covers the common case without
 * parsing every interval name.
 */
const TIMER_TICK_MAX_AGE_MS = 20 * 60 * 1000;

const UNRESOLVED_REF = /\$\{?(?:self|capability|infra|infrastructure|system|secret):/;

/** Worst of a set of statuses (fail > warn > ok). */
function worst(statuses: FleetFindingStatus[]): FleetFindingStatus {
  if (statuses.includes('fail')) return 'fail';
  if (statuses.includes('warn')) return 'warn';
  return 'ok';
}

/**
 * The DB schema the running CLI expects must actually exist on this box.
 * celilo only runs drizzle's `migrate()` on a FRESH database; an existing
 * install gets a hand-maintained CREATE/ALTER list in db/client.ts instead
 * (ISS-0100). When that list drifts from the shipped migrations — e.g. a new
 * migration adds a table nobody added to the list — the running code expects
 * a table/column the DB doesn't have, and features fail at runtime.
 *
 * This asserts the outcome directly (track-agnostic): every table + column in
 * the code's drizzle schema is present in the DB. A miss means migrations
 * haven't reached this box. It does NOT count migration rows — an existing DB
 * patched via the hand list legitimately lags `__drizzle_migrations` while
 * its schema is current, so presence is the honest signal.
 */
export function checkSchemaDrift(db: DbClient): FleetFinding {
  const { missingTables, missingColumns, tableCount, columnCount } = findSchemaDrift(db.$client);

  const detail: string[] = [];
  if (missingTables.length > 0) detail.push(`missing table(s): ${missingTables.join(', ')}`);
  if (missingColumns.length > 0) detail.push(`missing column(s): ${missingColumns.join(', ')}`);

  // Also name unapplied migrations. Presence is the honest signal for schema
  // objects, but a migration can carry an index or a data fix that presence
  // can't see — and an operator reading "migrations applied" deserves to know
  // when some aren't (celilo#604). Best-effort: an install layout where the
  // journal can't be found must not fail the check.
  let pending: string[] = [];
  try {
    pending = getMigrationStatus(db.$client, findMigrationsFolder()).pending;
  } catch {
    // No journal reachable — the presence check above still stands.
  }
  if (pending.length > 0) detail.push(`unapplied migration(s): ${pending.join(', ')}`);

  const status: FleetFindingStatus = detail.length > 0 ? 'fail' : 'ok';

  return {
    id: 'schema',
    title: 'database schema matches the running CLI (migrations applied)',
    status,
    // Say tables AND columns: "all 35 tables present" reads as though columns
    // went unchecked, which is what sent a rollout to sqlite3 over SSH to
    // confirm a column migration the doctor had in fact already verified.
    summary:
      status === 'ok'
        ? `all ${tableCount} schema tables and ${columnCount} columns present, no unapplied migrations`
        : 'database schema is behind the running CLI — migrations not applied',
    detail,
    remediation:
      status === 'ok'
        ? null
        : 'run `celilo system migrate` to apply pending migrations on this box (`celilo system migrate --status` names them) — see ISS-0100',
    autoFixable: false,
  };
}

interface HeartbeatRow {
  dispatcher_id: string;
  last_heartbeat: number;
  started_at: number;
  pid: number;
  version: string;
}

export interface DispatcherCheckOptions {
  now?: number;
  /**
   * mtime (ms) of the installed dispatcher code (`@celilo/event-bus`
   * package.json). A dispatcher whose `started_at` predates this is
   * running stale in-memory code — the exact workstream-B orphan. Omit
   * to skip the staleness aspect (e.g. unit tests, or when the package
   * can't be located).
   */
  installedCodeMtimeMs?: number | null;
  /** Override for readInstalledUnit — tests point this at a temp home. */
  home?: string;
  platform?: SupervisorPlatform;
  /** Prefix for system-scope unit paths. Test seam — see getDaemonUnitPath. */
  systemRoot?: string;
  /**
   * Which pid each scope's unit supervises. Injected so the check is testable
   * without systemd. Returning null means "can't tell" — the check then makes
   * no supervision claim rather than guessing.
   */
  unitMainPid?: (scope: SupervisorScope) => number | null;
  /**
   * Probe launchd for the unit's health (darwin only — see launchdUnitStatus).
   * Injected so the crash-loop aspect is testable without launchd.
   */
  launchdProbe?: (scope: SupervisorScope) => LaunchdUnitStatus | null;
}

/**
 * The dispatcher check is four-part (design D5): a dispatcher is (1)
 * running, (2) the *supervised* one (survives reboot, not an orphan),
 * (3) running *current* code, and (4) actually emitting timer ticks +
 * draining deliveries. A naive "is a process up?" check reports green
 * while broken — that's the trap this exists to avoid.
 */
export function checkDispatcher(bus: Bus, opts: DispatcherCheckOptions = {}): FleetFinding {
  const now = opts.now ?? Date.now();
  const health = bus.health();
  const hb = bus.db
    .query<HeartbeatRow, []>(
      'SELECT dispatcher_id, last_heartbeat, started_at, pid, version FROM dispatcher_heartbeat ORDER BY last_heartbeat DESC LIMIT 1',
    )
    .get();

  const detail: string[] = [];
  const statuses: FleetFindingStatus[] = [];
  const remediations: string[] = [];

  // (1) running — a fresh heartbeat. health() already classifies a
  // stale/absent heartbeat as no_dispatcher.
  if (health.status === 'no_dispatcher' || !hb) {
    statuses.push('fail');
    detail.push(
      'no live dispatcher — heartbeat absent or stale (events are queueing, not delivered)',
    );
    remediations.push(
      'start the dispatcher: `systemctl --user enable --now celilo-events.service` (or `celilo events install-daemon` then enable it)',
    );
    // A unit can be INSTALLED and still dead: the supervisor respawns a
    // program that cannot start in the unit's environment and it dies again,
    // forever. On macOS the classic cause is a PATH-less
    // EnvironmentVariables dict — the global celilo wrapper needs
    // `command -v bun`, and launchd's default PATH has none (celilo#1373:
    // 22680 respawns on one Mac). Name the crash loop instead of telling
    // the operator to "start the dispatcher".
    if (opts.platform === 'darwin') {
      const probe = opts.launchdProbe ?? launchdUnitStatus;
      const st = probe('user');
      if (st && st.pid === null && st.lastExitStatus != null && st.lastExitStatus !== 0) {
        const fate =
          st.lastExitStatus & 0xff
            ? `signal ${st.lastExitStatus & 0xff}`
            : `exit code ${st.lastExitStatus >> 8}`;
        detail.push(
          `the installed launchd unit is crash-looping (last ${fate}) — launchd respawns it and it dies again, so no dispatcher ever serves the bus`,
        );
        remediations.push(
          "read the unit's stderr (`tail ~/Library/Logs/celilo-events.err.log`) and reinstall with `celilo events install-daemon` so the unit carries an environment that resolves bun (celilo#1373)",
        );
      }
    }
    // Without a heartbeat there's nothing more to assert about it.
    return {
      id: 'dispatcher',
      title: 'event dispatcher running, supervised & current',
      status: 'fail',
      summary: 'no live event dispatcher',
      detail,
      remediation: remediations.join('; '),
      autoFixable: false,
    };
  }

  const ageMs = health.lastHeartbeatAgeMs ?? now - hb.last_heartbeat;
  detail.push(
    `running (pid ${hb.pid}, heartbeat ${Math.round(ageMs / 1000)}s ago, code v${hb.version})`,
  );
  if (health.status === 'stuck') {
    statuses.push('warn');
    detail.push(
      `${health.stuckRunningCount} delivery(ies) stuck in 'running' — possible crashed handler`,
    );
    remediations.push('`celilo events repair` to sweep stuck deliveries');
  }

  // (1b) sole — a duplicate dispatcher can no longer START (it refuses), but one
  // stranded before that shipped keeps running, and it makes every other check
  // here ambiguous: `hb` is whichever of them wrote last. Fail rather than warn —
  // celilo-mgr ran two for 40 days precisely because nothing reported it (#580).
  if (health.dispatcherCount > 1) {
    statuses.push('fail');
    const pids = health.dispatchers.map((d) => d.pid).join(', ');
    detail.push(
      `${health.dispatcherCount} dispatchers are live on this bus (pids ${pids}) — only one may run`,
    );
    remediations.push(
      'stop the unsupervised one: compare `systemctl show celilo-events.service -p MainPID` against those pids and kill the pid systemd does not own',
    );
  }

  // (2) supervised — not just "a unit file exists on disk", but "the process
  // that is actually running IS the one an installed unit supervises".
  //
  // The file-exists test this replaces reported green on celilo-mgr while the
  // system unit was dead and a user-scope unit of the SAME NAME served
  // production (#610). A check whose entire job is catching an unsupervised
  // dispatcher cannot be satisfied by a file nobody is running.
  const installedScopes = SUPERVISOR_SCOPES.filter(
    (scope) =>
      readInstalledUnit({
        scope,
        home: opts.home,
        platform: opts.platform,
        systemRoot: opts.systemRoot,
      }).exists,
  );
  if (installedScopes.length === 0) {
    statuses.push('warn');
    detail.push('not under a supervisor unit — will not survive a reboot (orphan process)');
    remediations.push('`celilo events install-daemon` then enable the unit so it is supervised');
  } else {
    if (installedScopes.length > 1) {
      statuses.push('fail');
      detail.push(
        'both a user-scope AND a system-scope unit are installed — same unit name, different services; ' +
          'one will lose the race on every boot and retry forever',
      );
      remediations.push(
        'keep exactly one: `celilo events uninstall-daemon` (user) or `celilo events uninstall-daemon --system`, and disable it in systemd',
      );
    }
    // Only accuse when systemd actually answered. A null probe is ignorance,
    // not evidence of an orphan.
    const probe =
      opts.unitMainPid ?? ((scope: SupervisorScope) => unitMainPid(scope, opts.platform));
    const supervisedPids = installedScopes
      .map((scope) => ({ scope, pid: probe(scope) }))
      .filter((entry): entry is { scope: SupervisorScope; pid: number } => entry.pid !== null);
    if (supervisedPids.length > 0 && !supervisedPids.some((entry) => entry.pid === hb.pid)) {
      statuses.push('fail');
      detail.push(
        `the running dispatcher (pid ${hb.pid}) is not the process any installed unit supervises ` +
          `(${supervisedPids.map((e) => `${e.scope}=${e.pid}`).join(', ')}) — restarting the unit will not restart it`,
      );
      remediations.push(
        'stop the unsupervised process and let the unit own the dispatcher, or reinstall the unit for the scope that is actually running it',
      );
    }
  }

  // (3) current — started before the installed code was last written ⇒
  // running stale in-memory code (delivers, but may not emit new event
  // types like timer ticks). The workstream-B orphan, exactly.
  if (opts.installedCodeMtimeMs != null && hb.started_at < opts.installedCodeMtimeMs) {
    statuses.push('warn');
    const startedAgoMin = Math.round((now - hb.started_at) / 60000);
    detail.push(
      `started ${startedAgoMin}min ago — before the last code update; running stale code, restart to pick it up`,
    );
    remediations.push('restart the dispatcher: `systemctl --user restart celilo-events.service`');
  }

  // (4) emitting + delivering. Only assert ticks if something subscribes
  // to a timer (no subscriber ⇒ no expectation). Assert no piled-up
  // failed deliveries either way.
  const timerSub = bus.db
    .query<{ pattern: string }, []>(
      "SELECT pattern FROM subscribers WHERE pattern LIKE 'timer.tick.%' LIMIT 1",
    )
    .get();
  if (timerSub) {
    const latestTick = bus.recentEvents({ type: timerSub.pattern, limit: 1 })[0];
    if (!latestTick) {
      statuses.push('warn');
      detail.push(
        `a subscriber wants '${timerSub.pattern}' but no such tick has ever been emitted (refresh/DDNS not firing)`,
      );
      remediations.push('restart the dispatcher so it emits timer ticks');
    } else if (now - latestTick.emittedAt > TIMER_TICK_MAX_AGE_MS) {
      statuses.push('warn');
      const ageMin = Math.round((now - latestTick.emittedAt) / 60000);
      detail.push(
        `last '${timerSub.pattern}' was ${ageMin}min ago — dispatcher not emitting on schedule`,
      );
      remediations.push('restart the dispatcher so it resumes emitting timer ticks');
    }
  }

  // A TRUE total, not `failedDeliveries().length` — that saturates at its LIMIT
  // and printed a literal `50` on celilo-mgr that read as a count (celilo#623).
  // The newest row dates the backlog: a big total whose newest entry is days old
  // is drained history, not active bleeding.
  const { total: failedTotal } = bus.failedDeliveryTotals();
  if (failedTotal > 0) {
    statuses.push('warn');
    const newest = bus.failedDeliveries({ limit: 1 })[0];
    const age = newest?.finishedAt
      ? `, most recent ${Math.round((now - newest.finishedAt) / 60000)}min ago`
      : '';
    detail.push(`${failedTotal} failed/abandoned delivery(ies) total${age}`);
    const sample = describeError(newest?.lastError ?? null)?.split('\n')[0];
    if (sample) detail.push(`  example (newest, not the only one): ${sample}`);
    remediations.push('`celilo events list-failed` to see them; re-emit/repair as needed');
  }

  const status = worst(statuses);
  return {
    id: 'dispatcher',
    title: 'event dispatcher running, supervised & current',
    status,
    summary:
      status === 'ok'
        ? 'dispatcher healthy, supervised, current, and emitting'
        : 'dispatcher running but degraded',
    detail,
    remediation: remediations.length > 0 ? remediations.join('; ') : null,
    autoFixable: false,
  };
}

/**
 * The build-bus self-update chain, asserted outcome-first like every check
 * here. Four facts separate the ways `on_upstream_publish` silently does
 * nothing (celilo#1304: the CLI on a management box aged in place while the
 * doctor's dispatcher check read green — that check covers the event-bus
 * dispatcher, which no build-bus hook ever travels through):
 *
 *   1. no module declares the hook        → nothing can self-update; not in use
 *   2. `build-bus.publish` events arrive  → the webhook path works end to end
 *   3. hook-run rows exist for an event   → the dispatcher ran matching hooks
 *   4. the newest run exited 0            → the self-update itself succeeded
 *
 * Facts 3 and 4 read the `build_bus_hook_runs` ledger the dispatcher writes;
 * before that ledger existed, a failed self-update existed only as daemon
 * console output, which no surface reads.
 */
export function checkBuildBusPublishing(
  bus: Bus,
  db: DbClient,
  opts: { now?: number } = {},
): FleetFinding {
  const now = opts.now ?? Date.now();

  const hookModules = db
    .select({ id: modules.id })
    .from(modules)
    .where(inArray(modules.state, ['INSTALLED', 'VERIFIED']))
    .all()
    .filter((m) => hasUpstreamPublishHooks(db, m.id));

  const finding = (
    status: FleetFindingStatus,
    summary: string,
    detail: string[],
    remediation: string | null,
  ): FleetFinding => ({
    id: 'build-bus-publishing',
    title: 'build-bus webhooks arriving and self-update hooks running',
    status,
    summary,
    detail,
    remediation,
    autoFixable: false,
  });

  if (hookModules.length === 0) {
    return finding(
      'ok',
      'no installed module declares on_upstream_publish — build-bus self-update not in use here',
      [],
      null,
    );
  }

  const newestPublish = bus.recentEvents({ type: 'build-bus.publish', limit: 1 })[0];
  if (!newestPublish) {
    return finding(
      'warn',
      `${hookModules.map((m) => m.id).join(', ')} self-update can never fire: no build-bus publish has ever arrived`,
      [
        'The webhook receiver (`celilo subscribers serve`) either is not running here or was never sent a verified publish — and nothing in the deploy installs it.',
      ],
      "install the receiver under a supervisor unit: `celilo subscribers install-daemon --secret $CELILO_BUS_SECRET` (systemd/launchd; `--print` renders for Ansible), enable it per its next-steps output, and register this host in the publisher's build-bus-subscribers.json with the same secret",
    );
  }

  const runs = db
    .select()
    .from(buildBusHookRuns)
    .orderBy(desc(buildBusHookRuns.id))
    .limit(10)
    .all();

  const detail: string[] = [
    `newest publish: ${newestPublish.emittedAt ? `${Math.round((now - newestPublish.emittedAt) / 60000)}min ago` : 'unknown age'}`,
  ];

  const newest = runs[0];
  if (!newest) {
    return finding(
      'warn',
      'publish events arrive but no self-update hook has ever run',
      [
        ...detail,
        'The receiver emitted to the bus, yet the in-process hook dispatcher dispatched nothing — it was started with --no-dispatch, no module matched, or the running CLI predates the run ledger.',
      ],
      "check how the receiver daemon is started (drop --no-dispatch) and compare the publish events against the installed modules' on_upstream_publish match rules",
    );
  }

  const runAgeMin = newest.ranAt ? Math.round((now - newest.ranAt.getTime()) / 60000) : null;
  const runLine = `newest hook run: ${newest.packageName}@${newest.packageVersion} on ${newest.moduleId}${runAgeMin != null ? `, ${runAgeMin}min ago` : ''}`;

  if (newest.timedOut) {
    return finding(
      'warn',
      `self-update hook timed out after ${newest.durationMs}ms`,
      [
        ...detail,
        runLine,
        `  stderr tail: ${(newest.stderrTail ?? '(none)').trim().slice(0, 500)}`,
      ],
      'inspect the ledger: `select * from build_bus_hook_runs order by id desc limit 10` in the celilo DB',
    );
  }
  if (newest.exitCode !== 0) {
    return finding(
      'warn',
      `self-update hook exited ${newest.exitCode ?? 'without an exit code (spawn failure)'}`,
      [
        ...detail,
        runLine,
        `  stderr tail: ${(newest.stderrTail ?? '(none)').trim().slice(0, 500)}`,
      ],
      'fix the failing hook script, then re-publish or re-emit the build-bus.publish event',
    );
  }

  // A success that predates the newest publish means the newest publish was
  // dropped on the dispatch side even though earlier ones ran. The ledger's
  // ran_at has second resolution (unixepoch), so the event's millisecond
  // stamp is truncated to the same granularity before comparing — otherwise
  // a run in the same second as its event reads as older than it.
  const newestPublishSec = Math.floor((newestPublish.emittedAt ?? 0) / 1000) * 1000;
  if (newest.ranAt && newest.ranAt.getTime() < newestPublishSec) {
    return finding(
      'warn',
      'newest publish event predates the newest hook run — that publish dispatched nothing',
      [...detail, runLine],
      'check the receiver daemon logs for that event and the match rules against what the publisher sent',
    );
  }

  return finding(
    'ok',
    `last self-update hook succeeded (${newest.packageName}@${newest.packageVersion})`,
    [...detail, runLine],
    null,
  );
}

/**
 * Does module `id`'s installed manifest declare an `on_upstream_publish`
 * hook? Reads the manifest from disk exactly as the hook dispatcher's own
 * loader does, so the check and the dispatcher cannot disagree about what
 * "in use" means. A manifest that has vanished or fails to parse reads as
 * "no hooks" — the dispatcher skips it silently too (Rule 6.2: the skip is
 * visible here as fact 1 rather than as a crash).
 */
function hasUpstreamPublishHooks(db: DbClient, moduleId: string): boolean {
  const row = db
    .select({ sourcePath: modules.sourcePath })
    .from(modules)
    .where(eq(modules.id, moduleId))
    .get();
  if (!row?.sourcePath) return false;
  try {
    const parsed = parseYaml(readFileSync(join(row.sourcePath, 'manifest.yml'), 'utf-8')) as {
      hooks?: { on_upstream_publish?: unknown[] };
    };
    return (parsed.hooks?.on_upstream_publish ?? []).length > 0;
  } catch {
    return false;
  }
}

/** A deployed module + its parsed manifest (INSTALLED/VERIFIED only). */
interface DeployedModule {
  id: string;
  manifest: ModuleManifest;
}

function loadDeployedModules(db: DbClient): DeployedModule[] {
  const rows = db
    .select()
    .from(modules)
    .where(inArray(modules.state, ['INSTALLED', 'VERIFIED']))
    .all();
  return rows.map((m) => ({ id: m.id, manifest: m.manifestData as unknown as ModuleManifest }));
}

/**
 * The bus `subscribers` table must reflect what the deployed fleet's
 * manifests declare. A restore/migration starts it EMPTY (ISS-0088), so
 * every reactive subscription silently vanishes until a resync or a
 * redeploy. Missing rows fail; stale rows (a since-removed module) warn.
 *
 * Scoped to MODULE-owned rows — core's own subscribers share this bus and are
 * declared by no manifest.
 */
export function checkSubscribers(bus: Bus, db: DbClient): FleetFinding {
  const deployed = loadDeployedModules(db);

  // Expected: every (scoped name → pattern) the deployed manifests declare.
  const expected = new Map<string, string>();
  for (const mod of deployed) {
    const subs = mod.manifest.subscriptions ?? [];
    const modulePath = `${getModuleStoragePath()}/${mod.id}`;
    for (const sub of subs) {
      const resolved = resolveSubscription(sub, mod.id, modulePath);
      expected.set(resolved.name, resolved.pattern);
    }
  }

  const actualRows = bus.db
    .query<{ name: string; pattern: string; registered_by: string | null }, []>(
      'SELECT name, pattern, registered_by FROM subscribers',
    )
    .all();
  const actual = new Map(actualRows.map((r) => [r.name, r.pattern]));

  const missing: string[] = [];
  const mismatched: string[] = [];
  for (const [name, pattern] of expected) {
    const have = actual.get(name);
    if (have === undefined) missing.push(name);
    else if (have !== pattern) mismatched.push(`${name} (manifest: ${pattern}, bus: ${have})`);
  }
  // Only module-owned rows can be stale. celilo core registers subscribers on
  // the same bus (the alerting/backup/operations sweeps) that no manifest
  // declares, so judging every row against the manifest-derived map reported
  // them as drift on every install, forever (#624).
  //
  // INVARIANT this relies on: a module row is named `<registered_by>.<sub>` —
  // `resolveSubscription` (services/module-subscriptions.ts) derives both the
  // scoped name and registered_by from the same module id. A core row is NOT
  // scoped under its registrar's id (`celilo-alerting` registers
  // `celilo-alerting-sweep`, not `celilo-alerting.sweep`). That's what lets
  // provenance separate them with no allowlist to keep in sync.
  //
  // To break it: have a core registrar name a row under its own id — a future
  // `celilo-alerting.digest` would read as module-owned, find no deployed
  // module named `celilo-alerting`, and be reported stale. Which is #624 again.
  const stale = actualRows
    .filter((r) => r.registered_by !== null && r.name.startsWith(`${r.registered_by}.`))
    .map((r) => r.name)
    .filter((name) => !expected.has(name));

  const detail: string[] = [];
  const statuses: FleetFindingStatus[] = [];
  if (missing.length > 0) {
    statuses.push('fail');
    detail.push(
      `${missing.length} subscription(s) declared by the fleet but missing from the bus: ${missing.join(', ')}`,
    );
  }
  if (mismatched.length > 0) {
    statuses.push('warn');
    detail.push(
      `${mismatched.length} subscription(s) with a pattern the bus disagrees on: ${mismatched.join('; ')}`,
    );
  }
  if (stale.length > 0) {
    statuses.push('warn');
    detail.push(
      `${stale.length} subscriber(s) on the bus with no deployed module: ${stale.join(', ')}`,
    );
  }

  // `resync-subscriptions` only upserts from manifests — it never deletes, so
  // it cannot clear a stale row. Advertising that as the remedy (and as
  // auto-fixable) made `--fix` run a guaranteed no-op and call it a fix.
  const resyncable = missing.length > 0 || mismatched.length > 0;
  const remediations: string[] = [];
  if (resyncable) remediations.push('`celilo events resync-subscriptions` (safe, idempotent)');
  if (stale.length > 0)
    remediations.push(`\`celilo subscribers remove <name>\` for: ${stale.join(', ')}`);

  const status = worst(statuses);
  return {
    id: 'subscribers',
    title: 'bus subscribers reflect the deployed fleet',
    status,
    summary:
      status === 'ok'
        ? `${expected.size} subscription(s) match the deployed fleet`
        : 'bus subscribers drifted from the deployed fleet',
    detail,
    remediation: remediations.length > 0 ? remediations.join('; ') : null,
    autoFixable: resyncable,
  };
}

/** A `$capability:<name>.<path>` reference parsed out of a derive_from. */
interface CapabilityRef {
  variable: string;
  capability: string;
  path: string;
  /** A manifest default makes an absent/broken optional derivation safe. */
  hasFallback: boolean;
}

export type CapabilityDerivationReason = 'no-provider' | 'empty-value' | 'unresolved-ref';

/**
 * A broken `source: capability` derivation found on a consumer module.
 *  - `no-provider`: nothing in the capabilities map provides `capability`.
 *  - `empty-value`: the field exists but is null/undefined/empty.
 *  - `unresolved-ref`: the field is itself an unresolved template ref
 *    (e.g. authentik's `idp.dmz_ip = $self:caddy_dmz_ip`) — the chain is
 *    broken one+ hops upstream.
 */
export interface CapabilityDerivationProblem {
  consumerModule: string;
  variable: string;
  capability: string;
  path: string;
  reason: CapabilityDerivationReason;
  /** The offending value, for `unresolved-ref`. */
  value?: string;
}

function parseCapabilityRefs(manifest: ModuleManifest): CapabilityRef[] {
  const refs: CapabilityRef[] = [];
  for (const v of manifest.variables?.owns ?? []) {
    if (v.source !== 'capability' || !v.derive_from) continue;
    const re = /\$\{?capability:([\w-]+)\.([\w.]+)/g;
    let m: RegExpExecArray | null = re.exec(v.derive_from);
    while (m !== null) {
      refs.push({
        variable: v.name,
        capability: m[1],
        path: m[2],
        hasFallback: Object.prototype.hasOwnProperty.call(v, 'default'),
      });
      m = re.exec(v.derive_from);
    }
  }
  return refs;
}

/** Walk a dotted path into a JSON object; undefined if any hop is absent. */
function getNested(data: Record<string, unknown>, path: string): unknown {
  let cur: unknown = data;
  for (const seg of path.split('.')) {
    if (cur == null || typeof cur !== 'object') return undefined;
    cur = (cur as Record<string, unknown>)[seg];
  }
  return cur;
}

/**
 * The shared capability-derivation predicate (design D4: build once, call
 * from the detector AND the preventer). For a consumer manifest and a map
 * of capability-name → data, return every `source: capability` derivation
 * that won't resolve.
 *
 * The `capabilities` map can be either RAW (the doctor reads capability
 * rows straight from the DB, so a still-derived field shows up as
 * `unresolved-ref`) or RESOLVED (deploy preflight / generate pass
 * `ResolutionContext.capabilities`, where `$self:` refs are already
 * substituted against the provider's config — so a broken upstream link
 * shows up as `empty-value` or `unresolved-ref`). Callers decide severity:
 * the doctor treats `unresolved-ref` as "needs the ISS-0114 chain trace"
 * (a note), while preflight/generate treat every reason as a hard error.
 */
export function findBrokenCapabilityDerivations(
  consumerModule: string,
  manifest: ModuleManifest,
  capabilities: Record<string, Record<string, unknown> | undefined>,
): CapabilityDerivationProblem[] {
  const problems: CapabilityDerivationProblem[] = [];
  for (const ref of parseCapabilityRefs(manifest)) {
    const base = {
      consumerModule,
      variable: ref.variable,
      capability: ref.capability,
      path: ref.path,
    };
    const data = capabilities[ref.capability];
    if (!data) {
      if (ref.hasFallback) continue;
      problems.push({ ...base, reason: 'no-provider' });
      continue;
    }
    const value = getNested(data, ref.path);
    if (value === undefined || value === null || value === '') {
      if (ref.hasFallback) continue;
      problems.push({ ...base, reason: 'empty-value' });
      continue;
    }
    if (typeof value === 'string' && UNRESOLVED_REF.test(value)) {
      if (ref.hasFallback) continue;
      problems.push({ ...base, reason: 'unresolved-ref', value });
    }
  }
  return problems;
}

/** One-line human description of a broken derivation, shared by all callers. */
export function describeCapabilityProblem(p: CapabilityDerivationProblem): string {
  const head = `${p.consumerModule}.${p.variable} derives from $capability:${p.capability}.${p.path}`;
  switch (p.reason) {
    case 'no-provider':
      return `${head}, but no deployed module provides '${p.capability}' — deploy/redeploy its provider first`;
    case 'empty-value':
      return `${head}, but the provider's '${p.capability}' data has no value there — redeploy the provider so it re-registers`;
    case 'unresolved-ref':
      return `${head}, which resolves to an unresolved ref (${p.value}) — its own upstream chain is broken; redeploy the provider chain (provider → consumer)`;
  }
}

/**
 * Every `source: capability` variable a deployed module derives must have
 * a live provider whose capability data carries the referenced field
 * (the forgejo `$self:idp_dmz_ip not found` class, ISS-0095/ISS-0115).
 *
 * Reads RAW capability data, so a present-but-still-derived field (e.g.
 * authentik's `idp.dmz_ip = $self:caddy_dmz_ip`) can't be verified here
 * without the backward chain-walker (ISS-0114). Rather than ship a second
 * walker (design D2.1), those `unresolved-ref` cases are flagged as "needs
 * the chain trace" — a note pointing at `celilo capability chain`, not a
 * false-positive fail. (Deploy preflight + generate run the same predicate
 * against the RESOLVED context, where the same break IS a hard error.)
 */
export function checkCapabilityProviders(db: DbClient): FleetFinding {
  const deployed = loadDeployedModules(db);
  const capRows = db
    .select({ name: capabilitiesTable.capabilityName, data: capabilitiesTable.data })
    .from(capabilitiesTable)
    .all();
  const rawMap: Record<string, Record<string, unknown>> = {};
  for (const r of capRows) rawMap[r.name] = r.data;

  const breaks: string[] = [];
  const traceNeeded: string[] = [];
  let refCount = 0;

  for (const mod of deployed) {
    const problems = findBrokenCapabilityDerivations(mod.id, mod.manifest, rawMap);
    refCount += parseCapabilityRefs(mod.manifest).length;
    for (const p of problems) {
      if (p.reason === 'unresolved-ref') {
        traceNeeded.push(
          `${p.consumerModule}.${p.variable} ← ${p.capability}.${p.path} (= ${p.value})`,
        );
      } else {
        breaks.push(describeCapabilityProblem(p));
      }
    }
  }

  const detail: string[] = [];
  let status: FleetFindingStatus = 'ok';
  if (breaks.length > 0) {
    status = 'fail';
    detail.push(...breaks);
  }
  if (traceNeeded.length > 0) {
    detail.push(
      // celilo#1308: `celilo capability chain` no longer exists (the ISS-0115
      // rework removed it), so prescribing it sent the operator to an
      // "unknown command" error. No CLI command traces a chain; say so.
      `${traceNeeded.length} derived value(s) resolve through another capability. No CLI command traces a capability chain, so verify these by hand (ISS-0114): ${traceNeeded.join('; ')}`,
    );
  }

  return {
    id: 'capability-derived',
    title: 'capability-derived config has live providers',
    status,
    summary:
      status === 'ok'
        ? `${refCount} capability-derived reference(s) have providers`
        : 'capability-derived config is missing a provider',
    detail,
    remediation:
      status === 'ok'
        ? null
        : 'redeploy the provider module(s) so they re-register capability data, then redeploy the consumer (provider → consumer order)',
    autoFixable: false,
  };
}

/**
 * Internal split-horizon DNS records for service hostnames must resolve to
 * the firewall natIp (the LAN-reachable DNAT ingress), not a zone-side
 * container IP a LAN device can't route to (ISS-0094 / ISS-0111). Reads the
 * dns_internal ledger offline — every `registerRecord({type:'A'})` the
 * capability loader saw — and compares each to the natIp:
 *   - == natIp                              → ok
 *   - a segmented-zone system's container IP → fail (unroutable from the LAN)
 *   - an `internal`-zone system's IP         → ok (that zone IS the LAN)
 *   - anything else                          → warn (unknown / possibly stale)
 *
 * Skipped cleanly when no firewall advertises a natIp (a flat network has no
 * segmented zones, so container IPs are reachable).
 */
export async function checkServiceDns(db: DbClient): Promise<FleetFinding> {
  const base = {
    id: 'service-dns',
    title: 'service DNS points at the firewall natIp',
    autoFixable: false,
  } as const;

  // The ledger table may be absent on a DB whose schema is behind (ISS-0100).
  // Don't crash the whole doctor — the schema-drift check owns that signal.
  let records: ReturnType<typeof listDnsInternalRecords>;
  try {
    records = listDnsInternalRecords(db);
  } catch {
    return {
      ...base,
      status: 'ok',
      summary: 'internal-DNS ledger not present (schema behind — see the schema check)',
      detail: [],
      remediation: null,
    };
  }

  if (records.length === 0) {
    return {
      ...base,
      status: 'ok',
      summary: 'no internal DNS records registered',
      detail: [],
      remediation: null,
    };
  }

  const natIp = await resolveFirewallNatIp(db);
  if (!natIp) {
    return {
      ...base,
      status: 'ok',
      summary: `${records.length} internal DNS record(s); no firewall natIp to check against`,
      detail: ['no firewall advertises a natIp — flat network, container IPs are LAN-reachable'],
      remediation: null,
    };
  }

  // Map every deployed system's container IP → its zone, so a record can be
  // recognized as pointing at a segmented-zone container (the bug) vs an
  // internal-zone (LAN) system.
  const ipZone = new Map<string, { moduleId: string; zone: string }>();
  for (const mod of loadDeployedModules(db)) {
    for (const sys of getModuleSystems(mod.id, db)) {
      if (sys.ipv4_address) ipZone.set(sys.ipv4_address, { moduleId: mod.id, zone: sys.zone });
    }
  }

  const atContainer: string[] = [];
  const atOther: string[] = [];
  for (const r of records) {
    // `.infra.<zone>` system-identity records are intentionally container-IP
    // (zone-side names, not LAN-reachability records) — not subject to the
    // natIp rule.
    if (SYSTEM_IDENTITY_HOST.test(r.host)) continue;
    if (r.ip === natIp) continue;
    const owner = ipZone.get(r.ip);
    if (owner && owner.zone !== LAN_REACHABLE_ZONE) {
      atContainer.push(
        `${r.host} → ${r.ip} (${owner.moduleId}'s ${owner.zone}-zone container IP — a LAN device can't route there; should be the natIp ${natIp})`,
      );
    } else if (!owner) {
      atOther.push(`${r.host} → ${r.ip} (neither the natIp ${natIp} nor a known system IP)`);
    }
  }

  const detail: string[] = [];
  const statuses: FleetFindingStatus[] = [];
  if (atContainer.length > 0) {
    statuses.push('fail');
    detail.push(...atContainer);
  }
  if (atOther.length > 0) {
    statuses.push('warn');
    detail.push(...atOther);
  }

  const status = worst(statuses);
  return {
    ...base,
    status,
    summary:
      status === 'ok'
        ? `${records.length} internal DNS record(s) resolve to the natIp or a LAN-reachable system`
        : 'internal DNS records point at zone-side IPs unreachable from the LAN',
    detail,
    remediation:
      status === 'fail'
        ? 'redeploy the provider so it registers the record at the firewall natIp (firewall.exposeService result), not the container IP'
        : null,
  };
}

export interface RunFleetChecksOptions {
  now?: number;
  installedCodeMtimeMs?: number | null;
  /**
   * Where the host-liveness verdict gets its facts. Injected so the check can
   * be exercised without SSH or a Proxmox credential; defaults to the live
   * fleet (`collectHostLiveness`).
   */
  hostLiveness?: () => Promise<HostLivenessInputs>;
}

/**
 * Run every fleet check against the given handles. The caller owns
 * gating (skip when there's no celilo DB) and rendering; this just
 * returns the findings, in the order they're shown.
 */
/**
 * celilo's control plane must sit on a network celilo RECOGNIZES. When it doesn't,
 * two things break silently and in different subsystems:
 *
 *  1. the firewall's trusted sources don't cover it, so a default-DROP FORWARD
 *     chain blocks celilo's own SSH to every deployed box;
 *  2. the internal resolver has no split-horizon view for it, so it answers
 *     NOERROR with ZERO records — indistinguishable from "no such name". The name
 *     then falls through to public DNS and the box tries to hairpin off the WAN IP.
 *
 * Neither surfaces where the cause is. In production this presented as
 * `apt-get update` timing out against the celilo-hosted apt repo — three layers
 * from the actual misconfiguration, and it took a multi-step investigation to
 * locate precisely because every layer degraded politely. Hence this check.
 */
export function checkControlPlaneNetwork(db: DbClient): FleetFinding {
  const systems = getModuleSystems(CONTROL_PLANE_MODULE, db);
  const subnet = loadControlPlaneSubnet(db);

  if (subnet) {
    return {
      id: 'control-plane-network',
      title: "celilo's own network is recognized (firewall trust + internal DNS)",
      status: 'ok',
      summary: `control plane on ${subnet}`,
      detail: [],
      remediation: null,
      autoFixable: false,
    };
  }

  const notDeployed = systems.length === 0;
  const zones = [...new Set(systems.map((sys) => sys.zone).filter(Boolean))];
  const addresses = systems.map((sys) => sys.ipv4_address).filter(Boolean);

  const detail = notDeployed
    ? [`no deployed systems found for '${CONTROL_PLANE_MODULE}'`]
    : [
        `'${CONTROL_PLANE_MODULE}' is in zone(s): ${zones.join(', ') || '(unset)'}`,
        `address(es): ${addresses.join(', ') || '(unknown)'}`,
        'no network.<zone>.subnet is configured for that zone',
      ];
  detail.push(
    'consequence 1: firewall trusted sources fall back to network.internal.subnet — celilo may be unable to reach segmented zones',
    'consequence 2: the internal resolver has no view for this network — internal names resolve publicly and cannot be hairpinned',
  );

  return {
    id: 'control-plane-network',
    title: "celilo's own network is recognized (firewall trust + internal DNS)",
    status: 'warn',
    summary: notDeployed
      ? "cannot determine celilo's control-plane network — falling back to the internal subnet"
      : "celilo's control-plane network is not a configured zone",
    detail,
    remediation: notDeployed
      ? `deploy '${CONTROL_PLANE_MODULE}' so celilo knows where its control plane runs, or set network.internal.subnet if it lives on the internal LAN`
      : `run \`celilo system config set network.${zones[0] ?? '<zone>'}.subnet <cidr>\` for the control plane's network, then reconcile the firewall and the internal resolver`,
    autoFixable: false,
  };
}

/**
 * Any paused module is a doctor failure — no threshold, whatever its age
 * (openspec/changes/module-pause-lifecycle, design D7, closed at review).
 *
 * A pause deliberately switches OFF the alerting that would otherwise report
 * the module as down, so the paused-ness itself has to be the signal. A
 * duration threshold was considered and dropped: there is no number of hours
 * after which a deliberate outage becomes acceptable, and a configurable one is
 * just an invitation to tune the detector until it stops firing.
 *
 * The fleet had just run 20 hours of failing forgejo backups whose only symptom
 * was a column of `0 B` rows that read as healthy hourly cadence. Same failure
 * shape; this is the check that would have named it.
 */
export function checkPausedModules(db: DbClient): FleetFinding {
  const paused = listPausedModules(db);

  if (paused.length === 0) {
    return {
      id: 'paused-modules',
      title: 'No module is paused (a pause suppresses its own alerting)',
      status: 'ok',
      summary: 'nothing paused',
      detail: [],
      remediation: null,
      autoFixable: false,
    };
  }

  const names = paused.map((m) => describePausedModule(m));
  return {
    id: 'paused-modules',
    title: 'No module is paused (a pause suppresses its own alerting)',
    status: 'fail',
    summary: `${paused.length} module(s) paused: ${paused.map((m) => m.id).join(', ')}`,
    detail: [
      ...names.map((n) => `paused: ${n}`),
      'a paused module receives no dispatched work, runs no health checks, and has its alerts suppressed',
      'it is still deployed and may still be serving traffic — pause does not stop the data plane',
    ],
    remediation:
      'bring each back with "celilo module unpause <id>" (which redeploys it, rebinding its capabilities), or remove it if the pause was permanent',
    autoFixable: false,
  };
}

/** One module deployment, and the host it landed on. */
export interface HostPlacement {
  moduleId: string;
  /** The host's user-facing name — a pool hostname, or the container's. */
  hostname: string;
  infraType: 'machine' | 'container_service';
  /** Proxmox VMID for a celilo-provisioned container; null for a pool machine. */
  vmid: number | null;
}

/**
 * Everything the liveness verdict is computed from, injected so the check is a
 * pure function over data and needs neither SSH nor a Proxmox credential to
 * test.
 *
 * ⚠️ Every source here is ALLOWED TO BE ABSENT, and absent is not "fine".
 * A machine missing from `machines` was not probed; a node missing from `nodes`
 * was not reported. Neither means the host is up, and neither means it is down.
 * Conflating "I could not look" with "I looked and it was healthy" is the
 * failure this whole check exists to end — doctor said OK-with-warnings while a
 * node hosting two modules was offline.
 */
export interface HostLivenessInputs {
  placements: HostPlacement[];
  /** Machine-pool SSH probe results. A hostname absent here was NOT probed. */
  machines: Array<{ hostname: string; reachable: boolean }>;
  /** Proxmox node status. Empty when no container service is configured. */
  nodes: Array<{ node: string; online: boolean }>;
  /** VMID → node name, from `/cluster/resources`. Empty when unqueried. */
  guestNodes: Array<{ vmid: number; node: string }>;
}

type HostState = 'up' | 'down' | 'unknown';

interface HostVerdict {
  host: string;
  state: HostState;
  /** Why the state could not be determined. Set only when state is 'unknown'. */
  reason?: string;
}

function resolveHostState(placement: HostPlacement, inputs: HostLivenessInputs): HostVerdict {
  if (placement.infraType === 'machine') {
    const probe = inputs.machines.find((m) => m.hostname === placement.hostname);
    if (!probe) {
      return {
        host: placement.hostname,
        state: 'unknown',
        // Either the SSH probe did not run at all, or this hostname is no
        // longer in the machine pool — a stale `module_systems` row, which is
        // its own defect and worth surfacing rather than rounding off.
        reason: 'no probe result for this machine',
      };
    }
    return { host: placement.hostname, state: probe.reachable ? 'up' : 'down' };
  }

  // A container's liveness is its NODE's liveness. The guest being stopped is a
  // different condition with a different owner (`module pause --stop-infra`
  // stops guests deliberately), so this deliberately reads the node only.
  if (placement.vmid === null) {
    return {
      host: placement.hostname,
      state: 'unknown',
      reason: 'no VMID recorded — celilo has no liveness source for this provider',
    };
  }
  const guest = inputs.guestNodes.find((g) => g.vmid === placement.vmid);
  if (!guest) {
    return {
      host: placement.hostname,
      state: 'unknown',
      reason: `VMID ${placement.vmid} not present in the cluster's resources`,
    };
  }
  const node = inputs.nodes.find((n) => n.node === guest.node);
  if (!node) {
    return { host: guest.node, state: 'unknown', reason: 'the cluster reported no such node' };
  }
  return { host: guest.node, state: node.online ? 'up' : 'down' };
}

/**
 * Are the hosts this fleet's modules actually run on alive? (celilo#728)
 *
 * Every other `Fleet runtime` check is a control-plane concern — the
 * dispatcher, bus subscribers, capability-derived config, internal DNS. None of
 * them asked the most basic data-plane question, so `system doctor` reported
 * "OK with warnings" while a Proxmox node was OFFLINE with `celilo-apt-repo`
 * and `lunacycle` on it. The information was already in `proxmox node list`;
 * doctor simply never consulted it. It surfaced only because a release run got
 * an HTTP 502 from the apt repo that happened to live there — had nothing tried
 * to publish, the node could have stayed down indefinitely.
 *
 * A down host is a FAILURE, not a warning: it is strictly worse than the
 * conditions already reported as failures here, and every module on it is down
 * with it.
 *
 * A host celilo TRIED to verify and could not is a WARNING, and the reason is
 * named per host. This is not the same as "quiet because it might be fine": a
 * cluster that will not answer its own API is not obviously healthier than one
 * reporting a node offline, and the failure to answer may BE the outage this
 * check exists to catch. Reporting it as ok-with-a-note would rebuild the
 * defect one level down — a report reading healthy over something unmeasured.
 *
 * ⚠️ The thing that makes the warning safe to have is that it is not
 * permanent. A machine-only fleet produces NO unverified hosts at all: every
 * placement takes the probe path and resolves. The one standing source would be
 * a provider celilo cannot interrogate — today a DigitalOcean droplet, whose
 * client can verify the token but never reads droplet status. That is a gap to
 * close (its own change), not a reason to soften the signal here. A warning
 * that fires forever is what trains an operator to skim the whole report
 * (celilo#723, whose false positive was competing for attention in the very
 * output that missed the offline node) — so if this one ever becomes standing,
 * the fix is to teach celilo the missing provider, not to quieten it.
 */
export function checkHostLiveness(inputs: HostLivenessInputs): FleetFinding {
  const base = {
    id: 'host-liveness',
    title: 'The hosts running deployed modules are alive',
    autoFixable: false,
  } as const;

  if (inputs.placements.length === 0) {
    return {
      ...base,
      status: 'ok',
      summary: 'no modules are deployed to a host yet',
      detail: [],
      remediation: null,
    };
  }

  interface HostEntry {
    state: HostState;
    modules: Set<string>;
    reason?: string;
  }

  const modulesByHost = new Map<string, HostEntry>();
  for (const placement of inputs.placements) {
    const { host, state, reason } = resolveHostState(placement, inputs);
    const entry = modulesByHost.get(host) ?? { state, modules: new Set<string>(), reason };
    // A host resolved 'down' by any placement stays down — one authoritative
    // negative outranks an unknown from a sibling placement.
    if (state === 'down' || entry.state === 'unknown') {
      entry.state = state;
      entry.reason = reason;
    }
    entry.modules.add(placement.moduleId);
    modulesByHost.set(host, entry);
  }

  const describe = (host: string, e: HostEntry) => `${host}: ${[...e.modules].sort().join(', ')}`;
  const describeUnverified = (host: string, e: HostEntry) =>
    `unverified — ${describe(host, e)}${e.reason ? ` (${e.reason})` : ''}`;

  const down = [...modulesByHost].filter(([, e]) => e.state === 'down');
  const unknown = [...modulesByHost].filter(([, e]) => e.state === 'unknown');
  const up = [...modulesByHost].filter(([, e]) => e.state === 'up');

  if (down.length > 0) {
    const affected = down.reduce((n, [, e]) => n + e.modules.size, 0);
    return {
      ...base,
      status: 'fail',
      summary: `${down.length} host(s) down, ${affected} module(s) unreachable: ${down
        .map(([host]) => host)
        .join(', ')}`,
      detail: [
        ...down.map(([host, e]) => `DOWN ${describe(host, e)}`),
        ...unknown.map(([host, e]) => describeUnverified(host, e)),
        'every module listed against a down host is down with it, whatever its own status says',
      ],
      remediation:
        'bring the host back, then confirm with "celilo proxmox node list" (container services) or "celilo machine status <hostname>" (pool machines)',
    };
  }

  if (unknown.length > 0) {
    // WARN, not ok-with-a-note. celilo tried and could not find out, and the
    // reason it could not may be the outage itself — a cluster that will not
    // answer its own API is not evidence of health. Reporting this quietly
    // would rebuild #728 one level down.
    const affected = unknown.reduce((n, [, e]) => n + e.modules.size, 0);
    return {
      ...base,
      status: 'warn',
      summary: `${up.length} host(s) up, ${unknown.length} could not be verified (${affected} module(s))`,
      // Named and reasoned, never counted: "1 not verified" tells an operator
      // neither which host nor what to do about it.
      detail: unknown.map(([host, e]) => describeUnverified(host, e)),
      remediation:
        'check the host directly — "celilo proxmox node list" for a container service, "celilo machine status <hostname>" for a pool machine; a host celilo cannot reach is not a host known to be healthy',
    };
  }

  return {
    ...base,
    status: 'ok',
    summary: `${up.length} host(s) up`,
    detail: [],
    remediation: null,
  };
}

/**
 * Read the liveness facts off the live fleet.
 *
 * Every source degrades to ABSENT rather than to a cheerful default. A Proxmox
 * cluster that cannot be reached, or a fleet with no container service at all,
 * contributes no node rows — and `checkHostLiveness` reads that as unverified,
 * never as healthy. That distinction is the whole point of the check.
 */
export async function collectHostLiveness(db: DbClient): Promise<HostLivenessInputs> {
  const placements: HostPlacement[] = listAllModuleSystems(db).map((s) => ({
    moduleId: s.moduleId,
    hostname: s.hostname,
    infraType: s.infraType,
    vmid: s.vmid ?? null,
  }));

  // Nothing deployed — skip the probes entirely rather than SSH a fleet of none.
  if (placements.length === 0) {
    return { placements, machines: [], nodes: [], guestNodes: [] };
  }

  let machines: HostLivenessInputs['machines'] = [];
  try {
    machines = (await probeMachines()).map((m) => ({
      hostname: m.hostname,
      reachable: m.reachable,
    }));
  } catch {
    // Leave it empty: unprobed, which reports as unverified rather than up.
  }

  const nodes: HostLivenessInputs['nodes'] = [];
  const guestNodes: HostLivenessInputs['guestNodes'] = [];
  try {
    for (const service of await listContainerServices()) {
      if (service.providerName !== 'proxmox') continue;
      const creds = (await getServiceCredentials(service.id)) as ProxmoxCredentials;
      const result = await new ProxmoxClient(creds).clusterResources();
      if (!result.success) continue;
      for (const row of result.data) {
        if (row.type === 'node' && row.node) {
          nodes.push({ node: row.node, online: row.status === 'online' });
        } else if (typeof row.vmid === 'number' && row.node) {
          guestNodes.push({ vmid: row.vmid, node: row.node });
        }
      }
    }
  } catch {
    // Same rule: unreachable is unverified, not healthy.
  }

  return { placements, machines, nodes, guestNodes };
}

export async function runFleetChecks(
  bus: Bus,
  db: DbClient,
  opts: RunFleetChecksOptions = {},
): Promise<FleetFinding[]> {
  const hostLiveness = opts.hostLiveness ?? (() => collectHostLiveness(db));
  return [
    checkSchemaDrift(db),
    checkDispatcher(bus, { now: opts.now, installedCodeMtimeMs: opts.installedCodeMtimeMs }),
    checkBuildBusPublishing(bus, db, { now: opts.now }),
    checkSubscribers(bus, db),
    checkCapabilityProviders(db),
    checkControlPlaneNetwork(db),
    checkPausedModules(db),
    checkHostLiveness(await hostLiveness()),
    await checkServiceDns(db),
  ];
}