stack/packages/mosaic/src/fleet/fleet-reconciler.acceptance.spec.ts

import { chmod, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { Command } from 'commander';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { registerFleetCommand, type CommandResult, type CommandRunner } from '../commands/fleet.js';
import {
  executeFleetReconcile,
  type FleetReconcileCommandResult,
  type FleetReconcileDeps,
  type FleetReconcileResult,
} from './fleet-reconciler.js';
import {
  parseRosterV2,
  renderRosterV2Yaml,
  type FleetRosterV2,
  type FleetRosterV2Agent,
} from './roster-v2.js';
import { FleetTmuxRuntimeTransport } from './tmux-runtime-transport.js';

const holderIdentity = '11111111-1111-4111-8111-111111111111';
const stoppedAgent: FleetRosterV2Agent = {
  name: 'coder0',
  alias: 'Coder 0',
  className: 'code',
  runtime: 'pi',
  provider: 'openai',
  model: 'gpt-5.6-sol',
  reasoning: 'high',
  toolPolicy: 'code',
  workingDirectory: '/srv/mosaic',
  persistentPersona: false,
  resetBetweenTasks: true,
  lifecycle: { enabled: true, desiredState: 'stopped' },
  launch: { yolo: true },
};

const baseRoster: FleetRosterV2 = {
  version: 2,
  generation: 7,
  transport: 'tmux',
  tmux: { socketName: 'mosaic-fleet', holderSession: '_holder' },
  defaults: { workingDirectory: '/srv/mosaic', runtime: 'pi' },
  runtimes: { pi: { resetCommand: '/new' } },
  agents: [stoppedAgent],
};

interface InjectedLifecycleFailure {
  readonly action: 'start' | 'stop' | 'restart';
  readonly service: string;
  readonly diagnostic: string;
}

class FakeLifecycleHost {
  readonly calls: string[][] = [];
  readonly sessions = new Set<string>();
  readonly activeServices = new Set<string>();
  private failure: InjectedLifecycleFailure | undefined;

  constructor(readonly roster: FleetRosterV2) {
    this.sessions.add(roster.tmux.holderSession);
  }

  injectFailure(failure: InjectedLifecycleFailure): void {
    this.failure = failure;
  }

  readonly run = async (
    command: string,
    args: readonly string[],
  ): Promise<FleetReconcileCommandResult> => {
    this.calls.push([command, ...args]);
    if (command === 'tmux') return this.runTmux(args);
    if (command === 'systemctl') return this.runSystemctl(args);
    return { stdout: '', stderr: 'unsupported fake command', exitCode: 127 };
  };

  private runTmux(args: readonly string[]): FleetReconcileCommandResult {
    if (args.includes('list-sessions')) {
      return { stdout: `${[...this.sessions].join('\n')}\n`, stderr: '', exitCode: 0 };
    }
    if (args.includes('has-session')) {
      const targetArgument = args[args.indexOf('-t') + 1];
      const sessionName = targetArgument?.replace(/^=/, '').split(':')[0];
      return {
        stdout: '',
        stderr: '',
        exitCode: sessionName !== undefined && this.sessions.has(sessionName) ? 0 : 1,
      };
    }
    if (args.includes('show-environment')) {
      return {
        stdout: [
          'HOME=/home/mosaic',
          `MOSAIC_FLEET_OWNER=${holderIdentity}`,
          `MOSAIC_TMUX_HOLDER=${this.roster.tmux.holderSession}`,
          `MOSAIC_TMUX_SOCKET=${this.roster.tmux.socketName}`,
          'PATH=/usr/bin:/bin',
          'PWD=/home/mosaic',
          '',
        ].join('\n'),
        stderr: '',
        exitCode: 0,
      };
    }
    return { stdout: '', stderr: 'destructive tmux action rejected by fake', exitCode: 125 };
  }

  private runSystemctl(args: readonly string[]): FleetReconcileCommandResult {
    const action = args[1];
    const service = args[2];
    if (action === 'show' && service !== undefined) {
      return {
        stdout: `ActiveState=${this.activeServices.has(service) ? 'active' : 'inactive'}\n`,
        stderr: '',
        exitCode: 0,
      };
    }
    if (
      (action === 'start' || action === 'stop' || action === 'restart') &&
      service !== undefined
    ) {
      this.applyLifecycleEffect(action, service);
      if (this.failure?.action === action && this.failure.service === service) {
        const diagnostic = this.failure.diagnostic;
        this.failure = undefined;
        return { stdout: '', stderr: diagnostic, exitCode: 1 };
      }
      return { stdout: '', stderr: '', exitCode: 0 };
    }
    return { stdout: '', stderr: 'unsupported fake systemctl action', exitCode: 125 };
  }

  private applyLifecycleEffect(action: 'start' | 'stop' | 'restart', service: string): void {
    if (service === 'mosaic-tmux-holder.service') return;
    const match = /^mosaic-agent@(.+)\.service$/.exec(service);
    if (!match) return;
    const agentName = match[1];
    if (agentName === undefined) return;
    if (action === 'stop') {
      this.activeServices.delete(service);
      this.sessions.delete(agentName);
      return;
    }
    this.activeServices.add(service);
    this.sessions.add(agentName);
  }
}

const cleanupDirectories: string[] = [];

afterEach(async (): Promise<void> => {
  vi.restoreAllMocks();
  process.exitCode = undefined;
  await Promise.all(
    cleanupDirectories.splice(0).map(async (directory: string): Promise<void> => {
      await rm(directory, { recursive: true, force: true });
    }),
  );
});

function reconcileDeps(host: FakeLifecycleHost): FleetReconcileDeps {
  return {
    runner: host.run,
    homeDirectory: '/home/mosaic',
    readHolderIdentity: async () => holderIdentity,
    validateRoster: async () => undefined,
    prepareProjections: async () => [{ agentName: 'coder0' }],
    applyProjection: async () => undefined,
    readRoster: async () => host.roster,
    acquireMutationLock: async () => async () => undefined,
    // Hermetic broker observation (#1297 F3): without this, the plan probes
    // the REAL host filesystem, so the "stable JSON" fixtures answered true
    // on any machine with a live lease broker and false elsewhere. Pointing
    // both paths at fixtures that do not exist pins socketPresent:false and
    // unitInstalled:false on every host, which is what these fixtures assert.
    brokerSocketEnv: {
      MOSAIC_LEASE_BROKER_SOCKET: '/nonexistent/mosaic-lease/broker.sock',
      XDG_CONFIG_HOME: '/nonexistent/mosaic-config',
      XDG_RUNTIME_DIR: '/nonexistent/run',
    },
  };
}

async function execute(
  host: FakeLifecycleHost,
  command: 'apply' | 'reconcile' | 'restart' | 'status' | 'stop',
  agentName?: string,
): Promise<FleetReconcileResult> {
  return executeFleetReconcile({
    roster: host.roster,
    command,
    ...(agentName === undefined ? {} : { agentName }),
    ...(command === 'status' ? {} : { expectedGeneration: host.roster.generation }),
    deps: reconcileDeps(host),
  });
}

async function fixtureHome(roster: FleetRosterV2): Promise<string> {
  const home = await mkdtemp(join(tmpdir(), 'mosaic-reconciler-acceptance-'));
  cleanupDirectories.push(home);
  const fleetDirectory = join(home, 'fleet');
  await mkdir(fleetDirectory, { mode: 0o700 });
  await chmod(home, 0o700);
  await chmod(fleetDirectory, 0o700);
  await writeFile(join(fleetDirectory, 'roster.yaml'), renderRosterV2Yaml(roster), { mode: 0o600 });
  return home;
}

async function fixtureLegacyRoster(): Promise<string> {
  const home = await mkdtemp(join(tmpdir(), 'mosaic-reconciler-acceptance-v1-'));
  cleanupDirectories.push(home);
  const fleetDirectory = join(home, 'fleet');
  await mkdir(fleetDirectory, { mode: 0o700 });
  await chmod(home, 0o700);
  await chmod(fleetDirectory, 0o700);
  const rosterPath = join(fleetDirectory, 'roster.yaml');
  await writeFile(
    rosterPath,
    [
      'version: 1',
      'transport: tmux',
      'tmux:',
      '  holder_session: _holder',
      'agents:',
      '  - name: coder0',
      '    runtime: pi',
      '    class: code',
      '',
    ].join('\n'),
    { mode: 0o600 },
  );
  return rosterPath;
}

function cliProgram(home: string, host: FakeLifecycleHost): Command {
  const program = new Command();
  program.exitOverride();
  registerFleetCommand(program, {
    mosaicHome: home,
    runner: async (command: string, args: string[]): Promise<CommandResult> =>
      host.run(command, args),
    reconcileDeps: reconcileDeps(host),
  });
  return program;
}

function captureJson(): string[] {
  const output: string[] = [];
  vi.spyOn(console, 'log').mockImplementation((line: string): void => {
    output.push(line);
  });
  return output;
}

describe('FCM-M3-002 reconciler lifecycle acceptance', (): void => {
  it('observes named-socket drift through canonical roster-v2 parsing without runtime mutation', async (): Promise<void> => {
    const roster: FleetRosterV2 = {
      ...baseRoster,
      agents: [
        stoppedAgent,
        {
          ...stoppedAgent,
          name: 'reviewer0',
          alias: 'Reviewer 0',
          lifecycle: { enabled: true, desiredState: 'running' },
        },
        {
          ...stoppedAgent,
          name: 'validator0',
          alias: 'Validator 0',
          lifecycle: { enabled: false, desiredState: 'stopped' },
        },
      ],
    };
    const home = await fixtureHome(roster);
    const host = new FakeLifecycleHost(roster);
    host.sessions.add('coder0');
    host.sessions.add('validator0');
    host.sessions.add('coder0-shadow');
    const output = captureJson();

    await cliProgram(home, host).parseAsync(['node', 'mosaic', 'fleet', 'status']);

    expect(output).toHaveLength(1);
    expect(JSON.parse(output[0] ?? '{}')).toMatchObject({
      plan: {
        agents: [
          { name: 'coder0', drift: ['unexpected-session'] },
          { name: 'reviewer0', drift: ['missing-session'] },
          { name: 'validator0', drift: ['unexpected-session', 'disabled-running'] },
        ],
        unmanagedSessions: ['coder0-shadow'],
      },
    });
    expect(host.calls[0]).toEqual([
      'tmux',
      '-L',
      'mosaic-fleet',
      'list-sessions',
      '-F',
      '#{session_name}',
    ]);
    expect(
      host.calls.every(
        (call: string[]): boolean =>
          call[0] !== 'tmux' || (call[1] === '-L' && call[2] === 'mosaic-fleet'),
      ),
    ).toBe(true);
    expect(
      host.calls.every((call: string[]): boolean => call[0] !== 'systemctl' || call[2] === 'show'),
    ).toBe(true);
    expect(process.exitCode).toBe(0);
  });

  it('rejects a missing canonical roster-v2 tmux socket through parseRosterV2', (): void => {
    const source = renderRosterV2Yaml(baseRoster).replace(/^  socket_name:.*\n/m, '');
    expect(() => parseRosterV2(source, 'yaml')).toThrow(
      'Roster v2 tmux socket_name is required and must be a string.',
    );
  });

  it('accepts an explicit empty canonical roster-v2 socket as the literal default server', (): void => {
    const source = renderRosterV2Yaml(baseRoster).replace(
      /^  socket_name:.*$/m,
      '  socket_name: ""',
    );
    expect(parseRosterV2(source, 'yaml').tmux.socketName).toBe('');
  });

  it('omits -L for the literal default tmux server at the runtime transport boundary', async (): Promise<void> => {
    const rosterPath = await fixtureLegacyRoster();
    const runner = vi.fn<CommandRunner>(
      async (): Promise<CommandResult> => ({
        stdout: '111 pi 0 0 0 0\n',
        stderr: '',
        exitCode: 0,
      }),
    );
    const transport = new FleetTmuxRuntimeTransport({
      mosaicHome: '/unused',
      rosterPath,
      runner,
    });

    await expect(transport.verifySession('coder0')).resolves.toEqual({
      id: 'coder0',
      runtimeId: 'pi',
      socketName: '',
    });
    expect(runner).toHaveBeenCalledTimes(1);
    expect(runner).toHaveBeenCalledWith('tmux', [
      'list-panes',
      '-t',
      '=coder0:0.0',
      '-F',
      '#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}',
    ]);
    expect(runner.mock.calls[0]?.[1]).not.toContain('-L');
  });

  it('classifies unmanaged near-collisions and stops only the exact roster-owned service', async (): Promise<void> => {
    const host = new FakeLifecycleHost(baseRoster);
    host.sessions.add('coder0');
    host.sessions.add('coder0-shadow');
    host.sessions.add('unmanaged');
    host.activeServices.add('[email protected]');
    host.activeServices.add('[email protected]');

    const observed = await execute(host, 'status');
    const stopped = await execute(host, 'stop', 'coder0');

    expect(observed.plan.unmanagedSessions).toEqual(['coder0-shadow', 'unmanaged']);
    expect(stopped).toMatchObject({ applied: true, lifecycle: 'complete' });
    expect(host.activeServices.has('[email protected]')).toBe(false);
    expect(host.activeServices.has('[email protected]')).toBe(true);
    expect(host.sessions.has('coder0-shadow')).toBe(true);
    expect(host.sessions.has('unmanaged')).toBe(true);
    expect(host.calls).toContainEqual([
      'systemctl',
      '--user',
      'stop',
      '[email protected]',
    ]);
    expect(
      host.calls.some(
        (call: string[]): boolean =>
          call.includes('kill-session') ||
          call.includes('coder0-shadow.service') ||
          call.includes('unmanaged.service'),
      ),
    ).toBe(false);
  });

  it('preserves persisted stopped state through apply, reconcile, restart failure, and recovery reconcile', async (): Promise<void> => {
    const host = new FakeLifecycleHost(baseRoster);
    host.sessions.add('coder0');
    host.activeServices.add('[email protected]');

    const applied = await execute(host, 'apply');
    const reconciled = await execute(host, 'reconcile');
    host.injectFailure({
      action: 'restart',
      service: '[email protected]',
      diagnostic: 'crash after effect: TOKEN=acceptance-secret',
    });
    const partialRestart = await execute(host, 'restart', 'coder0');

    expect(applied).toMatchObject({ applied: true, lifecycle: 'complete' });
    expect(reconciled).toMatchObject({ applied: true, lifecycle: 'complete' });
    expect(host.roster.agents[0]?.lifecycle.desiredState).toBe('stopped');
    expect(partialRestart).toMatchObject({
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'incomplete',
      recovery: {
        code: 'lifecycle-apply-failed',
        action: 'rerun-after-inspecting-owned-resources',
      },
    });
    expect(host.activeServices.has('[email protected]')).toBe(true);

    const recovered = await execute(host, 'reconcile');

    expect(recovered).toMatchObject({ applied: true, lifecycle: 'complete' });
    expect(host.roster.agents[0]?.lifecycle.desiredState).toBe('stopped');
    expect(host.activeServices.has('[email protected]')).toBe(false);
    expect(host.sessions.has('coder0')).toBe(false);
    const destructiveCalls = host.calls.filter(
      (call: string[]): boolean =>
        call[0] === 'systemctl' && ['start', 'stop', 'restart'].includes(call[2] ?? ''),
    );
    expect(
      destructiveCalls.every(
        (call: string[]): boolean => call[3] === '[email protected]',
      ),
    ).toBe(true);
    expect(destructiveCalls.some((call: string[]): boolean => call[2] === 'start')).toBe(false);
  });

  it('emits stable non-zero redacted JSON for a partial lifecycle effect', async (): Promise<void> => {
    const home = await fixtureHome(baseRoster);
    const host = new FakeLifecycleHost(baseRoster);
    host.injectFailure({
      action: 'restart',
      service: '[email protected]',
      diagnostic: 'simulated runner stderr with PASSWORD=acceptance-secret',
    });
    const output = captureJson();

    await cliProgram(home, host).parseAsync([
      'node',
      'mosaic',
      'fleet',
      'restart',
      'coder0',
      '--expected-generation',
      '7',
    ]);

    const line = output.at(-1) ?? '';
    expect(output).toHaveLength(1);
    expect(JSON.parse(line)).toEqual({
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'incomplete',
      plan: {
        generation: 7,
        holder: 'owned',
        broker: { unitInstalled: false, socketPresent: false },
        agents: [
          {
            name: 'coder0',
            desiredState: 'stopped',
            enabled: true,
            systemd: 'inactive',
            tmux: 'missing',
            drift: [],
          },
        ],
        unmanagedSessions: [],
      },
      recovery: {
        code: 'lifecycle-apply-failed',
        action: 'rerun-after-inspecting-owned-resources',
      },
    });
    expect(process.exitCode).toBe(1);
    expect(line).not.toContain('PASSWORD');
    expect(line).not.toContain('acceptance-secret');
    expect(line).not.toContain('simulated runner stderr');
  });
});