feat(fleet): add generation-guarded agent CRUD (#773)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful

This commit was merged in pull request #773.
This commit is contained in:
2026-07-15 12:03:05 +00:00
parent 191efaefeb
commit bc5e73629e
10 changed files with 1866 additions and 0 deletions

View File

@@ -0,0 +1,220 @@
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import {
FleetAgentMutationError,
executeFleetAgentMutation,
planFleetAgentMutation,
type FleetAgentMutationRequest,
} from './fleet-agent-crud.js';
import { parseRosterV2, renderRosterV2Yaml } from './roster-v2.js';
const roster = parseRosterV2(`
version: 2
generation: 7
transport: tmux
tmux:
socket_name: mosaic-fleet
holder_session: _holder
defaults:
working_directory: /srv/mosaic
runtime: pi
runtimes:
pi:
reset_command: /new
agents:
- name: orchestrator
alias: Orchestrator
class: orchestrator
runtime: pi
provider: openai
model: gpt-5.6-sol
reasoning: high
tool_policy: orchestrator
working_directory: /srv/mosaic
persistent_persona: true
reset_between_tasks: false
lifecycle:
enabled: true
desired_state: stopped
launch:
yolo: true
`);
function createRequest(): FleetAgentMutationRequest {
return {
operation: 'create',
expectedGeneration: 7,
agent: {
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,
launch: { yolo: true },
},
};
}
let cleanup: string | undefined;
afterEach(async (): Promise<void> => {
if (cleanup) await rm(cleanup, { recursive: true, force: true });
cleanup = undefined;
});
async function semanticDirs(): Promise<{ rolesDir: string; overrideDir: string }> {
cleanup = await mkdtemp(join(tmpdir(), 'mosaic-fleet-crud-'));
const rolesDir = join(cleanup, 'roles');
const overrideDir = join(cleanup, 'roles.local');
await mkdir(rolesDir, { recursive: true });
await mkdir(overrideDir, { recursive: true });
for (const klass of ['orchestrator', 'code']) {
await writeFile(join(rolesDir, `${klass}.md`), `# ${klass}\n\n(\`class: ${klass}\`)\n`);
}
return { rolesDir, overrideDir };
}
describe('fleet agent CRUD plan', (): void => {
it('creates stopped-by-default desired state and a deterministic generation plan', (): void => {
const plan = planFleetAgentMutation(roster, createRequest());
expect(plan).toMatchObject({
operation: 'create',
currentGeneration: 7,
nextGeneration: 8,
changed: true,
agent: { name: 'coder0', lifecycle: { enabled: true, desiredState: 'stopped' } },
});
});
it('rejects a stale generation before proposing effects', (): void => {
expect((): void => {
planFleetAgentMutation(roster, { ...createRequest(), expectedGeneration: 6 });
}).toThrow(FleetAgentMutationError);
});
it('treats an equivalent retry as an idempotent no-op', (): void => {
const created = planFleetAgentMutation(roster, createRequest()).roster;
const retried = planFleetAgentMutation(created, {
...createRequest(),
expectedGeneration: created.generation,
});
expect(retried).toMatchObject({ changed: false, currentGeneration: 8, nextGeneration: 8 });
});
});
describe('fleet agent CRUD execution', (): void => {
it('keeps roster and projections unchanged for dry-run', async (): Promise<void> => {
const dirs = await semanticDirs();
const home = join(cleanup!, 'mosaic');
const rosterPath = join(home, 'fleet', 'roster.yaml');
await mkdir(join(home, 'fleet', 'agents'), { recursive: true, mode: 0o700 });
await writeFile(rosterPath, 'before\n', { mode: 0o600 });
const result = await executeFleetAgentMutation({
roster,
request: createRequest(),
mosaicHome: home,
rosterPath,
agentEnvDir: join(home, 'fleet', 'agents'),
rolesDir: dirs.rolesDir,
overrideDir: dirs.overrideDir,
dryRun: true,
});
expect(result.applied).toBe(false);
expect(await readFile(rosterPath, 'utf8')).toBe('before\n');
await expect(
readFile(join(home, 'fleet', 'agents', 'coder0.env.generated'), 'utf8'),
).rejects.toThrow();
});
it('fails closed when another writer owns the mutation lock', async (): Promise<void> => {
const dirs = await semanticDirs();
const home = join(cleanup!, 'mosaic');
const rosterPath = join(home, 'fleet', 'roster.yaml');
await mkdir(join(home, 'fleet', 'agents'), { recursive: true, mode: 0o700 });
await writeFile(rosterPath, renderRosterV2Yaml(roster), { mode: 0o600 });
await writeFile(`${rosterPath}.mutation.lock`, 'owned\n', { mode: 0o600 });
await expect(
executeFleetAgentMutation({
roster,
request: createRequest(),
mosaicHome: home,
rosterPath,
agentEnvDir: join(home, 'fleet', 'agents'),
rolesDir: dirs.rolesDir,
overrideDir: dirs.overrideDir,
}),
).rejects.toMatchObject({ code: 'concurrent-mutation' });
expect(await readFile(rosterPath, 'utf8')).toBe(renderRosterV2Yaml(roster));
});
it('deletes only the removed agent generated projection when it is stale or absent', async (): Promise<void> => {
const dirs = await semanticDirs();
const home = join(cleanup!, 'mosaic');
const rosterPath = join(home, 'fleet', 'roster.yaml');
const agentEnvDir = join(home, 'fleet', 'agents');
const created = planFleetAgentMutation(roster, createRequest()).roster;
await mkdir(agentEnvDir, { recursive: true, mode: 0o700 });
await writeFile(rosterPath, renderRosterV2Yaml(created), { mode: 0o600 });
await writeFile(join(agentEnvDir, 'coder0.env.local'), 'MOSAIC_RUNTIME_BIN=/usr/bin/pi\n', {
mode: 0o600,
});
const result = await executeFleetAgentMutation({
roster: created,
request: { operation: 'delete', expectedGeneration: 8, name: 'coder0' },
mosaicHome: home,
rosterPath,
agentEnvDir,
rolesDir: dirs.rolesDir,
overrideDir: dirs.overrideDir,
});
expect(result).toMatchObject({ applied: true, plan: { nextGeneration: 9 } });
await expect(readFile(join(agentEnvDir, 'coder0.env.generated'), 'utf8')).rejects.toThrow();
expect(await readFile(join(agentEnvDir, 'coder0.env.local'), 'utf8')).toBe(
'MOSAIC_RUNTIME_BIN=/usr/bin/pi\n',
);
});
it('returns redacted projection recovery after the authoritative roster write', async (): Promise<void> => {
const dirs = await semanticDirs();
const home = join(cleanup!, 'mosaic');
const rosterPath = join(home, 'fleet', 'roster.yaml');
await mkdir(join(home, 'fleet', 'agents'), { recursive: true, mode: 0o700 });
await writeFile(rosterPath, renderRosterV2Yaml(roster), { mode: 0o600 });
const result = await executeFleetAgentMutation({
roster,
request: createRequest(),
mosaicHome: home,
rosterPath,
agentEnvDir: join(home, 'fleet', 'agents'),
rolesDir: dirs.rolesDir,
overrideDir: dirs.overrideDir,
projectionApplier: async (): Promise<void> => {
throw new Error('MOSAIC_AGENT_COMMAND=must-not-leak');
},
});
expect(result).toMatchObject({
applied: false,
authoritativeRoster: 'committed',
projections: 'incomplete',
recovery: { code: 'projection-apply-failed', action: 'regenerate-projections-from-roster' },
});
expect(JSON.stringify(result)).not.toContain('must-not-leak');
expect(parseRosterV2(await readFile(rosterPath, 'utf8'), 'yaml').generation).toBe(8);
});
});

View File

@@ -0,0 +1,397 @@
import { open, readFile, unlink } from 'node:fs/promises';
import { dirname, join } from 'node:path';
import {
applyPreparedAgentEnvironmentProjection,
prepareAgentGeneratedProjectionDeletion,
prepareAgentEnvironmentProjection,
type PreparedAgentEnvironmentProjection,
writeManagedFleetRoster,
} from './generated-env-boundary.js';
import {
parseRosterV2,
renderRosterV2Yaml,
validateRosterV2Semantics,
type FleetRosterV2,
type FleetRosterV2Agent,
type FleetRosterV2Launch,
type FleetRosterV2Lifecycle,
type RosterV2ReasoningLevel,
type RosterV2RuntimeName,
} from './roster-v2.js';
export type FleetAgentMutationOperation = 'create' | 'update' | 'delete';
export interface FleetAgentMutationAgent {
readonly name: string;
readonly alias: string;
readonly className: string;
readonly runtime: RosterV2RuntimeName;
readonly provider: string;
readonly model: string;
readonly reasoning: RosterV2ReasoningLevel;
readonly toolPolicy: string;
readonly workingDirectory: string;
readonly persistentPersona: boolean;
readonly resetBetweenTasks: boolean;
readonly launch: FleetRosterV2Launch;
}
export interface FleetAgentMutationRequest {
readonly operation: FleetAgentMutationOperation;
readonly expectedGeneration: number;
readonly name?: string;
readonly agent?: FleetAgentMutationAgent;
readonly persistedStart?: boolean;
}
export interface FleetAgentMutationPlan {
readonly operation: FleetAgentMutationOperation;
readonly currentGeneration: number;
readonly nextGeneration: number;
readonly changed: boolean;
readonly roster: FleetRosterV2;
readonly agent?: FleetRosterV2Agent;
}
export type FleetAgentMutationAuthoritativeRosterState = 'unchanged' | 'committed';
export type FleetAgentMutationProjectionState = 'not-applied' | 'complete' | 'incomplete';
export interface FleetAgentMutationResult {
/** True only when every requested roster and projection write completed. */
readonly applied: boolean;
/** Whether the authoritative roster changed on disk. */
readonly authoritativeRoster: FleetAgentMutationAuthoritativeRosterState;
/** Whether derived projections were skipped, complete, or require recovery. */
readonly projections: FleetAgentMutationProjectionState;
readonly plan: FleetAgentMutationPlan;
readonly recovery?: FleetAgentMutationRecovery;
}
export interface FleetAgentMutationRecovery {
readonly code: 'projection-apply-failed';
readonly rosterPath: string;
readonly action: 'regenerate-projections-from-roster';
}
export interface FleetAgentMutationOptions {
readonly roster: FleetRosterV2;
readonly request: FleetAgentMutationRequest;
readonly mosaicHome: string;
readonly rosterPath: string;
readonly agentEnvDir: string;
readonly rolesDir: string;
readonly overrideDir: string;
readonly dryRun?: boolean;
readonly projectionApplier?: (prepared: PreparedAgentEnvironmentProjection) => Promise<unknown>;
}
export class FleetAgentMutationError extends Error {
constructor(
readonly code:
| 'stale-generation'
| 'agent-conflict'
| 'agent-not-found'
| 'invalid-request'
| 'concurrent-mutation'
| 'projection-apply-failed',
message: string,
) {
super(message);
this.name = FleetAgentMutationError.name;
}
}
/** Plans a generation-guarded desired-state mutation without writing any file. */
export function planFleetAgentMutation(
roster: FleetRosterV2,
request: FleetAgentMutationRequest,
): FleetAgentMutationPlan {
if (request.expectedGeneration !== roster.generation) {
throw new FleetAgentMutationError(
'stale-generation',
`Expected roster generation ${request.expectedGeneration}; current generation is ${roster.generation}.`,
);
}
if (request.operation === 'create') return planCreate(roster, request);
if (request.operation === 'update') return planUpdate(roster, request);
return planDelete(roster, request);
}
/**
* Validates the complete proposed roster and its deterministic projections before
* changing the roster authority. The only post-roster write is a derived projection.
*/
export async function executeFleetAgentMutation(
options: FleetAgentMutationOptions,
): Promise<FleetAgentMutationResult> {
const plan = planFleetAgentMutation(options.roster, options.request);
await validateRosterV2Semantics(plan.roster, {
rolesDir: options.rolesDir,
overrideDir: options.overrideDir,
});
const prepared = await prepareProjections(plan.roster, options.mosaicHome, options.agentEnvDir);
if (plan.operation === 'delete' && plan.agent) {
// Delete validates only the exact generated projection. Local overrides,
// legacy input, and quarantine records are retained operator state.
await prepareAgentGeneratedProjectionDeletion({
mosaicHome: options.mosaicHome,
agentEnvDir: options.agentEnvDir,
agentName: plan.agent.name,
});
}
if (options.dryRun === true || !plan.changed) {
return { applied: false, authoritativeRoster: 'unchanged', projections: 'not-applied', plan };
}
const lockPath = `${options.rosterPath}.mutation.lock`;
const release = await acquireMutationLock(lockPath);
try {
const persisted = parseRosterV2(await readFile(options.rosterPath, 'utf8'), 'yaml');
if (persisted.generation !== options.roster.generation) {
throw new FleetAgentMutationError(
'stale-generation',
`Expected roster generation ${options.roster.generation}; current generation is ${persisted.generation}.`,
);
}
await writeManagedFleetRoster(
options.mosaicHome,
options.rosterPath,
renderRosterV2Yaml(plan.roster),
);
const apply = options.projectionApplier ?? applyPreparedAgentEnvironmentProjection;
try {
for (const projection of prepared) await apply(projection);
if (plan.operation === 'delete' && plan.agent) {
await prepareAgentGeneratedProjectionDeletion({
mosaicHome: options.mosaicHome,
agentEnvDir: options.agentEnvDir,
agentName: plan.agent.name,
});
await removeGeneratedProjection(options.agentEnvDir, plan.agent.name);
}
} catch {
throw new FleetAgentMutationError(
'projection-apply-failed',
`Projection application failed after roster write; regenerate projections from ${options.rosterPath}.`,
);
}
return { applied: true, authoritativeRoster: 'committed', projections: 'complete', plan };
} catch (error: unknown) {
if (error instanceof FleetAgentMutationError && error.code === 'projection-apply-failed') {
return {
applied: false,
authoritativeRoster: 'committed',
projections: 'incomplete',
plan,
recovery: {
code: 'projection-apply-failed',
rosterPath: options.rosterPath,
action: 'regenerate-projections-from-roster',
},
};
}
throw error;
} finally {
await release();
}
}
function planCreate(
roster: FleetRosterV2,
request: FleetAgentMutationRequest,
): FleetAgentMutationPlan {
if (!request.agent)
throw new FleetAgentMutationError('invalid-request', 'Create requires an agent.');
const existing = roster.agents.find(
(agent: FleetRosterV2Agent): boolean => agent.name === request.agent?.name,
);
const created = toRosterAgent(request.agent, request.persistedStart === true);
if (existing) {
if (sameAgent(existing, created)) return unchangedPlan(roster, request.operation, existing);
throw new FleetAgentMutationError('agent-conflict', `Agent "${created.name}" already exists.`);
}
return changedPlan(roster, request.operation, [...roster.agents, created], created);
}
function planUpdate(
roster: FleetRosterV2,
request: FleetAgentMutationRequest,
): FleetAgentMutationPlan {
if (!request.name || !request.agent) {
throw new FleetAgentMutationError('invalid-request', 'Update requires name and agent.');
}
const existing = roster.agents.find(
(agent: FleetRosterV2Agent): boolean => agent.name === request.name,
);
if (!existing)
throw new FleetAgentMutationError(
'agent-not-found',
`Agent "${request.name}" is not in roster.`,
);
if (request.agent.name !== request.name) {
throw new FleetAgentMutationError(
'invalid-request',
'Agent names are immutable during update.',
);
}
const updated = {
...toRosterAgent(request.agent, existing.lifecycle.desiredState === 'running'),
lifecycle: existing.lifecycle,
};
if (sameAgent(existing, updated)) return unchangedPlan(roster, request.operation, existing);
return changedPlan(
roster,
request.operation,
roster.agents.map(
(agent: FleetRosterV2Agent): FleetRosterV2Agent =>
agent.name === request.name ? updated : agent,
),
updated,
);
}
function planDelete(
roster: FleetRosterV2,
request: FleetAgentMutationRequest,
): FleetAgentMutationPlan {
if (!request.name) throw new FleetAgentMutationError('invalid-request', 'Delete requires name.');
const existing = roster.agents.find(
(agent: FleetRosterV2Agent): boolean => agent.name === request.name,
);
if (!existing) return unchangedPlan(roster, request.operation);
return changedPlan(
roster,
request.operation,
roster.agents.filter((agent: FleetRosterV2Agent): boolean => agent.name !== request.name),
existing,
);
}
function toRosterAgent(
input: FleetAgentMutationAgent,
persistedStart: boolean,
): FleetRosterV2Agent {
const lifecycle: FleetRosterV2Lifecycle = {
enabled: true,
desiredState: persistedStart ? 'running' : 'stopped',
};
return { ...input, lifecycle };
}
function changedPlan(
roster: FleetRosterV2,
operation: FleetAgentMutationOperation,
agents: readonly FleetRosterV2Agent[],
agent?: FleetRosterV2Agent,
): FleetAgentMutationPlan {
const proposed = { ...roster, generation: roster.generation + 1, agents };
const normalized = parseRosterV2(renderRosterV2Yaml(proposed), 'yaml');
return {
operation,
currentGeneration: roster.generation,
nextGeneration: normalized.generation,
changed: true,
roster: normalized,
...(agent ? { agent } : {}),
};
}
function unchangedPlan(
roster: FleetRosterV2,
operation: FleetAgentMutationOperation,
agent?: FleetRosterV2Agent,
): FleetAgentMutationPlan {
return {
operation,
currentGeneration: roster.generation,
nextGeneration: roster.generation,
changed: false,
roster,
...(agent ? { agent } : {}),
};
}
function sameAgent(left: FleetRosterV2Agent, right: FleetRosterV2Agent): boolean {
return (
left.name === right.name &&
left.alias === right.alias &&
left.className === right.className &&
left.runtime === right.runtime &&
left.provider === right.provider &&
left.model === right.model &&
left.reasoning === right.reasoning &&
left.toolPolicy === right.toolPolicy &&
left.workingDirectory === right.workingDirectory &&
left.persistentPersona === right.persistentPersona &&
left.resetBetweenTasks === right.resetBetweenTasks &&
left.launch.yolo === right.launch.yolo &&
left.lifecycle.enabled === right.lifecycle.enabled &&
left.lifecycle.desiredState === right.lifecycle.desiredState
);
}
async function prepareProjections(
roster: FleetRosterV2,
mosaicHome: string,
agentEnvDir: string,
): Promise<readonly PreparedAgentEnvironmentProjection[]> {
const projections: PreparedAgentEnvironmentProjection[] = [];
for (const agent of roster.agents) {
projections.push(
await prepareAgentEnvironmentProjection({
mosaicHome,
agentEnvDir,
agentName: agent.name,
generated: generatedValues(roster, agent),
}),
);
}
return projections;
}
function generatedValues(
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,
};
}
async function removeGeneratedProjection(agentEnvDir: string, agentName: string): Promise<void> {
try {
await unlink(join(agentEnvDir, `${agentName}.env.generated`));
} catch (error: unknown) {
if (isMissingFile(error)) return;
throw error;
}
}
function isMissingFile(error: unknown): boolean {
return typeof error === 'object' && error !== null && 'code' in error && error.code === 'ENOENT';
}
async function acquireMutationLock(lockPath: string): Promise<() => Promise<void>> {
try {
const handle = await open(lockPath, 'wx', 0o600);
await handle.close();
} catch {
throw new FleetAgentMutationError(
'concurrent-mutation',
`Another roster mutation is in progress for ${dirname(lockPath)}.`,
);
}
return async (): Promise<void> => {
await unlink(lockPath).catch((): void => {});
};
}

