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

import { constants } from 'node:fs';
import { lstat, open, readFile, unlink, type FileHandle } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { homedir } from 'node:os';
import { join } from 'node:path';
import {
  applyPreparedAgentEnvironmentProjection,
  prepareAgentEnvironmentProjection,
  type PreparedAgentEnvironmentProjection,
} from './generated-env-boundary.js';
import {
  validateRosterV2Semantics,
  type FleetRosterV2,
  type FleetRosterV2Agent,
} from './roster-v2.js';

export type FleetReconcileCommand =
  | 'plan'
  | 'apply'
  | 'reconcile'
  | 'start'
  | 'stop'
  | 'restart'
  | 'status'
  | 'verify'
  | 'doctor';

export interface FleetReconcileCommandResult {
  readonly stdout: string;
  readonly stderr: string;
  readonly exitCode: number;
}

export type FleetReconcileRunner = (
  command: string,
  args: readonly string[],
) => Promise<FleetReconcileCommandResult>;

export interface FleetReconcileDeps {
  readonly runner: FleetReconcileRunner;
  readonly mosaicHome?: string;
  readonly rolesDir?: string;
  readonly overrideDir?: string;
  readonly homeDirectory?: string;
  readonly readHolderIdentity?: () => Promise<string>;
  readonly validateRoster?: (roster: FleetRosterV2) => Promise<void>;
  readonly prepareProjections?: (roster: FleetRosterV2) => Promise<readonly unknown[]>;
  readonly applyProjection?: (prepared: unknown) => Promise<unknown>;
  /** Canonical roster reader; mutation authority is reread under the private lock. */
  readonly readRoster?: () => Promise<FleetRosterV2>;
  /** Test seam; production uses a private exclusive roster-adjacent lock. */
  readonly acquireMutationLock?: () => Promise<() => Promise<void>>;
  /** Internal recursion guard for the under-lock canonical roster read. */
  readonly lockAlreadyHeld?: boolean;
}

export interface FleetReconcileRequest {
  readonly roster: FleetRosterV2;
  readonly command: FleetReconcileCommand;
  readonly agentName?: string;
  readonly expectedGeneration?: number;
  readonly deps: FleetReconcileDeps;
}

export interface FleetReconcileObservedAgent {
  readonly name: string;
  readonly desiredState: 'running' | 'stopped';
  readonly enabled: boolean;
  readonly systemd: 'active' | 'inactive' | 'unknown';
  readonly tmux: 'present' | 'missing';
  readonly drift: readonly ('missing-session' | 'unexpected-session' | 'disabled-running')[];
}

export interface FleetReconcilePlan {
  readonly generation: number;
  readonly holder: 'owned' | 'missing' | 'ownership-mismatch';
  readonly agents: readonly FleetReconcileObservedAgent[];
  readonly unmanagedSessions: readonly string[];
}

export type FleetReconcileProjectionState = 'not-applied' | 'complete' | 'incomplete';
export type FleetReconcileLifecycleState = 'not-applied' | 'complete' | 'incomplete';

export interface FleetReconcileResult {
  readonly applied: boolean;
  readonly authoritativeRoster: 'unchanged';
  readonly projections: FleetReconcileProjectionState;
  readonly lifecycle: FleetReconcileLifecycleState;
  readonly plan: FleetReconcilePlan;
  readonly recovery?: {
    readonly code: 'projection-apply-failed' | 'lifecycle-apply-failed';
    readonly action:
      | 'regenerate-projections-from-roster'
      | 'rerun-after-inspecting-owned-resources';
  };
  /** Additive: effects remain truthful when private lock cleanup cannot be proven. */
  readonly cleanup?: {
    readonly code: 'lock-cleanup-failed';
    readonly action: 'inspect-lock-before-retry';
  };
}

export class FleetReconcileError extends Error {
  constructor(
    readonly code:
      | 'missing-generation'
      | 'stale-generation'
      | 'concurrent-mutation'
      | 'unsafe-managed-path'
      | 'unsafe-lock'
      | 'lock-io-failed'
      | 'lock-cleanup-failed'
      | 'agent-not-found'
      | 'disabled-agent'
      | 'ownership-mismatch'
      | 'unmanaged-session'
      | 'lifecycle-precondition-failed',
    message: string,
  ) {
    super(message);
    this.name = FleetReconcileError.name;
  }
}

