import { constants } from 'node:fs';
import {
  access,
  chmod,
  copyFile,
  mkdir,
  open,
  readFile,
  stat,
  unlink,
  writeFile,
} from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { homedir, hostname, userInfo } from 'node:os';
import { dirname, join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { spawn } from 'node:child_process';
import * as readline from 'node:readline';
import type { Command } from 'commander';
import YAML from 'yaml';
import { resolveCommsBlock } from '../fleet/comms-onboarding.js';
import { registerFleetBacklogCommand } from './fleet-backlog.js';
import { registerFleetPersonaCommand } from './fleet-personas.js';
import { registerFleetProfileCommand } from './fleet-profiles.js';
import { registerFleetProvisionCommand } from './fleet-provision.js';

/**
 * A function that spawns a command with inherited stdio (TTY passthrough).
 * Used for interactive commands like `tmux attach` that need a real terminal.
 * Resolves with the process exit code.
 */
export type InteractiveRunner = (command: string, args: string[]) => Promise<number>;

export interface CommandResult {
  stdout: string;
  stderr: string;
  exitCode: number;
}

export type CommandRunner = (command: string, args: string[]) => Promise<CommandResult>;

/**
 * Injectable sleep helper used by the send --verify polling loop.
 * Tests stub this to avoid real delays; production uses the default
 * implementation backed by setTimeout.
 */
export type SleepFn = (ms: number) => Promise<void>;

export interface FleetCommandDeps {
  runner?: CommandRunner;
  /** Injectable interactive runner for commands needing inherited TTY (e.g., `tmux attach`). */
  interactiveRunner?: InteractiveRunner;
  /**
   * Injectable sleep function for the send --verify polling loop.
   * Defaults to a real setTimeout-based sleep. Tests stub this to avoid
   * real delays; the default is used in production.
   */
  sleepFn?: SleepFn;
  mosaicHome?: string;
  frameworkRoot?: string;
  /**
   * Injectable TTY check for `fleet init` wizard. Defaults to process.stdin.isTTY.
   * Tests stub this to simulate interactive or non-interactive environments.
   */
  isStdinTTY?: boolean;
}

interface RawFleetRoster {
  version?: unknown;
  transport?: unknown;
  tmux?: {
    socket_name?: unknown;
    socketName?: unknown;
    holder_session?: unknown;
    holderSession?: unknown;
  };
  defaults?: {
    working_directory?: unknown;
    workingDirectory?: unknown;
  };
  runtimes?: Record<string, { reset_command?: unknown; resetCommand?: unknown }>;
  agents?: Array<{
    name?: unknown;
    alias?: unknown;
    provider?: unknown;
    runtime?: unknown;
    class?: unknown;
    working_directory?: unknown;
    workingDirectory?: unknown;
    model_hint?: unknown;
    modelHint?: unknown;
    reasoning_level?: unknown;
    reasoningLevel?: unknown;
    tool_policy?: unknown;
    toolPolicy?: unknown;
    persistent_persona?: unknown;
    persistentPersona?: unknown;
    reset_between_tasks?: unknown;
    resetBetweenTasks?: unknown;
    kickstart_template?: unknown;
    kickstartTemplate?: unknown;
  }>;
}

export interface FleetAgent {
  name: string;
  alias?: string;
  provider?: string;
  runtime: string;
  className: string;
  workingDirectory?: string;
  modelHint?: string;
  reasoningLevel?: string;
  toolPolicy?: string;
  persistentPersona?: boolean | string;
  resetBetweenTasks?: boolean;
  kickstartTemplate?: string;
}

export interface FleetRoster {
  version: 1;
  transport: 'tmux';
  tmux: {
    socketName: string;
    holderSession: string;
  };
  defaults: {
    workingDirectory: string;
  };
  runtimes: Record<string, { resetCommand: string }>;
  agents: FleetAgent[];
}

export interface FleetPaths {
  mosaicHome: string;
  rosterPath: string;
  toolsDir: string;
  fleetToolsDir: string;
  tmuxToolsDir: string;
  systemdUserDir: string;
  agentEnvDir: string;
}

type FleetServiceAction = 'start' | 'stop' | 'restart' | 'status';

/**
 * The named tmux socket the canonical fleet uses. Kept as a public constant for
 * rosters/callers that explicitly want isolation; it is NO LONGER the silent
 * fallback for a socket-less roster (that now resolves to the default socket).
 */
export const DEFAULT_SOCKET_NAME = 'mosaic-fleet';
const DEFAULT_HOLDER_SESSION = '_holder';
const DEFAULT_WORKING_DIRECTORY = '~/src';

/**
 * tmux `-L` args for a socket name. An empty/absent socket ⇒ the LITERAL default
 * tmux socket (no `-L`), so spawn, observe (`fleet ps`/watch), and the onboarding
 * cheat-sheet all agree. A named socket ⇒ `-L <name>`. `DEFAULT_SOCKET_NAME`
 * remains a constant for callers that explicitly want mosaic-fleet; it is no
 * longer the silent fallback for a socket-less roster.
 */
export function socketArgs(socketName: string): string[] {
  return socketName ? ['-L', socketName] : [];
}

/**
 * Default poll interval (ms) between capture-pane checks in `send --verify`.
 * Kept short enough to react quickly while not hammering tmux on busy hosts.
 */
export const VERIFY_POLL_INTERVAL_MS = 400;

/**
 * Default total timeout (ms) for the `send --verify` polling loop.
 * Configurable via `--verify-timeout <ms>` on `agent send`.
 */
export const VERIFY_DEFAULT_TIMEOUT_MS = 6_000;
const DEFAULT_RUNTIME_RESETS: Record<string, { resetCommand: string }> = {
  claude: { resetCommand: '/clear' },
  codex: { resetCommand: '/clear' },
  opencode: { resetCommand: '/clear' },
  pi: { resetCommand: '/new' },
};

export function resolveFleetPaths(mosaicHome = defaultMosaicHome()): FleetPaths {
  return {
    mosaicHome,
    rosterPath: join(mosaicHome, 'fleet', 'roster.yaml'),
    toolsDir: join(mosaicHome, 'tools'),
    fleetToolsDir: join(mosaicHome, 'tools', 'fleet'),
    tmuxToolsDir: join(mosaicHome, 'tools', 'tmux'),
    systemdUserDir: join(homedir(), '.config', 'systemd', 'user'),
    agentEnvDir: join(mosaicHome, 'fleet', 'agents'),
  };
}

function defaultMosaicHome(): string {
  // Honor MOSAIC_HOME so the reader matches the writer sidecar (and the launcher),
  // even when MOSAIC_HOME is set in the shell without an explicit --mosaic-home flag.
  return process.env.MOSAIC_HOME ?? join(homedir(), '.config', 'mosaic');
}

function assertDefaultMosaicHomeForSystemd(mosaicHome: string): void {
  const literalHome = join(homedir(), '.config', 'mosaic');
  if (resolve(mosaicHome) !== resolve(literalHome)) {
    throw new Error(
      `install-systemd only supports the default Mosaic home (${literalHome}) because the user systemd units use %h/.config/mosaic paths.`,
    );
  }
}

export async function loadFleetRoster(path: string): Promise<FleetRoster> {
  const rawText = await readFile(path, 'utf8');
  const parsed = parseRosterText(rawText, path);
  return normalizeRoster(parsed);
}

export function getRosterAgent(roster: FleetRoster, name: string): FleetAgent {
  const agent = roster.agents.find((candidate) => candidate.name === name);
  if (!agent) {
    throw new Error(`Agent "${name}" is not in the fleet roster.`);
  }
  return agent;
}

// ---------------------------------------------------------------------------
// NORTH_STAR — machine-readable fleet planning source + Markdown projection
//
// docs/fleet/NORTH_STAR.yaml is the single source of truth. The Markdown file
// (docs/fleet/NORTH_STAR.md) is a deterministic, pure projection of the YAML —
// no network, no CLI, no clock. Edit the YAML, regenerate the .md.
// ---------------------------------------------------------------------------

export interface NorthStarIdText {
  id: string;
  text: string;
}

export interface NorthStarWorkstream {
  id: string;
  title: string;
}

export interface NorthStarGoal {
  id: string;
  title: string;
  phase: number;
  priority: string;
  depends_on: string[];
}

export interface NorthStarAssumption {
  id: string;
  vetoable: boolean;
  text: string;
}

export interface NorthStarSpend {
  advisory: boolean;
  note: string;
}

export interface NorthStar {
  version: number;
  mission: string;
  substrate: { note: string };
  standing_objectives: NorthStarIdText[];
  success_criteria: NorthStarIdText[];
  workstreams: NorthStarWorkstream[];
  goals: NorthStarGoal[];
  assumptions: NorthStarAssumption[];
  spend: NorthStarSpend;
}

/**
 * Parse + validate the NORTH_STAR YAML text into a typed NorthStar object.
 * Pure: no IO, no network. Throws a descriptive error when a required key is
 * missing or malformed so the generator/tests fail loudly rather than emit a
 * partial projection.
 */
export function parseNorthStar(rawText: string): NorthStar {
  const parsed = YAML.parse(rawText) as Record<string, unknown> | null;
  if (!parsed || typeof parsed !== 'object') {
    throw new Error('NORTH_STAR.yaml did not parse to a mapping.');
  }

  const requireString = (value: unknown, key: string): string => {
    if (typeof value !== 'string' || value.trim() === '') {
      throw new Error(`NORTH_STAR.yaml: "${key}" must be a non-empty string.`);
    }
    return value.trim();
  };

  const requireArray = (value: unknown, key: string): unknown[] => {
    if (!Array.isArray(value) || value.length === 0) {
      throw new Error(`NORTH_STAR.yaml: "${key}" must be a non-empty array.`);
    }
    return value;
  };

  const idText = (value: unknown, key: string, index: number): NorthStarIdText => {
    const row = value as Record<string, unknown>;
    return {
      id: requireString(row?.id, `${key}[${index}].id`),
      text: requireString(row?.text, `${key}[${index}].text`),
    };
  };

  const version = parsed.version;
  if (typeof version !== 'number') {
    throw new Error('NORTH_STAR.yaml: "version" must be a number.');
  }

  const substrate = parsed.substrate as Record<string, unknown> | undefined;

  const spendRaw = parsed.spend as Record<string, unknown> | undefined;
  if (!spendRaw || typeof spendRaw.advisory !== 'boolean') {
    throw new Error('NORTH_STAR.yaml: "spend.advisory" must be a boolean.');
  }

  return {
    version,
    mission: requireString(parsed.mission, 'mission'),
    substrate: { note: requireString(substrate?.note, 'substrate.note') },
    standing_objectives: requireArray(parsed.standing_objectives, 'standing_objectives').map(
      (row, i) => idText(row, 'standing_objectives', i),
    ),
    success_criteria: requireArray(parsed.success_criteria, 'success_criteria').map((row, i) =>
      idText(row, 'success_criteria', i),
    ),
    workstreams: requireArray(parsed.workstreams, 'workstreams').map((row, i) => {
      const ws = row as Record<string, unknown>;
      return {
        id: requireString(ws?.id, `workstreams[${i}].id`),
        title: requireString(ws?.title, `workstreams[${i}].title`),
      };
    }),
    goals: requireArray(parsed.goals, 'goals').map((row, i) => {
      const goal = row as Record<string, unknown>;
      const dependsRaw = goal?.depends_on ?? [];
      if (!Array.isArray(dependsRaw)) {
        throw new Error(`NORTH_STAR.yaml: goals[${i}].depends_on must be an array.`);
      }
      const phase = goal?.phase;
      if (typeof phase !== 'number') {
        throw new Error(`NORTH_STAR.yaml: goals[${i}].phase must be a number.`);
      }
      return {
        id: requireString(goal?.id, `goals[${i}].id`),
        title: requireString(goal?.title, `goals[${i}].title`),
        phase,
        priority: requireString(goal?.priority, `goals[${i}].priority`),
        depends_on: dependsRaw.map((dep, j) => requireString(dep, `goals[${i}].depends_on[${j}]`)),
      };
    }),
    assumptions: requireArray(parsed.assumptions, 'assumptions').map((row, i) => {
      const asm = row as Record<string, unknown>;
      if (typeof asm?.vetoable !== 'boolean') {
        throw new Error(`NORTH_STAR.yaml: assumptions[${i}].vetoable must be a boolean.`);
      }
      return {
        id: requireString(asm?.id, `assumptions[${i}].id`),
        vetoable: asm.vetoable,
        text: requireString(asm?.text, `assumptions[${i}].text`),
      };
    }),
    spend: {
      advisory: spendRaw.advisory,
      note: requireString(spendRaw?.note, 'spend.note'),
    },
  };
}

/**
 * Render a GitHub-Flavored-Markdown table with prettier-compatible column
 * alignment: each column is padded to the widest cell (minimum 3 for the
 * `---` divider) so the generated bytes survive `prettier --check` unchanged.
 * Pure; the row strings use the same single-code-unit dash/arrow glyphs that
 * prettier's string-width counts as width 1.
 */
function renderMarkdownTable(headers: string[], rows: string[][]): string[] {
  const widths = headers.map((header, col) =>
    Math.max(3, header.length, ...rows.map((row) => row[col]?.length ?? 0)),
  );
  const pad = (cell: string, col: number): string => cell.padEnd(widths[col] ?? 0, ' ');
  const formatRow = (cells: string[]): string =>
    `| ${cells.map((cell, col) => pad(cell, col)).join(' | ')} |`;
  const divider = `| ${widths.map((w) => '-'.repeat(w)).join(' | ')} |`;
  return [formatRow(headers), divider, ...rows.map(formatRow)];
}

/**
 * Deterministically project a parsed NorthStar into the Markdown doctrine doc.
 * Pure function of its input — same input always yields byte-identical output,
 * so the round-trip (YAML → render → write) is stable across runs. No clock, no
 * network, no CLI. Layout follows the repo's existing doctrine-doc convention
 * (heading, blockquote banner, then sections + tables, e.g. north-star.md /
 * mission-control/BOARD.md).
 */
export function renderNorthStarMarkdown(ns: NorthStar): string {
  const lines: string[] = [];

  lines.push('# Mosaic Fleet — NORTH STAR');
  lines.push('');
  lines.push('> **Generated file — do not edit by hand.**');
  lines.push(
    '> Projected deterministically from [`NORTH_STAR.yaml`](./NORTH_STAR.yaml) by the pure',
  );
  lines.push('> generator in `packages/mosaic/src/commands/fleet.ts` (`renderNorthStarMarkdown`).');
  lines.push('> Edit the YAML, then regenerate. Self-contained Mosaic — no Hermes dependency.');
  lines.push('');

  lines.push('## Mission');
  lines.push('');
  lines.push(ns.mission);
  lines.push('');

  lines.push('## Substrate');
  lines.push('');
  lines.push(ns.substrate.note);
  lines.push('');

  lines.push('## Standing objectives');
  lines.push('');
  for (const obj of ns.standing_objectives) {
    lines.push(`- **${obj.id}** — ${obj.text}`);
  }
  lines.push('');

  lines.push('## Success criteria');
  lines.push('');
  for (const ac of ns.success_criteria) {
    lines.push(`- **${ac.id}** — ${ac.text}`);
  }
  lines.push('');

  lines.push('## Workstreams');
  lines.push('');
  lines.push(
    ...renderMarkdownTable(
      ['id', 'title'],
      ns.workstreams.map((ws) => [ws.id, ws.title]),
    ),
  );
  lines.push('');

  lines.push('## Goals (backlog projection)');
  lines.push('');
  lines.push(
    ...renderMarkdownTable(
      ['id', 'title', 'phase', 'priority', 'depends_on'],
      ns.goals.map((goal) => [
        goal.id,
        goal.title,
        String(goal.phase),
        goal.priority,
        goal.depends_on.length > 0 ? goal.depends_on.join(', ') : '—',
      ]),
    ),
  );
  lines.push('');

  lines.push('## Assumptions (vetoable)');
  lines.push('');
  for (const asm of ns.assumptions) {
    const veto = asm.vetoable ? 'vetoable' : 'fixed';
    lines.push(`- **${asm.id}** (${veto}) — ${asm.text}`);
  }
  lines.push('');

  lines.push('## Spend');
  lines.push('');
  lines.push(`- **advisory:** ${ns.spend.advisory ? 'true' : 'false'}`);
  lines.push(`- ${ns.spend.note}`);

  // No trailing blank line: the writer appends a single newline, yielding the
  // one-newline EOF prettier expects (round-trip stays format:check-clean).
  return lines.join('\n');
}

/**
 * Resolve the repo's docs/fleet directory from this compiled module's location.
 * fleet.ts lives at packages/mosaic/src/commands; docs/fleet sits at the repo
 * root. Exposed so the generator + tests share one path resolution.
 */
export function resolveNorthStarPaths(repoRoot?: string): {
  yamlPath: string;
  markdownPath: string;
} {
  const root = repoRoot ?? resolve(dirname(fileURLToPath(import.meta.url)), '..', '..', '..', '..');
  const dir = join(root, 'docs', 'fleet');
  return {
    yamlPath: join(dir, 'NORTH_STAR.yaml'),
    markdownPath: join(dir, 'NORTH_STAR.md'),
  };
}

/**
 * Read NORTH_STAR.yaml, project it to Markdown, and write NORTH_STAR.md.
 * The only IO in the NORTH_STAR pipeline; the parse + render steps it composes
 * are pure. Returns the rendered Markdown so callers/tests can assert on it.
 */
export async function generateNorthStarMarkdown(repoRoot?: string): Promise<string> {
  const { yamlPath, markdownPath } = resolveNorthStarPaths(repoRoot);
  const rawText = await readFile(yamlPath, 'utf8');
  const ns = parseNorthStar(rawText);
  const markdown = renderNorthStarMarkdown(ns);
  await writeFile(markdownPath, `${markdown}\n`, 'utf8');
  return markdown;
}

export function generateAgentEnv(roster: FleetRoster, agent: FleetAgent): string {
  const workingDirectory = agent.workingDirectory ?? roster.defaults.workingDirectory;
  return [
    `MOSAIC_AGENT_NAME=${shellEnvValue(agent.name)}`,
    // Per-agent class → start-agent-session.sh / launcher reads this to inject the
    // matching persona contract for the agent's class (default `worker`).
    `MOSAIC_AGENT_CLASS=${shellEnvValue(agent.className)}`,
    `MOSAIC_AGENT_RUNTIME=${shellEnvValue(agent.runtime)}`,
    // Per-agent model hint → start-agent-session.sh appends `--model <hint>` to
    // the `mosaic yolo` launch so workers run on the roster's model (e.g. pi on
    // openai-codex/gpt-5.5:high). Empty when the agent declares no model_hint.
    `MOSAIC_AGENT_MODEL=${shellEnvValue(agent.modelHint ?? '')}`,
    ...(agent.reasoningLevel !== undefined
      ? [`MOSAIC_AGENT_REASONING=${shellEnvValue(agent.reasoningLevel)}`]
      : []),
    ...(agent.toolPolicy !== undefined
      ? [`MOSAIC_AGENT_TOOL_POLICY=${shellEnvValue(agent.toolPolicy)}`]
      : []),
    `MOSAIC_AGENT_WORKDIR=${shellEnvValue(expandHome(workingDirectory))}`,
    `MOSAIC_TMUX_SOCKET=${shellEnvValue(roster.tmux.socketName)}`,
    '',
  ].join('\n');
}

export function mergeAgentEnv(generatedEnv: string, existingEnv?: string): string {
  if (!existingEnv?.trim()) {
    return generatedEnv;
  }
  const generatedKeys = new Set(
    generatedEnv
      .split('\n')
      .map((line) => line.match(/^([A-Za-z_][A-Za-z0-9_]*)=/)?.[1])
      .filter((key): key is string => key !== undefined),
  );
  const preservedLines = existingEnv.split('\n').filter((line) => {
    if (!line.trim()) {
      return false;
    }
    const key = line.match(/^([A-Za-z_][A-Za-z0-9_]*)=/)?.[1];
    return key === undefined || !generatedKeys.has(key);
  });
  if (preservedLines.length === 0) {
    return generatedEnv;
  }
  return [generatedEnv.trimEnd(), ...preservedLines, ''].join('\n');
}

export function buildFleetServiceCommand(action: FleetServiceAction, agentName?: string): string[] {
  const service = agentName ? `mosaic-agent@${agentName}.service` : 'mosaic-tmux-holder.service';
  return ['systemctl', '--user', action, service];
}

/** Poll interval (ms) while waiting for an in-flight restart's lock to clear. */
export const RESTART_LOCK_POLL_INTERVAL_MS = 250;
/**
 * Maximum time (ms) a re-entrant restart waits for the in-flight restart to
 * finish before it breaks the lock and proceeds anyway. A bound is required so
 * a crashed holder of the lock can never deadlock the fleet permanently.
 */
export const RESTART_LOCK_MAX_WAIT_MS = 30_000;
/**
 * Age (ms) past which a restart lock is treated as stale (its owner died
 * without releasing it) and is broken immediately rather than waited on.
 */
export const RESTART_LOCK_STALE_MS = 60_000;

/**
 * Resolves the path of the cross-process restart lock for a given Mosaic home.
 * Kept strictly under `<mosaicHome>/fleet/run` (not the heartbeat env override)
 * so the lock is scoped to the same fleet the restart acts on.
 */
export function restartLockPath(mosaicHome: string): string {
  return join(mosaicHome, 'fleet', 'run', 'restart.lock');
}

/** A held restart lock; `release()` removes the lock file iff we still own it. */
interface RestartGuard {
  release(): Promise<void>;
}

/** Lock-file contents: pid (informational), timestamp, and a unique owner token. */
function formatRestartLockContent(token: string): string {
  return `${process.pid}\n${Date.now()}\n${token}\n`;
}

/**
 * Reads the owner token (line 3) from a lock file, or null if the file is
 * missing/unreadable/tokenless. The token is what makes release and break
 * ownership-safe: a process only ever acts on a lock whose token matches its own.
 */
async function readRestartLockToken(lockPath: string): Promise<string | null> {
  let raw: string;
  try {
    raw = await readFile(lockPath, 'utf8');
  } catch {
    return null;
  }
  const token = raw.split('\n')[2]?.trim();
  return token ? token : null;
}

/**
 * Returns true when a lock's contents are stale: older than RESTART_LOCK_STALE_MS,
 * or unparseable (a corrupt or partially written lock left by a crashed owner).
 */
function isRestartLockContentStale(raw: string, now: number): boolean {
  const stampLine = raw.split('\n')[1] ?? '';
  const stamp = Number.parseInt(stampLine.trim(), 10);
  if (!Number.isFinite(stamp)) {
    return true;
  }
  return now - stamp >= RESTART_LOCK_STALE_MS;
}

/**
 * Path of the short-lived registry mutex that guards EVERY transition of the
 * restart lock (acquire, release, takeover). Held only across a few filesystem
 * ops — never across the restart itself — so contention clears in microseconds.
 */
function restartMutexPath(lockPath: string): string {
  return `${lockPath}.mutex`;
}

/** Brief back-off between registry-mutex acquisition attempts (held microseconds). */
const RESTART_MUTEX_RETRY_MS = 20;

/**
 * Staleness for the internal mutex / reclaim locks, judged by the file's mtime
 * rather than its CONTENT. `open(path, 'wx')` creates the inode (with a fresh
 * mtime) before any token/timestamp is written into it, so a content-based check
 * would momentarily see that empty file as corrupt-and-stale and could reap a
 * lock another contender is still acquiring. mtime is set atomically at creation,
 * so a just-created lock always reads as live; only a lock whose holder died and
 * stopped touching it ages past the threshold. These locks are never held across
 * the restart itself (only a couple of filesystem ops), so any mtime this old can
 * belong only to a dead holder.
 */
async function isRestartLockPathStale(path: string, now: number): Promise<boolean> {
  try {
    const info = await stat(path);
    return now - info.mtimeMs >= RESTART_LOCK_STALE_MS;
  } catch (err) {
    if ((err as NodeJS.ErrnoException).code === 'ENOENT') {
      return false; // Gone, not stale — the caller will re-contend.
    }
    return false; // Can't stat — treat as live and back off rather than reap.
  }
}

/** Path of the reclaim lock that serializes reaping of a crashed-holder mutex. */
function restartReclaimPath(mutexPath: string): string {
  return `${mutexPath}.reclaim`;
}

/**
 * Reap a registry mutex left behind by a process that CRASHED mid-transition —
 * one whose file has aged past RESTART_LOCK_STALE_MS. Because the mutex is held
 * only for a couple of filesystem ops (no sleeps, never across the restart), a
 * mutex this old can only belong to a dead holder.
 *
 * The reap removes the dead mutex but never CREATES/holds it — acquisition stays
 * the single `open('wx')` create in {@link acquireRestartMutex}, so exactly one
 * contender wins ownership no matter how the reap and acquires interleave. The
 * removal is made conditional by a dedicated reclaim lock: while it is held the
 * dead mutex is stable (its dead holder will never touch it, and no other
 * reclaimer can race), so re-reading it and removing it only if it is STILL stale
 * is a true compare — a live holder's fresh mutex is never removed. This closes
 * the reclaim race a content-blind rename-and-restore left open (a third
 * contender slipping into the gap while a fresh mutex was moved aside).
 */
async function reclaimStaleRestartMutex(mutexPath: string): Promise<void> {
  const reclaimPath = restartReclaimPath(mutexPath);
  let handle: Awaited<ReturnType<typeof open>>;
  try {
    handle = await open(reclaimPath, 'wx');
  } catch (err) {
    if ((err as NodeJS.ErrnoException).code !== 'EEXIST') {
      throw err;
    }
    // Someone is already reclaiming. If their reclaim lock is itself stale by
    // mtime, its holder crashed mid-reap (the lock spans only a stat + unlink,
    // microseconds) — clear it so a later pass can retry. Otherwise a live
    // reclaimer has it; back off. Either way we do not reap the mutex this pass.
    if (await isRestartLockPathStale(reclaimPath, Date.now())) {
      await unlink(reclaimPath).catch(() => {});
    }
    return;
  }
  try {
    // Re-check the mutex UNDER the reclaim lock and remove it only if it is STILL
    // stale by mtime. A live holder's mutex is fresh and is left untouched; a dead
    // holder's mutex is stable here (its holder is gone and no other reclaimer can
    // race us), so this re-check is authoritative.
    if (await isRestartLockPathStale(mutexPath, Date.now())) {
      await unlink(mutexPath).catch(() => {});
    }
  } finally {
    await handle.close();
    await unlink(reclaimPath).catch(() => {});
  }
}

/**
 * Acquire the registry mutex, BLOCKING (with brief back-offs) until held, and
 * return a token-gated release. This is the single point of mutual exclusion for
 * the restart lock: acquire, release, and stale/timeout takeover all run under it,
 * so "read the lock, then mutate it" is atomic — no acquirer, releaser, or breaker
 * can ever interleave with another. A mutex left by a crashed holder is reclaimed
 * once it ages past the stale threshold.
 */
async function acquireRestartMutex(
  mutexPath: string,
  token: string,
): Promise<RestartGuard['release']> {
  for (;;) {
    let handle: Awaited<ReturnType<typeof open>>;
    try {
      handle = await open(mutexPath, 'wx');
    } catch (err) {
      if ((err as NodeJS.ErrnoException).code !== 'EEXIST') {
        throw err;
      }
      // Staleness is judged by mtime, not content, so a mutex that exists but has
      // not yet had its token written (the open-before-write window) reads as live
      // and is never wrongly reaped.
      if (!(await isRestartLockPathStale(mutexPath, Date.now()))) {
        // A live holder has it — it will be gone in microseconds. Back off briefly.
        await new Promise((resolve) => setTimeout(resolve, RESTART_MUTEX_RETRY_MS));
        continue;
      }
      await reclaimStaleRestartMutex(mutexPath);
      continue;
    }
    // We created the mutex. Populate it with our token; if writing fails, clean up
    // our own file so we never leak an empty mutex that a peer would have to reap.
    try {
      await handle.writeFile(formatRestartLockContent(token));
      await handle.close();
    } catch (err) {
      await handle.close().catch(() => {});
      await unlink(mutexPath).catch(() => {});
      throw err;
    }
    return async (): Promise<void> => {
      if ((await readRestartLockToken(mutexPath)) !== token) return;
      await unlink(mutexPath).catch(() => {});
    };
  }
}

/**
 * Acquire the fleet restart lock, serializing concurrent `mosaic fleet restart`
 * invocations across processes. Each restart tears the tmux holder (and the
 * agent sessions inside it) down and back up; without this guard a re-entrant
 * restart relaunches agents against a half-torn-down holder, which fails and
 * tight-loops. A re-entrant caller waits for the in-flight restart to release
 * the lock (clean shutdown settled) before proceeding, breaks a stale lock left
 * by a crashed owner, and after RESTART_LOCK_MAX_WAIT_MS breaks the lock to
 * avoid a permanent deadlock.
 *
 * Correctness rests on a single invariant: EVERY transition of the lock — taking
 * a free lock, taking over a stale/timed-out one, and releasing — happens under
 * the registry mutex. Because the check ("is the lock free / stale / fresh?") and
 * the mutation that follows it both run while the mutex is held, they are atomic:
 * no other acquirer, releaser, or breaker can slip in between. That is what makes
 * takeover a true compare-and-swap rather than a content-blind clobber — a normal
 * `open('wx')` acquirer cannot create a fresh lock in a gap, and the original
 * owner's `release()` (also mutex-gated and token-checked) cannot drop a lock a
 * breaker already took over. So no interleaving lets two restarts both own the
 * lock and run concurrently.
 */
export async function acquireRestartLock(
  mosaicHome: string,
  sleepFn: SleepFn,
): Promise<RestartGuard> {
  const token = randomUUID();
  const lockPath = restartLockPath(mosaicHome);
  const mutexPath = restartMutexPath(lockPath);
  await mkdir(dirname(lockPath), { recursive: true });
  const release = async (): Promise<void> => {
    // Mutex-gated and token-gated: only remove the lock if it is still ours. If
    // another caller took it over (after a stale/timeout break) the token no
    // longer matches and we leave their lock intact.
    const releaseMutex = await acquireRestartMutex(mutexPath, token);
    try {
      if ((await readRestartLockToken(lockPath)) === token) {
        await unlink(lockPath).catch(() => {});
      }
    } finally {
      await releaseMutex();
    }
  };
  const deadline = Date.now() + RESTART_LOCK_MAX_WAIT_MS;
  for (;;) {
    let owned = false;
    const releaseMutex = await acquireRestartMutex(mutexPath, token);
    try {
      // Read and (if appropriate) mutate the lock atomically under the mutex.
      let current: string | null = null;
      let absent = false;
      try {
        current = await readFile(lockPath, 'utf8');
      } catch (readErr) {
        if ((readErr as NodeJS.ErrnoException).code === 'ENOENT') {
          absent = true;
        } else {
          current = null; // Unreadable/corrupt: treat as stale.
        }
      }
      const now = Date.now();
      if (absent) {
        // Lock is free — take it.
        await writeFile(lockPath, formatRestartLockContent(token));
        owned = true;
      } else {
        const stale = current === null || isRestartLockContentStale(current, now);
        const timedOut = now >= deadline;
        if (stale || timedOut) {
          process.stderr.write(
            stale
              ? 'Breaking stale fleet restart lock.\n'
              : `Timed out after ${RESTART_LOCK_MAX_WAIT_MS}ms waiting for the in-flight fleet ` +
                  'restart; breaking the lock.\n',
          );
          // Takeover is just an overwrite — safe because we hold the mutex, so no
          // acquirer or releaser can touch the lock between our read and this write.
          await writeFile(lockPath, formatRestartLockContent(token));
          owned = true;
        }
        // else: a fresh restart owns it — wait below and re-evaluate.
      }
    } finally {
      await releaseMutex();
    }
    if (owned) {
      return { release };
    }
    await sleepFn(RESTART_LOCK_POLL_INTERVAL_MS);
  }
}

/**
 * Returns the systemctl --user enable command for a given unit.
 * Used by the install auto-enable step to persist units across reboots.
 */
export function buildSystemdEnableCommand(unit: string): string[] {
  return ['systemctl', '--user', 'enable', unit];
}

/**
 * Returns the systemctl --user disable command for a given unit.
 * Used by `fleet remove` so a removed agent's enabled unit cannot resurrect on
 * boot pointing at deleted config (boot-survival symmetry with enable-on-add).
 */
export function buildSystemdDisableCommand(unit: string): string[] {
  return ['systemctl', '--user', 'disable', unit];
}

/**
 * Returns the loginctl enable-linger command for a given user.
 * Linger allows user systemd services to survive logout.
 */
export function buildEnableLingerCommand(user: string): string[] {
  return ['loginctl', 'enable-linger', user];
}

/**
 * Enable fleet units for boot-survival after install.
 * Non-fatal: if systemctl enable returns non-zero, a warning is printed and we continue.
 * If opts.enable === false (--no-enable flag), the whole step is skipped.
 */
export async function enableFleetUnits(
  runner: CommandRunner,
  roster: FleetRoster,
  opts: { enable?: boolean },
): Promise<void> {
  if (opts.enable === false) {
    return;
  }
  try {
    let succeeded = 0;
    let failed = 0;

    const holderResult = await runner(
      ...splitCommand(buildSystemdEnableCommand('mosaic-tmux-holder.service')),
    );
    if (holderResult.exitCode === 0) {
      succeeded++;
    } else {
      failed++;
      process.stderr.write(
        `Warning: could not enable mosaic-tmux-holder.service: ${holderResult.stderr || holderResult.stdout || 'non-zero exit'}\n`,
      );
    }

    for (const agent of roster.agents) {
      const unit = `mosaic-agent@${agent.name}.service`;
      const result = await runner(...splitCommand(buildSystemdEnableCommand(unit)));
      if (result.exitCode === 0) {
        succeeded++;
      } else {
        failed++;
        process.stderr.write(
          `Warning: could not enable ${unit}: ${result.stderr || result.stdout || 'non-zero exit'}\n`,
        );
      }
    }

    if (succeeded > 0) {
      console.log(`Enabled ${succeeded} unit(s) for boot-survival.`);
    }
    if (failed > 0) {
      process.stderr.write(
        `Warning: ${failed} unit(s) could not be enabled (systemctl unavailable?). Run manually if needed.\n`,
      );
    }

    // Best-effort linger
    let username: string;
    try {
      username = userInfo().username;
    } catch {
      username = process.env['USER'] ?? process.env['LOGNAME'] ?? 'unknown';
    }
    const lingerResult = await runner(...splitCommand(buildEnableLingerCommand(username)));
    if (lingerResult.exitCode !== 0) {
      process.stderr.write(
        `Hint: run 'loginctl enable-linger ${username}' as root to survive logout.\n`,
      );
    }
  } catch (err) {
    process.stderr.write(
      `Warning: auto-enable step failed unexpectedly: ${err instanceof Error ? err.message : String(err)}\n`,
    );
  }
}

export function buildAgentSendCommand(
  paths: FleetPaths,
  agentName: string,
  message: string,
  socketName = '',
  sourceLabel = getDefaultOperatorSourceLabel(),
): string[] {
  return [
    join(paths.tmuxToolsDir, 'agent-send.sh'),
    ...socketArgs(socketName),
    '-S',
    sourceLabel,
    '-s',
    agentName,
    '-m',
    message,
  ];
}

export function getDefaultOperatorSourceLabel(): string {
  const shortHostname = hostname().split('.')[0] || 'localhost';
  return `${shortHostname}:operator`;
}

export function buildAgentResetCommand(
  paths: FleetPaths,
  agentName: string,
  resetCommand: string,
  socketName = '',
): string[] {
  return [
    join(paths.tmuxToolsDir, 'send-message.sh'),
    ...socketArgs(socketName),
    '-t',
    `=${agentName}`,
    '-m',
    resetCommand,
  ];
}

export function buildAgentTailCommand(agentName: string, lines: number, socketName = ''): string[] {
  return [
    'tmux',
    ...socketArgs(socketName),
    'capture-pane',
    '-t',
    `=${agentName}:0.0`,
    '-p',
    '-S',
    `-${lines}`,
  ];
}

// ---------------------------------------------------------------------------
// Fleet ps — phase 2 observability helpers
// ---------------------------------------------------------------------------

export const HEARTBEAT_INTERVAL_MS = 15_000;
export const HEARTBEAT_IDLE_THRESHOLD_SECONDS = 300;

/**
 * Heartbeat interval in ms, honoring MOSAIC_HEARTBEAT_INTERVAL (seconds) so the
 * `fleet ps` freshness threshold matches the writer sidecar's actual cadence
 * (start-agent-session.sh). Falls back to HEARTBEAT_INTERVAL_MS (15s).
 */
export function heartbeatIntervalMs(): number {
  const sec = Number.parseInt(process.env.MOSAIC_HEARTBEAT_INTERVAL ?? '', 10);
  return Number.isFinite(sec) && sec > 0 ? sec * 1000 : HEARTBEAT_INTERVAL_MS;
}

/** Activity threshold in seconds, honoring MOSAIC_HEARTBEAT_IDLE_THRESHOLD. */
export function idleThresholdSeconds(): number {
  const sec = Number.parseInt(process.env.MOSAIC_HEARTBEAT_IDLE_THRESHOLD ?? '', 10);
  return Number.isFinite(sec) && sec > 0 ? sec : HEARTBEAT_IDLE_THRESHOLD_SECONDS;
}
export const HEARTBEAT_HEALTHY_MULTIPLIER = 3;

export type ReadinessState = 'working' | 'available' | 'stuck' | 'stale' | 'dead' | 'unknown';

export interface ReadinessSignals {
  paneAlive: boolean;
  hbHealth: 'healthy' | 'stale' | 'unknown';
  hbStatus: 'ok' | 'busy' | null;
  idleSeconds: number | null;
}

export interface ReadinessThresholds {
  idleThresholdSeconds: number;
}

/**
 * Classify whether an agent is progressing based on already-parsed heartbeat/tmux signals.
 * Best-effort and runtime-agnostic: it never probes, never throws, and preserves existing
 * unknown/stale behavior when heartbeat data is absent or old.
 */
export function classifyReadiness(
  signals: Partial<ReadinessSignals> | null | undefined,
  thresholds: Partial<ReadinessThresholds> | null | undefined = {},
): ReadinessState {
  try {
    if (signals?.paneAlive !== true) return 'dead';
    if (signals.hbHealth === 'unknown' || signals.hbHealth === undefined) return 'unknown';
    if (signals.hbHealth === 'stale') return 'stale';
    if (signals.hbStatus === 'busy') return 'working';
    if (signals.idleSeconds === null || signals.idleSeconds === undefined) return 'working';

    const idleSeconds = Number.isFinite(signals.idleSeconds) ? signals.idleSeconds : null;
    if (idleSeconds === null) return 'working';

    const idleThreshold = Number.isFinite(thresholds?.idleThresholdSeconds)
      ? Number(thresholds?.idleThresholdSeconds)
      : idleThresholdSeconds();
    // Follow-up: stuck pending per-agent assignment awareness: assigned task + idle past threshold => stuck.
    if (idleSeconds >= idleThreshold) return 'available';
    return 'working';
  } catch {
    return 'unknown';
  }
}

export interface HeartbeatInfo {
  ts: Date | null;
  pid: number | null;
  status: 'ok' | 'busy' | null;
  /** healthy | stale | unknown */
  health: 'healthy' | 'stale' | 'unknown';
  ageMs: number | null;
  /** Model id the runtime self-reported in its heartbeat (native HB only), else null. */
  model: string | null;
}

export interface AgentPsRow {
  name: string;
  alias?: string;
  tenant_id: string;
  host: string;
  runtime: string;
  systemdActive: string;
  systemdEnabled: string;
  paneAlive: boolean;
  panePid: number | null;
  paneCommand: string | null;
  idleSeconds: number | null;
  heartbeat: HeartbeatInfo;
  readiness: ReadinessState;
  /** roster runtime !== actual pane command */
  driftFlag: boolean;
  /** active but UnitFileState=disabled */
  bootEnableWarning: boolean;
  /** true = came from roster; false = found on socket but not in roster */
  managed: boolean;
  /** "roster" = defined in roster.yaml; "socket" = discovered via tmux list-sessions */
  source: 'roster' | 'socket';
}

/**
 * Returns the systemd show command for an agent unit (active+enabled state).
 * Returns: `systemctl --user show <unit> -p ActiveState -p SubState -p UnitFileState`
 */
export function buildSystemdShowCommand(agentName: string): string[] {
  const unit = `mosaic-agent@${agentName}.service`;
  return [
    'systemctl',
    '--user',
    'show',
    unit,
    '-p',
    'ActiveState',
    '-p',
    'SubState',
    '-p',
    'UnitFileState',
  ];
}

/**
 * Returns the tmux list-panes command for an agent pane.
 * Format: `#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}`
 */
export function buildTmuxListPanesCommand(agentName: string, socketName = ''): string[] {
  return [
    'tmux',
    ...socketArgs(socketName),
    'list-panes',
    '-t',
    `=${agentName}:0.0`,
    '-F',
    '#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}',
  ];
}

/**
 * Returns the tmux list-sessions command to enumerate all sessions on a socket.
 * Format: `tmux -L <socket> list-sessions -F '#{session_name}'`
 * Used to discover ad-hoc sessions that are not in the roster.
 */
export function buildTmuxListSessionsCommand(socketName = ''): string[] {
  return ['tmux', ...socketArgs(socketName), 'list-sessions', '-F', '#{session_name}'];
}

/**
 * Parse the output of `tmux list-sessions -F '#{session_name}'` into an array of session names.
 * Returns an empty array on empty/blank output.
 */
export function parseTmuxListSessions(output: string): string[] {
  return output
    .split('\n')
    .map((line) => line.trim())
    .filter((line) => line.length > 0);
}

/**
 * Returns the heartbeat file path for an agent.
 */
export function heartbeatPath(agentName: string, mosaicHome = defaultMosaicHome()): string {
  // Honor MOSAIC_HEARTBEAT_RUN_DIR (the writer sidecar's override); otherwise the
  // canonical <mosaicHome>/fleet/run. Keeps reader and writer on the same path.
  const runDir = process.env.MOSAIC_HEARTBEAT_RUN_DIR ?? join(mosaicHome, 'fleet', 'run');
  return join(runDir, `${agentName}.hb`);
}

/**
 * Parse a heartbeat file's contents into a HeartbeatInfo.
 * File format (one key=value per line):
 *   ts=<iso8601>
 *   pid=<pid>
 *   status=<ok|busy>
 *   model=<model-id>   (optional — native runtime heartbeats self-report it)
 */
export function parseHeartbeat(content: string | null, nowMs = Date.now()): HeartbeatInfo {
  if (content === null) {
    return { ts: null, pid: null, status: null, health: 'unknown', ageMs: null, model: null };
  }
  const lines = content.split('\n');
  let ts: Date | null = null;
  let pid: number | null = null;
  let status: 'ok' | 'busy' | null = null;
  let model: string | null = null;
  for (const line of lines) {
    const [key, ...rest] = line.split('=');
    const val = rest.join('=').trim();
    if (key === 'ts' && val) {
      const d = new Date(val);
      if (!Number.isNaN(d.getTime())) ts = d;
    } else if (key === 'pid' && val) {
      const n = Number.parseInt(val, 10);
      if (Number.isFinite(n)) pid = n;
    } else if (key === 'status' && (val === 'ok' || val === 'busy')) {
      status = val;
    } else if (key === 'model' && val) {
      model = val;
    }
  }
  const thresholdMs = heartbeatIntervalMs() * HEARTBEAT_HEALTHY_MULTIPLIER;
  let health: 'healthy' | 'stale' | 'unknown' = 'unknown';
  let ageMs: number | null = null;
  if (ts !== null) {
    ageMs = nowMs - ts.getTime();
    health = ageMs <= thresholdMs ? 'healthy' : 'stale';
  }
  return { ts, pid, status, health, ageMs, model };
}

/**
 * Parse the output of `systemctl --user show ... -p ActiveState -p SubState -p UnitFileState`
 * Returns an object with the three properties.
 */
export function parseSystemdShow(output: string): {
  ActiveState: string;
  SubState: string;
  UnitFileState: string;
} {
  const result: Record<string, string> = {};
  for (const line of output.split('\n')) {
    const eq = line.indexOf('=');
    if (eq !== -1) {
      result[line.slice(0, eq)] = line.slice(eq + 1).trim();
    }
  }
  return {
    ActiveState: result['ActiveState'] ?? 'unknown',
    SubState: result['SubState'] ?? 'unknown',
    UnitFileState: result['UnitFileState'] ?? 'unknown',
  };
}

/**
 * Parse the output of `tmux list-panes -F '#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}'`
 * Activity fields are Unix epoch timestamps (seconds), ordered most precise to coarsest.
 */
export function parseTmuxListPanes(
  output: string,
  nowMs = Date.now(),
): { pid: number | null; command: string | null; dead: boolean; idleSeconds: number | null } {
  const line = output.trim().split('\n')[0];
  if (!line) {
    return { pid: null, command: null, dead: true, idleSeconds: null };
  }
  // format: <pid> <command> <dead(0|1)> <pane_activity> <window_activity> <session_activity>
  const parts = line.split(' ');
  const pid = parts[0] ? (Number.isFinite(Number(parts[0])) ? Number(parts[0]) : null) : null;
  const command = parts[1] ?? null;
  const dead = parts[2] === '1';
  const activityEpoch = parts
    .slice(3, 6)
    .map((part) => (part ? Number(part) : NaN))
    .find((epoch) => Number.isFinite(epoch) && epoch > 0);
  const idleSeconds = activityEpoch
    ? Math.max(0, Math.floor((nowMs - activityEpoch * 1000) / 1000))
    : null;
  return { pid, command, dead, idleSeconds };
}

/**
 * Maps each known runtime to the set of acceptable pane commands.
 * A pane running any of these commands for the given runtime is NOT considered drifted.
 * Runtimes launched via `mosaic yolo` wrap in node, so 'node' is acceptable for most.
 * The dogfood runtime accepts python3/python (the canary-pi dogfood stub).
 */
export const RUNTIME_ACCEPTABLE_COMMANDS: Record<string, readonly string[]> = {
  claude: ['claude', 'node'],
  codex: ['codex', 'node'],
  opencode: ['opencode', 'node'],
  pi: ['pi', 'node'],
  dogfood: ['python3', 'python'],
};

/**
 * Determine if there is a runtime drift: roster says one runtime but the pane
 * is actually running something from a different runtime. We detect this by
 * checking if the pane command doesn't match a known acceptable command for the
 * roster's declared runtime.
 *
 * Known acceptable commands per runtime (see RUNTIME_ACCEPTABLE_COMMANDS):
 *   claude   → claude, node  (node covers mosaic yolo wrapper)
 *   codex    → codex, node
 *   opencode → opencode, node
 *   pi       → pi, node      (python3 still flags drift for canary-pi dogfood stub)
 *   dogfood  → python3, python
 *
 * If the pane is running something else (e.g., python3/dogfood-agent.py) for
 * an agent whose roster runtime is "pi", that's a drift.
 */
export function detectDrift(rosterRuntime: string, paneCommand: string | null): boolean {
  if (!paneCommand) return false;
  const acceptable = RUNTIME_ACCEPTABLE_COMMANDS[rosterRuntime];
  if (!acceptable) return false;
  return !acceptable.includes(paneCommand);
}

/**
 * Returns the default tenant_id (OS username) and host (short hostname).
 * These MUST appear in every --json record for multi-tenant/multi-host zero-foreclosure.
 */
export function getDefaultTenantAndHost(): { tenant_id: string; host: string } {
  let tenant_id: string;
  try {
    tenant_id = userInfo().username;
  } catch {
    tenant_id = process.env['USER'] ?? process.env['LOGNAME'] ?? 'unknown';
  }
  const host = hostname().split('.')[0] || 'localhost';
  return { tenant_id, host };
}

/**
 * Builds the command to create a grouped viewer session targeting an agent session.
 * A grouped session shares the same windows as the target but gets INDEPENDENT sizing,
 * so attaching the viewer never resizes the agent's window.
 *
 * The viewer session name is derived from the agent name and a unique suffix (typically
 * the caller's PID) so multiple concurrent watchers don't collide.
 *
 * Usage sequence:
 *   1. Run buildAgentWatchCreateViewerCommand → create grouped session (via capturing runner).
 *   2. Run buildAgentWatchAttachCommand → attach -r to the viewer session (via interactiveRunner).
 *   3. Run buildAgentWatchKillViewerCommand → kill the viewer session on detach (via capturing runner).
 */
export function buildAgentWatchCreateViewerCommand(
  agentName: string,
  viewerSessionName: string,
  socketName = '',
): string[] {
  return [
    'tmux',
    ...socketArgs(socketName),
    'new-session',
    '-d',
    '-t',
    `=${agentName}`,
    '-s',
    viewerSessionName,
  ];
}

/**
 * Builds the interactive attach command for a viewer session (read-only).
 * Must be run via interactiveRunner (stdio: 'inherit').
 */
export function buildAgentWatchAttachCommand(viewerSessionName: string, socketName = ''): string[] {
  return ['tmux', ...socketArgs(socketName), 'attach', '-r', '-t', viewerSessionName];
}

/**
 * Builds the kill-session command to clean up a viewer session after detach.
 * Keeps the agent session intact.
 */
export function buildAgentWatchKillViewerCommand(
  viewerSessionName: string,
  socketName = '',
): string[] {
  return ['tmux', ...socketArgs(socketName), 'kill-session', '-t', viewerSessionName];
}

/**
 * Returns a unique viewer session name for a given agent.
 * Uses process.pid so concurrent watchers produce distinct names.
 */
export function buildViewerSessionName(agentName: string): string {
  return `${agentName}-watch-${process.pid}`;
}

/**
 * @deprecated Use buildAgentWatchCreateViewerCommand + buildAgentWatchAttachCommand +
 * buildAgentWatchKillViewerCommand instead. This bare attach targets the agent session
 * directly and can resize it when the viewer terminal is smaller than the agent's window.
 *
 * Kept for backward compatibility only.
 */
export function buildAgentWatchCommand(agentName: string, socketName = ''): string[] {
  return ['tmux', ...socketArgs(socketName), 'attach', '-r', '-t', `=${agentName}`];
}

/**
 * Builds the capture-pane command used to verify that agent send was accepted
 * (not left as an unsubmitted draft). Captures the last N lines and checks for
 * the draft heuristic.
 */
export function buildAgentVerifyAcceptedCommand(
  agentName: string,
  socketName = '',
  lines = 5,
): string[] {
  return [
    'tmux',
    ...socketArgs(socketName),
    'capture-pane',
    '-t',
    `=${agentName}:0.0`,
    '-p',
    '-S',
    `-${lines}`,
  ];
}

/**
 * Result of a send-verify check.
 * - 'accepted': positive evidence that the message was accepted (response content present).
 * - 'draft': last non-empty line matches the draft heuristic (unsubmitted input).
 * - 'unverifiable': pane did not change after send (stale or blank) — we cannot determine
 *   acceptance; fails closed per FR-5.
 */
export type SendVerifyResult = 'accepted' | 'draft' | 'unverifiable';

/**
 * Classify the result of a send-verify check by comparing BEFORE and AFTER pane snapshots.
 *
 * This is the primary classifier for `send --verify`. It addresses the stale-pane
 * false-success problem: if the pane content did not change after the send, the new
 * message never registered in the TUI (wedged pane, send dropped, etc.).
 *
 * Classification logic:
 *   'unverifiable' — AFTER is blank/empty OR AFTER == BEFORE (no pane change after send).
 *   'draft'        — AFTER differs from BEFORE AND the last non-empty line of AFTER starts
 *                    with the draft pattern ("> "); message was typed but not submitted.
 *   'accepted'     — AFTER differs from BEFORE AND AFTER does not end in a draft line;
 *                    positive evidence that the TUI accepted the message.
 *
 * NOTE on blank AFTER: Full-screen TUIs (claude, codex, opencode, pi) render blank for
 * `tmux capture-pane`. A blank AFTER is indistinguishable from a wedged pane, so it
 * is always classified 'unverifiable' (fail-closed).
 *
 * NOTE on definitive acceptance: Phase-2 can only observe the pane surface — there is no
 * runtime acknowledgement (heartbeat-ack) at this phase. The pane-change check is the best
 * signal available against an opaque TUI. Definitive acceptance ultimately requires a
 * runtime acknowledgement (Phase-3 heartbeat-ack).
 *
 * Draft heuristic: a last non-empty line (after stripping ANSI escapes) that starts
 * with "> " is treated as an unsubmitted input line. This pattern is specific to
 * pi/claude TUIs; draft detection for codex/opencode TUIs is best-effort only.
 *
 * FR-5 requires `send --verify` to return non-zero when delivery cannot be verified.
 *
 * @param before  Pane snapshot captured BEFORE the send command.
 * @param after   Pane snapshot captured AFTER the send command (after the delay).
 */
export function classifySendResult(before: string, after: string): SendVerifyResult {
  const afterLines = after.split('\n').filter((l) => l.trim().length > 0);
  // Blank/empty AFTER => full-screen TUI rendered blank, or pane is wedged => unverifiable.
  if (afterLines.length === 0) return 'unverifiable';
  // No change => message didn't register in the TUI (stale/wedged pane) => unverifiable.
  if (after === before) return 'unverifiable';
  // AFTER differs from BEFORE — check whether the pane is now showing a draft line.
  const lastLine = afterLines[afterLines.length - 1]!;
  const stripped = lastLine.replace(/\x1b\[[0-9;]*m/g, '').trim();
  // Heuristic: if stripped last line starts with "> " — that's the common draft pattern
  // in pi/claude TUIs for showing user input before submission.
  // NOTE: this heuristic is pi/claude-specific; draft detection for codex/opencode
  // TUIs is best-effort only and may miss other unsubmitted-input indicators.
  if (/^>\s/.test(stripped)) return 'draft';
  return 'accepted';
}

/**
 * Check whether a send was accepted (not left as draft), using only the AFTER snapshot.
 *
 * @deprecated Prefer classifySendResult(before, after) which guards against stale-pane
 * false-successes. This single-snapshot variant cannot detect a wedged pane that still
 * shows old non-empty content — it will incorrectly return 'accepted' in that case.
 *
 * Retained for unit-test compatibility with single-snapshot assertions.
 *
 * Returns:
 *   'unverifiable' — blank/empty capture (full-screen TUIs render blank; we cannot tell).
 *   'draft'        — last non-empty line matches the draft heuristic.
 *   'accepted'     — non-blank and not a draft line (but may be stale — see above).
 */
export function isSendAccepted(capturedOutput: string): SendVerifyResult {
  const lines = capturedOutput.split('\n').filter((l) => l.trim().length > 0);
  // Blank/empty capture => full-screen TUI rendered blank => unverifiable.
  // This is the known-unverifiable case; fail closed (not treated as success).
  if (lines.length === 0) return 'unverifiable';
  const lastLine = lines[lines.length - 1]!;
  const stripped = lastLine.replace(/\x1b\[[0-9;]*m/g, '').trim();
  // Heuristic: if stripped last line starts with "> " — that's the common draft pattern
  // in pi/claude TUIs for showing user input before submission.
  // NOTE: this heuristic is pi/claude-specific; draft detection for codex/opencode
  // TUIs is best-effort only and may miss other unsubmitted-input indicators.
  if (/^>\s/.test(stripped)) return 'draft';
  return 'accepted';
}

export function registerFleetCommand(program: Command, deps: FleetCommandDeps = {}): Command {
  const runner = deps.runner ?? runCommand;
  const sleepFn = deps.sleepFn ?? defaultSleep;
  const paths = resolveFleetPaths(deps.mosaicHome);
  const frameworkRoot = deps.frameworkRoot ?? resolveFrameworkRoot();

  const cmd = program
    .command('fleet')
    .description('Manage the local Mosaic tmux fleet canary')
    .option('--mosaic-home <path>', 'Mosaic home directory', paths.mosaicHome)
    .option('--roster <path>', 'Fleet roster path');

  cmd
    .command('init')
    .description('Initialize a local fleet roster')
    .option(
      '--profile <name>',
      `Roster profile: ${FLEET_PROFILES.join(', ')} (skips interactive wizard)`,
    )
    .option('--write', 'Write the roster to Mosaic home')
    .option('--force', 'Overwrite an existing roster when used with --write')
    .action(async (opts: { profile?: string; write?: boolean; force?: boolean }) => {
      const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
      const activePaths = resolveFleetPaths(commandOpts.mosaicHome);

      let profile: FleetProfile;
      if (opts.profile !== undefined) {
        // Explicit --profile flag: validate and use it (non-interactive path).
        profile = parseInitProfile(opts.profile);
      } else {
        // No --profile: use wizard when stdin is a TTY, else default to 'general'.
        const isTTY = deps.isStdinTTY ?? process.stdin.isTTY ?? false;
        if (isTTY) {
          profile = await promptFleetProfile();
        } else {
          process.stderr.write(
            'Note: stdin is not a TTY; defaulting to fleet profile "general". ' +
              'Use --profile <name> to select a different preset.\n',
          );
          profile = 'general';
        }
      }

      const source = join(frameworkRoot, 'fleet', 'examples', resolvePresetFilename(profile));
      const content = await readFile(source, 'utf8');

      if (!opts.write) {
        console.log(content.trimEnd());
        return;
      }

      const destination = commandOpts.roster ?? activePaths.rosterPath;
      if (!opts.force && (await canRead(destination))) {
        throw new Error(
          `Fleet roster already exists: ${destination}. Re-run with --force to overwrite.`,
        );
      }
      await mkdir(dirname(destination), { recursive: true });
      await writeFile(destination, content);

      // Guarantee the two-agent floor: exactly one orchestrator AND at least
      // one enhancer for every profile except the sanctioned no-orchestrator
      // `minimal` preset. A mismatch means a corrupted/edited preset — fail hard
      // rather than write a malformed fleet.
      const written = await loadFleetRoster(destination);
      const orchCount = countOrchestrators(written);
      const enhancerCount = countEnhancers(written);
      if (profile === 'minimal') {
        console.log(
          `Initialized ${profile} fleet: ${written.agents.length} agent(s) (no orchestrator). Next: mosaic fleet install`,
        );
      } else if (orchCount !== 1) {
        throw new Error(
          `Fleet init failed: the "${profile}" roster has ${orchCount} orchestrator agent(s), ` +
            `expected exactly 1 (R5). The preset may be corrupted — re-install the framework.`,
        );
      } else if (enhancerCount < 1) {
        throw new Error(
          `Fleet init failed: the "${profile}" roster has no enhancer agent. Every fleet keeps an ` +
            `orchestrator + enhancer minimum (two-agent floor). The preset may be corrupted — ` +
            `re-install the framework.`,
        );
      } else {
        const workerCount = written.agents.length - 1 - enhancerCount;
        console.log(
          `Initialized ${profile} fleet: 1 orchestrator + ${enhancerCount} enhancer(s) + ` +
            `${workerCount} worker(s). Next: mosaic fleet install`,
        );
      }
    });

  cmd
    .command('install')
    .description('Install local fleet tools and user systemd units')
    .option('--no-enable', 'Skip enabling units for boot-survival')
    .action(async (opts: { enable?: boolean }) => {
      await installFleet(cmd, frameworkRoot);
      const roster = await loadRosterForCommand(cmd);
      await enableFleetUnits(runner, roster, opts);
    });

  cmd
    .command('install-systemd')
    .description('Install local fleet tools and user systemd units')
    .option('--no-enable', 'Skip enabling units for boot-survival')
    .action(async (opts: { enable?: boolean }) => {
      await installFleet(cmd, frameworkRoot);
      const roster = await loadRosterForCommand(cmd);
      await enableFleetUnits(runner, roster, opts);
    });

  for (const action of ['start', 'stop', 'restart'] as const) {
    cmd
      .command(`${action} [agent]`)
      .description(`${action} the fleet holder or one agent`)
      .action(async (agent?: string) => {
        const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
        const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
        const roster = await loadRosterForCommand(cmd);
        if (agent) {
          getRosterAgent(roster, agent);
          // Single-agent restart is guarded too: it can race a full restart that
          // is tearing the shared holder down.
          if (action === 'restart') {
            const guard = await acquireRestartLock(activePaths.mosaicHome, sleepFn);
            try {
              await runChecked(runner, buildFleetServiceCommand(action, agent));
            } finally {
              await guard.release();
            }
            return;
          }
          await runChecked(runner, buildFleetServiceCommand(action, agent));
          return;
        }
        if (action === 'stop') {
          await stopFleetBestEffort(
            runner,
            roster.agents.map((rosterAgent) => rosterAgent.name),
          );
          return;
        }
        if (action === 'restart') {
          // Serialize the holder+agents teardown/relaunch behind the restart lock
          // so a re-entrant restart waits for clean shutdown before relaunching,
          // instead of racing a half-torn-down holder into a tight loop.
          const guard = await acquireRestartLock(activePaths.mosaicHome, sleepFn);
          try {
            await runChecked(runner, buildFleetServiceCommand(action));
            for (const rosterAgent of roster.agents) {
              await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
            }
          } finally {
            await guard.release();
          }
          return;
        }
        await runChecked(runner, buildFleetServiceCommand(action));
        for (const rosterAgent of roster.agents) {
          await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
        }
      });
  }

  cmd
    .command('status [agent]')
    .description('Show fleet holder or agent systemd status')
    .option('--json', 'Print JSON status')
    .action(async (agent: string | undefined, opts: { json?: boolean }) => {
      if (agent) {
        const roster = await loadRosterForCommand(cmd);
        getRosterAgent(roster, agent);
      }
      const result = await runner(...splitCommand(buildFleetServiceCommand('status', agent)));
      if (opts.json) {
        console.log(
          JSON.stringify({
            exitCode: result.exitCode,
            stdout: result.stdout,
            stderr: result.stderr,
          }),
        );
        setExitCodeFromResult(result);
        return;
      }
      writeCommandOutput(result);
    });

  cmd
    .command('verify')
    .description('Verify the local canary holder and roster sessions on the isolated socket')
    .action(async () => {
      const roster = await loadRosterForCommand(cmd);
      const socketName = roster.tmux.socketName;
      await runChecked(runner, [
        'tmux',
        ...socketArgs(socketName),
        'has-session',
        '-t',
        `=${roster.tmux.holderSession}:0.0`,
      ]);
      for (const agent of roster.agents) {
        await runChecked(runner, [
          'tmux',
          ...socketArgs(socketName),
          'has-session',
          '-t',
          `=${agent.name}:0.0`,
        ]);
      }
      console.log(`Verified fleet on tmux socket ${socketName}.`);
    });

  cmd
    .command('ps')
    .description(
      'Show real-time status for all roster agents and unmanaged socket sessions (systemd + tmux + heartbeat)',
    )
    .option('--json', 'Print JSON array')
    .action(async (opts: { json?: boolean }) => {
      const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
      const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
      const roster = await loadRosterForCommand(cmd);
      const { tenant_id, host } = getDefaultTenantAndHost();
      const nowMs = Date.now();

      const rows: AgentPsRow[] = [];
      const readinessThresholds = {
        idleThresholdSeconds: idleThresholdSeconds(),
      };

      // Build the set of roster agent names for quick lookup when filtering socket sessions.
      const rosterAgentNames = new Set(roster.agents.map((a) => a.name));

      for (const agent of roster.agents) {
        // systemd show
        const showResult = await runner(...splitCommand(buildSystemdShowCommand(agent.name)));
        const sysInfo = parseSystemdShow(showResult.stdout);

        // tmux list-panes
        const panesResult = await runner(
          ...splitCommand(buildTmuxListPanesCommand(agent.name, roster.tmux.socketName)),
        );
        const paneInfo = parseTmuxListPanes(panesResult.stdout, nowMs);

        // heartbeat
        const hbFile = heartbeatPath(agent.name, activePaths.mosaicHome);
        let hbContent: string | null = null;
        try {
          hbContent = await readFile(hbFile, 'utf8');
        } catch {
          hbContent = null;
        }
        const hb = parseHeartbeat(hbContent, nowMs);

        // drift and boot-enable
        const driftFlag = detectDrift(agent.runtime, paneInfo.command);
        const bootEnableWarning =
          sysInfo.ActiveState === 'active' && sysInfo.UnitFileState === 'disabled';

        const paneAlive = !paneInfo.dead;
        const readiness = classifyReadiness(
          {
            paneAlive,
            hbHealth: hb.health,
            hbStatus: hb.status,
            idleSeconds: paneInfo.idleSeconds,
          },
          readinessThresholds,
        );

        rows.push({
          name: agent.name,
          alias: agent.alias,
          tenant_id,
          host,
          runtime: agent.runtime,
          systemdActive: sysInfo.ActiveState,
          systemdEnabled: sysInfo.UnitFileState,
          paneAlive,
          panePid: paneInfo.pid,
          paneCommand: paneInfo.command,
          idleSeconds: paneInfo.idleSeconds,
          heartbeat: hb,
          readiness,
          driftFlag,
          bootEnableWarning,
          managed: true,
          source: 'roster',
        });
      }

      // Enumerate all live sessions on the socket to surface unmanaged (ad-hoc) sessions.
      // If list-sessions fails (socket not up), silently skip — show roster rows only.
      try {
        const listSessionsResult = await runner(
          ...splitCommand(buildTmuxListSessionsCommand(roster.tmux.socketName)),
        );
        if (listSessionsResult.exitCode === 0) {
          const socketSessions = parseTmuxListSessions(listSessionsResult.stdout);
          const holderSession = roster.tmux.holderSession;

          for (const sessionName of socketSessions) {
            // Skip roster agents (already in rows) and the holder session (infrastructure).
            if (rosterAgentNames.has(sessionName) || sessionName === holderSession) {
              continue;
            }

            // tmux list-panes for pane info
            const panesResult = await runner(
              ...splitCommand(buildTmuxListPanesCommand(sessionName, roster.tmux.socketName)),
            );
            const paneInfo = parseTmuxListPanes(panesResult.stdout, nowMs);

            // heartbeat — try reading the .hb file using the same path convention
            const hbFile = heartbeatPath(sessionName, activePaths.mosaicHome);
            let hbContent: string | null = null;
            try {
              hbContent = await readFile(hbFile, 'utf8');
            } catch {
              hbContent = null;
            }
            const hb = parseHeartbeat(hbContent, nowMs);

            // systemd — check if mosaic-agent@<name>.service exists (usually inactive for ad-hoc)
            const showResult = await runner(...splitCommand(buildSystemdShowCommand(sessionName)));
            const sysInfo = parseSystemdShow(showResult.stdout);

            const bootEnableWarning =
              sysInfo.ActiveState === 'active' && sysInfo.UnitFileState === 'disabled';

            const paneAlive = !paneInfo.dead;
            const readiness = classifyReadiness(
              {
                paneAlive,
                hbHealth: hb.health,
                hbStatus: hb.status,
                idleSeconds: paneInfo.idleSeconds,
              },
              readinessThresholds,
            );

            rows.push({
              name: sessionName,
              tenant_id,
              host,
              // runtime unknown — not in roster
              runtime: 'unknown',
              systemdActive: sysInfo.ActiveState,
              systemdEnabled: sysInfo.UnitFileState,
              paneAlive,
              panePid: paneInfo.pid,
              paneCommand: paneInfo.command,
              idleSeconds: paneInfo.idleSeconds,
              heartbeat: hb,
              readiness,
              // No roster runtime to compare — drift is not meaningful for unmanaged sessions
              driftFlag: false,
              bootEnableWarning,
              managed: false,
              source: 'socket',
            });
          }
        }
      } catch {
        // list-sessions failed (socket missing or permission error) — show roster rows only
      }

      if (opts.json) {
        console.log(JSON.stringify(rows, null, 2));
        return;
      }

      // Table output
      const header = [
        'NAME'.padEnd(18),
        'TENANT'.padEnd(12),
        'HOST'.padEnd(12),
        'RUNTIME'.padEnd(10),
        'SYSTEMD'.padEnd(16),
        'PANE'.padEnd(8),
        'PID'.padEnd(8),
        'IDLE'.padEnd(8),
        'HB'.padEnd(12),
        'MODEL'.padEnd(22),
        'FLAGS',
      ].join(' ');
      console.log(header);
      console.log('-'.repeat(header.length));

      for (const row of rows) {
        const systemd = `${row.systemdActive}/${row.systemdEnabled}`;
        const pane = row.paneAlive ? 'alive' : 'dead';
        const pid = row.panePid !== null ? String(row.panePid) : '-';
        const idle = row.idleSeconds !== null ? `${row.idleSeconds}s` : '-';
        const hbAge =
          row.heartbeat.ageMs !== null
            ? `${Math.round(row.heartbeat.ageMs / 1000)}s/${row.readiness}`
            : `unknown`;
        const model = row.heartbeat.model ?? '-';
        const flags: string[] = [];
        if (!row.managed) flags.push('UNMANAGED');
        if (row.driftFlag) flags.push('DRIFT');
        if (row.bootEnableWarning) flags.push('BOOT-ENABLE');

        console.log(
          [
            (row.alias ?? row.name).padEnd(18),
            row.tenant_id.padEnd(12),
            row.host.padEnd(12),
            row.runtime.padEnd(10),
            systemd.padEnd(16),
            pane.padEnd(8),
            pid.padEnd(8),
            idle.padEnd(8),
            hbAge.padEnd(12),
            model.padEnd(22),
            flags.join(','),
          ].join(' '),
        );
      }
    });

  cmd
    .command('add <name>')
    .description('Add a new agent to the fleet roster and optionally start it')
    .requiredOption('--runtime <runtime>', `Agent runtime (${VALID_FLEET_RUNTIMES.join(', ')})`)
    .requiredOption('--class <class>', 'Agent class (e.g. worker, orchestrator, canary)')
    .option('--model <hint>', 'Model hint for the agent')
    .option('--working-dir <path>', 'Working directory for the agent')
    .option('--no-start', 'Skip starting the agent after adding')
    .action(
      async (
        name: string,
        opts: {
          runtime: string;
          class: string;
          model?: string;
          workingDir?: string;
          start: boolean;
        },
      ) => {
        if (!VALID_FLEET_RUNTIMES.includes(opts.runtime)) {
          throw new Error(
            `Invalid runtime "${opts.runtime}". Valid runtimes: ${VALID_FLEET_RUNTIMES.join(', ')}.`,
          );
        }
        const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
        const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
        const rosterPath = await resolveRosterPath(commandOpts.mosaicHome, commandOpts.roster);
        const roster = await loadFleetRoster(rosterPath);

        const newAgent: FleetAgent = {
          name,
          runtime: opts.runtime,
          className: opts.class,
          ...(opts.workingDir !== undefined && { workingDirectory: opts.workingDir }),
          ...(opts.model !== undefined && { modelHint: opts.model }),
        };

        const updatedRoster = addAgentToRoster(roster, newAgent);
        await writeFile(rosterPath, serializeRosterToYaml(updatedRoster));

        const envPath = join(activePaths.agentEnvDir, `${name}.env`);
        const existingEnv = (await canRead(envPath)) ? await readFile(envPath, 'utf8') : undefined;
        await mkdir(activePaths.agentEnvDir, { recursive: true });
        await writeFile(
          envPath,
          mergeAgentEnv(generateAgentEnv(updatedRoster, newAgent), existingEnv),
        );

        console.log(`Added ${name} (${opts.runtime}/${opts.class}) to the fleet.`);

        // Enable the unit for boot-survival (non-fatal) — symmetry with
        // disable-on-remove. Independent of --start so a queued agent still
        // survives a reboot once its unit exists.
        try {
          const enableResult = await runner(
            ...splitCommand(buildSystemdEnableCommand(`mosaic-agent@${name}.service`)),
          );
          if (enableResult.exitCode !== 0) {
            process.stderr.write(
              `Warning: could not enable mosaic-agent@${name}.service: ${enableResult.stderr || enableResult.stdout || 'non-zero exit'}\n`,
            );
          }
        } catch (err) {
          process.stderr.write(
            `Warning: enable command failed for ${name}: ${err instanceof Error ? err.message : String(err)}\n`,
          );
        }

        if (opts.start !== false) {
          await runChecked(runner, buildFleetServiceCommand('start', name));
          console.log(`Started mosaic-agent@${name}.service.`);
        } else {
          console.log(`Agent queued (--no-start); run: mosaic fleet start ${name}`);
        }
      },
    );

  cmd
    .command('remove <name>')
    .description('Remove an agent from the fleet roster')
    .option('--keep-files', 'Skip deleting env and heartbeat files')
    .action(async (name: string, opts: { keepFiles?: boolean }) => {
      const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
      const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
      const rosterPath = await resolveRosterPath(commandOpts.mosaicHome, commandOpts.roster);
      const roster = await loadFleetRoster(rosterPath);

      // Guard: throws if removing leaves 0 orchestrators or agent not in roster
      const updatedRoster = removeAgentFromRoster(roster, name);

      // Stop agent (non-fatal)
      try {
        const stopResult = await runner(...splitCommand(buildFleetServiceCommand('stop', name)));
        if (stopResult.exitCode !== 0) {
          process.stderr.write(
            `Warning: could not stop mosaic-agent@${name}.service: ${stopResult.stderr || stopResult.stdout || 'non-zero exit'}\n`,
          );
        }
      } catch (err) {
        process.stderr.write(
          `Warning: stop command failed for ${name}: ${err instanceof Error ? err.message : String(err)}\n`,
        );
      }

      // Disable the unit (non-fatal) so an enabled instance cannot resurrect on
      // boot pointing at the now-deleted config — boot-survival symmetry with
      // enable-on-add. Skipped only when --keep-files keeps the config in place.
      if (!opts.keepFiles) {
        try {
          const disableResult = await runner(
            ...splitCommand(buildSystemdDisableCommand(`mosaic-agent@${name}.service`)),
          );
          if (disableResult.exitCode !== 0) {
            process.stderr.write(
              `Warning: could not disable mosaic-agent@${name}.service: ${disableResult.stderr || disableResult.stdout || 'non-zero exit'}\n`,
            );
          }
        } catch (err) {
          process.stderr.write(
            `Warning: disable command failed for ${name}: ${err instanceof Error ? err.message : String(err)}\n`,
          );
        }
      }

      // Write updated roster
      await writeFile(rosterPath, serializeRosterToYaml(updatedRoster));

      // Delete env and heartbeat files (best-effort, non-fatal)
      if (!opts.keepFiles) {
        try {
          await unlink(join(activePaths.agentEnvDir, `${name}.env`));
        } catch {
          // best-effort
        }
        try {
          await unlink(heartbeatPath(name, activePaths.mosaicHome));
        } catch {
          // best-effort
        }
      }

      console.log(`Removed ${name} from the fleet.`);
    });

  // Mosaic-native backlog of record (card A4). Resolves the active --mosaic-home
  // (parent flag) at call time so the embedded PGlite store lands under the same
  // fleet/ directory as the roster and heartbeats.
  registerFleetBacklogCommand(cmd, () => cmd.opts<{ mosaicHome: string }>().mosaicHome);

  // System-type profiles (H2): declarative persona roster + topology, resolved
  // from <mosaicHome>/fleet/profiles/*.yaml using the same --mosaic-home flag.
  registerFleetProfileCommand(cmd, () => cmd.opts<{ mosaicHome: string }>().mosaicHome);

  // Update-surviving persona overrides (H4): baseline fleet/roles/ ⊕ the
  // PRESERVE-protected fleet/roles.local/ override layer, resolved via the same
  // --mosaic-home flag.
  registerFleetPersonaCommand(cmd, () => cmd.opts<{ mosaicHome: string }>().mosaicHome);

  // Provisioning (H3): materialize a concrete roster.yaml from a system-type
  // profile. DRY-RUN by default; --write persists under the same --mosaic-home.
  registerFleetProvisionCommand(cmd, () => cmd.opts<{ mosaicHome: string }>().mosaicHome);

  return cmd;
}

export function registerFleetAgentCommands(
  agentCommand: Command,
  deps: FleetCommandDeps = {},
): void {
  const runner = deps.runner ?? runCommand;
  const iRunner = deps.interactiveRunner ?? spawnInteractive;
  const sleepFn = deps.sleepFn ?? defaultSleep;

  agentCommand
    .command('roster')
    .description('List agents from the local fleet roster')
    .option('--json', 'Print JSON')
    .action(async (opts: { json?: boolean }) => {
      const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
      if (opts.json) {
        console.log(JSON.stringify(roster, null, 2));
        return;
      }
      for (const agent of roster.agents) {
        console.log(`${agent.name}\t${agent.runtime}\t${agent.className}`);
      }
    });

  agentCommand
    .command('comms-block <role>')
    .description(
      "Print the Fleet Comms cheat-sheet for a roster role (preview a peer's peer-reach view)",
    )
    .option('--host <host>', 'Override the fleet host (preview a cross-host peer view)')
    .action((role: string, opts: { host?: string }) => {
      const mosaicHome = resolveMosaicHomeFromCommand(agentCommand, deps.mosaicHome);
      const res = resolveCommsBlock(mosaicHome, role, opts.host);
      if (!res.ok) {
        console.error(`[mosaic] comms-block: ${res.error}`);
        process.exitCode = 1;
        return;
      }
      console.log(res.output);
    });

  agentCommand
    .command('status [agent]')
    .description('Show tmux status for the local fleet or one agent')
    .option('--json', 'Print JSON')
    .action(async (agent: string | undefined, opts: { json?: boolean }) => {
      const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
      if (agent) {
        getRosterAgent(roster, agent);
      }
      const command = agent
        ? ['tmux', ...socketArgs(roster.tmux.socketName), 'has-session', '-t', `=${agent}:0.0`]
        : ['tmux', ...socketArgs(roster.tmux.socketName), 'ls'];
      const result = await runner(...splitCommand(command));
      if (opts.json) {
        console.log(
          JSON.stringify({
            exitCode: result.exitCode,
            stdout: result.stdout,
            stderr: result.stderr,
          }),
        );
        setExitCodeFromResult(result);
        return;
      }
      writeCommandOutput(result);
    });

  agentCommand
    .command('send <agent>')
    .description('Send a message to a local fleet agent')
    .requiredOption('--message <text>', 'Message text')
    .option('--source-label <label>', 'Source label for the message preamble')
    .option('--source <label>', 'Alias for --source-label')
    .option(
      '--verify',
      'Verify message was accepted (not left as a draft); exit non-zero if unverifiable',
    )
    .option(
      '--verify-timeout <ms>',
      `Maximum time (ms) to poll for pane change when --verify is set (default: ${VERIFY_DEFAULT_TIMEOUT_MS})`,
      String(VERIFY_DEFAULT_TIMEOUT_MS),
    )
    .action(
      async (
        agent: string,
        opts: {
          message: string;
          sourceLabel?: string;
          source?: string;
          verify?: boolean;
          verifyTimeout?: string;
        },
      ) => {
        const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
        getRosterAgent(roster, agent);
        const paths = resolveFleetPaths(
          resolveMosaicHomeFromCommand(agentCommand, deps.mosaicHome),
        );
        const sourceLabel = opts.sourceLabel ?? opts.source ?? getDefaultOperatorSourceLabel();
        if (opts.verify) {
          const parsedTimeout =
            opts.verifyTimeout !== undefined ? Number.parseInt(opts.verifyTimeout, 10) : Number.NaN;
          const timeoutMs = Number.isFinite(parsedTimeout)
            ? Math.max(0, parsedTimeout)
            : VERIFY_DEFAULT_TIMEOUT_MS;

          // Capture BEFORE snapshot so we can detect stale-pane false-successes.
          // A wedged pane that still shows old non-empty content must not be reported
          // as 'accepted' — we compare BEFORE vs AFTER to guard against that case.
          const beforeResult = await runner(
            ...splitCommand(buildAgentVerifyAcceptedCommand(agent, roster.tmux.socketName)),
          );
          if (beforeResult.exitCode !== 0) {
            throw new Error(
              `send --verify: could not capture pane output before send (tmux exited ${beforeResult.exitCode}).`,
            );
          }
          const beforeSnapshot = beforeResult.stdout;

          await runChecked(
            runner,
            buildAgentSendCommand(paths, agent, opts.message, roster.tmux.socketName, sourceLabel),
          );

          // Bounded polling loop: poll capture-pane every VERIFY_POLL_INTERVAL_MS up to
          // timeoutMs. Return immediately when the pane shows 'accepted' or 'draft';
          // keep polling while 'unverifiable' (no pane change yet). Fail closed after
          // timeout with the existing "no pane change after send" message.
          const deadline = Date.now() + timeoutMs;
          let verifyResult: SendVerifyResult = 'unverifiable';

          while (true) {
            await sleepFn(VERIFY_POLL_INTERVAL_MS);
            const afterResult = await runner(
              ...splitCommand(buildAgentVerifyAcceptedCommand(agent, roster.tmux.socketName)),
            );
            if (afterResult.exitCode !== 0) {
              throw new Error(
                `send --verify: could not capture pane output to verify acceptance (tmux exited ${afterResult.exitCode}).`,
              );
            }
            verifyResult = classifySendResult(beforeSnapshot, afterResult.stdout);
            // Definitive result — stop polling immediately.
            if (verifyResult === 'accepted' || verifyResult === 'draft') {
              break;
            }
            // Still unverifiable — check if we have time left to poll again.
            if (Date.now() >= deadline) {
              break;
            }
          }

          if (verifyResult === 'draft') {
            process.exitCode = 1;
            process.stderr.write(
              `send --verify: message left as unsubmitted draft in agent "${agent}".\n`,
            );
          } else if (verifyResult === 'unverifiable') {
            process.exitCode = 1;
            process.stderr.write(
              `send --verify: could not verify delivery (no pane change after send) for agent "${agent}".\n`,
            );
          }
        } else {
          await runChecked(
            runner,
            buildAgentSendCommand(paths, agent, opts.message, roster.tmux.socketName, sourceLabel),
          );
        }
      },
    );

  agentCommand
    .command('watch <agent>')
    .description('Open a read-only view of a fleet agent tmux session (cannot send keystrokes)')
    .action(async (agent: string) => {
      const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
      getRosterAgent(roster, agent);

      // Use a GROUPED VIEWER SESSION to prevent the observer from resizing the agent's
      // window. A bare `tmux attach -r` against the agent session itself still lets the
      // client shrink the session to its terminal size; a grouped session gets INDEPENDENT
      // sizing so the agent's window is never affected by the viewer's terminal dimensions.
      //
      // Sequence:
      //   1. Create a throwaway grouped session targeting the agent (capturing runner).
      //   2. Attach -r (read-only) to the viewer session (interactiveRunner / TTY).
      //   3. Kill the viewer session on detach so stale sessions don't accumulate.
      const viewerName = buildViewerSessionName(agent);
      const socketName = roster.tmux.socketName;

      await runChecked(runner, buildAgentWatchCreateViewerCommand(agent, viewerName, socketName));

      let exitCode = 0;
      try {
        const [bin, args] = splitCommand(buildAgentWatchAttachCommand(viewerName, socketName));
        exitCode = await iRunner(bin, args);
      } finally {
        // ALWAYS clean up the viewer session — even if attach threw or the process was
        // interrupted — so stale grouped *-watch-* sessions never accumulate. Errors here
        // are intentionally suppressed; the agent session is unaffected.
        const killResult = await runner(
          ...splitCommand(buildAgentWatchKillViewerCommand(viewerName, socketName)),
        );
        void killResult;
      }

      if (exitCode !== 0) {
        process.exitCode = exitCode;
      }
    });

  agentCommand
    .command('reset <agent>')
    .description('Reset a local fleet agent by sending the runtime reset command')
    .option('--clear', 'Send /clear')
    .option('--new', 'Send /new')
    .action(async (agent: string, opts: { clear?: boolean; new?: boolean }) => {
      const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
      const rosterAgent = getRosterAgent(roster, agent);
      const paths = resolveFleetPaths(resolveMosaicHomeFromCommand(agentCommand, deps.mosaicHome));
      const resetCommand = opts.clear
        ? '/clear'
        : opts.new
          ? '/new'
          : (roster.runtimes[rosterAgent.runtime]?.resetCommand ?? '/clear');
      await runChecked(
        runner,
        buildAgentResetCommand(paths, agent, resetCommand, roster.tmux.socketName),
      );
    });

  agentCommand
    .command('tail <agent>')
    .description('Print recent pane output for a local fleet agent')
    .option('-n, --lines <number>', 'Number of pane history lines', '80')
    .action(async (agent: string, opts: { lines: string }) => {
      const roster = await loadRosterFromAgentCommand(agentCommand, deps.mosaicHome);
      getRosterAgent(roster, agent);
      const lines = Number.parseInt(opts.lines, 10);
      const result = await runner(
        ...splitCommand(
          buildAgentTailCommand(agent, Number.isFinite(lines) ? lines : 80, roster.tmux.socketName),
        ),
      );
      writeCommandOutput(result);
    });
}

async function installFleet(cmd: Command, frameworkRoot: string): Promise<void> {
  const activePaths = resolveFleetPaths(cmd.opts<{ mosaicHome: string }>().mosaicHome);
  assertDefaultMosaicHomeForSystemd(activePaths.mosaicHome);
  const roster = await loadRosterForCommand(cmd);
  await mkdir(activePaths.fleetToolsDir, { recursive: true });
  await mkdir(activePaths.tmuxToolsDir, { recursive: true });
  await mkdir(activePaths.systemdUserDir, { recursive: true });
  await mkdir(activePaths.agentEnvDir, { recursive: true });

  const startAgentSessionPath = join(activePaths.fleetToolsDir, 'start-agent-session.sh');
  const startInteractionServicePath = join(
    activePaths.fleetToolsDir,
    'start-interaction-service.sh',
  );
  const printInteractionPolicyPath = join(
    activePaths.fleetToolsDir,
    'print-interaction-effective-policy.sh',
  );
  const sendMessagePath = join(activePaths.tmuxToolsDir, 'send-message.sh');
  const agentSendPath = join(activePaths.tmuxToolsDir, 'agent-send.sh');
  const executableToolPaths = [
    startAgentSessionPath,
    startInteractionServicePath,
    printInteractionPolicyPath,
    sendMessagePath,
    agentSendPath,
  ];
  await copyFile(
    join(frameworkRoot, 'tools', 'fleet', 'start-agent-session.sh'),
    startAgentSessionPath,
  );
  await copyFile(
    join(frameworkRoot, 'tools', 'fleet', 'start-interaction-service.sh'),
    startInteractionServicePath,
  );
  await copyFile(
    join(frameworkRoot, 'tools', 'fleet', 'print-interaction-effective-policy.sh'),
    printInteractionPolicyPath,
  );
  await copyFile(join(frameworkRoot, 'tools', 'tmux', 'send-message.sh'), sendMessagePath);
  await copyFile(join(frameworkRoot, 'tools', 'tmux', 'agent-send.sh'), agentSendPath);
  for (const toolPath of executableToolPaths) {
    await chmod(toolPath, 0o755);
  }
  await copyFile(
    join(frameworkRoot, 'systemd', 'user', 'mosaic-tmux-holder.service'),
    join(activePaths.systemdUserDir, 'mosaic-tmux-holder.service'),
  );
  await copyFile(
    join(frameworkRoot, 'systemd', 'user', 'mosaic-agent@.service'),
    join(activePaths.systemdUserDir, 'mosaic-agent@.service'),
  );
  await copyFile(
    join(frameworkRoot, 'systemd', 'user', 'mosaic-interaction-agent@.service'),
    join(activePaths.systemdUserDir, 'mosaic-interaction-agent@.service'),
  );

  for (const agent of roster.agents) {
    const envPath = join(activePaths.agentEnvDir, `${agent.name}.env`);
    const existingEnv = (await canRead(envPath)) ? await readFile(envPath, 'utf8') : undefined;
    await writeFile(envPath, mergeAgentEnv(generateAgentEnv(roster, agent), existingEnv));
  }

  console.log(`Installed fleet files for ${roster.agents.length} agent(s).`);
}

async function loadRosterForCommand(cmd: Command): Promise<FleetRoster> {
  const opts = cmd.opts<{ mosaicHome: string; roster?: string }>();
  return loadFleetRoster(await resolveRosterPath(opts.mosaicHome, opts.roster));
}

async function loadRosterFromAgentCommand(
  command: Command,
  mosaicHomeOverride?: string,
): Promise<FleetRoster> {
  const opts = command.optsWithGlobals<{ mosaicHome?: string; roster?: string }>();
  const mosaicHome = opts.mosaicHome ?? mosaicHomeOverride ?? defaultMosaicHome();
  return loadFleetRoster(await resolveRosterPath(mosaicHome, opts.roster));
}

function resolveMosaicHomeFromCommand(command: Command, override?: string): string {
  const opts = command.optsWithGlobals<{ mosaicHome?: string }>();
  return opts.mosaicHome ?? override ?? defaultMosaicHome();
}

function parseRosterText(text: string, path: string): RawFleetRoster {
  const trimmed = text.trim();
  if (path.endsWith('.json')) {
    return JSON.parse(trimmed) as RawFleetRoster;
  }
  return YAML.parse(trimmed) as RawFleetRoster;
}

function normalizeRoster(raw: RawFleetRoster): FleetRoster {
  assertObject(raw, 'Fleet roster');
  assertKnownKeys(raw, 'Fleet roster', [
    'version',
    'transport',
    'tmux',
    'defaults',
    'runtimes',
    'agents',
  ]);
  if (raw.tmux !== undefined) {
    assertObject(raw.tmux, 'Fleet roster tmux');
    assertKnownKeys(raw.tmux, 'Fleet roster tmux', [
      'socket_name',
      'socketName',
      'holder_session',
      'holderSession',
    ]);
  }
  if (raw.defaults !== undefined) {
    assertObject(raw.defaults, 'Fleet roster defaults');
    assertKnownKeys(raw.defaults, 'Fleet roster defaults', [
      'working_directory',
      'workingDirectory',
    ]);
  }
  if (raw.runtimes !== undefined) {
    assertObject(raw.runtimes, 'Fleet roster runtimes');
    for (const [runtime, config] of Object.entries(raw.runtimes)) {
      assertObject(config, `Fleet roster runtime "${runtime}"`);
      assertKnownKeys(config, `Fleet roster runtime "${runtime}"`, [
        'reset_command',
        'resetCommand',
      ]);
    }
  }
  if (raw.version !== 1) {
    throw new Error('Fleet roster version must be 1.');
  }
  if (raw.transport !== 'tmux') {
    throw new Error('Fleet roster transport must be "tmux".');
  }
  if (!Array.isArray(raw.agents) || raw.agents.length === 0) {
    throw new Error('Fleet roster must define at least one agent.');
  }

  const agents = raw.agents.map(normalizeAgent);
  assertUniqueAgentNames(agents);

  return {
    version: 1,
    transport: 'tmux',
    tmux: {
      // Absent socket_name ⇒ '' (the literal default tmux socket, no -L) — NOT
      // mosaic-fleet. Shipped presets set socket_name explicitly, so they are
      // unaffected; only socket-less rosters get default-socket behavior.
      socketName: stringValue(
        raw.tmux?.socket_name ?? raw.tmux?.socketName,
        '',
        'Fleet roster tmux socket_name',
      ),
      holderSession: stringValue(
        raw.tmux?.holder_session ?? raw.tmux?.holderSession,
        DEFAULT_HOLDER_SESSION,
        'Fleet roster tmux holder_session',
      ),
    },
    defaults: {
      workingDirectory: stringValue(
        raw.defaults?.working_directory ?? raw.defaults?.workingDirectory,
        DEFAULT_WORKING_DIRECTORY,
        'Fleet roster defaults working_directory',
      ),
    },
    runtimes: normalizeRuntimes(raw.runtimes as RawFleetRoster['runtimes']),
    agents,
  };
}

function normalizeAgent(raw: NonNullable<RawFleetRoster['agents']>[number]): FleetAgent {
  assertObject(raw, 'Fleet roster agent');
  assertKnownKeys(raw, 'Fleet roster agent', [
    'name',
    'alias',
    'provider',
    'runtime',
    'class',
    'working_directory',
    'workingDirectory',
    'model_hint',
    'modelHint',
    'reasoning_level',
    'reasoningLevel',
    'tool_policy',
    'toolPolicy',
    'persistent_persona',
    'persistentPersona',
    'reset_between_tasks',
    'resetBetweenTasks',
    'kickstart_template',
    'kickstartTemplate',
  ]);
  const name = stringValue(raw.name, '', 'Fleet roster agent name');
  const runtime = stringValue(
    raw.runtime,
    '',
    `Fleet roster agent "${name || '<unknown>'}" runtime`,
  );
  if (!name || !/^[A-Za-z0-9_.-]+$/.test(name)) {
    throw new Error(`Invalid fleet agent name: ${name || '<empty>'}`);
  }
  if (!runtime) {
    throw new Error(`Fleet agent "${name}" must define a runtime.`);
  }
  return {
    name,
    alias: optionalString(raw.alias, `Fleet roster agent "${name}" alias`),
    provider: optionalString(raw.provider, `Fleet roster agent "${name}" provider`),
    runtime,
    className: stringValue(raw.class, 'worker', `Fleet roster agent "${name}" class`),
    workingDirectory: optionalString(
      raw.working_directory ?? raw.workingDirectory,
      `Fleet roster agent "${name}" working_directory`,
    ),
    modelHint: optionalString(
      raw.model_hint ?? raw.modelHint,
      `Fleet roster agent "${name}" model_hint`,
    ),
    reasoningLevel: optionalString(
      raw.reasoning_level ?? raw.reasoningLevel,
      `Fleet roster agent "${name}" reasoning_level`,
    ),
    toolPolicy: optionalString(
      raw.tool_policy ?? raw.toolPolicy,
      `Fleet roster agent "${name}" tool_policy`,
    ),
    persistentPersona: optionalBooleanOrString(
      raw.persistent_persona ?? raw.persistentPersona,
      `Fleet roster agent "${name}" persistent_persona`,
    ),
    resetBetweenTasks: optionalBoolean(
      raw.reset_between_tasks ?? raw.resetBetweenTasks,
      `Fleet roster agent "${name}" reset_between_tasks`,
    ),
    kickstartTemplate: optionalString(
      raw.kickstart_template ?? raw.kickstartTemplate,
      `Fleet roster agent "${name}" kickstart_template`,
    ),
  };
}

function normalizeRuntimes(
  raw: RawFleetRoster['runtimes'] | undefined,
): Record<string, { resetCommand: string }> {
  const result: Record<string, { resetCommand: string }> = { ...DEFAULT_RUNTIME_RESETS };
  for (const [runtime, config] of Object.entries(raw ?? {})) {
    result[runtime] = {
      resetCommand: stringValue(
        config.reset_command ?? config.resetCommand,
        '/clear',
        `Fleet roster runtime "${runtime}" reset_command`,
      ),
    };
  }
  return result;
}

function assertObject(value: unknown, label: string): asserts value is Record<string, unknown> {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    throw new Error(`${label} must be an object.`);
  }
}

function assertKnownKeys(
  value: Record<string, unknown>,
  label: string,
  allowedKeys: readonly string[],
): void {
  const allowed = new Set(allowedKeys);
  const unknownKeys = Object.keys(value).filter((key) => !allowed.has(key));
  if (unknownKeys.length > 0) {
    throw new Error(`${label} has unknown field(s): ${unknownKeys.join(', ')}.`);
  }
}

function assertUniqueAgentNames(agents: FleetAgent[]): void {
  const seen = new Set<string>();
  for (const agent of agents) {
    if (seen.has(agent.name)) {
      throw new Error(`Fleet roster has duplicate agent name: ${agent.name}.`);
    }
    seen.add(agent.name);
  }
}

function stringValue(value: unknown, fallback = '', label = 'Value'): string {
  if (value === undefined) {
    return fallback;
  }
  if (typeof value !== 'string') {
    throw new Error(`${label} must be a string.`);
  }
  return value;
}

function optionalString(value: unknown, label = 'Value'): string | undefined {
  if (value === undefined) {
    return undefined;
  }
  if (typeof value !== 'string') {
    throw new Error(`${label} must be a string.`);
  }
  return value;
}

function optionalBoolean(value: unknown, label = 'Value'): boolean | undefined {
  if (value === undefined) {
    return undefined;
  }
  if (typeof value !== 'boolean') {
    throw new Error(`${label} must be a boolean.`);
  }
  return value;
}

function optionalBooleanOrString(value: unknown, label = 'Value'): boolean | string | undefined {
  if (value === undefined) {
    return undefined;
  }
  if (typeof value !== 'boolean' && typeof value !== 'string') {
    throw new Error(`${label} must be a boolean or string.`);
  }
  return value;
}

function expandHome(path: string): string {
  return path === '~' || path.startsWith('~/') ? join(homedir(), path.slice(2)) : path;
}

function shellEnvValue(value: string): string {
  // Empty ⇒ a bare `VAR=` (unambiguous empty in a systemd EnvironmentFile and
  // when shell-sourced). Quoting it as '' risks a literal two-char value (e.g.
  // a tmux socket named "''"), which would defeat the default-socket behavior.
  if (value === '') {
    return '';
  }
  if (/^[A-Za-z0-9_./:=@+-]+$/.test(value)) {
    return value;
  }
  return `'${value.replaceAll("'", "'\"'\"'")}'`;
}

async function stopFleetBestEffort(runner: CommandRunner, agentNames: string[]): Promise<void> {
  const failures: string[] = [];
  for (const agentName of agentNames) {
    const command = buildFleetServiceCommand('stop', agentName);
    const result = await runner(...splitCommand(command));
    writeSuccessfulCommandOutput(result);
    if (result.exitCode !== 0) {
      failures.push(result.stderr || result.stdout || `Command failed: ${command.join(' ')}`);
    }
  }

  const holderCommand = buildFleetServiceCommand('stop');
  const holderResult = await runner(...splitCommand(holderCommand));
  writeSuccessfulCommandOutput(holderResult);
  if (holderResult.exitCode !== 0) {
    failures.push(
      holderResult.stderr || holderResult.stdout || `Command failed: ${holderCommand.join(' ')}`,
    );
  }

  if (failures.length > 0) {
    throw new Error(
      `Fleet stop completed with ${failures.length} failure(s): ${failures.join('; ')}`,
    );
  }
}

async function runChecked(runner: CommandRunner, command: string[]): Promise<void> {
  const result = await runner(...splitCommand(command));
  if (result.exitCode !== 0) {
    throw new Error(result.stderr || result.stdout || `Command failed: ${command.join(' ')}`);
  }
  if (result.stdout) {
    process.stdout.write(result.stdout);
  }
}

function splitCommand(command: string[]): [string, string[]] {
  const [bin, ...args] = command;
  if (!bin) {
    throw new Error('Cannot run an empty command.');
  }
  return [bin, args];
}

/** All supported fleet profile names. */
export type FleetProfile =
  | 'general'
  | 'coding'
  | 'research'
  | 'hybrid'
  | 'minimal'
  | 'local-canary';

/** The list of all valid fleet profile names, for wizard menus and error messages. */
export const FLEET_PROFILES: readonly FleetProfile[] = [
  'general',
  'coding',
  'research',
  'hybrid',
  'minimal',
  'local-canary',
];

/**
 * Maps a fleet profile name to its example YAML filename (without the path).
 * Pure function — testable without I/O.
 */
export function resolvePresetFilename(profile: FleetProfile): string {
  return `${profile}.yaml`;
}

/**
 * Validate and normalise a fleet profile name string.
 * Throws with a clear message on unknown values.
 */
export function parseInitProfile(profile: string): FleetProfile {
  if ((FLEET_PROFILES as readonly string[]).includes(profile)) {
    return profile as FleetProfile;
  }
  throw new Error(`Unsupported fleet profile "${profile}". Use: ${FLEET_PROFILES.join(', ')}.`);
}

/**
 * Count orchestrator agents in a parsed roster.
 * Returns the count; callers assert === 1.
 */
export function countOrchestrators(roster: FleetRoster): number {
  return roster.agents.filter((a) => a.className === 'orchestrator').length;
}

/**
 * Count enhancer agents in a parsed roster. The two-agent floor (north-star)
 * requires every non-minimal fleet to carry at least one enhancer alongside the
 * sole orchestrator.
 */
export function countEnhancers(roster: FleetRoster): number {
  return roster.agents.filter((a) => a.className === 'enhancer').length;
}

/** Valid runtime identifiers for fleet agents. */
export const VALID_FLEET_RUNTIMES: readonly string[] = [
  'pi',
  'claude',
  'codex',
  'opencode',
  'dogfood',
];

/**
 * Add a new agent to a fleet roster (immutable — returns a new FleetRoster).
 * Throws on invalid name, duplicate name.
 */
export function addAgentToRoster(roster: FleetRoster, agent: FleetAgent): FleetRoster {
  if (!agent.name || !/^[A-Za-z0-9_.-]+$/.test(agent.name)) {
    throw new Error(`Invalid fleet agent name: ${agent.name || '<empty>'}`);
  }
  if (roster.agents.some((a) => a.name === agent.name)) {
    throw new Error(`Agent "${agent.name}" already exists in the fleet roster.`);
  }
  return {
    ...roster,
    agents: [...roster.agents, agent],
  };
}

/**
 * Remove an agent from a fleet roster (immutable — returns a new FleetRoster).
 * Throws if the agent is not found, or if removal would leave zero orchestrators.
 */
export function removeAgentFromRoster(roster: FleetRoster, name: string): FleetRoster {
  const agent = roster.agents.find((a) => a.name === name);
  if (!agent) {
    throw new Error(`Agent "${name}" is not in the fleet roster.`);
  }
  const remaining = roster.agents.filter((a) => a.name !== name);
  const remainingOrchCount = remaining.filter((a) => a.className === 'orchestrator').length;
  if (remainingOrchCount === 0 && agent.className === 'orchestrator') {
    throw new Error(
      `Cannot remove agent "${name}": it is the sole orchestrator. Add another orchestrator first (R5).`,
    );
  }
  // Two-agent floor: never drop the last enhancer (the continuous-improvement
  // loop). Symmetric with the sole-orchestrator guard.
  const remainingEnhancerCount = remaining.filter((a) => a.className === 'enhancer').length;
  if (remainingEnhancerCount === 0 && agent.className === 'enhancer') {
    throw new Error(
      `Cannot remove agent "${name}": it is the sole enhancer. Every fleet keeps at least one ` +
        `enhancer (two-agent floor). Add another enhancer first.`,
    );
  }
  return {
    ...roster,
    agents: remaining,
  };
}

/**
 * Serialize a FleetRoster to YAML text (snake_case keys).
 * The output is parseable by loadFleetRoster.
 */
export function serializeRosterToYaml(roster: FleetRoster): string {
  const agents = roster.agents.map((agent) => {
    const raw: Record<string, unknown> = {
      name: agent.name,
      runtime: agent.runtime,
      class: agent.className,
    };
    if (agent.alias !== undefined) {
      raw['alias'] = agent.alias;
    }
    if (agent.provider !== undefined) {
      raw['provider'] = agent.provider;
    }
    if (agent.workingDirectory !== undefined) {
      raw['working_directory'] = agent.workingDirectory;
    }
    if (agent.modelHint !== undefined) {
      raw['model_hint'] = agent.modelHint;
    }
    if (agent.reasoningLevel !== undefined) {
      raw['reasoning_level'] = agent.reasoningLevel;
    }
    if (agent.toolPolicy !== undefined) {
      raw['tool_policy'] = agent.toolPolicy;
    }
    if (agent.persistentPersona !== undefined) {
      raw['persistent_persona'] = agent.persistentPersona;
    }
    if (agent.resetBetweenTasks !== undefined) {
      raw['reset_between_tasks'] = agent.resetBetweenTasks;
    }
    if (agent.kickstartTemplate !== undefined) {
      raw['kickstart_template'] = agent.kickstartTemplate;
    }
    return raw;
  });

  const runtimes: Record<string, { reset_command: string }> = {};
  for (const [runtime, config] of Object.entries(roster.runtimes)) {
    runtimes[runtime] = { reset_command: config.resetCommand };
  }

  const raw: Record<string, unknown> = {
    version: roster.version,
    transport: roster.transport,
    tmux: {
      socket_name: roster.tmux.socketName,
      holder_session: roster.tmux.holderSession,
    },
    defaults: {
      working_directory: roster.defaults.workingDirectory,
    },
    runtimes,
    agents,
  };

  return YAML.stringify(raw);
}

/**
 * Prompt interactively for a fleet profile via stdin readline.
 * AI-free: no LLM calls — pure readline menu.
 * Resolves with the chosen profile string, or rejects on I/O error.
 */
function promptFleetProfile(): Promise<FleetProfile> {
  return new Promise((resolve, reject) => {
    const rl = readline.createInterface({ input: process.stdin, output: process.stdout });
    const menu = [
      '',
      'Choose a fleet configuration type:',
      '  1) general      — orchestrator + generalist worker',
      '  2) coding       — orchestrator + coder0 + coder1 + reviewer',
      '  3) research     — orchestrator + researcher0 + researcher1 + analyst',
      '  4) hybrid       — orchestrator + coder0 + researcher0 + reviewer',
      '  5) minimal      — single canary-pi agent (no orchestrator)',
      '  6) local-canary — legacy canary preset with lead + coder + reviewer',
      '',
    ].join('\n');
    process.stdout.write(menu);
    rl.question('Enter number or name [1]: ', (answer) => {
      rl.close();
      const trimmed = answer.trim();
      // Map numeric shortcut → name
      const byNumber: Record<string, FleetProfile> = {
        '1': 'general',
        '2': 'coding',
        '3': 'research',
        '4': 'hybrid',
        '5': 'minimal',
        '6': 'local-canary',
        '': 'general', // default on empty enter
      };
      if (trimmed in byNumber) {
        resolve(byNumber[trimmed]!);
        return;
      }
      try {
        resolve(parseInitProfile(trimmed));
      } catch (err) {
        reject(err);
      }
    });
  });
}

function writeCommandOutput(result: CommandResult): void {
  if (result.stdout) {
    process.stdout.write(result.stdout);
  } else if (result.stderr) {
    process.stderr.write(result.stderr);
  }
  setExitCodeFromResult(result);
}

function writeSuccessfulCommandOutput(result: CommandResult): void {
  if (result.exitCode !== 0) {
    return;
  }
  if (result.stdout) {
    process.stdout.write(result.stdout);
  }
}

function setExitCodeFromResult(result: CommandResult): void {
  if (result.exitCode !== 0) {
    process.exitCode = result.exitCode;
  }
}

function runCommand(command: string, args: string[]): Promise<CommandResult> {
  return new Promise((resolvePromise) => {
    const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'pipe'] });
    let stdout = '';
    let stderr = '';
    child.stdout.on('data', (chunk: Buffer) => {
      stdout += chunk.toString('utf8');
    });
    child.stderr.on('data', (chunk: Buffer) => {
      stderr += chunk.toString('utf8');
    });
    child.on('error', (error) => {
      resolvePromise({ stdout, stderr: error.message, exitCode: 127 });
    });
    child.on('close', (code) => {
      resolvePromise({ stdout, stderr, exitCode: code ?? 1 });
    });
  });
}

