/**
 * Module Deployment Orchestration Service
 *
 * Coordinates full module deployment workflow across all phases
 */

import { existsSync, readdirSync, rmSync } from 'node:fs';
import { join } from 'node:path';
import { ProgressDisplay } from '@celilo/cli-display';
import { and, eq } from 'drizzle-orm';
import { generateInventory } from '../ansible/inventory';
import { FuelGauge } from '../cli/fuel-gauge';
import { log, setActiveDisplay } from '../cli/prompts';
import type { DbClient } from '../db/client';
import { capabilities, machines, modules, systemConfig } from '../db/schema';
import { loadCapabilityFunctions } from '../hooks/capability-loader';
import { invokeHook } from '../hooks/executor';
import { createHookStores } from '../hooks/hook-store';
import { createGaugeLogger } from '../hooks/logger';
import { describeArtifacts } from '../hooks/types';
import type { HookDefinition, HookLogger, HookResult } from '../hooks/types';
import { type ModuleManifest, getSingularSystemSpec } from '../manifest/schema';
import { decryptSecret } from '../secrets/encryption';
import { getOrCreateMasterKey } from '../secrets/master-key';
import { buildResolutionContext } from '../variables/context';
import { maybeRunAspectForTrigger } from './aspect-runner';
import { askConfirm } from './bus-interview';
import {
  autoDeriveMachineConfig,
  findEnsureOnProvider,
  interviewForEnsureInputs,
  interviewForMissingConfig,
  interviewForMissingSecrets,
} from './config-interview';
import { getContainerService } from './container-service';
import { bootstrapControlPlane } from './control-plane-bootstrap';
import { executeAnsible } from './deploy-ansible';
import { planDeployment } from './deploy-planner';
import { waitForSSH } from './deploy-ssh';
import { executeTerraform, parseTerraformOutputs } from './deploy-terraform';
import { validateAndPrepareDeployment } from './deploy-validation';
import { CONTROL_PLANE_MODULE_ID } from './deployed-systems';
import { getModuleSystems } from './deployed-systems';
import { E2E_CONFLICT_MESSAGE, runningE2eContainers } from './e2e-guard';
import { resolveInfrastructureVariables } from './infrastructure-variable-resolver';
import { InterviewAbandonedError, InterviewUnansweredError } from './interview-errors';
import { findMachineForModule } from './machine-pool';
import { ensureRequiredNetworks } from './network-ensure';
import {
  type ProviderBackfillResult,
  loadProviderBackfillPlan,
  runProviderBackfill,
} from './provider-arrival';
import { checkProxmoxReachable, formatProxmoxUnreachableError } from './proxmox-preflight';
import { remoteAccessPolicy } from './remote-access';
import { LOCAL_MACHINE_IP, deleteTemporarySshKey, writeTemporarySshKey } from './ssh-key-manager';
import { buildTerraformEnvForService } from './terraform-env';

export interface DeployResult {
  success: boolean;
  error?: string;
  phases: {
    validation?: boolean;
    autoGenerated?: boolean;
    planning?: boolean;
    terraformInit?: boolean;
    terraformPlan?: boolean;
    terraformApply?: boolean;
    sshWait?: boolean;
    ansible?: boolean;
  };
}

// `updateMachineAssignment` is deleted (celilo#773).
//
// It was the sole writer of `machines.assigned_module_ids`, it only ever
// APPENDED, and nothing ever removed an entry — `module remove` deletes the
// module row and never touches `machines`. So a machine accumulated the ids of
// modules that no longer existed and could never be freed except by editing the
// database.
//
// Nothing replaces it. Occupancy is now derived from `module_infrastructure`
// and `module_systems` at the point of use (`getModulesOnMachine`), both of
// which the deploy path already writes and both of which cascade on module
// removal — so the machine frees itself with no bookkeeping step to forget.

export interface DeployOptions {
  debug?: boolean;
  /**
   * Keep all sub-events visible during the deploy. With this off (the
   * default), the ProgressDisplay collapses each "… doing" block into
   * a single "✔ done" line on completion. Verbose mode disables that
   * collapse so the user can see every line that flowed through —
   * useful for debugging slow or hanging steps.
   */
  verbose?: boolean;
  /**
   * Exit cleanly after the initial config + secrets interview phase
   * — before any infrastructure code generation, terraform, ansible,
   * or hook execution. Manual validation tool: lets an operator
   * exercise the bus-mediated interview (config.required / secret.required
   * events fire, responders answer, values land in the encrypted store)
   * without actually deploying anything to a machine. The cross-module
   * `ensure` interview (which fires from hooks) is NOT exercised in
   * this mode — that requires a real hook run.
   */
  stopAfterInterview?: boolean;
  /**
   * D4 (control-plane-stops-building-modules): retain the generated project
   * after a successful deploy instead of deleting it. The tree is ephemeral
   * by design — rendered for this deploy from the signed payload plus live
   * config, used, then removed — because a persisted generated tree cannot be
   * validated (only its verbatim role assets have a meaningful digest, the
   * rest is rendered and SUPPOSED to differ), it holds rendered secrets, and
   * it is what the next deploy would ship. This flag keeps it for deliberate
   * inspection. It is always kept when the deploy fails.
   */
  keepGeneratedProject?: boolean;
}

/**
 * Per-attempt context the caller builds for `invokeHookWithEnsureRetry`.
 * Recreated for each retry so the gauge animation and the logger that
 * pipes into it stay paired — and so the interview prompts get a clean
 * terminal between attempts (no animation collision).
 */
interface HookAttempt {
  gauge: FuelGauge;
  logger: HookLogger;
  invokeOptions: Parameters<typeof invokeHook>[8];
}

/**
 * Run a hook, and if it fails because a capability call discovered that a
 * provider module is missing config, run the cross-module ensure interview
 * against the provider, optionally redeploy it, and retry the hook.
 *
 * The factory is called once per attempt: the helper owns the gauge
 * lifecycle so it can `stopSilent()` before running the interview (clean
 * prompts) and create a fresh gauge for the retry.
 *
 * Loop guard: each (provider, ensure, value) triple is only allowed to
 * trigger an interview once per call — a second hit means the post-action
 * didn't actually pick up the change, and we abort with a circular-ensure
 * error rather than spinning forever.
 *
 * See `apps/celilo/designs/CROSS_MODULE_CONFIG_INTERVIEW.md`.
 */
async function invokeHookWithEnsureRetry(
  modulePath: string,
  hookName: string,
  contractVersion: string,
  hookDef: HookDefinition,
  inputs: Record<string, unknown>,
  config: Record<string, unknown>,
  hookSecrets: Record<string, string>,
  buildAttempt: () => Promise<HookAttempt>,
  db: DbClient,
  deployOptions: DeployOptions,
): Promise<HookResult> {
  const seen = new Set<string>();
  let attempt = await buildAttempt();

  while (true) {
    const result = await invokeHook(
      modulePath,
      hookName,
      contractVersion,
      hookDef,
      inputs,
      config,
      hookSecrets,
      attempt.logger,
      attempt.invokeOptions,
    );

    if (result.success) {
      attempt.gauge.stop(true);
      return result;
    }

    if (!result.missingProviderInput) {
      attempt.gauge.stop(false);
      return result;
    }

    const m = result.missingProviderInput;
    const key = `${m.providerModuleId}|${m.ensureId}|${m.value}`;
    if (seen.has(key)) {
      attempt.gauge.stop(false);
      return {
        ...result,
        error: `Circular ensure: ${m.providerModuleId} still didn't satisfy "${m.ensureId}" for "${m.value}" after the interview ran. Check that the post action (e.g. redeploy_self) actually applies the change.`,
      };
    }
    seen.add(key);

    const ensure = findEnsureOnProvider(m.providerModuleId, m.ensureId, db);
    if (!ensure) {
      attempt.gauge.stop(false);
      return {
        ...result,
        error:
          `Module "${m.providerModuleId}" doesn't declare an "ensure" block ` +
          `for "${m.ensureId}". Original hook error: ${result.error}`,
      };
    }

    // Step the gauge out of the way so prompts render on a clean line.
    attempt.gauge.stopSilent();

    log.info('');
    log.info(
      `${m.providerModuleId} doesn't yet provide ` +
        `${m.ensureId} for "${m.value}" — running cross-module setup.`,
    );

    const interviewResult = await interviewForEnsureInputs(m.providerModuleId, ensure, m.value, db);

    if (!interviewResult.success) {
      return {
        ...result,
        error: `Cross-module interview failed: ${interviewResult.error ?? 'unknown'}`,
      };
    }

    for (const line of interviewResult.applied) {
      log.info(`  ✓ ${line}`);
    }

    if (ensure.post === 'redeploy_self') {
      log.info('');
      log.info(`Redeploying ${m.providerModuleId} to apply changes...`);
      const redeploy = await deployModule(m.providerModuleId, db, deployOptions);
      if (!redeploy.success) {
        return {
          ...result,
          error: `Failed to redeploy ${m.providerModuleId}: ${redeploy.error ?? 'unknown'}`,
        };
      }
    }

    // Build a fresh attempt for the retry — new gauge, new logger, new
    // capability bindings (capabilities close over the logger).
    attempt = await buildAttempt();
  }
}