/**
 * Reconciles only local roster-owned projections and exact service targets.
 * The roster is read-only desired state: no observed runtime result ever writes it.
 */
export async function executeFleetReconcile(
  request: FleetReconcileRequest,
): Promise<FleetReconcileResult> {
  const mutating =
    request.command === 'apply' ||
    request.command === 'reconcile' ||
    isLifecycleCommand(request.command);
  if (mutating && !request.deps.lockAlreadyHeld) {
    if (request.expectedGeneration === undefined) assertExpectedGeneration(request);
    const acquire =
      request.deps.acquireMutationLock ?? acquirePrivateReconcileLock(mosaicHomeFor(request.deps));
    const release = await acquire();
    let result: FleetReconcileResult | undefined;
    let primaryError: unknown;
    try {
      if (!request.deps.readRoster) {
        throw new FleetReconcileError(
          'lifecycle-precondition-failed',
          'Canonical roster state cannot be read for mutation.',
        );
      }
      const canonicalRoster = await request.deps.readRoster();
      result = await executeFleetReconcile({
        ...request,
        roster: canonicalRoster,
        deps: { ...request.deps, lockAlreadyHeld: true },
      });
    } catch (error: unknown) {
      primaryError = error;
    }
    try {
      await release();
    } catch (cleanupError: unknown) {
      if (result) {
        return {
          ...result,
          cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
        };
      }
      if (primaryError) throw primaryError;
      throw cleanupError;
    }
    if (primaryError) throw primaryError;
    return result as FleetReconcileResult;
  }

  assertExpectedGeneration(request);
  assertLocalRosterOnly(request.roster);
  const validateRoster = request.deps.validateRoster ?? defaultValidateRoster(request);
  await validateRoster(request.roster);
  assertLifecycleSocketAuthority(request);
  const plan = scopePlan(await observeFleet(request.roster, request.deps), request.agentName);

  if (request.command === 'status' || request.command === 'doctor') {
    return {
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'not-applied',
      plan,
    };
  }
  if (request.command === 'verify') {
    assertVerificationSafe(plan);
    return {
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'not-applied',
      plan,
    };
  }

  assertMutationSafe(plan, request.command);
  const prepareProjections = request.deps.prepareProjections ?? defaultPrepareProjections(request);
  const prepared = await prepareProjections(request.roster);
  if (request.command === 'plan') {
    return {
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'not-applied',
      plan,
    };
  }
  const targetAgents = targetAgentsFor(request.roster, request.agentName);
  const release = async (): Promise<void> => undefined;
  let result: FleetReconcileResult | undefined;
  let primaryError: unknown;
  try {
    if (request.command !== 'apply' && request.command !== 'reconcile') {
      result = await executeExplicitLifecycle(request, plan, targetAgents);
    } else {
      const applyProjection = request.deps.applyProjection ?? defaultApplyProjection;
      try {
        for (const projection of prepared) await applyProjection(projection);
      } catch {
        result = {
          applied: false,
          authoritativeRoster: 'unchanged',
          projections: 'incomplete',
          lifecycle: 'not-applied',
          plan,
          recovery: {
            code: 'projection-apply-failed',
            action: 'regenerate-projections-from-roster',
          },
        };
      }
      if (!result) {
        try {
          await applyDesiredLifecycle(request.roster, plan, request.deps);
          result = {
            applied: true,
            authoritativeRoster: 'unchanged',
            projections: 'complete',
            lifecycle: 'complete',
            plan,
          };
        } catch {
          result = {
            applied: false,
            authoritativeRoster: 'unchanged',
            projections: 'complete',
            lifecycle: 'incomplete',
            plan,
            recovery: {
              code: 'lifecycle-apply-failed',
              action: 'rerun-after-inspecting-owned-resources',
            },
          };
        }
      }
    }
  } catch (error: unknown) {
    primaryError = error;
  }

  try {
    await release();
  } catch (cleanupError: unknown) {
    if (result) {
      return {
        ...result,
        cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
      };
    }
    if (primaryError) throw primaryError;
    throw cleanupError;
  }
  if (primaryError) throw primaryError;
  return result as FleetReconcileResult;
}

