254 lines
15 KiB
JavaScript
254 lines
15 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs';
|
|
import { createRequire } from 'node:module';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { assertProcessing, initializeControl, readControl, transition } from '../src/state.mjs';
|
|
const api = await import('../src/containers.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
|
|
const { parse: parseYaml } = createRequire(import.meta.url)('yaml');
|
|
function fixture(t) { const root = mkdtempSync(join(tmpdir(), 'processing-container-')); initializeControl(root); t.after(() => rmSync(root, { recursive: true, force: true })); return root; }
|
|
const actor = { business: 'demo', human: 'jason', halted: true };
|
|
|
|
test('static Compose service has no data-root or control-directory mount', () => {
|
|
const compose = parseYaml(readFileSync(new URL('../../../compose.yaml', import.meta.url), 'utf8'));
|
|
const volumes = compose.services['mosaic-agent'].volumes;
|
|
const targets = volumes.map(volume => {
|
|
if (typeof volume !== 'string') return volume.target;
|
|
const searchFrom = volume.startsWith('${') ? volume.indexOf('}') + 1 : 0;
|
|
const sourceEnd = volume.indexOf(':', searchFrom);
|
|
return volume.slice(sourceEnd + 1).split(':')[0];
|
|
});
|
|
assert.deepEqual(targets, ['/home/node/.pi/agent/auth.json']);
|
|
});
|
|
|
|
test('source launch view maps the control directory so actual atomic state replacements stay path-visible', t => {
|
|
const root = fixture(t), initial = readControl(root);
|
|
const launch = api.workerLaunch(root, { kind: 'probe' });
|
|
const control = launch.mounts.find(mount => mount.target === '/var/lib/mosaic/control');
|
|
assert.deepEqual(control, {
|
|
source: join(root, 'control'), target: '/var/lib/mosaic/control', readOnly: true,
|
|
});
|
|
const file = join(control.source, 'processing.json');
|
|
const initialInode = statSync(file).ino;
|
|
|
|
const halted = transition(root, { business: 'demo', human: 'jason', halted: true });
|
|
assert.notEqual(statSync(file).ino, initialInode, 'the state writer replaces processing.json with a new inode');
|
|
const { changed: haltChanged, ...publishedHalt } = halted;
|
|
assert.equal(haltChanged, true);
|
|
assert.deepEqual(JSON.parse(readFileSync(file, 'utf8')), publishedHalt);
|
|
assert.throws(() => assertProcessing(root, initial.generation), error => error.code === 'processing-halted');
|
|
|
|
const haltedInode = statSync(file).ino;
|
|
const cleared = transition(root, { business: 'demo', human: 'jason', halted: false, generation: halted.generation });
|
|
assert.notEqual(statSync(file).ino, haltedInode, 'clearing also publishes by atomic replacement');
|
|
assert.equal(readControl(root).generation, cleared.generation);
|
|
assert.throws(() => assertProcessing(root, initial.generation), error => error.code === 'processing-generation-changed');
|
|
});
|
|
|
|
function workerTree(t) {
|
|
const root = fixture(t);
|
|
const files = {
|
|
task: join(root, 'runs', 'r-own', 'task.json'),
|
|
mission: join(root, 'runs', 'r-own', 'mission.json'),
|
|
evidence: join(root, 'runs', 'r-own', 'result.json'),
|
|
siblingEvidence: join(root, 'runs', 'r-sibling', 'result.json'),
|
|
workspace: join(root, 'workspaces', 'own'),
|
|
siblingWorkspace: join(root, 'workspaces', 'sibling'),
|
|
session: join(root, 'sessions', 'own'),
|
|
forkSource: join(root, 'sessions', 'base', 'turn.jsonl'),
|
|
forkSnapshot: join(root, 'runs', 'r-own', 'session-fork.jsonl'),
|
|
foreignForkSnapshot: join(root, 'runs', 'r-sibling', 'session-fork.jsonl'),
|
|
siblingSession: join(root, 'sessions', 'sibling'),
|
|
missionInteractive: join(root, 'agent-missions', 'alice.json'),
|
|
soul: join(root, 'agents', 'alice', 'SOUL.md'),
|
|
skill: join(root, 'skills-enabled', 'review'),
|
|
socket: join(root, 'bus', 'broker.sock'),
|
|
unrelatedUserFile: join(root, 'user', 'private.bin'),
|
|
};
|
|
for (const dir of [
|
|
join(root, 'runs', 'r-own'), join(root, 'runs', 'r-sibling'), files.workspace,
|
|
files.siblingWorkspace, files.session, join(root, 'sessions', 'base'),
|
|
files.siblingSession, join(root, 'user'), join(root, 'agent-missions'),
|
|
join(root, 'agents', 'alice'), files.skill, join(root, 'bus'),
|
|
]) mkdirSync(dir, { recursive: true });
|
|
for (const file of [files.task, files.mission, files.evidence, files.siblingEvidence,
|
|
files.forkSource, files.forkSnapshot, files.foreignForkSnapshot,
|
|
files.missionInteractive, files.soul, files.socket]) writeFileSync(file, '{}\n');
|
|
writeFileSync(join(root, 'user', 'USER.md'), '# User\n');
|
|
writeFileSync(files.unrelatedUserFile, 'not prompt context\n');
|
|
writeFileSync(join(files.skill, 'SKILL.md'), '# Review\n');
|
|
return { root, files };
|
|
}
|
|
|
|
test('task worker launch mounts only selected inputs RO and selected workspace/session RW', t => {
|
|
assert.equal(typeof api.workerLaunch, 'function');
|
|
const { root, files } = workerTree(t);
|
|
const launch = api.workerLaunch(root, {
|
|
kind: 'task', taskFile: files.task, missionFile: files.mission,
|
|
workspaceDir: files.workspace, sessionDir: files.session, sessionForkFile: files.forkSnapshot,
|
|
});
|
|
const byTarget = Object.fromEntries(launch.mounts.map(m => [m.target, m]));
|
|
assert.deepEqual(Object.keys(byTarget).sort(), [
|
|
'/var/lib/mosaic/control', '/var/lib/mosaic/input/mission.json',
|
|
'/var/lib/mosaic/input/session-fork.jsonl', '/var/lib/mosaic/input/task.json',
|
|
'/var/lib/mosaic/input/user/USER.md', '/var/lib/mosaic/session', '/workspace',
|
|
].sort());
|
|
for (const target of ['/var/lib/mosaic/control', '/var/lib/mosaic/input/mission.json',
|
|
'/var/lib/mosaic/input/session-fork.jsonl', '/var/lib/mosaic/input/task.json', '/var/lib/mosaic/input/user/USER.md']) {
|
|
assert.equal(byTarget[target].readOnly, true, target);
|
|
}
|
|
assert.equal(byTarget['/workspace'].readOnly, false);
|
|
assert.equal(byTarget['/var/lib/mosaic/session'].readOnly, false);
|
|
assert.equal(launch.environment.MOSAIC_TASK_FILE, '/var/lib/mosaic/input/task.json');
|
|
assert.equal(launch.environment.MOSAIC_MISSION_FILE, '/var/lib/mosaic/input/mission.json');
|
|
assert.equal(launch.environment.MOSAIC_WORKSPACE, '/workspace');
|
|
assert.equal(launch.environment.MOSAIC_SESSION_DIR, '/var/lib/mosaic/session');
|
|
assert.equal(launch.environment.MOSAIC_SESSION_FORK, '/var/lib/mosaic/input/session-fork.jsonl');
|
|
assert.equal(byTarget['/var/lib/mosaic/input/session-fork.jsonl'].source, files.forkSnapshot);
|
|
assert.equal(launch.environment.MOSAIC_SYSTEM_PROMPT_FILE, '/var/lib/mosaic/private/system-prompt.md');
|
|
const sources = launch.mounts.map(m => m.source);
|
|
assert.equal(sources.includes(root), false);
|
|
for (const hidden of [files.evidence, files.siblingEvidence, files.siblingWorkspace, files.siblingSession,
|
|
files.forkSource, files.foreignForkSnapshot, files.socket]) {
|
|
assert.equal(sources.includes(hidden), false, hidden);
|
|
}
|
|
assert.equal(sources.includes(files.unrelatedUserFile), false);
|
|
});
|
|
|
|
test('task fork slot refuses a shared session file or another run snapshot', t => {
|
|
const { root, files } = workerTree(t);
|
|
const input = { kind: 'task', taskFile: files.task, sessionDir: files.session };
|
|
assert.throws(() => api.workerLaunch(root, { ...input, sessionForkFile: files.forkSource }), /sessionForkFile/);
|
|
assert.throws(() => api.workerLaunch(root, { ...input, sessionForkFile: files.foreignForkSnapshot }), /sessionForkFile/);
|
|
});
|
|
|
|
test('probe refuses symlinked or non-file user markdown and colon mount sources', t => {
|
|
for (const [name, arrange, expected] of [
|
|
['symlinked user directory', root => {
|
|
mkdirSync(join(root, 'actual-user')); symlinkSync(join(root, 'actual-user'), join(root, 'user'));
|
|
}, /user context directory.*non-symbolic-link/],
|
|
['symlinked markdown', root => {
|
|
mkdirSync(join(root, 'user')); writeFileSync(join(root, 'context.md'), '# context\n');
|
|
symlinkSync(join(root, 'context.md'), join(root, 'user', 'USER.md'));
|
|
}, /user context file.*non-symbolic-link/],
|
|
['markdown-named directory', root => {
|
|
mkdirSync(join(root, 'user', 'USER.md'), { recursive: true });
|
|
}, /user context file.*non-symbolic-link/],
|
|
['colon source', root => {
|
|
mkdirSync(join(root, 'user')); writeFileSync(join(root, 'user', 'USER:.md'), '# context\n');
|
|
}, /mount source contains an unsupported colon/],
|
|
]) {
|
|
const root = fixture(t);
|
|
arrange(root);
|
|
assert.throws(() => api.workerLaunch(root, { kind: 'probe' }), expected, name);
|
|
}
|
|
});
|
|
|
|
test('interactive worker launch keeps role inputs RO and only its workspace/session RW', t => {
|
|
assert.equal(typeof api.workerLaunch, 'function');
|
|
const { root, files } = workerTree(t);
|
|
const launch = api.workerLaunch(root, {
|
|
kind: 'interactive', missionFile: files.missionInteractive, agentSoulFile: files.soul,
|
|
skillDirs: [files.skill], workspaceDir: files.workspace, sessionDir: files.session,
|
|
});
|
|
const byTarget = Object.fromEntries(launch.mounts.map(m => [m.target, m]));
|
|
assert.equal(byTarget['/var/lib/mosaic/input/mission.json'].readOnly, true);
|
|
assert.equal(byTarget['/var/lib/mosaic/input/agent-soul.md'].readOnly, true);
|
|
assert.equal(byTarget['/var/lib/mosaic/input/skills/review'].readOnly, true);
|
|
assert.equal(byTarget['/var/lib/mosaic/input/user/USER.md'].readOnly, true);
|
|
assert.equal(byTarget['/workspace'].readOnly, false);
|
|
assert.equal(byTarget['/var/lib/mosaic/session'].readOnly, false);
|
|
assert.equal(launch.environment.MOSAIC_AGENT_SOUL_FILE, '/var/lib/mosaic/input/agent-soul.md');
|
|
assert.equal(launch.environment.MOSAIC_SKILLS, '/var/lib/mosaic/input/skills/review');
|
|
assert.equal(launch.environment.MOSAIC_TASK_FILE, '');
|
|
assert.equal(launch.environment.MOSAIC_SESSION_FORK, '');
|
|
});
|
|
|
|
test('unsupported or incomplete worker launch refuses before runtime registration', t => {
|
|
assert.equal(typeof api.workerLaunch, 'function');
|
|
const { root, files } = workerTree(t);
|
|
assert.throws(() => api.workerLaunch(root, { kind: 'task' }), /taskFile/);
|
|
assert.throws(() => api.workerLaunch(root, { kind: 'probe', brokerSocket: files.socket }), /unsupported worker launch key/);
|
|
assert.throws(() => api.workerLaunch(root, { kind: 'interactive', workspaceDir: files.workspace }), /sessionDir/);
|
|
const runtimes = join(root, 'control', 'runtimes');
|
|
assert.throws(() => api.runContainer(root, { kind: 'task', business: 'instance', run: 'missing' }, {}), /worker launch/);
|
|
assert.equal(existsSync(runtimes) ? readdirSync(runtimes).length : 0, 0);
|
|
});
|
|
test('container admission publishes its generated name and root/runtime labels before dispatch', t => {
|
|
assert.equal(typeof api.prepareContainer, 'function');
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'run-1' });
|
|
const args = api.containerArgs(r, { args: ['request'], worker: api.workerLaunch(root, { kind: 'probe' }) });
|
|
assert.match(r.containerName, /^mosaic-[a-f0-9]{12}-[a-f0-9-]{36}$/);
|
|
assert.ok(args.includes(r.containerName)); assert.ok(args.includes(`mosaic.root=${r.root}`));
|
|
assert.ok(args.includes(`mosaic.runtime=${r.id}`)); assert.ok(args.includes(`MOSAIC_PROCESSING_GENERATION=${r.generation}`));
|
|
assert.deepEqual(args.slice(-2), ['mosaic-agent', 'request']);
|
|
transition(root, actor); assert.throws(() => api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'new' }), e => e.code === 'processing-halted');
|
|
});
|
|
function inspected(r, running = true, id = 'a'.repeat(64)) {
|
|
return { Id: id, Name: '/' + r.containerName, Config: { Labels: { 'mosaic.root': r.root, 'mosaic.runtime': r.id } }, State: { Running: running } };
|
|
}
|
|
test('container sweep sends TERM then KILL only after verifying exact id and both labels', async t => {
|
|
assert.equal(typeof api.createContainerStopper, 'function');
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
|
|
const calls = []; let running = true;
|
|
const stop = api.createContainerStopper({ docker: async args => {
|
|
calls.push(args);
|
|
if (args[0] === 'inspect') return JSON.stringify(inspected(r, running));
|
|
if (args.includes('KILL')) running = false;
|
|
return '';
|
|
} });
|
|
const result = await stop(r, { graceMs: 30, killWaitMs: 100 });
|
|
assert.equal(result.outcome, 'killed'); assert.deepEqual(result.signals, ['SIGTERM', 'SIGKILL']);
|
|
assert.deepEqual(calls.filter(c => c[0] === 'kill'), [['kill', '--signal', 'TERM', 'a'.repeat(64)], ['kill', '--signal', 'KILL', 'a'.repeat(64)]]);
|
|
});
|
|
test('foreign labels and a replaced container id stay unknown and are never signalled', async t => {
|
|
assert.equal(typeof api.createContainerStopper, 'function');
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
|
|
const calls = [];
|
|
const stop = api.createContainerStopper({ docker: async args => {
|
|
calls.push(args); const doc = inspected(r); doc.Config.Labels['mosaic.root'] = 'b'.repeat(64); return JSON.stringify(doc);
|
|
} });
|
|
assert.equal((await stop(r, { graceMs: 1, killWaitMs: 1 })).outcome, 'unknown');
|
|
assert.equal(calls.filter(c => c[0] === 'kill').length, 0);
|
|
let reads = 0; calls.length = 0;
|
|
const replacement = api.createContainerStopper({ docker: async args => {
|
|
calls.push(args); if (args[0] !== 'inspect') return '';
|
|
return JSON.stringify(inspected(r, true, (++reads > 1 ? 'b' : 'a').repeat(64)));
|
|
} });
|
|
assert.equal((await replacement(r, { graceMs: 1, killWaitMs: 1 })).outcome, 'unknown');
|
|
assert.equal(calls.filter(c => c[0] === 'kill').length, 0);
|
|
});
|
|
test('an unverified missing container remains unknown rather than claiming a pending start ended', async t => {
|
|
assert.equal(typeof api.createContainerStopper, 'function');
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'container', business: 'instance', run: 'one' });
|
|
const stop = api.createContainerStopper({ docker: async () => { throw new Error('no container'); } });
|
|
assert.equal((await stop(r, { graceMs: 5, killWaitMs: 5 })).outcome, 'unknown');
|
|
});
|
|
|
|
test('a removed --rm container is settled only by its trusted owner completion receipt', async t => {
|
|
const { updateRuntime } = await import('../src/runtimes.mjs');
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
|
|
let removed = false;
|
|
const stop = api.createContainerStopper({ dataRoot: root, docker: async args => {
|
|
if (args[0] === 'kill') { removed = true; updateRuntime(root, r.id, { phase: 'ended' }); return ''; }
|
|
if (removed) throw Error('removed by --rm');
|
|
return JSON.stringify(inspected(r));
|
|
} });
|
|
const result = await stop(r, { graceMs: 30, killWaitMs: 100 });
|
|
assert.equal(result.outcome, 'terminated'); assert.deepEqual(result.signals, ['SIGTERM']);
|
|
});
|
|
|
|
test('after signalling a verified container, successful empty Docker listing proves --rm removed that id', async t => {
|
|
const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
|
|
let removed = false;
|
|
const stop = api.createContainerStopper({ docker: async args => {
|
|
if (args[0] === 'kill') { removed = true; return ''; }
|
|
if (args[0] === 'ps') return '';
|
|
if (removed) throw Error('removed');
|
|
return JSON.stringify(inspected(r));
|
|
} });
|
|
const result = await stop(r, { graceMs: 20, killWaitMs: 100 });
|
|
assert.equal(result.outcome, 'terminated'); assert.deepEqual(result.signals, ['SIGTERM']);
|
|
});
|