245 lines
14 KiB
JavaScript
245 lines
14 KiB
JavaScript
import { randomUUID } from 'node:crypto';
|
|
import { spawnSync, execFile } from 'node:child_process';
|
|
import { existsSync, lstatSync, readdirSync, realpathSync } from 'node:fs';
|
|
import { isAbsolute, join, relative, resolve } from 'node:path';
|
|
import { promisify } from 'node:util';
|
|
import { initializeControl, assertProcessing } from './state.mjs';
|
|
import { registerRuntime, updateRuntime, runtime, rootTag } from './runtimes.mjs';
|
|
const execute = promisify(execFile);
|
|
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
|
|
const WORKER_KEYS = new Set([
|
|
'kind', 'taskFile', 'missionFile', 'workspaceDir', 'sessionDir',
|
|
'sessionForkFile', 'agentSoulFile', 'skillDirs',
|
|
]);
|
|
const WORKER_ENV = Object.freeze({
|
|
MOSAIC_SYSTEM_PROMPT_FILE: '/var/lib/mosaic/private/system-prompt.md',
|
|
MOSAIC_TASK_FILE: '',
|
|
MOSAIC_USER_DIR: '',
|
|
MOSAIC_MISSION_FILE: '',
|
|
MOSAIC_AGENT_SOUL_FILE: '',
|
|
MOSAIC_WORKSPACE: '',
|
|
MOSAIC_SESSION_DIR: '',
|
|
MOSAIC_SESSION_FORK: '',
|
|
MOSAIC_SKILLS: '',
|
|
});
|
|
|
|
function workerError(message) {
|
|
const error = new Error(`worker launch: ${message}`);
|
|
error.code = 'processing-worker-launch-invalid';
|
|
return error;
|
|
}
|
|
|
|
function components(root, source, what, type) {
|
|
if (typeof source !== 'string' || !isAbsolute(source)) throw workerError(`${what} must be an absolute path`);
|
|
let stat;
|
|
try { stat = lstatSync(source); }
|
|
catch { throw workerError(`${what} is missing: ${source}`); }
|
|
if (stat.isSymbolicLink() || (type === 'file' ? !stat.isFile() : !stat.isDirectory())) {
|
|
throw workerError(`${what} must be a non-symbolic-link ${type}: ${source}`);
|
|
}
|
|
const canonical = realpathSync(source);
|
|
const rel = relative(root, canonical);
|
|
if (rel === '' || rel === '..' || rel.startsWith('../') || isAbsolute(rel)) {
|
|
throw workerError(`${what} must be below the configured data root: ${source}`);
|
|
}
|
|
return { source: canonical, parts: rel.split('/'), stat };
|
|
}
|
|
|
|
function exactParts(found, expected, what) {
|
|
if (found.parts.length !== expected.length || expected.some((part, i) => part !== null && found.parts[i] !== part)) {
|
|
throw workerError(`${what} is outside its allowed data-root path: ${found.source}`);
|
|
}
|
|
return found;
|
|
}
|
|
|
|
function mount(source, target, readOnly) {
|
|
if (source.includes(':')) throw workerError(`mount source contains an unsupported colon: ${source}`);
|
|
return Object.freeze({ source, target, readOnly });
|
|
}
|
|
|
|
// Pure legacy-worker launch contract. Every host source occupies a typed slot
|
|
// with one fixed container target; arbitrary mounts and the data-root root are
|
|
// not accepted. The returned prompt path has no host source and is private to
|
|
// this one disposable container filesystem.
|
|
export function workerLaunch(dataRoot, input) {
|
|
if (!input || typeof input !== 'object' || Array.isArray(input)) throw workerError('worker launch input is required');
|
|
for (const key of Object.keys(input)) if (!WORKER_KEYS.has(key)) throw workerError(`unsupported worker launch key: ${key}`);
|
|
if (!['task', 'interactive', 'probe'].includes(input.kind)) throw workerError(`unsupported worker launch kind: ${JSON.stringify(input.kind)}`);
|
|
const root = realpathSync(resolve(dataRoot));
|
|
const environment = { ...WORKER_ENV };
|
|
const mounts = [];
|
|
const add = (source, target, readOnly = true) => mounts.push(mount(source, target, readOnly));
|
|
|
|
// Keep the host-owned directory as the read-only view: state publication
|
|
// renames a fresh inode over processing.json, which a file bind would pin.
|
|
// This package produces only processing.json, transient .state-* files, and
|
|
// runtimes/, stops/, run-stops/ metadata here; sockets live outside control/.
|
|
const control = exactParts(components(root, join(root, 'control'), 'control directory', 'directory'),
|
|
['control'], 'control directory');
|
|
add(control.source, '/var/lib/mosaic/control');
|
|
|
|
const userDir = join(root, 'user');
|
|
if (existsSync(userDir)) {
|
|
const user = exactParts(components(root, userDir, 'user context directory', 'directory'), ['user'], 'user context directory');
|
|
for (const entry of readdirSync(user.source, { withFileTypes: true }).filter(e => e.name.endsWith('.md')).sort((a, b) => a.name.localeCompare(b.name))) {
|
|
const file = exactParts(components(root, join(user.source, entry.name), 'user context file', 'file'), ['user', entry.name], 'user context file');
|
|
add(file.source, `/var/lib/mosaic/input/user/${entry.name}`);
|
|
}
|
|
environment.MOSAIC_USER_DIR = '/var/lib/mosaic/input/user';
|
|
}
|
|
|
|
if (input.kind === 'probe') {
|
|
for (const key of WORKER_KEYS) if (key !== 'kind' && input[key] !== undefined) throw workerError(`${key} is unsupported for probe launches`);
|
|
return Object.freeze({ mounts: Object.freeze(mounts), environment: Object.freeze(environment) });
|
|
}
|
|
|
|
if (input.kind === 'task') {
|
|
if (!input.taskFile) throw workerError('taskFile is required for task launches');
|
|
if (input.agentSoulFile !== undefined || input.skillDirs !== undefined) throw workerError('role inputs are unsupported for task launches');
|
|
const task = exactParts(components(root, input.taskFile, 'taskFile', 'file'), ['runs', null, 'task.json'], 'taskFile');
|
|
add(task.source, '/var/lib/mosaic/input/task.json');
|
|
environment.MOSAIC_TASK_FILE = '/var/lib/mosaic/input/task.json';
|
|
if (input.missionFile) {
|
|
const mission = exactParts(components(root, input.missionFile, 'missionFile', 'file'), ['runs', task.parts[1], 'mission.json'], 'missionFile');
|
|
add(mission.source, '/var/lib/mosaic/input/mission.json');
|
|
environment.MOSAIC_MISSION_FILE = '/var/lib/mosaic/input/mission.json';
|
|
}
|
|
if (input.workspaceDir) {
|
|
const workspace = components(root, input.workspaceDir, 'workspaceDir', 'directory');
|
|
const persistent = workspace.parts.length === 2 && workspace.parts[0] === 'workspaces';
|
|
const perRun = workspace.parts.length === 3 && workspace.parts[0] === 'runs' && workspace.parts[1] === task.parts[1] && workspace.parts[2] === 'workspace';
|
|
if (!persistent && !perRun) throw workerError(`workspaceDir is outside its allowed data-root path: ${workspace.source}`);
|
|
add(workspace.source, '/workspace', false);
|
|
environment.MOSAIC_WORKSPACE = '/workspace';
|
|
}
|
|
if (input.sessionDir) {
|
|
const session = exactParts(components(root, input.sessionDir, 'sessionDir', 'directory'), ['sessions', null], 'sessionDir');
|
|
add(session.source, '/var/lib/mosaic/session', false);
|
|
environment.MOSAIC_SESSION_DIR = '/var/lib/mosaic/session';
|
|
}
|
|
if (input.sessionForkFile) {
|
|
if (!input.sessionDir) throw workerError('sessionForkFile requires sessionDir');
|
|
const fork = exactParts(components(root, input.sessionForkFile, 'sessionForkFile', 'file'),
|
|
['runs', task.parts[1], 'session-fork.jsonl'], 'sessionForkFile');
|
|
add(fork.source, '/var/lib/mosaic/input/session-fork.jsonl');
|
|
environment.MOSAIC_SESSION_FORK = '/var/lib/mosaic/input/session-fork.jsonl';
|
|
}
|
|
} else {
|
|
if (input.taskFile !== undefined || input.sessionForkFile !== undefined) throw workerError('task inputs are unsupported for interactive launches');
|
|
if (!input.workspaceDir) throw workerError('workspaceDir is required for interactive launches');
|
|
if (!input.sessionDir) throw workerError('sessionDir is required for interactive launches');
|
|
const workspace = exactParts(components(root, input.workspaceDir, 'workspaceDir', 'directory'), ['workspaces', null], 'workspaceDir');
|
|
const session = exactParts(components(root, input.sessionDir, 'sessionDir', 'directory'), ['sessions', null], 'sessionDir');
|
|
add(workspace.source, '/workspace', false);
|
|
add(session.source, '/var/lib/mosaic/session', false);
|
|
environment.MOSAIC_WORKSPACE = '/workspace';
|
|
environment.MOSAIC_SESSION_DIR = '/var/lib/mosaic/session';
|
|
if (input.missionFile) {
|
|
const mission = exactParts(components(root, input.missionFile, 'missionFile', 'file'), ['agent-missions', null], 'missionFile');
|
|
if (!mission.parts[1].endsWith('.json')) throw workerError(`missionFile must be an agent mission JSON file: ${mission.source}`);
|
|
add(mission.source, '/var/lib/mosaic/input/mission.json');
|
|
environment.MOSAIC_MISSION_FILE = '/var/lib/mosaic/input/mission.json';
|
|
}
|
|
if (input.agentSoulFile) {
|
|
const soul = exactParts(components(root, input.agentSoulFile, 'agentSoulFile', 'file'), ['agents', null, 'SOUL.md'], 'agentSoulFile');
|
|
add(soul.source, '/var/lib/mosaic/input/agent-soul.md');
|
|
environment.MOSAIC_AGENT_SOUL_FILE = '/var/lib/mosaic/input/agent-soul.md';
|
|
}
|
|
const skills = input.skillDirs ?? [];
|
|
if (!Array.isArray(skills)) throw workerError('skillDirs must be an array');
|
|
const targets = [];
|
|
for (const skillDir of skills) {
|
|
const skill = exactParts(components(root, skillDir, 'skillDirs entry', 'directory'), ['skills-enabled', null], 'skillDirs entry');
|
|
const target = `/var/lib/mosaic/input/skills/${skill.parts[1]}`;
|
|
if (targets.includes(target)) throw workerError(`duplicate skill directory: ${skill.source}`);
|
|
targets.push(target);
|
|
add(skill.source, target);
|
|
}
|
|
environment.MOSAIC_SKILLS = targets.join(',');
|
|
}
|
|
return Object.freeze({ mounts: Object.freeze(mounts), environment: Object.freeze(environment) });
|
|
}
|
|
export function prepareContainer(dataRoot, record) {
|
|
initializeControl(dataRoot);
|
|
const state = assertProcessing(dataRoot);
|
|
const containerName = `mosaic-${rootTag(dataRoot).slice(0, 12)}-${randomUUID()}`;
|
|
const managed = registerRuntime(dataRoot, { ...record, containerName });
|
|
assertProcessing(dataRoot, state.generation);
|
|
return managed;
|
|
}
|
|
export function containerArgs(record, { tty = false, args = [], worker } = {}) {
|
|
if (!worker || !Array.isArray(worker.mounts) || !worker.environment) throw workerError('validated worker launch is required');
|
|
const volumes = worker.mounts.flatMap(({ source, target, readOnly }) => ['--volume', `${source}:${target}:${readOnly ? 'ro' : 'rw'}`]);
|
|
const environment = Object.entries(worker.environment).flatMap(([key, value]) => ['-e', `${key}=${value}`]);
|
|
return ['compose', 'run', '--rm', ...(!tty ? ['-T'] : []), '--name', record.containerName,
|
|
'--label', `mosaic.root=${record.root}`, '--label', `mosaic.runtime=${record.id}`,
|
|
'-e', `MOSAIC_PROCESSING_GENERATION=${record.generation}`, ...environment, ...volumes, 'mosaic-agent', ...args];
|
|
}
|
|
export function runContainer(dataRoot, record, { tty = false, args = [], worker, ...options } = {}) {
|
|
const launch = workerLaunch(dataRoot, worker);
|
|
const managed = prepareContainer(dataRoot, record);
|
|
assertProcessing(dataRoot, managed.generation);
|
|
// Keep this synchronous. The independent broker sweeps the prepublished name while this waits.
|
|
const proc = spawnSync('docker', containerArgs(managed, { tty, args, worker: launch }), options);
|
|
const current = runtime(dataRoot, managed.id);
|
|
// An interrupted Docker client is not proof its container exited. Keep that receipt for inspection.
|
|
if (!proc.error && proc.status === 0) updateRuntime(dataRoot, managed.id, { phase: 'ended' });
|
|
return { proc, managed, haltGeneration: current.haltGeneration ?? null };
|
|
}
|
|
export function createContainerStopper({ dataRoot = null, docker = async args => {
|
|
const r = await execute('docker', args, { encoding: 'utf8', timeout: 2000, maxBuffer: 131072 });
|
|
return r.stdout;
|
|
} } = {}) {
|
|
return async (record, { graceMs = 5000, killWaitMs = 2000 } = {}) => {
|
|
const signals = [];
|
|
let id = record.containerId ?? null, seen = false;
|
|
const start = Date.now(), termEnd = start + graceMs, end = termEnd + killWaitMs;
|
|
async function inspect(target) {
|
|
const doc = JSON.parse(await docker(['inspect', '--format', '{{json .}}', target]));
|
|
if (!doc || !/^[a-f0-9]{64}$/.test(doc.Id) || doc.Name !== '/' + record.containerName ||
|
|
doc.Config?.Labels?.['mosaic.root'] !== record.root || doc.Config?.Labels?.['mosaic.runtime'] !== record.id ||
|
|
typeof doc.State?.Running !== 'boolean' || (id && doc.Id !== id)) {
|
|
const error = new Error('container identity mismatch'); error.code = 'identity-mismatch'; throw error;
|
|
}
|
|
if (!id) {
|
|
id = doc.Id;
|
|
if (dataRoot) updateRuntime(dataRoot, record.id, { containerId: id });
|
|
}
|
|
seen = true;
|
|
return doc;
|
|
}
|
|
while (Date.now() <= end) {
|
|
try {
|
|
const doc = await inspect(id ?? record.containerName);
|
|
if (!doc.State.Running) return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
|
|
const next = !signals.length ? 'TERM' : Date.now() >= termEnd && !signals.includes('SIGKILL') ? 'KILL' : null;
|
|
if (next) {
|
|
// Revalidate immediately before each signal, targeting the immutable inspected id.
|
|
const checked = await inspect(id);
|
|
if (!checked.State.Running) continue;
|
|
await docker(['kill', '--signal', next, id]);
|
|
signals.push('SIG' + next);
|
|
}
|
|
} catch (e) {
|
|
// A missing name can still be a queued create. Only a container observed before a signal
|
|
// and then removed by --rm is known to have finished. Never signal on an inspect failure.
|
|
if (e.code === 'identity-mismatch') return { outcome: 'unknown', signals };
|
|
if (id && seen && signals.length) {
|
|
try {
|
|
const listed = await docker(['ps', '--all', '--no-trunc', '--filter', `id=${id}`, '--format', '{{.ID}}']);
|
|
if (listed.trim() === '') return { outcome: signals.includes('SIGKILL') ? 'killed' : 'terminated', signals };
|
|
} catch { /* a failed Docker request proves nothing about the workload */ }
|
|
}
|
|
if (dataRoot) {
|
|
try {
|
|
if (runtime(dataRoot, record.id).phase === 'ended')
|
|
return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
|
|
} catch { return { outcome: 'unknown', signals }; }
|
|
}
|
|
}
|
|
await pause(Math.min(20, Math.max(1, end - Date.now())));
|
|
}
|
|
return { outcome: 'unknown', signals };
|
|
};
|
|
}
|