function assertLocalRosterOnly(roster: FleetRosterV2): void {
  for (const agent of roster.agents) {
    const untypedAgent = agent as unknown as Record<string, unknown>;
    if (
      Object.hasOwn(untypedAgent, 'remote') ||
      Object.hasOwn(untypedAgent, 'ssh') ||
      Object.hasOwn(untypedAgent, 'connector') ||
      Object.hasOwn(untypedAgent, 'host')
    ) {
      throw new FleetReconcileError(
        'lifecycle-precondition-failed',
        'Remote or connector inventory cannot receive local lifecycle actions.',
      );
    }
  }
}

function assertExpectedGeneration(request: FleetReconcileRequest): void {
  if (isObservational(request.command)) return;
  if (request.expectedGeneration === undefined) {
    throw new FleetReconcileError(
      'missing-generation',
      'A roster generation is required for mutation.',
    );
  }
  if (request.expectedGeneration !== request.roster.generation) {
    throw new FleetReconcileError('stale-generation', 'The roster generation is stale.');
  }
}

function isObservational(command: FleetReconcileCommand): boolean {
  return command === 'plan' || command === 'status' || command === 'verify' || command === 'doctor';
}

async function observeFleet(
  roster: FleetRosterV2,
  deps: FleetReconcileDeps,
): Promise<FleetReconcilePlan> {
  const sessionsResult = await run(deps, 'tmux', [
    ...tmuxSocketArgs(roster.tmux.socketName),
    'list-sessions',
    '-F',
    '#{session_name}',
  ]);
  if (sessionsResult.exitCode !== 0) {
    return {
      generation: roster.generation,
      holder: 'missing',
      agents: await observeAgents(roster, deps, new Set<string>()),
      unmanagedSessions: [],
    };
  }

  const sessions = new Set(
    sessionsResult.stdout
      .split('\n')
      .map((value: string): string => value.trim())
      .filter((value: string): boolean => value.length > 0),
  );
  const knownSessions = new Set([
    roster.tmux.holderSession,
    ...roster.agents.map((agent) => agent.name),
  ]);
  const unmanagedSessions = [...sessions].filter(
    (session: string): boolean => !knownSessions.has(session),
  );
  const holder = await observeHolder(roster, deps, sessions);
  return {
    generation: roster.generation,
    holder,
    agents: await observeAgents(roster, deps, sessions),
    unmanagedSessions: Object.freeze(unmanagedSessions.sort()),
  };
}

async function observeHolder(
  roster: FleetRosterV2,
  deps: FleetReconcileDeps,
  sessions: ReadonlySet<string>,
): Promise<FleetReconcilePlan['holder']> {
  if (!sessions.has(roster.tmux.holderSession)) return 'ownership-mismatch';
  let owner: string;
  try {
    owner = await (deps.readHolderIdentity ?? defaultReadHolderIdentity(mosaicHomeFor(deps)))();
  } catch {
    return 'ownership-mismatch';
  }
  const environment = await run(deps, 'tmux', [
    ...tmuxSocketArgs(roster.tmux.socketName),
    'show-environment',
    '-g',
  ]);
  if (environment.exitCode !== 0) return 'ownership-mismatch';
  const homeDirectory = deps.homeDirectory ?? homedir();
  const expected = [
    `HOME=${homeDirectory}`,
    `MOSAIC_FLEET_OWNER=${owner}`,
    `MOSAIC_TMUX_HOLDER=${roster.tmux.holderSession}`,
    `MOSAIC_TMUX_SOCKET=${roster.tmux.socketName}`,
    'PATH=/usr/bin:/bin',
    `PWD=${homeDirectory}`,
  ].sort();
  const actual = environment.stdout
    .split('\n')
    .filter((line: string): boolean => line.length > 0)
    .sort();
  return sameStringArray(actual, expected) ? 'owned' : 'ownership-mismatch';
}