/**
 * A provider just arrived — re-run the consumers that were already here.
 *
 * The generic mirror of `consumer-cleanup.ts`. It replaces
 * `republishStaticWebConsumers`, which did this for `public_web` alone, and it
 * covers `firewall` for the first time — three modules provide `firewall` and
 * none of their consumers inherited anything on deploy (celilo#1011).
 *
 * Never fatal to the provider's own deploy. The provider deployed fine; what
 * can fail is one consumer's re-registration, and the honest report is that
 * consumer named with the command that retries it. Silence would report a clean
 * deploy over a fleet where some consumers never re-registered.
 */
// ponytail: fans out on EVERY deploy of a module that provides anything, so a
// caddy redeploy re-runs all eight `public_web` consumers' `on_install` even
// though caddy was already their provider and they have nothing to gain. The
// hooks are idempotent by contract and a provider deploy is already heavyweight,
// so this is accepted rather than guessed at — every predicate for "this
// consumer has something to gain" that does not need new state is wrong in some
// case (a redeploy of the incumbent provider looks identical to a first
// deploy). If it measurably hurts, the upgrade is to record which provider a
// consumer last bound to and re-run only where that changed.
async function backfillArrivedProvider(moduleId: string, db: DbClient): Promise<void> {
  const plan = loadProviderBackfillPlan(moduleId, db);
  if (plan.length === 0) return;

  const attempts = plan.filter((target) => !target.skip);
  const gauge = new FuelGauge(`Re-registering ${attempts.length} consumer(s) with ${moduleId}`, {
    skipAnimation: !process.stdout.isTTY,
  });
  gauge.start();
  let result: ProviderBackfillResult;
  try {
    const logger = createGaugeLogger(gauge, moduleId, 'provider_arrival');
    result = await runProviderBackfill(moduleId, plan, db, logger);
    gauge.stop(result.failures.length === 0);
  } catch (error) {
    gauge.stop(false);
    const msg = error instanceof Error ? error.message : String(error);
    log.warn(`Consumer re-registration against '${moduleId}' could not run: ${msg}`);
    return;
  }

  if (result.rerun.length > 0) {
    log.success(
      `Re-registered ${result.rerun.length} consumer(s) with '${moduleId}': ${result.rerun.join(', ')}`,
    );
  }
  // Re-surfaced outside the gauge: the gauge's own preview scrolls away, and a
  // consumer that never re-registered is exactly what an operator must still be
  // able to read once the deploy has finished.
  //
  // INFO, not warn, and the level is load-bearing. `unpause --cascade`
  // redeploys in topological order and clears `pausedAt` only AFTER each
  // redeploy succeeds (`module-pause.ts#executeUnpause`), so the provider's own
  // redeploy runs while every consumer in the cascade is still PAUSED. Warning
  // there would put one alarm per consumer immediately before the cascade
  // redeploys each of them successfully — an alarm about the normal path.
  //
  // Nothing is at risk either way: a paused module cannot return to service
  // without a redeploy, and that redeploy re-resolves its capabilities. That is
  // the same guarantee `remove-guard.ts` relies on to treat a paused module as
  // not a dependent. So this is worth saying and not worth warning about.
  for (const paused of result.skipped.filter((s) => s.reason === 'paused')) {
    log.info(
      `'${paused.consumerId}' is paused, so it was not re-registered with '${moduleId}'. It rebinds when it is unpaused and redeployed.`,
    );
  }
  for (const failure of result.failures) {
    log.warn(
      `'${failure.consumerId}' failed to re-register with '${moduleId}': ${failure.error}. Run \`celilo module deploy ${failure.consumerId}\` to retry.`,
    );
  }
}

/**
 * Orchestrate module deployment workflow
 * Orchestrator function - coordinates deployment phases
 *
 * @param moduleId - Module identifier
 * @param db - Database connection
 * @param options - Deployment options
 * @returns Deployment result with phase tracking
 */
/**
 * Public entry point. Wraps the core deploy work with event-bus
 * lifecycle emits (`deploy.started.<id>`, `deploy.completed.<id>`,
 * `deploy.failed.<id>`) so subscribers — production smoke tests,
 * alerting, etc. — can react. Bus emit failures are best-effort and
 * never affect the deploy outcome.
 */
/**
 * Files under `generated/terraform/` that outlive the deploy.
 *
 * D4 (control-plane-stops-building-modules) deletes `generated/` because it
 * holds RENDERED templates and secrets — a tree nothing can validate and what a
 * later deploy would wrongly ship. Terraform state is neither. It is the record
 * of what celilo actually BUILT, and `templates/generator.ts` calls it
 * "celilo's authoritative record of placement (ISS-0090)".
 *
 * Deleting it meant the next deploy started from empty state, failed to create
 * a container that already existed, auto-imported it, and planned a REPLACE —
 * destroying the container it was asked to converge (celilo#1374, which
 * destroyed signal's LXC and its signal-cli device identity).
 *
 * `.terraform/` is NOT preserved: it is the provider binary cache, ~19 MB per
 * provider, rebuilt by `terraform init`. `cross-module-read.ts` excludes it from
 * the backup envelope for the same reason.
 */
export const PRESERVED_GENERATED_FILES = [
  'terraform.tfstate',
  'terraform.tfstate.backup',
  '.terraform.lock.hcl',
];

/** True for the paths, relative to `generated/`, that survive the cleanup. */
export function isPreservedGeneratedPath(relativePath: string): boolean {
  const [dir, file, ...rest] = relativePath.split('/');
  if (dir !== 'terraform' || rest.length > 0) return false;
  return file !== undefined && PRESERVED_GENERATED_FILES.includes(file);
}

/**
 * Delete the rendered project but keep terraform's record of what exists.
 * Absent `generated/` is not an error — a deploy that failed early never made
 * one.
 */
export function pruneGeneratedProject(generatedPath: string): void {
  if (!existsSync(generatedPath)) return;

  for (const entry of readdirSync(generatedPath, { withFileTypes: true })) {
    if (entry.name !== 'terraform') {
      rmSync(join(generatedPath, entry.name), { recursive: true, force: true });
    }
  }

  const terraformDir = join(generatedPath, 'terraform');
  if (!existsSync(terraformDir)) return;

  for (const entry of readdirSync(terraformDir, { withFileTypes: true })) {
    if (!isPreservedGeneratedPath(`terraform/${entry.name}`)) {
      rmSync(join(terraformDir, entry.name), { recursive: true, force: true });
    }
  }
}

export async function deployModule(
  moduleId: string,
  db: DbClient,
  options: DeployOptions = {},
): Promise<DeployResult> {
  const startedAt = Date.now();
  const { emitDeployCompleted, emitDeployFailed, emitDeployStarted, emitHealthCheckFailed } =
    await import('./celilo-events');
  const { startOperation, completeOperation, failOperation } = await import('./module-operations');

  const { closeDeployWindows, ensureMonitorOnDeploy, openDeployWindow } = await import(
    './alerting/deploy-hooks'
  );

  const opId = startOperation(moduleId, 'deploy');
  emitDeployStarted({ module: moduleId, startedAt });

  // A deploy restarts services, which fails their own health checks. Without
  // this the operator is paged about the deploy they are personally running.
  openDeployWindow(db, moduleId, new Date());

  let result: DeployResult;
  try {
    result = await deployModuleImpl(moduleId, db, options);
  } catch (err) {
    const error = err instanceof Error ? err.message : String(err);
    // Close on the failure path too: a window left open would silence this
    // module forever, which is far worse than a few noisy alerts.
    closeDeployWindows(db, moduleId, new Date());
    failOperation(opId, err);
    emitDeployFailed({
      module: moduleId,
      startedAt,
      durationMs: Math.max(0, Date.now() - startedAt),
      error,
    });
    throw err;
  }

  const durationMs = Math.max(0, Date.now() - startedAt);
  closeDeployWindows(db, moduleId, new Date());

  if (result.success) {
    // D4 (control-plane-stops-building-modules): the generated project is
    // ephemeral. It was rendered for THIS deploy from the signed payload plus
    // live config; a persisted copy is a tree nothing can validate, it holds
    // rendered secrets, and it is what a later deploy would ship. Now that
    // the deploy that needed it is done — ansible, hooks, DNS backfill and
    // aspects all ran inside the impl — it goes. Kept on failure (the impl's
    // early returns) and under --keep.
    if (!options.keepGeneratedProject) {
      const deployed = db.select().from(modules).where(eq(modules.id, moduleId)).get();
      if (deployed) {
        const generatedPath = `${deployed.sourcePath}/generated`;
        try {
          pruneGeneratedProject(generatedPath);
          log.info(
            `Removed generated/ for '${moduleId}' (terraform state preserved; --keep retains all)`,
          );
        } catch (cleanupError) {
          const msg = cleanupError instanceof Error ? cleanupError.message : String(cleanupError);
          log.warn(`Failed to remove generated/ for '${moduleId}': ${msg}`);
        }
      }
    }
    // A module that declares a suggested interval starts being watched on its
    // first successful deploy — watching is the default, not a thing to
    // remember.
    ensureMonitorOnDeploy(db, moduleId);
    completeOperation(opId);
    emitDeployCompleted({ module: moduleId, startedAt, durationMs });
  } else {
    failOperation(opId, result.error ?? 'unknown error');
    emitDeployFailed({
      module: moduleId,
      startedAt,
      durationMs,
      error: result.error ?? 'unknown error',
    });
    // Health-check failures are a sub-class worth surfacing on their own
    // channel so subscribers can target them without parsing error strings.
    if (result.error?.toLowerCase().includes('health check')) {
      emitHealthCheckFailed({ module: moduleId, reason: result.error });
    }
  }
  return result;
}

