fargo
8e8879ca34
rev-security-03 REQUEST_CHANGES (review note on the brain), all four findings verified by measurement before fixing: F1 BLOCKER — brokerSocketPresent used access(path, S_IFSOCK). S_IFSOCK (0xC000) is a file-type constant, not an access() mode (0-7): on node 24.18.0 the call throws ERR_OUT_OF_RANGE, so the probe could NEVER return true and the swallow-catch made every un-seamed call report the broker absent — on this host (roster v1) 'mosaic fleet start' would have refused broker-absent forever. Now stat().isSocket(), matching the bash side's [ -S ]. New spec exercises the REAL probe against a REAL unix socket (true), a regular file (false), and an absent path (false) — no seam, so no seam can hide this again. F2 — placeUnitFile's single catch conflated destination-absent with unlink-FAILED; on unlink failure it copied through the still-live symlink (copy-through overwrite measured by the reviewer). Now: ENOENT is the only swallowed lstat outcome; unlink failure aborts with a named UnitPlacementError BEFORE any copy. New spec proves the residue target's bytes survive an unlink failure and the destination link is untouched. F3 — observeBroker defaulted unit/socket to false when seams unset, and production injects none: plan/status/doctor reported a healthy broker as absent. Seams still take precedence; with no seam the real stat() probes run. The v2 start path (and apply-with-running) now re-check the socket after enable+start with a NAMED lifecycle-precondition-failed refusal — previously only the v1 path had that protection, and the refusal was masked into the generic recoverable result. Acceptance fixtures gain hermetic brokerSocketEnv paths (the old fixture asserted the lying default). F4 (note) — unlink race stays inside the user-owned dir; no action. Local gates on this tree: vitest 1566/1566, typecheck, lint, prettier 3.8.1, verify-sanitized all pass. CI 2468's sanitization failure did not reproduce locally on the identical tree; watching the fresh pipeline.
3199 lines
118 KiB
TypeScript
3199 lines
118 KiB
TypeScript
import { constants, type Stats } from 'node:fs';
|
|
import {
|
|
access,
|
|
chmod,
|
|
copyFile,
|
|
lstat,
|
|
mkdir,
|
|
open,
|
|
readFile,
|
|
readlink,
|
|
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 { fleetAgentEnvDir } from '../fleet/brain-home.js';
|
|
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 {
|
|
FleetRosterConfigurationError,
|
|
getRosterAgent,
|
|
loadFleetRoster,
|
|
parseFleetRosterDocument,
|
|
readFleetRosterText,
|
|
resolveInstalledFleetRosterPath,
|
|
type FleetAgent,
|
|
type FleetRoster,
|
|
} from '../fleet/fleet-roster-v1.js';
|
|
export {
|
|
getRosterAgent,
|
|
loadFleetRoster,
|
|
resolveInstalledFleetRosterPath,
|
|
} from '../fleet/fleet-roster-v1.js';
|
|
export type { FleetAgent, FleetRoster } from '../fleet/fleet-roster-v1.js';
|
|
import { parseRosterV2 } from '../fleet/roster-v2.js';
|
|
import {
|
|
registerFleetAgentCrudCommands,
|
|
type FleetAgentCrudCommandDeps,
|
|
} from './fleet-agent-crud-command.js';
|
|
import {
|
|
registerFleetMigrationCommand,
|
|
type FleetMigrationCommandDeps,
|
|
} from './fleet-migration-command.js';
|
|
import {
|
|
executeReconcilerCommandJson,
|
|
registerFleetReconcilerCommands,
|
|
type FleetReconcilerCommandDeps,
|
|
} from './fleet-reconciler-command.js';
|
|
import { registerFleetRegenCommand } from './fleet-regen-command.js';
|
|
import { resolveCommsBlock } from '../fleet/comms-onboarding.js';
|
|
import {
|
|
applyPreparedAgentEnvironmentProjection,
|
|
ensureFleetHolderIdentity,
|
|
GENERATED_AGENT_ENV_SUPPORTED_RUNTIMES,
|
|
parseAgentEnvironment,
|
|
prepareAgentEnvironmentProjection,
|
|
renderGeneratedAgentEnvironment,
|
|
writeAgentEnvironmentProjection,
|
|
writeManagedFleetRoster,
|
|
} from '../fleet/generated-env-boundary.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;
|
|
/** Test seam for the #1292 fleet-start broker preflight (socket presence). */
|
|
checkBrokerSocket?: (path: string) => Promise<boolean> | boolean;
|
|
/** 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;
|
|
projectionApplier?: FleetAgentCrudCommandDeps['projectionApplier'];
|
|
reconcileDeps?: FleetReconcilerCommandDeps['reconcileDeps'];
|
|
migrationDeps?: Omit<FleetMigrationCommandDeps, 'mosaicHome'>;
|
|
}
|
|
|
|
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';
|
|
|
|
/**
|
|
* 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;
|
|
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: fleetAgentEnvDir(mosaicHome),
|
|
};
|
|
}
|
|
|
|
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.`,
|
|
);
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 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 {
|
|
return renderGeneratedAgentEnvironment(generateAgentEnvValues(roster, agent));
|
|
}
|
|
|
|
/**
|
|
* Compatibility shim for callers that previously merged site-owned data into a
|
|
* generated file. Local data is validated but never appended to the generated
|
|
* projection, preventing a local override from becoming hidden authority.
|
|
*/
|
|
export function mergeAgentEnv(generatedEnv: string, existingEnv?: string): string {
|
|
parseAgentEnvironment(generatedEnv, 'generated');
|
|
if (existingEnv?.trim()) parseAgentEnvironment(existingEnv, 'local');
|
|
return generatedEnv;
|
|
}
|
|
|
|
function generateAgentEnvValues(
|
|
roster: FleetRoster,
|
|
agent: FleetAgent,
|
|
): Readonly<Record<string, string>> {
|
|
const workingDirectory = agent.workingDirectory ?? roster.defaults.workingDirectory;
|
|
return {
|
|
MOSAIC_AGENT_NAME: agent.name,
|
|
MOSAIC_GIT_IDENTITY: agent.name,
|
|
MOSAIC_AGENT_CLASS: agent.className,
|
|
MOSAIC_AGENT_RUNTIME: agent.runtime,
|
|
MOSAIC_AGENT_MODEL: agent.modelHint ?? '',
|
|
MOSAIC_AGENT_REASONING: agent.reasoningLevel ?? '',
|
|
MOSAIC_AGENT_TOOL_POLICY: agent.toolPolicy ?? '',
|
|
MOSAIC_AGENT_WORKDIR: workingDirectory,
|
|
MOSAIC_TMUX_SOCKET: agent.socket ?? roster.tmux.socketName,
|
|
};
|
|
}
|
|
|
|
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];
|
|
}
|
|
|
|
/**
|
|
* Place a unit file into the ACTIVE systemd user directory, never through a
|
|
* symlink (#1292, measured 2026-08-17).
|
|
*
|
|
* ⚠ SET-INDEPENDENCE (fomo-lin, 2026-08-17): the set of unit names carrying
|
|
* by-path residue and the set of unit names this install copies are
|
|
* INDEPENDENT. Until 0.0.50 they were disjoint only by accident of which
|
|
* units the install happened to name — fomo-lin survived copy-through solely
|
|
* because its one by-path symlink (the broker) was the one unit the install
|
|
* did NOT copy. Adding the broker to the copy set made the intersection
|
|
* non-empty on the first run. Whoever adds a fifth unit to the placement
|
|
* list inherits this helper and its unlink step; do not place units with a
|
|
* bare copyFile.
|
|
*
|
|
* A host provisioned by the enable-by-path convention carries a symlink AT
|
|
* the unit-name path in ~/.config/systemd/user/ pointing at the shipped
|
|
* template under ~/.config/mosaic/systemd/user/. Node's copyFile FOLLOWS
|
|
* that link and overwrites the SEED template instead of placing the active
|
|
* unit (verified with fs.copyFile on a throwaway systemd user instance) —
|
|
* silent, rc=0, and it mutates the directory every later reseed reads from.
|
|
* The same measurement showed `systemctl enable <name>` does NOT rewrite an
|
|
* existing by-path wants-symlink, so reconciliation must be explicit.
|
|
*
|
|
* Placement therefore: if the destination is a symlink, unlink it first
|
|
* (unlink → copy — copy-then-unlink would mutate the seed and then destroy
|
|
* the evidence that it did); then copy. Also removes a stale
|
|
* `default.target.wants/<name>` symlink that points outside the active
|
|
* directory (readlink — NOT readFile, which follows the link and returns the
|
|
* target's CONTENT), so the subsequent enable-by-name recreates it against
|
|
* the active copy. Idempotent: on a clean or already-reconciled destination
|
|
* every step is a no-op (the copy rewrites identical bytes).
|
|
*
|
|
* Returns what was done, for assertions and install reporting.
|
|
*/
|
|
export interface PlaceUnitResult {
|
|
readonly unit: string;
|
|
readonly destination: string;
|
|
/** A symlink at the unit-name path was unlinked (by-path residue). */
|
|
readonly unlinkedDestinationSymlink: boolean;
|
|
/** A stale wants-symlink pointing outside the active dir was removed. */
|
|
readonly removedStaleWantsSymlink: boolean;
|
|
}
|
|
|
|
/**
|
|
* placeUnitFile failed. Thrown BEFORE any copy: no destination bytes were
|
|
* written, so a residue target cannot have been clobbered by a copy-through
|
|
* (#1297 review F2).
|
|
*/
|
|
export class UnitPlacementError extends Error {
|
|
constructor(
|
|
readonly unit: string,
|
|
message: string,
|
|
) {
|
|
super(message);
|
|
this.name = UnitPlacementError.name;
|
|
}
|
|
}
|
|
|
|
function isErrnoException(error: unknown): error is NodeJS.ErrnoException {
|
|
return error instanceof Error && 'code' in error && typeof error.code === 'string';
|
|
}
|
|
|
|
export async function placeUnitFile(
|
|
source: string,
|
|
systemdUserDir: string,
|
|
unit: string,
|
|
): Promise<PlaceUnitResult> {
|
|
const destination = join(systemdUserDir, unit);
|
|
let unlinkedDestinationSymlink = false;
|
|
// Destination-absent and unlink-FAILED are different outcomes and must not
|
|
// share a catch (#1297 review F2): a swallowed unlink error used to fall
|
|
// through to copyFile through the still-live symlink, silently reintroducing
|
|
// the exact copy-through this helper exists to prevent.
|
|
let destinationInfo: Stats | undefined;
|
|
try {
|
|
destinationInfo = await lstat(destination);
|
|
} catch (error) {
|
|
if (!isErrnoException(error) || error.code !== 'ENOENT') {
|
|
throw new UnitPlacementError(
|
|
unit,
|
|
`cannot inspect destination ${destination}: ${error instanceof Error ? error.message : String(error)}`,
|
|
);
|
|
}
|
|
// ENOENT: absent destination — nothing to unlink, copy below is safe.
|
|
}
|
|
if (destinationInfo?.isSymbolicLink()) {
|
|
try {
|
|
await unlink(destination);
|
|
} catch (error) {
|
|
// Abort BEFORE the copy: proceeding would run copyFile through the
|
|
// still-live symlink and overwrite the residue target's bytes.
|
|
throw new UnitPlacementError(
|
|
unit,
|
|
`cannot unlink destination symlink ${destination}: ${error instanceof Error ? error.message : String(error)}`,
|
|
);
|
|
}
|
|
unlinkedDestinationSymlink = true;
|
|
}
|
|
await copyFile(source, destination);
|
|
|
|
let removedStaleWantsSymlink = false;
|
|
const wantsLink = join(systemdUserDir, 'default.target.wants', unit);
|
|
try {
|
|
const wantsInfo = await lstat(wantsLink);
|
|
if (wantsInfo.isSymbolicLink()) {
|
|
// readlink — NOT readFile: readFile FOLLOWS the link and returns the
|
|
// target file's CONTENT, which is not the question being asked.
|
|
let target: string | undefined;
|
|
try {
|
|
target = await readlink(wantsLink);
|
|
} catch {
|
|
target = undefined;
|
|
}
|
|
// Normalize (systemctl writes absolute targets; a relative one resolves
|
|
// against the wants dir). A wants-symlink pointing anywhere other than
|
|
// the active copy (the by-path convention points at the seed template)
|
|
// survives enable-by-name unchanged — remove it so enable recreates it.
|
|
if (target !== undefined && resolve(dirname(wantsLink), target) !== destination) {
|
|
await unlink(wantsLink);
|
|
removedStaleWantsSymlink = true;
|
|
}
|
|
}
|
|
} catch {
|
|
// absent wants link — nothing to reconcile
|
|
}
|
|
|
|
return { unit, destination, unlinkedDestinationSymlink, removedStaleWantsSymlink };
|
|
}
|
|
|
|
/**
|
|
* 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: { readonly agents: readonly { readonly name: string }[] },
|
|
opts: { enable?: boolean },
|
|
): Promise<void> {
|
|
if (opts.enable === false) {
|
|
return;
|
|
}
|
|
try {
|
|
let succeeded = 0;
|
|
let failed = 0;
|
|
|
|
// The lease broker ships with the fleet and every gated runtime needs it
|
|
// (#1292): seats die at lease registration without it, and no documented
|
|
// path ever enabled it. Enabled first — alongside the holder — and the
|
|
// unit must have been placed by installFleet's placeUnitFile step.
|
|
const brokerResult = await runner(
|
|
...splitCommand(buildSystemdEnableCommand('mosaic-lease-broker.service')),
|
|
);
|
|
if (brokerResult.exitCode === 0) {
|
|
succeeded++;
|
|
} else {
|
|
failed++;
|
|
process.stderr.write(
|
|
`Warning: could not enable mosaic-lease-broker.service: ${brokerResult.stderr || brokerResult.stdout || 'non-zero exit'}\n`,
|
|
);
|
|
}
|
|
|
|
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.`,
|
|
);
|
|
}
|
|
if (resolve(destination) === resolve(activePaths.rosterPath)) {
|
|
await writeManagedFleetRoster(activePaths.mosaicHome, destination, content);
|
|
} else {
|
|
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, runner);
|
|
// Unit enablement needs agent names only, so it reads either version.
|
|
const roster = await loadRosterReadModel(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, runner);
|
|
// Unit enablement needs agent names only, so it reads either version.
|
|
const roster = await loadRosterReadModel(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`)
|
|
.option('--expected-generation <number>', 'Authoritative roster generation for roster-v2')
|
|
.option('--dry-run', 'Plan roster-v2 lifecycle effects without mutation')
|
|
.action(
|
|
async (
|
|
agent: string | undefined,
|
|
opts: { expectedGeneration?: string; dryRun?: boolean },
|
|
) => {
|
|
if (await usesRosterV2ControlPlane(cmd)) {
|
|
await executeReconcilerCommandJson(
|
|
cmd,
|
|
{ runner, mosaicHome: deps.mosaicHome, reconcileDeps: deps.reconcileDeps },
|
|
action,
|
|
opts,
|
|
agent,
|
|
);
|
|
return;
|
|
}
|
|
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 === 'start') {
|
|
// Broker preflight (#1292), re-probed on EVERY invocation: a
|
|
// gated runtime started without a live lease broker dies ~4s in
|
|
// while the unit reports active (RemainAfterExit) — enabling +
|
|
// starting here and then RE-CHECKING the socket refuses loudly
|
|
// instead of reporting rc0 over a doomed start. This is the
|
|
// second-start check as much as the first: it never trusts unit
|
|
// ActiveState.
|
|
await runChecked(runner, [
|
|
'systemctl',
|
|
'--user',
|
|
'enable',
|
|
'mosaic-lease-broker.service',
|
|
]);
|
|
await runChecked(runner, [
|
|
'systemctl',
|
|
'--user',
|
|
'start',
|
|
'mosaic-lease-broker.service',
|
|
]);
|
|
if (!(await brokerSocketPresent(deps))) {
|
|
console.error(
|
|
'[fleet] broker-absent: lease broker socket did not appear after enable+start (#1292).',
|
|
);
|
|
console.error(
|
|
'[fleet] remedy: mosaic fleet install (it reconciles either enable convention)',
|
|
);
|
|
process.exitCode = 1;
|
|
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 (await usesRosterV2ControlPlane(cmd)) {
|
|
await executeReconcilerCommandJson(
|
|
cmd,
|
|
{ runner, mosaicHome: deps.mosaicHome, reconcileDeps: deps.reconcileDeps },
|
|
'status',
|
|
{},
|
|
agent,
|
|
);
|
|
return;
|
|
}
|
|
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 () => {
|
|
if (await usesRosterV2ControlPlane(cmd)) {
|
|
await executeReconcilerCommandJson(
|
|
cmd,
|
|
{ runner, mosaicHome: deps.mosaicHome, reconcileDeps: deps.reconcileDeps },
|
|
'verify',
|
|
{},
|
|
);
|
|
return;
|
|
}
|
|
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);
|
|
// ps only reads, so it takes the version-agnostic read model rather than
|
|
// the v1 parser, which rejects a v2 roster outright.
|
|
const roster = await loadRosterReadModel(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 (await usesRosterV2ControlPlane(cmd)) {
|
|
// command.error, not a bare throw: this is operator guidance, and a
|
|
// bare throw reaches the top level uncaught and prints it under a Node
|
|
// stack trace. Measured on canary — the message is the whole point of
|
|
// the refusal, so it has to arrive readable.
|
|
cmd.error(rosterV2MutationGuidance('add', 'create', name), {
|
|
code: 'fleet.roster-v2',
|
|
exitCode: 1,
|
|
});
|
|
}
|
|
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 loadRosterAtPath(cmd, 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);
|
|
const generated = generateAgentEnvValues(updatedRoster, newAgent);
|
|
// Preflight every deterministic projection, local, legacy, quarantine,
|
|
// and managed-directory condition before roster authority changes.
|
|
const preparedProjection = await prepareAgentEnvironmentProjection({
|
|
mosaicHome: activePaths.mosaicHome,
|
|
agentEnvDir: activePaths.agentEnvDir,
|
|
agentName: name,
|
|
generated,
|
|
});
|
|
await writeFile(rosterPath, serializeRosterToYaml(updatedRoster));
|
|
await applyPreparedAgentEnvironmentProjection(preparedProjection);
|
|
|
|
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 }) => {
|
|
if (await usesRosterV2ControlPlane(cmd)) {
|
|
cmd.error(rosterV2MutationGuidance('remove', 'delete', name), {
|
|
code: 'fleet.roster-v2',
|
|
exitCode: 1,
|
|
});
|
|
}
|
|
const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
|
|
const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
|
|
const rosterPath = await resolveRosterPath(commandOpts.mosaicHome, commandOpts.roster);
|
|
const roster = await loadRosterAtPath(cmd, 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 the non-authoritative generated projection and heartbeat files
|
|
// (best-effort, non-fatal). Operator local/quarantine data is retained.
|
|
if (!opts.keepFiles) {
|
|
try {
|
|
await unlink(join(activePaths.agentEnvDir, `${name}.env.generated`));
|
|
} 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);
|
|
|
|
// Roster-v2 desired-state mutations belong directly to the fleet control
|
|
// plane; they do not share the root `mosaic agent` gateway-backed surface.
|
|
registerFleetAgentCrudCommands(cmd, deps);
|
|
registerFleetMigrationCommand(cmd, {
|
|
...deps.migrationDeps,
|
|
mosaicHome: deps.mosaicHome,
|
|
});
|
|
registerFleetReconcilerCommands(cmd, {
|
|
runner,
|
|
mosaicHome: deps.mosaicHome,
|
|
reconcileDeps: deps.reconcileDeps,
|
|
});
|
|
|
|
// Recovery (#791 PR3): rebuild roster-derived env projections from the roster
|
|
// SSOT. Projection-only and preview-first — never issues a lifecycle restart.
|
|
registerFleetRegenCommand(cmd, {
|
|
runner,
|
|
mosaicHome: deps.mosaicHome,
|
|
// Resolve personas the SAME way reconcile does: forward any configured roots
|
|
// so a custom-persona-root deployment cannot have reconcile accept a roster
|
|
// that regen then rejects against the default `<mosaicHome>/fleet/roles`.
|
|
rolesDir: deps.reconcileDeps?.rolesDir,
|
|
overrideDir: deps.reconcileDeps?.overrideDir,
|
|
});
|
|
|
|
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 <exact-member>')
|
|
.description('Print the Fleet Comms contract for one exact roster member')
|
|
.action((exactMember: string) => {
|
|
const mosaicHome = resolveMosaicHomeFromCommand(agentCommand, deps.mosaicHome);
|
|
const res = resolveCommsBlock(mosaicHome, exactMember);
|
|
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,
|
|
runner: CommandRunner,
|
|
): Promise<void> {
|
|
const activePaths = resolveFleetPaths(cmd.opts<{ mosaicHome: string }>().mosaicHome);
|
|
assertDefaultMosaicHomeForSystemd(activePaths.mosaicHome);
|
|
// Read model first: every file this function places is roster-independent, and
|
|
// the v1 parser would reject a v2 roster before any of them were written.
|
|
const roster = await loadRosterReadModel(cmd);
|
|
await ensureFleetHolderIdentity(activePaths.mosaicHome);
|
|
await mkdir(activePaths.fleetToolsDir, { recursive: true });
|
|
await mkdir(activePaths.tmuxToolsDir, { recursive: true });
|
|
await mkdir(activePaths.systemdUserDir, { recursive: true });
|
|
|
|
const startAgentSessionPath = join(activePaths.fleetToolsDir, 'start-agent-session.sh');
|
|
const startInteractionServicePath = join(
|
|
activePaths.fleetToolsDir,
|
|
'start-interaction-service.sh',
|
|
);
|
|
const startTmuxHolderPath = join(activePaths.fleetToolsDir, 'start-tmux-holder.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,
|
|
startTmuxHolderPath,
|
|
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', 'start-tmux-holder.sh'),
|
|
startTmuxHolderPath,
|
|
);
|
|
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);
|
|
}
|
|
// Unit placement (#1292): every unit goes through placeUnitFile — never a
|
|
// bare copyFile — so a by-path-enable symlink at the destination is
|
|
// unlinked rather than written through (copy-through would silently
|
|
// overwrite the SEED template, measured 2026-08-17). The lease broker unit
|
|
// is placed here too: previously the install named three units and omitted
|
|
// the broker entirely, which is why no documented path ever enabled it.
|
|
const placedUnits = await Promise.all(
|
|
[
|
|
'mosaic-tmux-holder.service',
|
|
'[email protected]',
|
|
'[email protected]',
|
|
'mosaic-lease-broker.service',
|
|
].map((unit) =>
|
|
placeUnitFile(join(frameworkRoot, 'systemd', 'user', unit), activePaths.systemdUserDir, unit),
|
|
),
|
|
);
|
|
const reconciled = placedUnits.filter(
|
|
(result) => result.unlinkedDestinationSymlink || result.removedStaleWantsSymlink,
|
|
);
|
|
if (reconciled.length > 0) {
|
|
console.log(
|
|
`Reconciled ${reconciled.length} unit placement(s) from by-path enable residue: ${reconciled.map((r) => r.unit).join(', ')}`,
|
|
);
|
|
}
|
|
// systemd will not see a replaced unit file without a reload; do it once
|
|
// after all placements, before any enable call below. runCommand never
|
|
// rejects (it resolves exitCode 127 on spawn error), so a plain await with
|
|
// an exitCode check matches the rest of this file's systemctl handling.
|
|
const reloadResult = await runner(...splitCommand(['systemctl', '--user', 'daemon-reload']));
|
|
if (reloadResult.exitCode !== 0) {
|
|
process.stderr.write(
|
|
`Warning: systemctl --user daemon-reload after unit placement failed (non-systemd host?): ${reloadResult.stderr || reloadResult.stdout || 'non-zero exit'}\n`,
|
|
);
|
|
}
|
|
|
|
// On roster v2 the reconciler owns the generated env: `apply` writes it and
|
|
// `regen` rebuilds it, both from projectRosterV2AgentGeneratedEnv. Writing it
|
|
// here too — necessarily through the v1 mapping — would be the third writer of
|
|
// one file and would break the #791 single-SSOT invariant. So v2 gets the tool
|
|
// files and the units, and nothing else.
|
|
if (roster.version === 2) {
|
|
console.log(
|
|
`Installed fleet tools and systemd units for ${roster.agents.length} agent(s). ` +
|
|
`Generated env is owned by the reconciler on roster v2 — run: mosaic fleet apply --expected-generation <n>`,
|
|
);
|
|
return;
|
|
}
|
|
|
|
const v1Roster = await loadRosterForCommand(cmd);
|
|
for (const agent of v1Roster.agents) {
|
|
await writeAgentEnvironmentProjection({
|
|
mosaicHome: activePaths.mosaicHome,
|
|
agentEnvDir: activePaths.agentEnvDir,
|
|
agentName: agent.name,
|
|
generated: generateAgentEnvValues(v1Roster, agent),
|
|
});
|
|
}
|
|
|
|
console.log(`Installed fleet files for ${v1Roster.agents.length} agent(s).`);
|
|
}
|
|
|
|
async function loadRosterForCommand(cmd: Command): Promise<FleetRoster> {
|
|
const opts = cmd.opts<{ mosaicHome: string; roster?: string }>();
|
|
return loadRosterAtPath(cmd, await resolveRosterPath(opts.mosaicHome, opts.roster));
|
|
}
|
|
|
|
/** Routes only a v2 roster to the M3 desired-state control plane; v1 aliases stay compatible. */
|
|
async function usesRosterV2ControlPlane(cmd: Command): Promise<boolean> {
|
|
const opts = cmd.opts<{ mosaicHome: string; roster?: string }>();
|
|
const path = await resolveRosterPath(opts.mosaicHome, opts.roster);
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = parseFleetRosterDocument(await readFleetRosterText(path), path);
|
|
} catch (error) {
|
|
reportFleetRosterConfigurationError(cmd, error);
|
|
}
|
|
return (
|
|
typeof parsed === 'object' &&
|
|
parsed !== null &&
|
|
!Array.isArray(parsed) &&
|
|
'version' in parsed &&
|
|
parsed.version === 2
|
|
);
|
|
}
|
|
|
|
/**
|
|
* `add`/`remove` and `create`/`delete` are not two spellings of one operation.
|
|
* The v1 pair edits the roster *and* drives systemd; the v2 pair is documented
|
|
* as changing desired state "without runtime actions", leaving convergence to
|
|
* `apply`. `add` also collects four fields where a v2 agent requires eleven, so
|
|
* routing it to `create` would mean inventing provider, alias, reasoning and
|
|
* tool-policy defaults on the operator's behalf. Refusing with the real command
|
|
* is honest; silently guessing an agent's provider is not.
|
|
*/
|
|
function rosterV2MutationGuidance(
|
|
v1Command: 'add' | 'remove',
|
|
v2Command: 'create' | 'delete',
|
|
name: string,
|
|
): string {
|
|
const target = v2Command === 'delete' ? ` ${name}` : '';
|
|
return (
|
|
`mosaic fleet ${v1Command} does not operate on a roster-v2 fleet. ` +
|
|
`Roster v2 separates desired state from convergence:\n` +
|
|
` 1. mosaic fleet ${v2Command}${target} --expected-generation <current> ` +
|
|
`${v2Command === 'create' ? "--agent '<json>' " : ''}` +
|
|
`(edits the roster only)\n` +
|
|
` 2. mosaic fleet apply --expected-generation <new> (converges systemd and tmux)\n` +
|
|
`Read the current generation with: mosaic fleet status`
|
|
);
|
|
}
|
|
|
|
/**
|
|
* The read-only fields shared by roster v1 and v2, for the commands that only
|
|
* ever *read* the roster (`ps`, and unit enablement inside `install`).
|
|
*
|
|
* This is deliberately NOT a v2→v1 downshift. A downshifted `FleetRoster` would
|
|
* be accepted by `generateAgentEnvValues`, and that would make a third writer of
|
|
* `fleet/agents/<name>.env.generated` — through the v1 mapping — breaking the
|
|
* #791 single-SSOT invariant that {@link projectRosterV2AgentGeneratedEnv} is
|
|
* documented to hold. Keeping the read model this small makes that misuse
|
|
* impossible: there is nothing here to write a roster or an env file back from.
|
|
*/
|
|
interface FleetRosterReadModel {
|
|
readonly version: 1 | 2;
|
|
readonly tmux: { readonly socketName: string; readonly holderSession: string };
|
|
readonly agents: readonly {
|
|
readonly name: string;
|
|
readonly alias?: string;
|
|
readonly runtime: string;
|
|
}[];
|
|
}
|
|
|
|
/** Reads either roster version into the shared read-only view. */
|
|
async function loadRosterReadModel(cmd: Command): Promise<FleetRosterReadModel> {
|
|
const opts = cmd.opts<{ mosaicHome: string; roster?: string }>();
|
|
const path = await resolveRosterPath(opts.mosaicHome, opts.roster);
|
|
if (!(await usesRosterV2ControlPlane(cmd))) {
|
|
const v1 = await loadRosterAtPath(cmd, path);
|
|
return {
|
|
version: 1,
|
|
tmux: { socketName: v1.tmux.socketName, holderSession: v1.tmux.holderSession },
|
|
agents: v1.agents,
|
|
};
|
|
}
|
|
try {
|
|
const v2 = parseRosterV2(await readFleetRosterText(path), 'yaml');
|
|
return {
|
|
version: 2,
|
|
tmux: { socketName: v2.tmux.socketName, holderSession: v2.tmux.holderSession },
|
|
agents: v2.agents,
|
|
};
|
|
} catch (error) {
|
|
reportFleetRosterConfigurationError(cmd, error);
|
|
}
|
|
}
|
|
|
|
async function loadRosterFromAgentCommand(
|
|
command: Command,
|
|
mosaicHomeOverride?: string,
|
|
): Promise<FleetRoster> {
|
|
const opts = command.optsWithGlobals<{ mosaicHome?: string; roster?: string }>();
|
|
const mosaicHome = opts.mosaicHome ?? mosaicHomeOverride ?? defaultMosaicHome();
|
|
return loadRosterAtPath(command, await resolveRosterPath(mosaicHome, opts.roster));
|
|
}
|
|
|
|
async function loadRosterAtPath(command: Command, path: string): Promise<FleetRoster> {
|
|
try {
|
|
return await loadFleetRoster(path);
|
|
} catch (error) {
|
|
reportFleetRosterConfigurationError(command, error);
|
|
}
|
|
}
|
|
|
|
function reportFleetRosterConfigurationError(command: Command, error: unknown): never {
|
|
if (error instanceof FleetRosterConfigurationError) {
|
|
command.error(error.message, { code: 'fleet.roster', exitCode: 1 });
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
function resolveMosaicHomeFromCommand(command: Command, override?: string): string {
|
|
const opts = command.optsWithGlobals<{ mosaicHome?: string }>();
|
|
return opts.mosaicHome ?? override ?? defaultMosaicHome();
|
|
}
|
|
|
|
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];
|
|
}
|
|
|
|
/**
|
|
* Lease-broker socket presence for the fleet-start preflight (#1292).
|
|
* Resolution precedence matches launch.ts's defaultLeaseBrokerSocket and
|
|
* start-agent-session.sh's broker_socket_path: explicit
|
|
* MOSAIC_LEASE_BROKER_SOCKET, else $XDG_RUNTIME_DIR/mosaic-lease/broker.sock,
|
|
* else /run/user/<uid>/mosaic-lease/broker.sock. Pure filesystem check — this
|
|
* deliberately does NOT consult systemd state: a unit can be active
|
|
* (RemainAfterExit) with no live socket, and the socket is the thing the
|
|
* gated runtime connects to. Injectable via deps for tests.
|
|
*/
|
|
export function resolveLeaseBrokerSocketForPreflight(
|
|
env: NodeJS.ProcessEnv = process.env,
|
|
uid: number = typeof process.getuid === 'function' ? process.getuid() : 0,
|
|
): string {
|
|
if (env['MOSAIC_LEASE_BROKER_SOCKET']) return env['MOSAIC_LEASE_BROKER_SOCKET'];
|
|
const runtimeDir = env['XDG_RUNTIME_DIR'] ?? `/run/user/${uid}`;
|
|
return join(runtimeDir, 'mosaic-lease', 'broker.sock');
|
|
}
|
|
|
|
export async function brokerSocketPresent(
|
|
deps: FleetCommandDeps,
|
|
env: NodeJS.ProcessEnv = process.env,
|
|
): Promise<boolean> {
|
|
const check = deps.checkBrokerSocket;
|
|
const socketPath = resolveLeaseBrokerSocketForPreflight(env);
|
|
if (check) return check(socketPath);
|
|
// S_IFSOCK (0xC000) is a file-TYPE constant, not an access() mode (0-7):
|
|
// access(path, S_IFSOCK) throws ERR_OUT_OF_RANGE on node >= 24 (measured on
|
|
// v24.18.0, #1297 review F1) and cannot succeed on any node — the old catch
|
|
// swallowed the throw, so this probe could NEVER return true and every
|
|
// un-seamed call reported the broker absent. stat() + isSocket() is the real
|
|
// check and matches the bash side's [ -S ].
|
|
try {
|
|
return (await stat(socketPath)).isSocket();
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/** 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. */
|
|
/** Runtime launchers accepted by `fleet add`; shared with the generated projection validator. */
|
|
export const VALID_FLEET_RUNTIMES: readonly string[] = GENERATED_AGENT_ENV_SUPPORTED_RUNTIMES;
|
|
|
|
/**
|
|
* 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,
|
|
};
|
|
}
|
|
|
|
function serializeConnector(
|
|
connector: NonNullable<FleetRoster['connector']>,
|
|
): Record<string, unknown> {
|
|
if (connector.kind === 'tmux') return { kind: 'tmux' };
|
|
if (connector.kind === 'discord') {
|
|
return { kind: 'discord', discord: { channel_id: connector.discord.channelId } };
|
|
}
|
|
return {
|
|
kind: 'matrix',
|
|
matrix: {
|
|
homeserver_url: connector.matrix.homeserverUrl,
|
|
user_id: connector.matrix.userId,
|
|
room_id: connector.matrix.roomId,
|
|
},
|
|
};
|
|
}
|
|
|
|
/**
|
|
* 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.host !== undefined) {
|
|
raw['host'] = agent.host;
|
|
}
|
|
if (agent.ssh !== undefined) {
|
|
raw['ssh'] = agent.ssh;
|
|
}
|
|
if (agent.socket !== undefined) {
|
|
raw['socket'] = agent.socket;
|
|
}
|
|
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,
|
|
...(roster.connector ? { connector: serializeConnector(roster.connector) } : {}),
|
|
};
|
|
|
|
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;
|
|
}
|
|
return resolveInstalledFleetRosterPath(mosaicHome);
|
|
}
|