async function observeAgents(
  roster: FleetRosterV2,
  deps: FleetReconcileDeps,
  sessions: ReadonlySet<string>,
): Promise<readonly FleetReconcileObservedAgent[]> {
  return Promise.all(
    roster.agents.map(async (agent: FleetRosterV2Agent): Promise<FleetReconcileObservedAgent> => {
      const service = await run(deps, 'systemctl', [
        '--user',
        'show',
        `mosaic-agent@${agent.name}.service`,
        '-p',
        'ActiveState',
      ]);
      const active = /^ActiveState=active$/m.test(service.stdout)
        ? 'active'
        : service.exitCode === 0
          ? 'inactive'
          : 'unknown';
      const tmux = sessions.has(agent.name) ? 'present' : 'missing';
      const drift: Array<'missing-session' | 'unexpected-session' | 'disabled-running'> = [];
      if (
        agent.lifecycle.enabled &&
        agent.lifecycle.desiredState === 'running' &&
        tmux === 'missing'
      ) {
        drift.push('missing-session');
      }
      if (agent.lifecycle.desiredState === 'stopped' && tmux === 'present')
        drift.push('unexpected-session');
      if (!agent.lifecycle.enabled && tmux === 'present') drift.push('disabled-running');
      return {
        name: agent.name,
        desiredState: agent.lifecycle.desiredState,
        enabled: agent.lifecycle.enabled,
        systemd: active,
        tmux,
        drift: Object.freeze(drift),
      };
    }),
  );
}

function assertMutationSafe(plan: FleetReconcilePlan, command: FleetReconcileCommand): void {
  if (plan.holder === 'ownership-mismatch') {
    throw new FleetReconcileError(
      'ownership-mismatch',
      'The named tmux server ownership cannot be proven.',
    );
  }
  if (plan.unmanagedSessions.length > 0 && affectsHolder(command)) {
    throw new FleetReconcileError(
      'unmanaged-session',
      'Unmanaged sessions are present on the named socket.',
    );
  }
}

function affectsHolder(command: FleetReconcileCommand): boolean {
  return (
    command === 'apply' || command === 'reconcile' || command === 'start' || command === 'restart'
  );
}

function scopePlan(plan: FleetReconcilePlan, agentName?: string): FleetReconcilePlan {
  if (agentName === undefined) return plan;
  const agent = plan.agents.find(
    (candidate: FleetReconcileObservedAgent): boolean => candidate.name === agentName,
  );
  if (!agent)
    throw new FleetReconcileError('agent-not-found', 'The lifecycle target is not roster-owned.');
  return { ...plan, agents: [agent] };
}

function assertVerificationSafe(plan: FleetReconcilePlan): void {
  if (plan.holder !== 'owned' || plan.unmanagedSessions.length > 0) {
    throw new FleetReconcileError(
      'lifecycle-precondition-failed',
      'Fleet ownership cannot be verified.',
    );
  }
  if (plan.agents.some((agent: FleetReconcileObservedAgent): boolean => agent.drift.length > 0)) {
    throw new FleetReconcileError(
      'lifecycle-precondition-failed',
      'Fleet drift prevents verification.',
    );
  }
}

function assertLifecycleSocketAuthority(request: FleetReconcileRequest): void {
  const usesFixedLifecycleUnits =
    request.command === 'apply' ||
    request.command === 'reconcile' ||
    isLifecycleCommand(request.command);
  if (usesFixedLifecycleUnits && request.roster.tmux.socketName === '') {
    throw new FleetReconcileError(
      'lifecycle-precondition-failed',
      'Default-server lifecycle mutation is unsupported by the fixed named-socket systemd units.',
    );
  }
}

function targetAgentsFor(roster: FleetRosterV2, agentName?: string): readonly FleetRosterV2Agent[] {
  if (agentName === undefined) return roster.agents;
  const agent = roster.agents.find(
    (candidate: FleetRosterV2Agent): boolean => candidate.name === agentName,
  );
  if (!agent)
    throw new FleetReconcileError('agent-not-found', 'The lifecycle target is not roster-owned.');
  return [agent];
}