/**
 * Headless bound for the repoint confirm. A staged responder replies in
 * milliseconds; this exists so an attached-but-unstaged responder produces a
 * defined skip instead of hanging until the suite budget kills the deploy
 * (module-orchestrator-primitives design.md D5, "never a hang"). On a TTY
 * there is no bound: the operator IS the responder and may take minutes.
 */
const DNS_REPOINT_HEADLESS_TIMEOUT_MS = 60_000;

/**
 * The deploy-time DNS repoint for a `dns_internal` provider
 * (module-orchestrator-primitives slice 3, design D5). celilo owns the whole
 * operation because every input is already celilo's: the resolver address is
 * the system celilo just placed, and the prior primary/fallback are celilo's
 * own rows. knot's on_install used to do this through the CLI, one key at a
 * time, under a catch that reported success over a repoint that never
 * happened.
 *
 * D5 is a recorded DELIBERATE EXCEPTION to the framework-does-not-infer-intent
 * rule. What makes it acceptable is the interview: the deploy ASKS before it
 * touches the fleet resolver, and a declined repoint is an operator decision,
 * not an error. Do not cite this as precedent (design.md D5, "So do not cite
 * this as precedent").
 */
export interface DnsRepointPlan {
  serverIp: string;
  priorPrimary: string | undefined;
  existingFallback: string | undefined;
  /** Merged dns.fallback when the repoint demotes the prior primary; undefined leaves the key untouched. */
  newFallback: string | undefined;
  /** False when dns.primary already names this resolver: the deploy asks nothing. */
  needsRepoint: boolean;
}

/**
 * Pure half of the repoint. The demote keeps an existing fallback (D5's
 * secondary observation: knot's one-key-at-a-time hook overwrote it, dropping
 * e.g. 8.8.8.8 from the fleet's resolver list).
 */
export function planDnsRepoint(args: {
  serverIp: string;
  priorPrimary: string | undefined;
  existingFallback: string | undefined;
}): DnsRepointPlan {
  const { serverIp, priorPrimary, existingFallback } = args;
  if (priorPrimary === serverIp) {
    return {
      serverIp,
      priorPrimary,
      existingFallback,
      newFallback: undefined,
      needsRepoint: false,
    };
  }
  const newFallback = priorPrimary ? mergeFallback(existingFallback, priorPrimary) : undefined;
  return { serverIp, priorPrimary, existingFallback, newFallback, needsRepoint: true };
}

/** Comma-separated, demoted primary first, deduped (public-dns-probe.ts parses commas). */
function mergeFallback(existing: string | undefined, demoted: string): string {
  const entries = [demoted];
  for (const part of (existing ?? '').split(',')) {
    const ip = part.trim();
    if (ip && !entries.includes(ip)) entries.push(ip);
  }
  return entries.join(',');
}

function readSystemConfigValue(db: DbClient, key: string): string | undefined {
  const row = db.select().from(systemConfig).where(eq(systemConfig.key, key)).get();
  return row?.value && row.value.length > 0 ? row.value : undefined;
}

function writeSystemConfigValue(db: DbClient, key: string, value: string): void {
  db.insert(systemConfig)
    .values({ key, value })
    .onConflictDoUpdate({ target: systemConfig.key, set: { value } })
    .run();
}

/**
 * Ask, then perform (or skip) the repoint for a just-deployed `dns_internal`
 * provider. Exported for tests; `deployModuleImpl` calls this after the DNS
 * backfill when the deployed module provides `dns_internal`.
 */
export async function repointDnsPrimaryForProvider(
  moduleId: string,
  db: DbClient,
  ask: typeof askConfirm = askConfirm,
): Promise<void> {
  const serverIp = getModuleSystems(moduleId, db)[0]?.ipv4_address;
  if (!serverIp) {
    log.warn(
      `dns_internal provider '${moduleId}' has no deployed system — skipping dns.primary repoint`,
    );
    return;
  }
  const plan = planDnsRepoint({
    serverIp,
    priorPrimary: readSystemConfigValue(db, 'dns.primary'),
    existingFallback: readSystemConfigValue(db, 'dns.fallback'),
  });
  if (!plan.needsRepoint) {
    log.info(`dns.primary already points at ${serverIp} — no repoint needed`);
    return;
  }

  const demoteNote = plan.newFallback
    ? ` The old primary ${plan.priorPrimary} moves to dns.fallback (${plan.newFallback}).`
    : '';
  const opts = {
    scope: `deploy:${moduleId}`,
    key: 'repoint_dns_primary',
    message: `Deploying '${moduleId}' made it the fleet's internal DNS resolver. Point dns.primary at ${serverIp}?${demoteNote}`,
    defaultValue: false,
  };
  // Called DIRECTLY, not inside withInterviewSession: deployModuleImpl already
  // opened the terminal responder above and closes it in its finally. A second
  // responder instance would race the first on stdin (design.md D5).
  let confirmed: boolean;
  try {
    confirmed = await ask(
      process.stdin.isTTY ? opts : { ...opts, timeoutMs: DNS_REPOINT_HEADLESS_TIMEOUT_MS },
    );
  } catch (error) {
    if (error instanceof InterviewUnansweredError || error instanceof InterviewAbandonedError) {
      log.warn(
        `dns.primary repoint skipped — nobody answered the confirm (${opts.scope}:${opts.key}). Stage the answer (celilo events respond --values) or set dns.primary manually later.`,
      );
      return;
    }
    throw error;
  }

  if (!confirmed) {
    log.info('dns.primary repoint declined — fleet resolver left unchanged');
    return;
  }

  writeSystemConfigValue(db, 'dns.primary', serverIp);
  if (plan.newFallback !== undefined) {
    writeSystemConfigValue(db, 'dns.fallback', plan.newFallback);
  }
  log.success(
    `dns.primary repointed at internal resolver ${serverIp}${plan.newFallback ? ` (dns.fallback: ${plan.newFallback})` : ''}`,
  );
}

