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; export interface FleetReconcileDeps { readonly runner: FleetReconcileRunner; readonly mosaicHome?: string; readonly rolesDir?: string; readonly overrideDir?: string; readonly homeDirectory?: string; readonly readHolderIdentity?: () => Promise; readonly validateRoster?: (roster: FleetRosterV2) => Promise; readonly prepareProjections?: (roster: FleetRosterV2) => Promise; readonly applyProjection?: (prepared: unknown) => Promise; /** Canonical roster reader; mutation authority is reread under the private lock. */ readonly readRoster?: () => Promise; /** Test seam; production uses a private exclusive roster-adjacent lock. */ readonly acquireMutationLock?: () => Promise<() => Promise>; /** 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 { 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 => 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; 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 { 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()), 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, ): Promise { 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, ): Promise { return Promise.all( roster.agents.map(async (agent: FleetRosterV2Agent): Promise => { 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 { 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 { 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 { return async (roster: FleetRosterV2): Promise => { 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'), }); }; } function defaultPrepareProjections( request: FleetReconcileRequest, ): (roster: FleetRosterV2) => Promise { return async (roster: FleetRosterV2): Promise => { const mosaicHome = mosaicHomeFor(request.deps); return Promise.all( roster.agents.map( (agent: FleetRosterV2Agent): Promise => prepareAgentEnvironmentProjection({ mosaicHome, agentEnvDir: join(mosaicHome, 'fleet', 'agents'), agentName: agent.name, generated: { 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, }, }), ), ); }; } async function defaultApplyProjection(prepared: unknown): Promise { return applyPreparedAgentEnvironmentProjection(prepared as PreparedAgentEnvironmentProjection); } function mosaicHomeFor(deps: FleetReconcileDeps): string { return deps.mosaicHome ?? join(homedir(), '.config', 'mosaic'); } /** Acquires a private lock only after proving the canonical managed path. */ export function acquirePrivateReconcileLock( mosaicHome: string, openLock: typeof open = open, ): () => Promise<() => Promise> { const fleetDir = join(mosaicHome, 'fleet'); const lockPath = join(fleetDir, 'roster.yaml.reconcile.lock'); return async (): Promise<() => Promise> => { await assertPrivateManagedDirectory(mosaicHome); await assertPrivateManagedDirectory(fleetDir); await assertSafeLockLeafIfPresent(lockPath); let handle: FileHandle; try { handle = await openLock(lockPath, 'wx', 0o600); } catch (error: unknown) { if (isCode(error, 'EEXIST')) { await assertSafeLockLeafIfPresent(lockPath); throw new FleetReconcileError( 'concurrent-mutation', 'Another roster reconciliation is in progress.', ); } throw new FleetReconcileError('lock-io-failed', 'The reconciliation lock cannot be created.'); } const token = randomUUID(); try { await handle.writeFile(`${token}\n`, 'utf8'); const opened = await handle.stat(); await handle.close(); await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'unsafe-lock'); return async (): Promise => { try { await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'lock-cleanup-failed'); await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'lock-cleanup-failed'); await unlink(lockPath); } catch (error: unknown) { if (error instanceof FleetReconcileError) throw error; throw new FleetReconcileError( 'lock-cleanup-failed', 'The reconciliation lock cleanup failed.', ); } }; } catch (error: unknown) { await handle.close().catch((): void => {}); if (error instanceof FleetReconcileError) throw error; throw new FleetReconcileError( 'lock-io-failed', 'The reconciliation lock cannot be initialized.', ); } }; } async function assertPrivateManagedDirectory(path: string): Promise { 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): Promise { try { const metadata = await lstat(lockPath); if (!metadata.isFile() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) { throw new FleetReconcileError('unsafe-lock', 'The reconciliation lock path is unsafe.'); } } catch (error: unknown) { if (isCode(error, 'ENOENT')) return; if (error instanceof FleetReconcileError) throw error; throw new FleetReconcileError('unsafe-lock', 'The reconciliation lock path is unavailable.'); } } async function assertLockOwnership( lockPath: string, device: number, inode: number, token: string, failureCode: 'unsafe-lock' | 'lock-cleanup-failed', ): Promise { 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 reconciliation lock 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 reconciliation lock ownership changed.'); } } finally { await handle.close(); } } catch (error: unknown) { if (error instanceof FleetReconcileError) throw error; throw new FleetReconcileError( failureCode, 'The reconciliation lock 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 { return async (): Promise => { 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 { return deps.runner(command, args); } async function runChecked( deps: FleetReconcileDeps, command: string, args: readonly string[], ): Promise { 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]) ); }