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

import { chmod, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import {
  acquirePrivateReconcileLock,
  FleetReconcileError,
  executeFleetReconcile,
  type FleetReconcileCommand,
  type FleetReconcileDeps,
} from './fleet-reconciler.js';
import type { FleetRosterV2 } from './roster-v2.js';

let cleanup: string | undefined;

afterEach(async (): Promise<void> => {
  if (cleanup) await rm(cleanup, { recursive: true, force: true });
  cleanup = undefined;
});

async function lockHome(): Promise<string> {
  cleanup = await mkdtemp(join(tmpdir(), 'mosaic-reconcile-lock-'));
  const fleet = join(cleanup, 'fleet');
  await mkdir(fleet, { mode: 0o700 });
  await chmod(cleanup, 0o700);
  await chmod(fleet, 0o700);
  return cleanup;
}

const roster: FleetRosterV2 = {
  version: 2,
  generation: 7,
  transport: 'tmux',
  tmux: { socketName: 'mosaic-fleet', holderSession: '_holder' },
  defaults: { workingDirectory: '/srv/mosaic', runtime: 'pi' },
  runtimes: { pi: { resetCommand: '/new' } },
  agents: [
    {
      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 },
    },
  ],
};

function deps(overrides: Partial<FleetReconcileDeps> = {}): FleetReconcileDeps {
  return {
    runner: async (command, args) => {
      if (command === 'tmux' && args.includes('list-sessions')) {
        return { stdout: '_holder\ncoder0\n', stderr: '', exitCode: 0 };
      }
      if (command === 'tmux' && args.includes('show-environment')) {
        return {
          stdout:
            'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
          stderr: '',
          exitCode: 0,
        };
      }
      return { stdout: '', stderr: '', exitCode: 0 };
    },
    homeDirectory: '/home/mosaic',
    readHolderIdentity: async () => '11111111-1111-4111-8111-111111111111',
    validateRoster: async () => undefined,
    prepareProjections: async () => [{ agentName: 'coder0' }],
    applyProjection: async () => undefined,
    readRoster: async () => roster,
    acquireMutationLock: async () => async () => undefined,
    ...overrides,
  };
}

async function run(command: FleetReconcileCommand, overrides: Partial<FleetReconcileDeps> = {}) {
  return executeFleetReconcile({
    roster,
    command,
    ...(command === 'status' || command === 'verify' || command === 'doctor'
      ? {}
      : { expectedGeneration: 7 }),
    deps: deps({ readRoster: async () => roster, ...overrides }),
  });
}