View File

@@ -17,6 +17,12 @@ export interface AgentEnvironmentProjectionOptions {
readonly generated: Readonly<Record<string, string>>;
}
export interface AgentGeneratedProjectionDeletionOptions {
readonly mosaicHome: string;
readonly agentEnvDir: string;
readonly agentName: string;
}
export interface AgentEnvironmentProjectionResult {
readonly generatedPath: string;
readonly localPath: string;
@@ -185,6 +191,23 @@ export async function prepareAgentEnvironmentProjection(
};
}
/**
* Validates only the exact generated projection eligible for a roster delete.
* Local overrides, legacy input, and quarantine records are operator-retained and
* intentionally excluded from delete validation and cleanup.
*/
export async function prepareAgentGeneratedProjectionDeletion(
options: AgentGeneratedProjectionDeletionOptions,
): Promise<string> {
if (!AGENT_NAME.test(options.agentName)) {
throw new AgentEnvBoundaryError('unsafe-agent-name', 'MOSAIC_AGENT_NAME', options.agentName);
}
await validatePrivateProjectionDirectory(options.mosaicHome, options.agentEnvDir);
const generatedPath = join(options.agentEnvDir, `${options.agentName}.env.generated`);
await assertPrivateRegularFileIfPresent(generatedPath);
return generatedPath;
}
/** Applies a previously prepared deterministic projection. */
export async function applyPreparedAgentEnvironmentProjection(
prepared: PreparedAgentEnvironmentProjection,