async function executeExplicitLifecycle(
  request: FleetReconcileRequest,
  plan: FleetReconcilePlan,
  agents: readonly FleetRosterV2Agent[],
): Promise<FleetReconcileResult> {
  if (request.command === 'start') {
    for (const agent of agents) {
      if (!agent.lifecycle.enabled) {
        throw new FleetReconcileError(
          'disabled-agent',
          'A disabled roster agent cannot be started.',
        );
      }
    }
  }
  try {
    if (request.command === 'start' && plan.holder === 'missing') {
      await runChecked(request.deps, 'systemctl', [
        '--user',
        'start',
        'mosaic-tmux-holder.service',
      ]);
    }
    for (const agent of agents) {
      await runChecked(request.deps, 'systemctl', [
        '--user',
        request.command,
        `mosaic-agent@${agent.name}.service`,
      ]);
    }
  } catch {
    return {
      applied: false,
      authoritativeRoster: 'unchanged',
      projections: 'not-applied',
      lifecycle: 'incomplete',
      plan,
      recovery: {
        code: 'lifecycle-apply-failed',
        action: 'rerun-after-inspecting-owned-resources',
      },
    };
  }
  return {
    applied: true,
    authoritativeRoster: 'unchanged',
    projections: 'not-applied',
    lifecycle: 'complete',
    plan,
  };
}

async function applyDesiredLifecycle(
  roster: FleetRosterV2,
  plan: FleetReconcilePlan,
  deps: FleetReconcileDeps,
): Promise<void> {
  const needsRunningAgent = roster.agents.some(
    (agent: FleetRosterV2Agent): boolean =>
      agent.lifecycle.enabled && agent.lifecycle.desiredState === 'running',
  );
  if (needsRunningAgent && plan.holder === 'missing') {
    await runChecked(deps, 'systemctl', ['--user', 'start', 'mosaic-tmux-holder.service']);
  }
  for (const agent of roster.agents) {
    const action =
      agent.lifecycle.enabled && agent.lifecycle.desiredState === 'running' ? 'start' : 'stop';
    await runChecked(deps, 'systemctl', ['--user', action, `mosaic-agent@${agent.name}.service`]);
  }
}

function defaultValidateRoster(
  request: FleetReconcileRequest,
): (roster: FleetRosterV2) => Promise<void> {
  return async (roster: FleetRosterV2): Promise<void> => {
    const mosaicHome = mosaicHomeFor(request.deps);
    await validateRosterV2Semantics(roster, {
      rolesDir: request.deps.rolesDir ?? join(mosaicHome, 'fleet', 'roles'),
      overrideDir: request.deps.overrideDir ?? join(mosaicHome, 'fleet', 'roles.local'),
    });
  };
}

/**
 * The single roster-v2 → generated-projection mapping. Both the reconciler's
 * apply path ({@link defaultPrepareProjections}) and the recovery-framed
 * `mosaic fleet regen` command derive their generated env from THIS one
 * function, so the two paths can never drift (the #791 single-SSOT invariant).
 * Pure: no IO, no clock — a deterministic function of the roster alone.
 */
export function projectRosterV2AgentGeneratedEnv(
  roster: FleetRosterV2,
  agent: FleetRosterV2Agent,
): Readonly<Record<string, string>> {
  return {
    MOSAIC_AGENT_NAME: agent.name,
    MOSAIC_AGENT_CLASS: agent.className,
    MOSAIC_AGENT_RUNTIME: agent.runtime,
    MOSAIC_AGENT_MODEL: agent.model,
    MOSAIC_AGENT_REASONING: agent.reasoning,
    MOSAIC_AGENT_TOOL_POLICY: agent.toolPolicy,
    MOSAIC_AGENT_WORKDIR: agent.workingDirectory,
    MOSAIC_TMUX_SOCKET: roster.tmux.socketName,
  };
}

function defaultPrepareProjections(
  request: FleetReconcileRequest,
): (roster: FleetRosterV2) => Promise<readonly PreparedAgentEnvironmentProjection[]> {
  return async (roster: FleetRosterV2): Promise<readonly PreparedAgentEnvironmentProjection[]> => {
    const mosaicHome = mosaicHomeFor(request.deps);
    return Promise.all(
      roster.agents.map(
        (agent: FleetRosterV2Agent): Promise<PreparedAgentEnvironmentProjection> =>
          prepareAgentEnvironmentProjection({
            mosaicHome,
            agentEnvDir: join(mosaicHome, 'fleet', 'agents'),
            agentName: agent.name,
            generated: projectRosterV2AgentGeneratedEnv(roster, agent),
          }),
      ),
    );
  };
}