function resolveFrameworkRoot(): string {
  const currentFile = fileURLToPath(import.meta.url);
  return resolve(dirname(currentFile), '..', '..', 'framework');
}

/**
 * Default InteractiveRunner implementation: spawns the command with inherited
 * stdio so the terminal is passed through to the child process. This is required
 * for commands like `tmux attach` that are full-screen interactive and cannot be
 * captured through a pipe.
 */
function spawnInteractive(command: string, args: string[]): Promise<number> {
  return new Promise((resolvePromise) => {
    const child = spawn(command, args, { stdio: 'inherit' });
    child.on('error', () => {
      resolvePromise(127);
    });
    child.on('close', (code) => {
      resolvePromise(code ?? 1);
    });
  });
}

/**
 * Default SleepFn implementation backed by setTimeout.
 * Tests inject a stub to avoid real delays in the send --verify polling loop.
 */
function defaultSleep(ms: number): Promise<void> {
  return new Promise<void>((res) => setTimeout(res, ms));
}

async function canRead(path: string): Promise<boolean> {
  try {
    await access(path, constants.R_OK);
    return true;
  } catch {
    return false;
  }
}

export async function resolveRosterPath(
  mosaicHome: string,
  explicitPath?: string,
): Promise<string> {
  if (explicitPath) {
    return explicitPath;
  }
  const yamlPath = resolveFleetPaths(mosaicHome).rosterPath;
  if (await canRead(yamlPath)) {
    return yamlPath;
  }
  const jsonPath = join(mosaicHome, 'fleet', 'roster.json');
  return jsonPath;
}