async function deployModuleImpl(
  moduleId: string,
  db: DbClient,
  options: DeployOptions = {},
): Promise<DeployResult> {
  const phases: DeployResult['phases'] = {};

  // When stdout isn't a TTY (cele2e subprocess, CI), emit structured
  // [progress:*] markers that cele2e parses. --verbose forces render
  // mode but with isTTY=false, which disables cursor magic — every
  // sub-event stays visible after each step completes (no collapse-
  // on-success). On a TTY without --verbose, mode is 'auto' and the
  // ProgressDisplay picks animated gauges. Non-TTY wins if both are
  // set (protocol output is what cele2e expects regardless).
  const nonTTY = !process.stdout.isTTY;
  const displayMode = nonTTY ? 'protocol' : options.verbose ? 'render' : 'auto';
  const display = new ProgressDisplay({
    mode: displayMode,
    out: options.verbose
      ? { write: process.stdout.write.bind(process.stdout), isTTY: false }
      : undefined,
  });
  setActiveDisplay(display);

  // Terminal-responder: when running on a TTY, this subscribes to
  // `config.required.*` / `secret.required.*` / `ensure.required.*`
  // events and prompts on the terminal. Other responder shapes (Claude
  // subagent, `celilo events respond` from another shell, autoresponder
  // daemon) compete on the bus; first reply wins. See
  // infra/openspec/changes/interactive-deploys-via-event-bus/proposal.md.
  const terminalResponder = process.stdin.isTTY
    ? (await import('./terminal-responder')).startTerminalResponder()
    : null;

  try {
    // Live and e2e environments are mutually exclusive — refuse if an e2e
    // stack is up (shared docker networks/hostnames). See e2e-guard.ts.
    // Pre-flight checks the same condition; this is the defense-in-depth
    // guard for callers that skip pre-flight.
    if (runningE2eContainers().length > 0) {
      return { success: false, error: E2E_CONFLICT_MESSAGE, phases };
    }

    // Every network this module REQUIRES must be defined before ANYTHING else
    // (openspec/changes/networks-are-declared-not-written/specs/network-declaration/spec.md).
    //
    // First, not merely "before the hooks". `validateAndPrepareDeployment`
    // generates templates itself when no variable is missing, and generation is
    // where `$system:` derivations resolve and get persisted. A module reading
    // `network.<name>.subnet` would derive it BEFORE the network existed, get
    // nothing, and — because generation does not run twice — never get it. The
    // hook would then fail on an unset value the deploy had just been told to
    // supply. Ensuring the network up front is what makes "the value is
    // available before any hook runs" true rather than nearly true.
    const declaringModule = await db.select().from(modules).where(eq(modules.id, moduleId)).get();
    if (declaringModule?.manifestData) {
      const networksEnsured = await ensureRequiredNetworks(
        moduleId,
        declaringModule.manifestData as ModuleManifest,
        db,
      );
      if (!networksEnsured.success) {
        return { success: false, error: networksEnsured.error, phases };
      }
      for (const line of networksEnsured.applied) {
        log.success(`network defined: ${line}`);
      }
    }

    const validation = await validateAndPrepareDeployment(moduleId, db);
    phases.validation = validation.success;
    phases.autoGenerated = validation.autoGenerated;
    if (!validation.success) {
      return {
        success: false,
        error: validation.error,
        phases,
      };
    }

    // Get module and paths (needed for secret generation)
    const module = await db.select().from(modules).where(eq(modules.id, moduleId)).get();
    if (!module) {
      return {
        success: false,
        error: `Module '${moduleId}' not found`,
        phases,
      };
    }

    const generatedPath = `${module.sourcePath}/generated`;
    const manifest = module.manifestData as ModuleManifest;

    // Handle missing required variables
    if (validation.missingVariables && validation.missingVariables.length > 0) {
      // Separate auto-generatable secrets from other missing variables
      const autoGeneratableSecrets = validation.missingVariables.filter(
        (v) => v.source === 'secret' && v.generate,
      );
      let remainingMissing = validation.missingVariables.filter(
        (v) => !(v.source === 'secret' && v.generate),
      );

      // Auto-generate secrets that have a generate field (works in both interactive and non-interactive)
      if (autoGeneratableSecrets.length > 0) {
        const autoGenResult = await interviewForMissingSecrets(
          moduleId,
          autoGeneratableSecrets,
          db,
        );
        if (!autoGenResult.success) {
          return {
            success: false,
            error: autoGenResult.error || 'Secret auto-generation failed',
            phases,
          };
        }
        log.success(`Auto-generated ${autoGenResult.configured.length} secret(s)`);

        // Regenerate secrets.yml with newly auto-generated secrets
        const { generateAnsibleSecrets } = await import('../ansible/secrets');
        const ansiblePath = join(generatedPath, 'ansible');
        const secretsYamlPath = join(ansiblePath, 'inventory', 'group_vars', 'all', 'secrets.yml');
        await generateAnsibleSecrets(moduleId, secretsYamlPath, db);
      }

      // Always auto-derive $machine: variables — even in non-interactive mode.
      // Machine data (interfaces, zone IPs) is catalogued at machine add time and
      // should never require user prompting.
      const machineDerivable = remainingMissing.filter((v) =>
        v.derive_from?.startsWith('$machine:'),
      );
      if (machineDerivable.length > 0) {
        const isFirewall = manifest.provides?.capabilities?.some((cap) => cap.name === 'firewall');
        const moduleZone = getSingularSystemSpec(manifest)?.zone;
        const matchedMachine = await findMachineForModule(
          moduleId,
          moduleZone,
          isFirewall ? 'router' : undefined,
        );
        if (matchedMachine) {
          const derived = await autoDeriveMachineConfig(
            moduleId,
            machineDerivable,
            db,
            matchedMachine,
          );
          remainingMissing = remainingMissing.filter((v) => !derived.configured.includes(v.name));
        }
      }

      // Bus-mediated interview: any remaining missing config produces
      // `config.required.*` / `secret.required.*` events. A responder
      // (terminal, Claude subagent, `events respond`) answers; the
      // deploy waits indefinitely if none does. Operators see stuck
      // queries via `celilo events list-pending`.
      // Separate secrets from regular config.
      const secrets = remainingMissing.filter((v) => v.source === 'secret');
      const regularConfig = remainingMissing.filter((v) => v.source !== 'secret');

      // Handle secrets with schema-aware interview
      if (secrets.length > 0) {
        const secretResult = await interviewForMissingSecrets(moduleId, secrets, db);
        if (!secretResult.success) {
          return {
            success: false,
            error: secretResult.error || 'Secret configuration failed',
            phases,
          };
        }
        log.success(`Configured ${secretResult.configured.length} secret(s)`);

        // Regenerate secrets.yml now that secrets exist
        // Place in group_vars/all/ so Ansible auto-loads vault-encrypted vars
        const { generateAnsibleSecrets } = await import('../ansible/secrets');
        const ansiblePath = join(generatedPath, 'ansible');
        const secretsYamlPath = join(ansiblePath, 'inventory', 'group_vars', 'all', 'secrets.yml');
        const secretsFileResult = await generateAnsibleSecrets(moduleId, secretsYamlPath, db);
        if (!secretsFileResult.success) {
          return {
            success: false,
            error: `Failed to generate secrets.yml: ${secretsFileResult.error}`,
            phases,
          };
        }
      }

      // Handle regular config with standard interview
      if (regularConfig.length > 0) {
        // Look up machine for $machine: derivation (earmarked or best match from pool)
        const isFirewall = manifest.provides?.capabilities?.some((cap) => cap.name === 'firewall');
        const moduleZone = getSingularSystemSpec(manifest)?.zone;
        const matchedMachine = await findMachineForModule(
          moduleId,
          moduleZone,
          isFirewall ? 'router' : undefined,
        );
        const configResult = await interviewForMissingConfig(
          moduleId,
          regularConfig,
          db,
          matchedMachine,
        );

        if (!configResult.success) {
          return {
            success: false,
            error: configResult.error || 'Configuration failed',
            phases,
          };
        }
        log.success(`Configured ${configResult.configured.length} variable(s)`);
      }
    }

    // --stop-after-interview: bail before any infrastructure work.
    // The bus-mediated interview has fired and any responders have
    // answered; the operator can inspect the encrypted store and
    // module config to confirm what landed without spinning up
    // terraform / ansible / actual hooks. The required-network
    // interview above HAS run by this point; the cross-module
    // `ensure` events fire later from hook execution, so those are
    // NOT exercised here — that requires a real run.
    if (options.stopAfterInterview) {
      log.info('--stop-after-interview: exiting before infrastructure phase.');
      return {
        success: true,
        phases: { ...phases, validation: true },
      };
    }

    // If validation returned early (missing variables were present), generation
    // was deferred until after the interview. Run it now that all vars are set.
    if (!validation.autoGenerated) {
      const { generateTemplates } = await import('../templates/generator');
      const regenResult = await generateTemplates({
        moduleId,
        modulePath: module.sourcePath,
        outputPath: generatedPath,
        db,
        skipVariableValidation: false,
      });
      if (!regenResult.success) {
        return {
          success: false,
          error: `Generation failed: ${regenResult.error || 'Unknown error'}`,
          phases,
        };
      }
    }

    log.success('Templates generated');

    // The generated-tree staleness pre-flight (D6 of module-integrity-rigor)
    // retired here (D5 of control-plane-stops-building-modules): the tree is
    // rendered for THIS deploy and deleted when it succeeds, so nothing
    // persists that could be stale relative to what ships.

    // Run validate_config hook if defined (e.g., credential validation via Playwright)
    if (manifest.hooks?.validate_config) {
      const hookDef = manifest.hooks.validate_config;
      const gauge = new FuelGauge('Validating configuration', {
        skipAnimation: !process.stdout.isTTY,
      });
      gauge.start();
      const hookLogger = createGaugeLogger(gauge, moduleId, 'validate_config');

      try {
        // Build config and secrets for hook context
        const { moduleConfigs: moduleConfigsTable, secrets: secretsTable } = await import(
          '../db/schema'
        );
        const configs = await db
          .select()
          .from(moduleConfigsTable)
          .where(eq(moduleConfigsTable.moduleId, moduleId))
          .all();
        const configMap: Record<string, unknown> = {};
        for (const c of configs) {
          configMap[c.key] = c.valueJson ? JSON.parse(c.valueJson) : c.value;
        }

        const secretRecords = await db
          .select()
          .from(secretsTable)
          .where(eq(secretsTable.moduleId, moduleId))
          .all();
        const masterKey = await getOrCreateMasterKey();
        const secretMap: Record<string, string> = {};
        for (const s of secretRecords) {
          secretMap[s.name] = decryptSecret(
            { encryptedValue: s.encryptedValue, iv: s.iv, authTag: s.authTag },
            masterKey,
          );
        }

        // Load capability functions for this module (e.g., dns_registrar).
        // hookLogger is captured by the auto-logging wrapper (HOOK_API_V2 D6).
        hookLogger.info('Loading capability functions...');
        const capabilityFunctions = await loadCapabilityFunctions(moduleId, db, hookLogger);
        const capNames = Object.keys(capabilityFunctions);
        hookLogger.info(
          `Loaded capabilities: ${capNames.length > 0 ? capNames.join(', ') : '(none)'}`,
        );

        const hookResult = await invokeHook(
          module.sourcePath,
          'validate_config',
          manifest.celilo_contract,
          hookDef,
          {},
          configMap,
          secretMap,
          hookLogger,
          {
            debug: options.debug,
            capabilities: capabilityFunctions,
            requiredCapabilities: manifest.requires.capabilities.map((c) => c.name),
            systems: getModuleSystems(moduleId, db),
            remoteAccess: remoteAccessPolicy(moduleId, db),
            hookStores: () => createHookStores(db, moduleId),
          },
        );

        if (!hookResult.success) {
          gauge.stop(false);
          const errorMsg =
            (hookResult.error || 'Configuration validation failed') +
            describeArtifacts(hookResult.artifactPaths);
          return {
            success: false,
            error: errorMsg,
            phases,
          };
        }

        // Hook return values are no longer persisted (hook-owned-state D5):
        // a hook that needs to persist state does so explicitly through the
        // `context.config` / `context.secrets` accessors inside its own
        // process. Nothing here sweeps `hookResult.outputs` into secrets.

        gauge.stop(true);
      } catch (error) {
        gauge.stop(false);
        return {
          success: false,
          error: `Configuration validation failed: ${error instanceof Error ? error.message : String(error)}`,
          phases,
        };
      }
    }

    // Config-only modules (no infrastructure requirements) don't need terraform/ansible
    const isConfigOnly = !getSingularSystemSpec(manifest);
    if (isConfigOnly) {
      log.success('Config-only module — no infrastructure deployment needed');

      // Run on_install hook for config-only modules (e.g. publishing static files to caddy)
      if (manifest.hooks?.on_install) {
        const onInstallDef = manifest.hooks.on_install;
        log.success('Running on_install hook');

        const { moduleConfigs: pcTable, secrets: secretsTable } = await import('../db/schema');
        const installConfigs = db
          .select()
          .from(pcTable)
          .where(eq(pcTable.moduleId, moduleId))
          .all();
        const installConfigMap: Record<string, unknown> = {};
        for (const c of installConfigs) {
          installConfigMap[c.key] = c.valueJson ? JSON.parse(c.valueJson) : c.value;
        }

        const installContext = await buildResolutionContext(moduleId, db);
        for (const [key, value] of Object.entries(installContext.selfConfig)) {
          if (
            !(key in installConfigMap) ||
            (typeof installConfigMap[key] === 'string' &&
              (installConfigMap[key] as string).startsWith('$'))
          ) {
            installConfigMap[key] = value;
          }
        }

        const installSecrets = db
          .select()
          .from(secretsTable)
          .where(eq(secretsTable.moduleId, moduleId))
          .all();
        const installMasterKey = await getOrCreateMasterKey();
        const installSecretMap: Record<string, string> = {};
        for (const s of installSecrets) {
          installSecretMap[s.name] = decryptSecret(
            { encryptedValue: s.encryptedValue, iv: s.iv, authTag: s.authTag },
            installMasterKey,
          );
        }

        const installResult = await invokeHookWithEnsureRetry(
          module.sourcePath,
          'on_install',
          manifest.celilo_contract,
          onInstallDef,
          {},
          installConfigMap,
          installSecretMap,
          async () => {
            const gauge = new FuelGauge(`${moduleId}: on_install`, {
              skipAnimation: !process.stdout.isTTY,
            });
            gauge.start();
            const logger = createGaugeLogger(gauge, moduleId, 'on_install');
            const capFns = await loadCapabilityFunctions(moduleId, db, logger);
            return {
              gauge,
              logger,
              invokeOptions: {
                debug: options.debug,
                capabilities: capFns,
                requiredCapabilities: (manifest.requires?.capabilities ?? []).map((c) => c.name),
                systems: getModuleSystems(moduleId, db),
                remoteAccess: remoteAccessPolicy(moduleId, db),
                hookStores: () => createHookStores(db, moduleId),
              },
            };
          },
          db,
          options,
        );

        if (!installResult.success) {
          log.error(`on_install hook failed: ${installResult.error}`);
          log.error('Module remains in DEPLOYING state. Fix the issue and retry:');
          log.error(`  celilo module run-hook ${moduleId} on_install`);
          log.error(`  celilo module deploy ${moduleId}`);
          return { success: false, phases, error: installResult.error };
        }
      }

      // Transition to INSTALLED. A successful deploy ends any recorded
      // failure: leaving a stale errorMessage would print `Error:` beside a
      // healthy state in `module list` / `module status` (celilo#1363).
      db.update(modules)
        .set({ state: 'INSTALLED', errorMessage: null })
        .where(eq(modules.id, moduleId))
        .run();

      // Run health checks for config-only modules too
      if (manifest.hooks?.health_check) {
        const { runModuleHealthCheck } = await import('./health-runner');
        const healthResult = await runModuleHealthCheck(moduleId, db, {
          debug: options.debug,
          noInteractive: !process.stdout.isTTY,
        });
        if (healthResult.status === 'error') {
          return { success: false, phases, error: `Health check failed: ${healthResult.error}` };
        }
        if (healthResult.status === 'unhealthy') {
          const failedChecks = healthResult.checks
            .filter((c) => c.status === 'fail')
            .map((c) => `  ✗ ${c.name}: ${c.message}`)
            .join('\n');
          return { success: false, phases, error: `Health checks failed:\n${failedChecks}` };
        }
      }

      // Fan out the module's base-module aspect (if any) per the
      // on_install trigger. Failures here don't fail the primary
      // deploy — D4 in openspec/specs/base-module-aspects/spec.md: aspects are idempotent
      // and forward-progress; a partial fleet update is expected to
      // converge on the next fan-out. We log the result instead.
      const aspectOutcome = await maybeRunAspectForTrigger({
        moduleId,
        manifest,
        trigger: 'on_install',
        db,
      });
      if (aspectOutcome.ran && !aspectOutcome.success) {
        log.warn(
          `Base-module aspect fan-out for '${moduleId}' failed: ${aspectOutcome.runResult?.error ?? 'unknown'}. Primary deploy succeeded; re-run \`celilo fleet redeploy ${moduleId}\` after fixing.`,
        );
      } else if (aspectOutcome.ran) {
        log.success(`Base-module aspect fan-out for '${moduleId}' completed`);
      }

      // Whatever this module provides, the consumers that were already here
      // never got to ask it for anything. Re-run them so they register through
      // the path that worked the first time (design D7).
      await backfillArrivedProvider(moduleId, db);

      // Mirror the infrastructure-path success message at the end of
      // a successful deploy. Without this, config-only deploys end
      // abruptly with whatever the last hook line was — operator sees
      // no clear "this finished cleanly" signal.
      log.success(`Module '${moduleId}' deployed successfully`);
      return {
        success: true,
        phases: {
          ...phases,
          planning: true,
        },
      };
    }

    const plan = await planDeployment(moduleId, generatedPath, manifest, db);
    phases.planning = true;

    let terraformOutputs: Record<string, unknown> | null = null;
    if (plan.needsTerraform) {
      // Compose TF_VAR_* env vars from the bound container service's
      // credentials. (Empty for machine-pool deployments.)
      let terraformEnvVars: Record<string, string> = {};
      if (plan.infrastructure?.type === 'container_service' && plan.infrastructure.serviceId) {
        const service = await getContainerService(plan.infrastructure.serviceId);
        if (!service) {
          return {
            success: false,
            error: `Container service not found: ${plan.infrastructure.serviceId}`,
            phases,
          };
        }
        terraformEnvVars = await buildTerraformEnvForService(plan.infrastructure.serviceId);

        // Fail fast if the proxmox API host is unreachable (e.g. VPN
        // down). The terraform provider would otherwise stall on a
        // SYN_SENT connect for ~60s with no visible feedback.
        if (service.providerName === 'proxmox' && terraformEnvVars.TF_VAR_proxmox_api_url) {
          const probe = await checkProxmoxReachable(terraformEnvVars.TF_VAR_proxmox_api_url);
          if (!probe.reachable) {
            return {
              success: false,
              error: formatProxmoxUnreachableError(probe),
              phases,
            };
          }
        }
      }

      const terraformResult = await executeTerraform(generatedPath, phases, terraformEnvVars, {
        noInteractive: !process.stdout.isTTY,
      });

      if (!terraformResult.success) {
        return {
          success: false,
          error: terraformResult.error || 'Terraform execution failed',
          phases,
        };
      }

      // Parse Terraform outputs for infrastructure variable resolution
      const terraformDir = join(generatedPath, 'terraform');
      terraformOutputs = await parseTerraformOutputs(terraformDir);

      // Recovery: For Proxmox, ensure vmid/container_ip are in module_configs
      // This handles state drift scenarios where container was deleted and recreated
      if (plan.infrastructure?.type === 'container_service' && plan.infrastructure.serviceId) {
        const service = await getContainerService(plan.infrastructure.serviceId);
        if (service && service.providerName === 'proxmox') {
          const { ensureProxmoxConfigFromState } = await import('./proxmox-state-recovery');
          await ensureProxmoxConfigFromState(moduleId, terraformDir, db);
        }
      }
    } else {
      log.success('Infrastructure up-to-date (skipping Terraform)');
      phases.terraformInit = true;
      phases.terraformPlan = true;
      phases.terraformApply = true;
    }

    // Resolve Infrastructure Variables
    // This happens after Terraform (if applicable) but before Ansible
    // - Machine infrastructure: reads from machine record
    // - Proxmox container: reads from IPAM allocation (from generate phase)
    // - Digital Ocean container: reads from Terraform outputs (from above)
    const resolution = await resolveInfrastructureVariables(
      moduleId,
      manifest,
      terraformOutputs,
      db,
    );

    // Record the module's deployed system(s) now that the IP is known for every
    // provider type (machine / proxmox IPAM / DO outputs). This populates
    // ctx.systems for on_install and is the source of truth for system.created
    // and DNS (openspec/specs/module-systems-addressing/spec.md). API-only modules record none.
    const { recordDeployedSystemForModule } = await import('./deployed-systems');
    const recordedSystems = await recordDeployedSystemForModule(
      moduleId,
      manifest,
      plan.infrastructure,
      db,
    );
    if (recordedSystems.length > 0) {
      log.success(
        `Recorded ${recordedSystems.length} deployed system(s): ${recordedSystems.map((s) => `${s.hostname} (${s.ipv4_address})`).join(', ')}`,
      );
    }

    if (Object.keys(resolution.resolved).length > 0) {
      // Build complete message with all variables
      const lines = ['Infrastructure variables resolved:'];
      for (const [key, value] of Object.entries(resolution.resolved)) {
        lines.push(`  ${key} = ${value}`);
      }
      if (resolution.skipped.length > 0) {
        // These are infrastructure-managed vars with no value this deploy (e.g.
        // vmid/target_node on a machine deploy). NOT honored operator overrides —
        // such keys are rejected at `config set` (ISS-0069); don't imply otherwise.
        lines.push(
          `  (auto-managed, not applicable this deploy: ${resolution.skipped.join(', ')})`,
        );
      }
      log.success(lines.join('\n'));

      // Regenerate Ansible inventory with resolved infrastructure variables
      const inventoryResult = await generateInventory(
        moduleId,
        generatedPath,
        db,
        plan.infrastructure,
      );

      if (!inventoryResult.success) {
        log.error(`Inventory generation error: ${inventoryResult.error}`);
        if (inventoryResult.details) {
          log.error(`Error details: ${inventoryResult.details}`);
        }
        return {
          success: false,
          error: `Failed to regenerate inventory: ${inventoryResult.error}`,
          phases,
        };
      }
      log.success('Ansible inventory updated');
    }

    // Invoke container_created hooks on capability providers
    // This runs after Terraform (VPS exists) and after infrastructure variable resolution (IP known)
    // Example: dns-external requires dns_registrar → namecheap's container_created hook
    //          gets called with the VPS IP to enable API access and whitelist the IP
    if (manifest.requires?.capabilities && manifest.requires.capabilities.length > 0) {
      const { moduleConfigs: pcTable } = await import('../db/schema');
      const ipConfig = db
        .select()
        .from(pcTable)
        .where(and(eq(pcTable.moduleId, moduleId), eq(pcTable.key, 'ip.primary')))
        .get();
      const vpsIp = ipConfig?.value || '';

      for (const requiredCap of manifest.requires.capabilities) {
        // Find the provider module
        const capRecord = db
          .select()
          .from(capabilities)
          .where(eq(capabilities.capabilityName, requiredCap.name))
          .get();

        if (!capRecord) continue;

        const providerModule = db
          .select()
          .from(modules)
          .where(eq(modules.id, capRecord.moduleId))
          .get();

        if (!providerModule) continue;

        const providerManifest = providerModule.manifestData as ModuleManifest;
        const containerCreatedHook = providerManifest.hooks?.container_created;

        if (!containerCreatedHook) continue;

        log.message(
          `Running container_created hook on ${capRecord.moduleId} (provides ${requiredCap.name})`,
        );

        const gauge = new FuelGauge(`${capRecord.moduleId}: container_created`, {
          skipAnimation: !process.stdout.isTTY,
        });
        gauge.start();
        const hookLogger = createGaugeLogger(gauge, capRecord.moduleId, 'container_created');

        // Build provider's config and secrets
        const { secrets: secretsTable } = await import('../db/schema');
        const providerConfigs = db
          .select()
          .from(pcTable)
          .where(eq(pcTable.moduleId, capRecord.moduleId))
          .all();
        const providerConfigMap: Record<string, unknown> = {};
        for (const c of providerConfigs) {
          providerConfigMap[c.key] = c.valueJson ? JSON.parse(c.valueJson) : c.value;
        }

        const providerSecretRecords = db
          .select()
          .from(secretsTable)
          .where(eq(secretsTable.moduleId, capRecord.moduleId))
          .all();
        const providerMasterKey = await getOrCreateMasterKey();
        const providerSecretMap: Record<string, string> = {};
        for (const s of providerSecretRecords) {
          providerSecretMap[s.name] = decryptSecret(
            {
              encryptedValue: s.encryptedValue,
              iv: s.iv,
              authTag: s.authTag,
            },
            providerMasterKey,
          );
        }

        const capFunctions = await loadCapabilityFunctions(capRecord.moduleId, db, hookLogger);

        const hookResult = await invokeHook(
          providerModule.sourcePath,
          'container_created',
          providerManifest.celilo_contract,
          containerCreatedHook,
          { vps_ip: vpsIp },
          providerConfigMap,
          providerSecretMap,
          hookLogger,
          {
            debug: options.debug,
            capabilities: capFunctions,
            requiredCapabilities: providerManifest.requires.capabilities.map((c) => c.name),
            systems: getModuleSystems(capRecord.moduleId, db),
            remoteAccess: remoteAccessPolicy(capRecord.moduleId, db),
            hookStores: () => createHookStores(db, capRecord.moduleId),
          },
        );

        if (!hookResult.success) {
          gauge.stop(false);
          return {
            success: false,
            error: `${capRecord.moduleId} container_created hook failed: ${hookResult.error}`,
            phases,
          };
        }

        // The return-value channel is gone (hook-owned-state D5/D3): hook
        // outputs are no longer persisted anywhere. A provider secret the
        // framework itself must record goes through the named
        // `recordProviderSecret` path (secrets/storage.ts) — never through a
        // hook return value.

        gauge.stop(true);
      }
    }

    if (plan.needsSSHWait) {
      // Re-extract target host to get infrastructure-derived IP (ip.primary)
      // This must happen AFTER infrastructure variable resolution
      const { extractTargetHost } = await import('./deploy-planner');
      const targetHost = await extractTargetHost(moduleId, db);
      // Longer timeout for cloud providers (Digital Ocean droplets can take 2-3 minutes)
      const sshResult = await waitForSSH(targetHost.ip, targetHost.user, 180);
      phases.sshWait = sshResult.success;
      if (!sshResult.success) {
        return {
          success: false,
          error: sshResult.error || 'SSH connection failed',
          phases,
        };
      }
    } else {
      phases.sshWait = true;
    }

    // Apply the fleet's aspects to the systems THIS deploy just created, before
    // the module's own playbook runs against them (celilo#902, design D3).
    //
    // A base-module aspect fans out to the systems that exist when its PROVIDER
    // deploys and to nothing afterwards, so a host provisioned here would keep
    // its boot-time /etc/resolv.conf and be unable to resolve any fleet-internal
    // name — silently, with a green deploy on top. This is the inbound half.
    //
    // Placed here on purpose: after `waitForSSH`, so the host is proven
    // reachable and the aspect's Ansible does not fail UNREACHABLE on a
    // still-booting LXC; before `executeAnsible`, so the module's playbook and
    // its `on_install` hook see a correctly configured host rather than
    // configuring themselves on top of a broken one.
    //
    // FAILURE IS FATAL HERE, and that is DELIBERATELY the opposite of the
    // outbound direction below, where a failed fan-out never fails the
    // provider's own deploy (D4). The asymmetry is the point and is not an
    // oversight to be tidied up: an inbound aspect is a PREREQUISITE of the host
    // being configured right now, so continuing would configure a host celilo
    // knows is misconfigured — exactly the outcome celilo#902 produced. Nothing
    // is rolled back: the rows, the guest and the IPAM allocation all persist,
    // and re-running this deploy after fixing the cause converges on the same
    // system.
    if (recordedSystems.length > 0) {
      const { reconcileAspectsForSystems } = await import('./aspect-runner');
      const reconcile = await reconcileAspectsForSystems({
        systems: recordedSystems.map((sys) => ({ hostname: sys.hostname, zone: sys.zone })),
        db,
        // This module's own aspect is fanned out by `on_install` below, across
        // the whole fleet rather than only these hosts.
        excludeModuleIds: [moduleId],
      });
      if (reconcile.failures.length > 0) {
        // Name the PROVIDING module, not the one being deployed — otherwise
        // this reads as a defect in `moduleId` and the operator debugs the
        // wrong thing.
        const detail = reconcile.failures
          .map(
            (f) =>
              `  ✗ '${f.providerModuleId}' aspect '${f.role}' on ${f.hostnames.join(', ')}: ${f.error ?? 'unknown error'}`,
          )
          .join('\n');
        return {
          success: false,
          phases,
          error: `Fleet aspects could not be applied to the system(s) this deploy created:\n${detail}\nThese are prerequisites, so '${moduleId}' was not configured on top of them. Fix the cause and re-run \`celilo module deploy ${moduleId}\` — the system is kept and will be reused.`,
        };
      }
      const applied = reconcile.outcomes.filter((o) => o.ran);
      if (applied.length > 0) {
        log.success(
          `Applied ${applied.length} fleet aspect(s) to the new system(s): ${applied.map((o) => `${o.providerModuleId}/${o.role}`).join(', ')}`,
        );
      }
    }

    let machineId: string | undefined;
    if (plan.infrastructure?.type === 'machine' && plan.infrastructure.machineId) {
      // The management box deploys to ITSELF (registered in the machine pool as
      // 127.0.0.1) over Ansible's local connection — it has no stored SSH key,
      // and pinning one here throws before Ansible ever runs. `inventory.ts` and
      // `aspect-runner.ts` both already skip the key for the local box; this call
      // site did not, so celilo-mgmt could not complete a self-deploy.
      const machine = db
        .select()
        .from(machines)
        .where(eq(machines.id, plan.infrastructure.machineId))
        .get();
      if (machine?.ipAddress !== LOCAL_MACHINE_IP) {
        machineId = plan.infrastructure.machineId;
        await writeTemporarySshKey(machineId);
      }
    }

    // A public_web provider's deploy no longer plans the fleet's static content
    // here. Core used to walk every declared route row, resolve every module's
    // built site, and throw for the whole deploy when any one of them was
    // missing — celilo#1383, where two out-of-tree modules made the public
    // ingress undeployable. The provider now renders its own desired state in
    // its hook and hands it to the converge, which runs the same role tasks
    // (providers-converge-declared-state, design D4/D6). on_install runs
    // immediately after this play, so /srv/www still converges on a rebuilt
    // host — through one path instead of two.
    //
    // A provider whose role predates that converge would deploy cleanly and
    // then never converge its static content again: core no longer writes the
    // release set, and the old role has no task that reads what the provider
    // rendered. Nothing would report it — the deploy succeeds, the publish
    // succeeds, and /srv/www silently stops tracking what is declared. So the
    // version skew is refused here, by the same tree scan the converge itself
    // uses, and the message names the upgrade that fixes it.
    if (manifest.provides?.capabilities?.some((cap) => cap.name === 'public_web')) {
      const { ansibleTreeMentions } = await import('./provider-converge');
      const ansiblePath = join(generatedPath, 'ansible');
      if (existsSync(ansiblePath) && !ansibleTreeMentions(ansiblePath, 'provider_config_files')) {
        return {
          success: false,
          error: `The installed version of '${moduleId}' predates the provider converge: its role never reads provider_config_files, so celilo would deploy it and then have no way to place the config or the sites it renders. Upgrade it first (\`celilo module upgrade ${moduleId}\`), then deploy.`,
          phases,
        };
      }
    }

    try {
      const ansibleResult = await executeAnsible(generatedPath, {
        noInteractive: !process.stdout.isTTY,
      });
      phases.ansible = ansibleResult.success;
      if (!ansibleResult.success) {
        return {
          success: false,
          error: ansibleResult.error || 'Ansible deployment failed',
          phases,
        };
      }

      if (plan.infrastructure?.type === 'container_service' && plan.infrastructure.serviceId) {
        // Placeholder branch retained below; the machine branch is gone because
        // occupancy is no longer a stored fact to update (celilo#773).
      }
      if (plan.infrastructure?.type === 'container_service' && plan.infrastructure.serviceId) {
        // TODO: extract Terraform outputs and persist them on
        // module_infrastructure.containerMetadata. Until that lands,
        // the deploy still succeeds — we just don't track which
        // container ID/IP got assigned per module in the DB.
      }

      // celilo initialises its own box, in its own process.
      //
      // This was `celilo-mgmt`'s `on_install` hook until celilo#1225. The hook
      // reached every one of these operations by spawning the `celilo` CLI,
      // which a jailed hook cannot do — the mount set binds no `/usr/bin`, no
      // `/usr/local/bin` and no shell — so the deploy ran Ansible to completion
      // and then died in its own install hook on any host with a jail backend.
      //
      // It runs at the point the hook used to, for the reason the hook ran
      // there: the Ansible role has just installed celilo and started the
      // dispatcher, and none of this can be read before that.
      if (moduleId === CONTROL_PLANE_MODULE_ID) {
        log.success('Initializing celilo management state');
        try {
          const bootstrap = await bootstrapControlPlane({ db });

          log.info(`  DNS: primary ${bootstrap.dns.primary}, fallback ${bootstrap.dns.fallback}`);
          for (const applied of bootstrap.network.applied) {
            log.info(`  Network: ${applied}`);
          }
          if (bootstrap.network.skipped) {
            log.warn(`  Network not recorded: ${bootstrap.network.skipped}`);
          }
          // The operator authorises this key on every machine celilo manages,
          // so it is printed rather than merely stored. It was the loudest line
          // the hook produced and it stays the loudest line here.
          log.success(
            `celilo fleet SSH key — add this public key to machines celilo will manage:\n${bootstrap.fleetKey.publicKey}`,
          );

          if (bootstrap.dispatcher.status === 'fail') {
            // A management plane with no dispatcher is broken: event-driven
            // reconciles never deliver. Fail the install, do not whisper.
            log.error(`Dispatcher check failed: ${bootstrap.dispatcher.summary}`);
            return {
              success: false,
              phases,
              error:
                `celilo-mgmt: ${bootstrap.dispatcher.summary}. ${bootstrap.dispatcher.remediation ?? ''}`.trim(),
            };
          }
          log.success(`Event dispatcher is live (${bootstrap.dispatcher.summary})`);
        } catch (err) {
          const detail = err instanceof Error ? err.message : String(err);
          log.error(`Initializing celilo management state failed: ${detail}`);
          return { success: false, phases, error: detail };
        }
      }

      // Run on_install hook (post-deploy actions like port forwarding, DNS registration)
      if (manifest.hooks?.on_install) {
        const onInstallDef = manifest.hooks.on_install;
        log.success('Running on_install hook');

        const { moduleConfigs: pcTable, secrets: secretsTable } = await import('../db/schema');
        const installConfigs = db
          .select()
          .from(pcTable)
          .where(eq(pcTable.moduleId, moduleId))
          .all();
        const installConfigMap: Record<string, unknown> = {};
        for (const c of installConfigs) {
          installConfigMap[c.key] = c.valueJson ? JSON.parse(c.valueJson) : c.value;
        }

        // Resolve capability-derived variables (e.g. source: capability with derive_from).
        // buildResolutionContext resolves $self: refs in capability data and applies
        // declarative derivations so that $capability:dns_registrar.primary_domain
        // yields "iamtheinternet.org" rather than the raw "$self:primary_domain" template.
        const installContext = await buildResolutionContext(moduleId, db);
        for (const [key, value] of Object.entries(installContext.selfConfig)) {
          if (
            !(key in installConfigMap) ||
            (typeof installConfigMap[key] === 'string' &&
              (installConfigMap[key] as string).startsWith('$'))
          ) {
            installConfigMap[key] = value;
          }
        }

        // Inject the machine's IP as ip.primary for hooks that still read it
        // (firewall NAT target, etc.). The host's address for hooks now comes
        // from ctx.systems (openspec/specs/module-systems-addressing/spec.md), recorded into
        // module_systems during generate — no target_ip is written here.
        if (machineId) {
          const { getMachine } = await import('./machine-pool');
          const deployMachine = await getMachine(machineId);
          if (deployMachine) {
            installConfigMap['ip.primary'] = deployMachine.ipAddress;
          }
        }

        const installSecrets = db
          .select()
          .from(secretsTable)
          .where(eq(secretsTable.moduleId, moduleId))
          .all();
        const installMasterKey = await getOrCreateMasterKey();
        const installSecretMap: Record<string, string> = {};
        for (const s of installSecrets) {
          installSecretMap[s.name] = decryptSecret(
            { encryptedValue: s.encryptedValue, iv: s.iv, authTag: s.authTag },
            installMasterKey,
          );
        }

        const installResult = await invokeHookWithEnsureRetry(
          module.sourcePath,
          'on_install',
          manifest.celilo_contract,
          onInstallDef,
          {},
          installConfigMap,
          installSecretMap,
          async () => {
            const gauge = new FuelGauge(`${moduleId}: on_install`, {
              skipAnimation: !process.stdout.isTTY,
            });
            gauge.start();
            const logger = createGaugeLogger(gauge, moduleId, 'on_install');
            const capFns = await loadCapabilityFunctions(moduleId, db, logger);
            return {
              gauge,
              logger,
              invokeOptions: {
                debug: options.debug,
                capabilities: capFns,
                requiredCapabilities: manifest.requires.capabilities.map((c) => c.name),
                systems: getModuleSystems(moduleId, db),
                remoteAccess: remoteAccessPolicy(moduleId, db),
                hookStores: () => createHookStores(db, moduleId),
              },
            };
          },
          db,
          options,
        );

        if (!installResult.success) {
          log.error(`on_install hook failed: ${installResult.error}`);
          log.error('Module remains in DEPLOYED state. Fix the issue and retry:');
          log.error(`  celilo module run-hook ${moduleId} on_install`);
          log.error(`  celilo module deploy ${moduleId}`);
          return {
            success: false,
            phases,
            error: installResult.error,
          };
        }

        // Hook return values are no longer persisted (hook-owned-state D5).
        // caddy's `public_ip` moved to `context.config.set` + a config read
        // (task 5.1); no other hook may rely on its return reaching the store.
      }

      // Transition to INSTALLED; see the first transition for errorMessage (celilo#1363).
      db.update(modules)
        .set({ state: 'INSTALLED', errorMessage: null })
        .where(eq(modules.id, moduleId))
        .run();

      // Run health checks after successful deployment
      if (manifest.hooks?.health_check) {
        const { runModuleHealthCheck } = await import('./health-runner');
        const healthResult = await runModuleHealthCheck(moduleId, db, {
          debug: options.debug,
          noInteractive: !process.stdout.isTTY,
        });

        if (healthResult.status === 'error') {
          return {
            success: false,
            phases,
            error: `Health check failed: ${healthResult.error}`,
          };
        }
        if (healthResult.status === 'unhealthy') {
          const failedChecks = healthResult.checks
            .filter((c) => c.status === 'fail')
            .map((c) => `  ✗ ${c.name}: ${c.message}`)
            .join('\n');
          return {
            success: false,
            phases,
            error: `Health checks failed:\n${failedChecks}`,
          };
        }
      }

      // Auto-register module hostname in internal DNS (if available).
      // When a dns_internal provider deploys, backfill DNS for every
      // already-deployed system by invoking its own on_system_event hook per
      // host — the event path below only covers systems that deploy AFTER the
      // provider (deliveries bind at emit time). Non-providers skip this
      // entirely; their registration rides the system.created event below.
      // openspec/specs/event-driven-hook-subscriptions/spec.md.
      const { isDnsInternalProvider, backfillProviderDns, backfillWebRouteDns } = await import(
        './dns-provider-backfill'
      );
      if (isDnsInternalProvider(moduleId, db)) {
        // FuelGauge so the per-host hook invocations nest as sub-events under
        // one step rather than leaking to scrollback.
        const dnsGauge = new FuelGauge(`Backfilling internal DNS via ${moduleId}`, {
          skipAnimation: !process.stdout.isTTY,
        });
        dnsGauge.start();
        try {
          const dnsLogger = createGaugeLogger(dnsGauge, moduleId, 'dns_backfill');
          // Per-system records (bare hostnames) ...
          await backfillProviderDns(moduleId, db, dnsLogger);
          // ... and published web-route FQDNs (apt.celilo.computer), which the
          // per-system path doesn't cover (ISS-0029).
          await backfillWebRouteDns(moduleId, db, dnsLogger);
          dnsGauge.stop(true);
        } catch (error) {
          dnsGauge.stop(false);
          const msg = error instanceof Error ? error.message : String(error);
          log.warn(
            `DNS backfill failed for '${moduleId}': ${msg}. Some internal DNS records may need manual setup.`,
          );
        }

        // The dns_internal handoff (module-orchestrator-primitives slice 3,
        // design D5): celilo repoints dns.primary at the resolver it just
        // deployed, gated on an interview. knot's on_install no longer does
        // this. Runs after the backfill so a failed deploy never repoints the
        // fleet at a resolver that never came up.
        await repointDnsPrimaryForProvider(moduleId, db);
      }

      // Announce each deployed system on the bus (D5/D6) — one
      // system.created.<module> per host, so a dns_internal provider's
      // subscription registers every one (openspec/specs/module-systems-addressing/spec.md,
      // openspec/specs/internal-dns-split-horizon/spec.md, openspec/specs/event-driven-hook-subscriptions/spec.md).
      try {
        const { emitSystemCreated } = await import('./celilo-events');
        for (const sys of getModuleSystems(moduleId, db)) {
          emitSystemCreated({
            module: moduleId,
            hostname: sys.hostname,
            targetIp: sys.ipv4_address,
          });
        }
      } catch (error) {
        // Best-effort: a bus hiccup must not fail an otherwise-good deploy.
        const msg = error instanceof Error ? error.message : String(error);
        log.warn(`Failed to emit system.created for ${moduleId}: ${msg}`);
      }

      // Fan out the module's base-module aspect (if any) per the
      // on_install trigger. Failures here don't fail the primary
      // deploy — D4 in openspec/specs/base-module-aspects/spec.md: aspects are idempotent
      // and forward-progress; a partial fleet update is expected to
      // converge on the next fan-out. We log the result instead.
      const aspectOutcome = await maybeRunAspectForTrigger({
        moduleId,
        manifest,
        trigger: 'on_install',
        db,
      });
      if (aspectOutcome.ran && !aspectOutcome.success) {
        log.warn(
          `Base-module aspect fan-out for '${moduleId}' failed: ${aspectOutcome.runResult?.error ?? 'unknown'}. Primary deploy succeeded; re-run \`celilo fleet redeploy ${moduleId}\` after fixing.`,
        );
      } else if (aspectOutcome.ran) {
        log.success(`Base-module aspect fan-out for '${moduleId}' completed`);
      }

      // Whatever this module provides, the consumers that were already here
      // never got to ask it for anything. Re-run them so they register through
      // the path that worked the first time (design D7).
      await backfillArrivedProvider(moduleId, db);

      log.success(`Module '${moduleId}' deployed successfully`);
      return {
        success: true,
        phases,
      };
    } finally {
      // Clean up temporary SSH key
      if (machineId) {
        deleteTemporarySshKey(machineId);
      }
    }
  } catch (error) {
    return {
      success: false,
      error: error instanceof Error ? error.message : 'Unknown error',
      phases,
    };
  } finally {
    display.flush();
    setActiveDisplay(null);
    terminalResponder?.close();
  }
}