async function defaultApplyProjection(prepared: unknown): Promise<unknown> {
  return applyPreparedAgentEnvironmentProjection(prepared as PreparedAgentEnvironmentProjection);
}

function mosaicHomeFor(deps: FleetReconcileDeps): string {
  return deps.mosaicHome ?? join(homedir(), '.config', 'mosaic');
}

/** Acquires the private reconcile lock only after proving the canonical managed path. */
export function acquirePrivateReconcileLock(
  mosaicHome: string,
  openLock: typeof open = open,
): () => Promise<() => Promise<void>> {
  return acquirePrivateManagedRosterLock(
    mosaicHome,
    'roster.yaml.reconcile.lock',
    'Another roster reconciliation is in progress.',
    openLock,
  );
}

/**
 * Acquires the private roster MUTATION lock (`roster.yaml.mutation.lock`) with the
 * exact same ownership-proof discipline as the reconcile lock. Exposed so the
 * recovery-framed `mosaic fleet regen` can serialize its projection rewrites
 * against agent CRUD — which holds this very lock while it rewrites the roster and
 * the same derived projections — reusing this hardened acquirer rather than
 * reimplementing it.
 *
 * The release PROVES ownership (device/inode + token) before unlinking, so a lock
 * that was already cleared and replaced by the time release runs is never deleted
 * out from under the new owner (it fails closed as `lock-cleanup-failed` instead),
 * and that cleanup fault is propagated to the caller rather than silently dropped.
 * CRUD contends on the identical path with a plain empty-file `wx` create; the two
 * never co-own it (whoever wins `wx` owns it; the loser gets `concurrent-mutation`),
 * so the token this writes is only ever read back by the same regen invocation that
 * wrote it.
 *
 * SCOPE of the guarantee (matches the merged reconcile lock exactly — see
 * {@link acquirePrivateManagedRosterLock}): the ownership proof closes the
 * *reachable* window — a stale-lock reaper or operator cleared our lock and another
 * writer took it BEFORE our release began — not the residual sub-instruction window
 * between the final check and the path-based `unlink`. Closing that residual fully
 * requires an fd-held advisory lock adopted by every fleet writer (CRUD, reconcile,
 * regen), which is a cross-cutting mechanism change out of scope for this
 * projection-only recovery command; it is unreachable within the `wx` writer
 * protocol regardless (no Mosaic writer removes a lock it does not own, so only
 * external interference can vacate our inode mid-release).
 */
export function acquirePrivateRosterMutationLock(
  mosaicHome: string,
  openLock: typeof open = open,
): () => Promise<() => Promise<void>> {
  return acquirePrivateManagedRosterLock(
    mosaicHome,
    'roster.yaml.mutation.lock',
    'Another roster mutation is in progress.',
    openLock,
  );
}

/**
 * Shared body for the ownership-proving managed roster locks. Acquires `lockLeaf`
 * under `<mosaicHome>/fleet` with a private `wx` create, writes an ownership token,
 * and returns a release that re-proves device/inode + token before unlinking so it
 * can never remove a lock it no longer owns. Non-blocking: throws `concurrent-mutation`
 * (with `busyMessage`) on contention rather than waiting.
 */
