feat(fleet): reconcile local roster state (#785)

packages/mosaic/src/commands/fleet.ts

@@ -22,6 +22,11 @@ import {
   registerFleetAgentCrudCommands,
   type FleetAgentCrudCommandDeps,
 } from './fleet-agent-crud-command.js';
+import {
+  executeReconcilerCommandJson,
+  registerFleetReconcilerCommands,
+  type FleetReconcilerCommandDeps,
+} from './fleet-reconciler-command.js';
 import { resolveCommsBlock } from '../fleet/comms-onboarding.js';
 import {
   applyPreparedAgentEnvironmentProjection,
@@ -78,6 +83,7 @@ export interface FleetCommandDeps {
    */
   isStdinTTY?: boolean;
   projectionApplier?: FleetAgentCrudCommandDeps['projectionApplier'];
+  reconcileDeps?: FleetReconcilerCommandDeps['reconcileDeps'];
 }
 
 interface RawFleetRoster {
@@ -1606,53 +1612,70 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
     cmd
       .command(`${action} [agent]`)
       .description(`${action} the fleet holder or one agent`)
-      .action(async (agent?: string) => {
-        const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
-        const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
-        const roster = await loadRosterForCommand(cmd);
-        if (agent) {
-          getRosterAgent(roster, agent);
-          // Single-agent restart is guarded too: it can race a full restart that
-          // is tearing the shared holder down.
+      .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 === '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, agent));
+              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, agent));
-          return;
-        }
-        if (action === 'stop') {
-          await stopFleetBestEffort(
-            runner,
-            roster.agents.map((rosterAgent) => rosterAgent.name),
-          );
-          return;
-        }
-        if (action === 'restart') {
-          // Serialize the holder+agents teardown/relaunch behind the restart lock
-          // so a re-entrant restart waits for clean shutdown before relaunching,
-          // instead of racing a half-torn-down holder into a tight loop.
-          const guard = await acquireRestartLock(activePaths.mosaicHome, sleepFn);
-          try {
-            await runChecked(runner, buildFleetServiceCommand(action));
-            for (const rosterAgent of roster.agents) {
-              await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
-            }
-          } finally {
-            await guard.release();
+          await runChecked(runner, buildFleetServiceCommand(action));
+          for (const rosterAgent of roster.agents) {
+            await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
           }
-          return;
-        }
-        await runChecked(runner, buildFleetServiceCommand(action));
-        for (const rosterAgent of roster.agents) {
-          await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
-        }
-      });
+        },
+      );
   }
 
   cmd
@@ -1660,6 +1683,16 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
     .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);
@@ -1683,6 +1716,15 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
     .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, [
@@ -2082,6 +2124,11 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
   // 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);
+  registerFleetReconcilerCommands(cmd, {
+    runner,
+    mosaicHome: deps.mosaicHome,
+    reconcileDeps: deps.reconcileDeps,
+  });
 
   return cmd;
 }
@@ -2415,6 +2462,20 @@ async function loadRosterForCommand(cmd: Command): Promise<FleetRoster> {
   return loadFleetRoster(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);
+  const parsed: unknown = YAML.parse(await readFile(path, 'utf8'));
+  return (
+    typeof parsed === 'object' &&
+    parsed !== null &&
+    !Array.isArray(parsed) &&
+    'version' in parsed &&
+    parsed.version === 2
+  );
+}
+
 async function loadRosterFromAgentCommand(
   command: Command,
   mosaicHomeOverride?: string,