describe('fleet roster-owned reconciler', (): void => {
  // ── #1292: broker as first-class plan member + broker-first start ordering ──

  it('reports broker unit and socket state in the plan (socket is the signal, not unit state)', async (): Promise<void> => {
    const result = await run('status', {
      statPath: async () => true,
      checkBrokerSocket: async () => true,
    });
    expect(result.plan.broker).toEqual({ unitInstalled: true, socketPresent: true });
  });

  it('reports a dead broker as socketPresent=false even when the unit is installed (enabled-but-dead is the #1292 shape)', async (): Promise<void> => {
    const result = await run('status', {
      statPath: async () => true,
      checkBrokerSocket: async () => false,
    });
    expect(result.plan.broker).toEqual({ unitInstalled: true, socketPresent: false });
  });

  it('reports broker-absent when neither seam is present (defaults false, never guesses healthy)', async (): Promise<void> => {
    const result = await run('status');
    expect(result.plan.broker).toEqual({ unitInstalled: false, socketPresent: false });
  });

  it('command start enables and starts the broker BEFORE the holder and any agent unit', async (): Promise<void> => {
    const calls: string[][] = [];
    const result = await run('start', {
      runner: async (command, args) => {
        calls.push([command, ...args]);
        if (command === 'tmux' && args.includes('list-sessions')) {
          return { stdout: '_holder\ncoder0\n', stderr: '', exitCode: 0 };
        }
        if (command === 'tmux' && args.includes('show-environment')) {
          return {
            stdout:
              'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
            stderr: '',
            exitCode: 0,
          };
        }
        return { stdout: '', stderr: '', exitCode: 0 };
      },
    });
    expect(result.lifecycle).toBe('complete');
    const brokerEnable = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user enable mosaic-lease-broker.service',
    );
    const brokerStart = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user start mosaic-lease-broker.service',
    );
    const holderStart = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user start mosaic-tmux-holder.service',
    );
    const agentStart = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user start [email protected]',
    );
    expect(brokerEnable).toBeGreaterThanOrEqual(0);
    expect(brokerStart).toBeGreaterThan(brokerEnable);
    // Holder start may be absent (holder 'owned' in this fixture); if present it must follow the broker.
    if (holderStart >= 0) expect(holderStart).toBeGreaterThan(brokerStart);
    expect(agentStart).toBeGreaterThan(brokerStart);
  });

  it('apply with a running desired agent also enables and starts the broker first', async (): Promise<void> => {
    const calls: string[][] = [];
    const runningRoster: FleetRosterV2 = {
      ...roster,
      agents: roster.agents.map((agent) =>
        agent.name === 'coder0'
          ? { ...agent, lifecycle: { enabled: true, desiredState: 'running' as const } }
          : agent,
      ),
    };
    const result = await executeFleetReconcile({
      roster: runningRoster,
      command: 'apply',
      expectedGeneration: 7,
      deps: deps({
        readRoster: async () => runningRoster,
        runner: async (command, args) => {
          calls.push([command, ...args]);
          if (command === 'tmux' && args.includes('list-sessions')) {
            return { stdout: '_holder\n', stderr: '', exitCode: 0 };
          }
          if (command === 'tmux' && args.includes('show-environment')) {
            return {
              stdout:
                'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
              stderr: '',
              exitCode: 0,
            };
          }
          return { stdout: '', stderr: '', exitCode: 0 };
        },
      }),
    });
    expect(result.applied).toBe(true);
    const brokerStart = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user start mosaic-lease-broker.service',
    );
    const agentStart = calls.findIndex(
      (c) => c.join(' ') === 'systemctl --user start [email protected]',
    );
    expect(brokerStart).toBeGreaterThanOrEqual(0);
    expect(agentStart).toBeGreaterThan(brokerStart);
  });

  it('fails closed on a symlinked fleet ancestor without touching its target', async (): Promise<void> => {
    const home = await lockHome();
    const fleet = join(home, 'fleet');
    const attacker = await mkdtemp(join(tmpdir(), 'mosaic-reconcile-attacker-'));
    await writeFile(join(attacker, 'sentinel'), 'unchanged\n', { mode: 0o600 });
    try {
      await rm(fleet, { recursive: true });
      await symlink(attacker, fleet, 'dir');
      await expect(acquirePrivateReconcileLock(home)()).rejects.toMatchObject({
        code: 'unsafe-managed-path',
      });
      expect(await readFile(join(attacker, 'sentinel'), 'utf8')).toBe('unchanged\n');
    } finally {
      await rm(attacker, { recursive: true, force: true });
    }
  });

  it('rejects unsafe ancestors, symlink leaves, EEXIST, and non-EEXIST lock creation failures', async (): Promise<void> => {
    const home = await lockHome();
    const fleet = join(home, 'fleet');
    const lockPath = join(fleet, 'roster.yaml.reconcile.lock');

    await chmod(fleet, 0o770);
    await expect(acquirePrivateReconcileLock(home)()).rejects.toMatchObject({
      code: 'unsafe-managed-path',
    });
    await chmod(fleet, 0o700);

    const target = join(home, 'target');
    await writeFile(target, 'target\n', { mode: 0o600 });
    await symlink(target, lockPath);
    await expect(acquirePrivateReconcileLock(home)()).rejects.toMatchObject({
      code: 'unsafe-lock',
    });
    await rm(lockPath);

    await writeFile(lockPath, 'other\n', { mode: 0o600 });
    await expect(acquirePrivateReconcileLock(home)()).rejects.toMatchObject({
      code: 'concurrent-mutation',
    });
    await rm(lockPath);

    const ioFailure = Object.assign(new Error('injected I/O failure'), { code: 'EIO' });
    await expect(
      acquirePrivateReconcileLock(home, async () => Promise.reject(ioFailure))(),
    ).rejects.toMatchObject({ code: 'lock-io-failed' });
  });

  it('does not unlink a replacement lock and normally releases its own lock', async (): Promise<void> => {
    const home = await lockHome();
    const lockPath = join(home, 'fleet', 'roster.yaml.reconcile.lock');
    const release = await acquirePrivateReconcileLock(home)();
    await rm(lockPath);
    await writeFile(lockPath, 'replacement\n', { mode: 0o600 });
    await expect(release()).rejects.toMatchObject({ code: 'lock-cleanup-failed' });
    expect(await readFile(lockPath, 'utf8')).toBe('replacement\n');

    await rm(lockPath);
    const normalRelease = await acquirePrivateReconcileLock(home)();
    await normalRelease();
    await expect(readFile(lockPath, 'utf8')).rejects.toThrow();
  });
  it('requires a generation before any mutating preflight or effect', async (): Promise<void> => {
    let effects = 0;
    await expect(
      executeFleetReconcile({
        roster,
        command: 'apply',
        deps: deps({
          validateRoster: async () => {
            effects += 1;
          },
        }),
      }),
    ).rejects.toMatchObject({ code: 'missing-generation' });
    expect(effects).toBe(0);
  });

  it('fences mutation against the canonical roster reread under lock', async (): Promise<void> => {
    let effects = 0;
    await expect(
      executeFleetReconcile({
        roster,
        command: 'apply',
        expectedGeneration: 7,
        deps: deps({
          readRoster: async () => ({ ...roster, generation: 8 }),
          runner: async () => {
            effects += 1;
            return { stdout: '', stderr: '', exitCode: 0 };
          },
          applyProjection: async () => {
            effects += 1;
          },
        }),
      }),
    ).rejects.toMatchObject({ code: 'stale-generation' });
    expect(effects).toBe(0);
  });

  it('rejects stale generation before projection or lifecycle effects', async (): Promise<void> => {
    let effects = 0;
    await expect(
      executeFleetReconcile({
        roster,
        command: 'apply',
        expectedGeneration: 6,
        deps: deps({
          validateRoster: async () => {
            effects += 1;
          },
        }),
      }),
    ).rejects.toMatchObject({ code: 'stale-generation' });
    expect(effects).toBe(0);
  });

  it('denies a concurrent mutation lock without effects and leaves observations lock-free', async (): Promise<void> => {
    let effects = 0;
    const busy = async (): Promise<() => Promise<void>> => {
      throw new FleetReconcileError('concurrent-mutation', 'busy');
    };
    await expect(
      run('apply', {
        acquireMutationLock: busy,
        applyProjection: async () => {
          effects += 1;
        },
      }),
    ).rejects.toMatchObject({ code: 'concurrent-mutation' });
    expect(effects).toBe(0);

    await expect(run('doctor', { acquireMutationLock: busy })).resolves.toMatchObject({
      applied: false,
      lifecycle: 'not-applied',
    });
  });

  it('always releases the mutation lock after success and partial failure', async (): Promise<void> => {
    let releases = 0;
    const lock = async (): Promise<() => Promise<void>> => async (): Promise<void> => {
      releases += 1;
    };
    await run('apply', { acquireMutationLock: lock });
    await run('apply', {
      acquireMutationLock: lock,
      applyProjection: async () => {
        throw new Error('injected failure');
      },
    });
    expect(releases).toBe(2);
  });

  it('preserves stopped desired state during apply', async (): Promise<void> => {
    const calls: string[][] = [];
    const result = await run('apply', {
      runner: async (command, args) => {
        calls.push([command, ...args]);
        if (command === 'tmux' && args.includes('list-sessions')) {
          return { stdout: '_holder\ncoder0\n', stderr: '', exitCode: 0 };
        }
        if (command === 'tmux' && args.includes('show-environment')) {
          return {
            stdout:
              'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
            stderr: '',
            exitCode: 0,
          };
        }
        return { stdout: '', stderr: '', exitCode: 0 };
      },
    });

    expect(result.lifecycle).toBe('complete');
    expect(calls).not.toContainEqual([
      'systemctl',
      '--user',
      'start',
      '[email protected]',
    ]);
  });

  it.each(['plan', 'status', 'doctor', 'verify'] as const)(
    'keeps default-server %s observational and free of lifecycle effects',
    async (command) => {
      const calls: string[][] = [];
      const defaultServerRoster: FleetRosterV2 = {
        ...roster,
        tmux: { ...roster.tmux, socketName: '' },
      };

      await expect(
        executeFleetReconcile({
          roster: defaultServerRoster,
          command,
          deps: deps({
            runner: async (executable, args) => {
              calls.push([executable, ...args]);
              if (executable === 'tmux' && args.includes('list-sessions')) {
                return { stdout: '_holder\n', stderr: '', exitCode: 0 };
              }
              if (executable === 'tmux' && args.includes('show-environment')) {
                return {
                  stdout:
                    'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
                  stderr: '',
                  exitCode: 0,
                };
              }
              return { stdout: 'ActiveState=inactive\n', stderr: '', exitCode: 0 };
            },
          }),
        }),
      ).resolves.toMatchObject({ applied: false, lifecycle: 'not-applied' });
      expect(
        calls.some(
          ([executable, , action]): boolean =>
            executable === 'systemctl' &&
            (action === 'start' || action === 'stop' || action === 'restart'),
        ),
      ).toBe(false);
    },
  );

  it.each(['start', 'stop', 'restart', 'apply', 'reconcile'] as const)(
    'fails closed before %s can route fixed named-socket services for a default-server roster',
    async (command) => {
      const calls: string[][] = [];
      let projectionPrepares = 0;
      let projectionApplies = 0;
      const defaultServerRoster: FleetRosterV2 = {
        ...roster,
        tmux: { ...roster.tmux, socketName: '' },
        agents: [
          {
            ...roster.agents[0]!,
            lifecycle: {
              enabled: true,
              desiredState: command === 'start' || command === 'restart' ? 'running' : 'stopped',
            },
          },
        ],
      };

      await expect(
        executeFleetReconcile({
          roster: defaultServerRoster,
          command,
          expectedGeneration: 7,
          deps: deps({
            readRoster: async () => defaultServerRoster,
            prepareProjections: async () => {
              projectionPrepares += 1;
              return [{ agentName: 'coder0' }];
            },
            applyProjection: async () => {
              projectionApplies += 1;
            },
            runner: async (executable, args) => {
              calls.push([executable, ...args]);
              return { stdout: '', stderr: '', exitCode: 0 };
            },
          }),
        }),
      ).rejects.toMatchObject({ code: 'lifecycle-precondition-failed' });
      expect(projectionPrepares).toBe(0);
      expect(projectionApplies).toBe(0);
      expect(calls).toEqual([]);
    },
  );

  it('starts only an explicitly running roster agent with exact systemd targets', async (): Promise<void> => {
    const calls: string[][] = [];
    const runningRoster: FleetRosterV2 = {
      ...roster,
      agents: [{ ...roster.agents[0]!, lifecycle: { enabled: true, desiredState: 'running' } }],
    };
    const result = await executeFleetReconcile({
      roster: runningRoster,
      command: 'apply',
      expectedGeneration: 7,
      deps: deps({
        readRoster: async () => runningRoster,
        runner: async (command, args) => {
          calls.push([command, ...args]);
          if (command === 'tmux' && args.includes('list-sessions')) {
            return { stdout: '', stderr: '', exitCode: 1 };
          }
          return { stdout: '', stderr: '', exitCode: 0 };
        },
      }),
    });

    expect(result.lifecycle).toBe('complete');
    expect(calls).toContainEqual(['systemctl', '--user', 'start', 'mosaic-tmux-holder.service']);
    expect(calls).toContainEqual(['systemctl', '--user', 'start', '[email protected]']);
  });

  it('rejects a fake holder with contaminated global state before projection application', async (): Promise<void> => {
    let projectionApplied = false;
    await expect(
      run('apply', {
        runner: async (command, args) => {
          if (command === 'tmux' && args.includes('list-sessions')) {
            return { stdout: '_holder\ncoder0\n', stderr: '', exitCode: 0 };
          }
          if (command === 'tmux' && args.includes('show-environment')) {
            return { stdout: 'MOSAIC_FLEET_OWNER=forged\n', stderr: '', exitCode: 0 };
          }
          return { stdout: '', stderr: '', exitCode: 0 };
        },
        applyProjection: async () => {
          projectionApplied = true;
        },
      }),
    ).rejects.toMatchObject({ code: 'ownership-mismatch' });
    expect(projectionApplied).toBe(false);
  });

  it('reports but never adopts unmanaged sessions', async (): Promise<void> => {
    await expect(
      run('apply', {
        runner: async (command, args) => {
          if (command === 'tmux' && args.includes('list-sessions')) {
            return { stdout: '_holder\ncoder0\nother\n', stderr: '', exitCode: 0 };
          }
          if (command === 'tmux' && args.includes('show-environment')) {
            return {
              stdout:
                'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n',
              stderr: '',
              exitCode: 0,
            };
          }
          return { stdout: '', stderr: '', exitCode: 0 };
        },
      }),
    ).rejects.toMatchObject({ code: 'unmanaged-session' } satisfies Partial<FleetReconcileError>);
  });

  it('rejects remote inventory before a local lifecycle command is constructed', async (): Promise<void> => {
    const calls: string[][] = [];
    const remoteRoster = {
      ...roster,
      agents: [{ ...roster.agents[0]!, remote: { host: 'inventory-only' } }],
    } as unknown as FleetRosterV2;

    await expect(
      executeFleetReconcile({
        roster: remoteRoster,
        command: 'apply',
        expectedGeneration: 7,
        deps: deps({
          readRoster: async () => remoteRoster,
          runner: async (command, args) => {
            calls.push([command, ...args]);
            return { stdout: '', stderr: '', exitCode: 0 };
          },
        }),
      }),
    ).rejects.toMatchObject({ code: 'lifecycle-precondition-failed' });
    expect(calls).toEqual([]);
  });

  it('plans without applying projections or lifecycle effects', async (): Promise<void> => {
    let applied = false;
    const result = await run('plan', {
      applyProjection: async () => {
        applied = true;
      },
    });

    expect(result.applied).toBe(false);
    expect(result.lifecycle).toBe('not-applied');
    expect(applied).toBe(false);
  });

  it('reports a truthful partial result if projection application fails', async (): Promise<void> => {
    const result = await run('apply', {
      applyProjection: async () => {
        throw new Error('injected failure');
      },
    });

    expect(result).toMatchObject({
      applied: false,
      projections: 'incomplete',
      lifecycle: 'not-applied',
      recovery: { code: 'projection-apply-failed', action: 'regenerate-projections-from-roster' },
    });
  });

  it('adds cleanup diagnostics without masking projection or lifecycle partial truth', async (): Promise<void> => {
    const failingRelease = async (): Promise<never> => {
      throw new FleetReconcileError('lock-cleanup-failed', 'injected');
    };
    const projectionPartial = await run('apply', {
      applyProjection: async () => {
        throw new Error('injected projection failure');
      },
      acquireMutationLock: async () => failingRelease,
    });
    expect(projectionPartial).toMatchObject({
      projections: 'incomplete',
      lifecycle: 'not-applied',
      recovery: { code: 'projection-apply-failed' },
      cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
    });

    const lifecyclePartial = await run('apply', {
      runner: async () => ({ stdout: '', stderr: '', exitCode: 1 }),
      acquireMutationLock: async () => failingRelease,
    });
    expect(lifecyclePartial).toMatchObject({
      projections: 'complete',
      lifecycle: 'incomplete',
      recovery: { code: 'lifecycle-apply-failed' },
      cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
    });
  });

  it('adds cleanup diagnostics after successful effects without changing effect completion', async (): Promise<void> => {
    const result = await run('apply', {
      acquireMutationLock: async () => async () => {
        throw new FleetReconcileError('lock-cleanup-failed', 'injected');
      },
    });
    expect(result).toMatchObject({
      applied: true,
      projections: 'complete',
      lifecycle: 'complete',
      cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
    });
  });
});