function acquirePrivateManagedRosterLock(
  mosaicHome: string,
  lockLeaf: string,
  busyMessage: string,
  openLock: typeof open,
): () => Promise<() => Promise<void>> {
  const fleetDir = join(mosaicHome, 'fleet');
  const lockPath = join(fleetDir, lockLeaf);
  // The message-producing helpers below serve BOTH managed locks, so every fault
  // names the actual lock file (`fleet/<leaf>`) — an operator needs to know WHICH
  // lock is stale, not a hardcoded "reconciliation lock".
  const lockLabel = `fleet/${lockLeaf}`;
  return async (): Promise<() => Promise<void>> => {
    await assertPrivateManagedDirectory(mosaicHome);
    await assertPrivateManagedDirectory(fleetDir);
    await assertSafeLockLeafIfPresent(lockPath, lockLabel);

    let handle: FileHandle;
    try {
      handle = await openLock(lockPath, 'wx', 0o600);
    } catch (error: unknown) {
      if (isCode(error, 'EEXIST')) {
        await assertSafeLockLeafIfPresent(lockPath, lockLabel);
        throw new FleetReconcileError('concurrent-mutation', busyMessage);
      }
      throw new FleetReconcileError('lock-io-failed', `The ${lockLabel} lock cannot be created.`);
    }

    // Identity of the lock WE exclusively created (the `wx` create guaranteed it is
    // ours). Captured up front so both the release closure and the init-failure
    // cleanup below can prove ownership by device/inode before touching the file.
    const created = await handle.stat().catch((): undefined => undefined);

    const token = randomUUID();
    let tokenPersisted = false;
    try {
      await handle.writeFile(`${token}\n`, 'utf8');
      tokenPersisted = true;
      await handle.close();
      if (!created)
        throw new FleetReconcileError(
          'lock-io-failed',
          `The ${lockLabel} lock cannot be initialized.`,
        );
      await assertLockOwnership(
        lockPath,
        created.dev,
        created.ino,
        token,
        'unsafe-lock',
        lockLabel,
      );
      return async (): Promise<void> => {
        try {
          await assertLockOwnership(
            lockPath,
            created.dev,
            created.ino,
            token,
            'lock-cleanup-failed',
            lockLabel,
          );
          await unlink(lockPath);
        } catch (error: unknown) {
          if (error instanceof FleetReconcileError) throw error;
          throw new FleetReconcileError('lock-cleanup-failed', `The ${lockLabel} cleanup failed.`);
        }
      };
    } catch (error: unknown) {
      await handle.close().catch((): void => {});
      // Best-effort: remove the lock THIS invocation created so a transient init
      // fault does not strand a lock that would block future regen and agent CRUD.
      // dev/ino-guarded so it can never delete a replacement; swallowed so cleanup
      // failure never masks the initialization error being surfaced. The token is
      // passed only once persisted, so the fallback path (when the post-create stat
      // failed and dev/ino is unavailable) can still prove ownership by content.
      await removeOwnedLockLeafBestEffort(lockPath, created, tokenPersisted ? token : undefined);
      if (error instanceof FleetReconcileError) throw error;
      throw new FleetReconcileError(
        'lock-io-failed',
        `The ${lockLabel} lock cannot be initialized.`,
      );
    }
  };
}

/**
 * Best-effort removal of a managed lock leaf THIS process created, used only on the
 * initialization-failure path. Two independent ownership proofs, so a lock that was
 * already replaced is never deleted:
 *
 *   - primary: the captured device/inode of the file we exclusively `wx`-created; or
 *   - fallback: when the post-create stat itself failed (dev/ino unavailable), the
 *     random `ownershipToken` we persisted — only OUR lock carries it, so a CRUD
 *     (empty) or a differently-tokened replacement is never removed.
 *
 * Every fault is swallowed because the caller is already surfacing the initialization
 * error and a leftover lock is recoverable via the documented runbook. If BOTH proofs
 * are unavailable (stat failed AND the token write never landed) the lock is left in
 * place rather than risk deleting another writer's file — a doubly-degenerate case
 * requiring two independent fs faults on a just-created fd.
 */
async function removeOwnedLockLeafBestEffort(
  lockPath: string,
  created: { dev: number; ino: number } | undefined,
  ownershipToken: string | undefined,
): Promise<void> {
  try {
    const current = await lstat(lockPath);
    if (!current.isFile() || current.isSymbolicLink()) return;
    if (created) {
      if (current.dev === created.dev && current.ino === created.ino) {
        await unlink(lockPath);
      }
      return;
    }
    if (ownershipToken !== undefined) {
      const contents = await readFile(lockPath, 'utf8');
      if (contents.trim() === ownershipToken) {
        await unlink(lockPath);
      }
    }
  } catch {
    // Swallowed: leftover lock is recoverable; do not mask the init error.
  }
}

async function assertPrivateManagedDirectory(path: string): Promise<void> {
  try {
    const metadata = await lstat(path);
    if (!metadata.isDirectory() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
      throw new FleetReconcileError('unsafe-managed-path', 'The managed lock ancestor is unsafe.');
    }
  } catch (error: unknown) {
    if (error instanceof FleetReconcileError) throw error;
    throw new FleetReconcileError(
      'unsafe-managed-path',
      'The managed lock ancestor is unavailable.',
    );
  }
}

async function assertSafeLockLeafIfPresent(lockPath: string, lockLabel: string): Promise<void> {
  try {
    const metadata = await lstat(lockPath);
    if (!metadata.isFile() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
      throw new FleetReconcileError('unsafe-lock', `The ${lockLabel} lock path is unsafe.`);
    }
  } catch (error: unknown) {
    if (isCode(error, 'ENOENT')) return;
    if (error instanceof FleetReconcileError) throw error;
    throw new FleetReconcileError('unsafe-lock', `The ${lockLabel} lock path is unavailable.`);
  }
}

async function assertLockOwnership(
  lockPath: string,
  device: number,
  inode: number,
  token: string,
  failureCode: 'unsafe-lock' | 'lock-cleanup-failed',
  lockLabel: string,
): Promise<void> {
  try {
    const metadata = await lstat(lockPath);
    if (
      !metadata.isFile() ||
      metadata.isSymbolicLink() ||
      (metadata.mode & 0o077) !== 0 ||
      metadata.dev !== device ||
      metadata.ino !== inode
    ) {
      throw new FleetReconcileError(failureCode, `The ${lockLabel} ownership changed.`);
    }
    const handle = await open(lockPath, constants.O_RDONLY | constants.O_NOFOLLOW);
    try {
      const opened = await handle.stat();
      const contents = await handle.readFile({ encoding: 'utf8' });
      if (opened.dev !== device || opened.ino !== inode || contents !== `${token}\n`) {
        throw new FleetReconcileError(failureCode, `The ${lockLabel} ownership changed.`);
      }
    } finally {
      await handle.close();
    }
  } catch (error: unknown) {
    if (error instanceof FleetReconcileError) throw error;
    throw new FleetReconcileError(failureCode, `The ${lockLabel} ownership cannot be proven.`);
  }
}

function isLifecycleCommand(command: FleetReconcileCommand): boolean {
  return command === 'start' || command === 'stop' || command === 'restart';
}

function isCode(error: unknown, code: string): boolean {
  return typeof error === 'object' && error !== null && 'code' in error && error.code === code;
}

function defaultReadHolderIdentity(mosaicHome: string): () => Promise<string> {
  return async (): Promise<string> => {
    const fleetDir = join(mosaicHome, 'fleet');
    const runDir = join(fleetDir, 'run');
    for (const directory of [mosaicHome, fleetDir, runDir]) {
      const metadata = await lstat(directory);
      if (!metadata.isDirectory() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
        throw new Error('Unsafe holder identity ancestor.');
      }
    }
    const identityPath = join(runDir, 'holder-owner');
    const metadata = await lstat(identityPath);
    if (!metadata.isFile() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
      throw new Error('Unsafe holder identity.');
    }
    const value = (await readFile(identityPath, 'utf8')).trim();
    if (!/^[a-f0-9-]{36}$/.test(value)) throw new Error('Malformed holder identity.');
    return value;
  };
}

async function run(
  deps: FleetReconcileDeps,
  command: string,
  args: readonly string[],
): Promise<FleetReconcileCommandResult> {
  return deps.runner(command, args);
}

async function runChecked(
  deps: FleetReconcileDeps,
  command: string,
  args: readonly string[],
): Promise<void> {
  const result = await run(deps, command, args);
  if (result.exitCode !== 0) throw new Error('Lifecycle action failed.');
}

function tmuxSocketArgs(socketName: string): readonly string[] {
  return socketName === '' ? [] : ['-L', socketName];
}

function sameStringArray(left: readonly string[], right: readonly string[]): boolean {
  return (
    left.length === right.length &&
    left.every((value: string, index: number): boolean => value === right[index])
  );
}