297 lines
18 KiB
JavaScript
297 lines
18 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { spawnSync } from 'node:child_process';
|
|
import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
|
|
import { join } from 'node:path';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { fixture, tmp, withoutHostCredentials, fakeHarnesses } from './helpers.mjs';
|
|
import { bootConfig, loadSystem } from '../src/config.mjs';
|
|
import { startHost, startTimeOf } from '../src/host.mjs';
|
|
import { createLauncher } from '../src/launcher.mjs';
|
|
import { Client } from '../../bus/src/client.mjs';
|
|
import { gitEnv } from '../../seat/src/checkout.mjs';
|
|
|
|
async function setup(t, edit = d => d, { revoked = false } = {}) {
|
|
const root = tmp(t, 'request-launch-'), f = fixture(root, 'demo', edit);
|
|
withoutHostCredentials(f.env, { piKey: true });
|
|
const repo = f.doc.projects.stack.root;
|
|
for (const args of [['init','-qb','main'],['-c','user.name=test','-c','[email protected]','commit','--allow-empty','-qm','seed']]) {
|
|
const r = spawnSync('git', ['-C', repo, ...args], { env: gitEnv(), encoding: 'utf8' }); assert.equal(r.status, 0, r.stderr);
|
|
}
|
|
const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
|
|
const adapter = join(root, 'adapter.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
|
|
const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
|
|
const system = loadSystem({ env: f.env }), host = await startHost({ boot: bootConfig({ system, businessId: 'demo', env: f.env }), business: 'demo' });
|
|
const launcher = createLauncher({ host, system, env: f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 30 } });
|
|
await launcher.listen();
|
|
t.after(() => host.close(0));
|
|
if (revoked) {
|
|
const { DatabaseSync } = await import('node:sqlite');
|
|
const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite'));
|
|
db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('fixture-revoked',?,'demo','launch.revoked',NULL,'{}')").run(new Date().toISOString());
|
|
db.close();
|
|
}
|
|
const bound = {};
|
|
for (const role of ['chief','pm']) {
|
|
bound[role] = await host.bindLaunch({ business: 'demo', role, run: role+'-fixture', harness: 'pi', pid: process.pid, startTime: startTimeOf(process.pid) });
|
|
const c = new Client({ path: host.path, cap: bound[role].cap }); await c.call('role.claim');
|
|
}
|
|
return { root, f, host, launcher, bound };
|
|
}
|
|
|
|
test('PM cannot use continuation launch recovery to gain role.launch', async t => {
|
|
const x = await setup(t);
|
|
const out = await x.launcher.fromSession(x.bound.pm.cap, 'coder');
|
|
assert.equal(out.ok, false);
|
|
assert.ok(['decision-required','not-authorized'].includes(out.error), out.error);
|
|
});
|
|
|
|
test('chief launch refusals are returned unchanged and stored through the host request path', async t => {
|
|
const x = await setup(t);
|
|
const c = new Client({ path: x.host.path, cap: x.bound.pm.cap });
|
|
// Host operations are not broker verbs and cannot be invoked by a managed session.
|
|
await assert.rejects(c.call('request.launchResult', { instance: 'coder', code: 'capacity-full' }), /unknown-verb/);
|
|
const out = await x.launcher.fromSession(x.bound.chief.cap, 'absent');
|
|
assert.equal(out.error, 'instance-not-listed');
|
|
const pending = await x.host.op({ op: 'request.launches', business: 'demo' });
|
|
assert.deepEqual(pending, []);
|
|
});
|
|
|
|
async function until(fn) {
|
|
const end = Date.now() + 10000;
|
|
while (Date.now() < end) { const value = await fn(); if (value) return value; await new Promise(r => setTimeout(r, 25)); }
|
|
throw Error('request launch did not settle');
|
|
}
|
|
|
|
test('the real chief launches an existing worker with the admitted project in its manifest', async t => {
|
|
const x = await setup(t);
|
|
const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder');
|
|
assert.equal(out.ok, true, out.error);
|
|
const session = x.launcher.registry.sessions.find(s => s.run === out.result.run);
|
|
const manifest = JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json')));
|
|
assert.deepEqual(manifest.projects, ['stack']);
|
|
assert.equal(manifest.definition, 'coder');
|
|
assert.ok(manifest.typed.includes('push_working'));
|
|
await x.host.close(0);
|
|
});
|
|
|
|
test('revoked launch and full model limits preserve their existing refusal codes', async t => {
|
|
for (const code of ['launch-revoked','capacity-full']) await t.test(code, async t => {
|
|
const x = await setup(t, doc => {
|
|
if (code === 'capacity-full') doc.launch.limits.models.zai['glm-5.3-flash'] = 0;
|
|
return doc;
|
|
}, { revoked: code === 'launch-revoked' });
|
|
const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder');
|
|
assert.equal(out.error, code);
|
|
assert.equal(x.launcher.registry.sessions.length, 0);
|
|
});
|
|
});
|
|
|
|
function requestHost(x) {
|
|
const operations = {
|
|
identity: m => x.bus.identity(m.cap), authorizeLaunch: m => x.bus.authorizeLaunch(m.cap, m.instance),
|
|
refuse: m => x.bus.refuse(m.cap, m.code), credentialStatus: m => x.bus.credentialStatus(m.business, m.role),
|
|
modelPreflight: m => x.bus.modelPreflight(m), mayLaunch: m => x.bus.mayLaunch(m),
|
|
landHeld: m => x.bus.landHeld(m.record), landHolders: m => x.bus.landHolders(m.business),
|
|
sessionClears: m => x.bus.sessionClears(m.business),
|
|
threadsState: m => x.bus.threadsState(m.business), lifecycleRequests: m => x.bus.lifecycleRequests(m.business),
|
|
'request.admission': m => x.bus.admission(m), 'request.launches': m => x.bus.requestLaunches(m.business),
|
|
'request.launchResult': m => x.bus.requestLaunchResult(m.record), 'request.manifest': m => x.bus.requestManifest(m.record),
|
|
'request.release': m => x.bus.requestRelease(m.cap, m.record), 'request.chief': m => x.bus.requestChief(m.business),
|
|
endLaunch: m => x.bus.endLaunch(m.record),
|
|
};
|
|
return { business: 'demo', path: x.bus.path, dataRoot: x.f.dataRoot, beforeClose: () => {},
|
|
op: m => Promise.resolve().then(() => {
|
|
assert.ok(operations[m.op], `unexpected host operation ${m.op}`); return operations[m.op](m);
|
|
}),
|
|
bindLaunch: record => { const out = x.bus.bindLaunch(record); x.caps[record.role] = out.cap; return out; } };
|
|
}
|
|
|
|
test('an old product manifest drains its active turn before the chief relaunches through normal admission', { timeout: 30000 }, async t => {
|
|
const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
|
|
let launcher, worker, host, entered, released;
|
|
const f = await requestFixture(t, 'autonomous', async x => {
|
|
withoutHostCredentials(x.f.env, { piKey: true });
|
|
const adapter = join(x.f.dataRoot, 'hold.sh'), source = join(x.f.dataRoot, 'hold.mjs');
|
|
writeFileSync(source, `import {writeFileSync,existsSync} from 'node:fs';
|
|
const marker=process.env.MOSAIC_TURN_MARKER;
|
|
if ((process.env.MOSAIC_REQUEST??'').includes('HOLD-REQUEST-TURN')) {
|
|
writeFileSync(marker+'.entered','entered');
|
|
while(!existsSync(marker+'.release')) await new Promise(r=>setTimeout(r,25));
|
|
}
|
|
writeFileSync(marker,'done'); process.stdout.write('turn completed');`);
|
|
writeFileSync(adapter, `exec '${process.execPath}' '${source}'\n`);
|
|
const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
|
|
host = requestHost(x);
|
|
launcher = createLauncher({ host, system: x.system, env: x.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [source], sessionDefaults: { pollInterval: 25 } });
|
|
t.after(() => launcher.close());
|
|
x.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit', exitCode: 0 });
|
|
const out = await launcher.fromSession(x.caps.chief, 'pm'); assert.equal(out.ok, true, out.error);
|
|
worker = launcher.registry.sessions.find(s => s.run === out.result.run);
|
|
entered = join(worker.runDir, 'turn.done.entered'); released = join(worker.runDir, 'turn.done.release');
|
|
await until(async () => { try { x.call('pm', 'state'); return true; } catch { return false; } });
|
|
x.call('human', 'message.send', { to: 'pm', body: 'HOLD-REQUEST-TURN' });
|
|
const { existsSync } = await import('node:fs'); await until(() => existsSync(entered));
|
|
});
|
|
const original = JSON.parse(readFileSync(join(worker.runDir, 'bundle/manifest.json')));
|
|
assert.deepEqual(original.projects, ['stack']);
|
|
await f.bus.requestLifecycle.recover();
|
|
await launcher.tick();
|
|
const chief = f.call('chief', 'message.receive');
|
|
assert.ok(chief.some(m => m.body.includes('lacks admitted project timer')));
|
|
const refused = await launcher.fromSession(f.caps.chief, 'pm');
|
|
assert.equal(refused.error, 'request-manifest-draining');
|
|
assert.ok(launcher.registry.sessions.some(s => s.run === worker.run), 'active turn remains alive');
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'request-manifest-draining');
|
|
writeFileSync(released, 'release');
|
|
// The runtime keeps the same drain intent while the turn finishes; the chief path owns the release.
|
|
await until(async () => { await launcher.tick(); return !launcher.registry.sessions.some(s => s.run === worker.run); });
|
|
const next = await launcher.fromSession(f.caps.chief, 'pm'); assert.equal(next.ok, true, next.error);
|
|
const successor = launcher.registry.sessions.find(s => s.run === next.result.run);
|
|
assert.deepEqual(JSON.parse(readFileSync(join(successor.runDir, 'bundle/manifest.json'))).projects, ['stack','timer']);
|
|
const { existsSync } = await import('node:fs');
|
|
assert.equal(existsSync(join(f.f.dataRoot, 'workspaces/demo/pm/timer/.git')), true);
|
|
await launcher.close();
|
|
});
|
|
|
|
test('an unavailable chief records its exact refusal, stays quiet, and retries the same path on admission revision change', { timeout: 30000 }, async t => {
|
|
const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
|
|
const f = await requestFixture(t); withoutHostCredentials(f.f.env, { piKey: true });
|
|
const { dirname } = await import('node:path');
|
|
const config = join(dirname(f.f.config), 'businesses/demo.json');
|
|
const doc = JSON.parse(readFileSync(config));
|
|
doc.launch.limits.models.zai['glm-5.3-flash'] = 0; writeFileSync(config, JSON.stringify(doc));
|
|
f.bus.broker.endLaunch({ business: 'demo', run: 'chief-run', reason: 'exit', exitCode: 0 });
|
|
const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
|
|
const adapter = join(f.f.dataRoot, 'chief.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
|
|
const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
|
|
const launcher = createLauncher({ host: requestHost({ bus: f.bus, f: f.f, caps: f.caps }), system: f.system, env: f.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 25 } });
|
|
t.after(() => launcher.close());
|
|
await launcher.listen();
|
|
await f.bus.requestLifecycle.recover(); await launcher.tick();
|
|
assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal.code, 'capacity-full');
|
|
const attempts = f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length;
|
|
for (let n = 0; n < 3; n++) await launcher.tick();
|
|
assert.equal(f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length, attempts);
|
|
doc.launch.limits.models.zai['glm-5.3-flash'] = 1; writeFileSync(config, JSON.stringify(doc));
|
|
await launcher.tick();
|
|
assert.equal(launcher.registry.sessions.length, 1, JSON.stringify(await f.call('pm', 'request.get', { request: f.request })));
|
|
assert.equal(launcher.registry.sessions[0].instance, 'chief');
|
|
assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal, null);
|
|
await launcher.close();
|
|
});
|
|
|
|
test('archived product mappings stay out of new manifests while the business still serves live projects', async t => {
|
|
const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
|
|
const x = await requestFixture(t);
|
|
withoutHostCredentials(x.f.env, { piKey: true });
|
|
const project = x.store.get("SELECT id FROM projects WHERE repo='timer'");
|
|
assert.ok(project);
|
|
await x.call('human', 'project.archive', { project: project.id });
|
|
const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
|
|
const adapter = join(x.f.dataRoot, 'archive-launch.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
|
|
const launcher = createLauncher({ host: requestHost({ bus: x.bus, f: x.f, caps: x.caps }), system: x.system, env: x.f.env,
|
|
makeLoader: await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter }), harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 25 } });
|
|
t.after(() => launcher.close()); await launcher.listen();
|
|
const out = await launcher.fromSession(x.caps.chief, 'coder');
|
|
assert.equal(out.ok, true, out.error);
|
|
const session = launcher.registry.sessions.find(s => s.run === out.result.run);
|
|
assert.deepEqual(JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json'))).projects, ['stack']);
|
|
await launcher.close();
|
|
});
|
|
|
|
// A resting request has an outstanding PM continuation and no running chief. These
|
|
// tests use the same broker throughout; only the launcher is replaced for --no-launch.
|
|
async function restingRequest(t, { launches = true, brokenHolders = false } = {}) {
|
|
let launcher;
|
|
// Close the launcher before requestFixture closes its bus and removes its root.
|
|
t.after(async () => { if (launcher) await launcher.close(); });
|
|
const { requestFixture } = await import('../../bus/tests/request-fixture.mjs');
|
|
const f = await requestFixture(t);
|
|
withoutHostCredentials(f.f.env, { piKey: true });
|
|
f.bus.broker.endLaunch({ business: 'demo', run: 'chief-run', reason: 'exit', exitCode: 0 });
|
|
await f.bus.requestLifecycle.recover();
|
|
const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url));
|
|
const adapter = join(f.f.dataRoot, 'resume-chief.sh');
|
|
writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`);
|
|
// The real bus host creates this before constructing its read-only launcher.
|
|
mkdirSync(join(f.f.dataRoot, 'bus-host'), { recursive: true, mode: 0o700 });
|
|
const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter });
|
|
const host = requestHost({ bus: f.bus, f: f.f, caps: f.caps });
|
|
const op = host.op;
|
|
let chiefChecks = 0, inChiefLaunch = false;
|
|
host.op = m => {
|
|
if (m.op === 'request.chief') { chiefChecks++; inChiefLaunch = true; }
|
|
if (brokenHolders && inChiefLaunch && m.op === 'landHolders') return Promise.resolve({ malformed: true });
|
|
return op(m);
|
|
};
|
|
async function start(enabled) {
|
|
launcher = createLauncher({ host, system: f.system, env: f.f.env, launches: enabled, makeLoader,
|
|
harnessVersion: 'fixture', readOnly: [fake], threadTickMs: 3600000, sessionDefaults: { pollInterval: 25 } });
|
|
await launcher.listen();
|
|
return launcher;
|
|
}
|
|
await start(launches);
|
|
return { f, host, start, get launcher() { return launcher; }, chiefChecks: () => chiefChecks,
|
|
async replace(enabled = true) { await launcher.close(); launcher = null; return start(enabled); } };
|
|
}
|
|
|
|
async function expectQuietTicks(x, code) {
|
|
await x.launcher.tick();
|
|
const { f } = x;
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, code);
|
|
const effects = f.journal.effects('demo', f.request).length;
|
|
const messages = f.store.get('SELECT count(*) n FROM messages').n;
|
|
for (let i = 0; i < 5; i++) await x.launcher.tick();
|
|
assert.equal(x.launcher.registry.sessions.length, 0, 'ticks do not launch');
|
|
assert.equal(x.chiefChecks(), 0, 'ticks never enter chief launch');
|
|
assert.equal(f.journal.effects('demo', f.request).length, effects, 'identical refusal receipts are deduplicated');
|
|
assert.equal(f.store.get('SELECT count(*) n FROM messages').n, messages, 'ticks do not flood messages');
|
|
}
|
|
|
|
test('a resting chief launches on a later tick after processing resumes', { timeout: 30000 }, async t => {
|
|
const x = await restingRequest(t), { f } = x;
|
|
const halted = await f.call('human', 'processing.halt');
|
|
await expectQuietTicks(x, 'processing-halted');
|
|
await f.call('human', 'processing.resume', { generation: halted.generation });
|
|
await x.launcher.tick();
|
|
assert.equal(x.launcher.registry.sessions.length, 1, 'resume must launch without restarting the bus');
|
|
assert.equal(x.launcher.registry.sessions[0].instance, 'chief');
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal, null);
|
|
});
|
|
|
|
test('a no-launch host remains quiet and an enabled launcher can subsequently start the chief', { timeout: 30000 }, async t => {
|
|
const x = await restingRequest(t, { launches: false });
|
|
await expectQuietTicks(x, 'launches-disabled');
|
|
await x.replace();
|
|
await x.launcher.tick();
|
|
assert.equal(x.launcher.registry.sessions.length, 1, 'enabling launches must not require a bus restart');
|
|
assert.equal(x.launcher.registry.sessions[0].instance, 'chief');
|
|
assert.equal((await x.f.call('human', 'request.get', { request: x.f.request })).launch_refusal, null);
|
|
});
|
|
|
|
test('a real launch failure after resume consumes the attempt at the unchanged prerequisite', { timeout: 30000 }, async t => {
|
|
const x = await restingRequest(t, { brokenHolders: true }), { f } = x;
|
|
const halted = await f.call('human', 'processing.halt');
|
|
await expectQuietTicks(x, 'processing-halted');
|
|
await f.call('human', 'processing.resume', { generation: halted.generation });
|
|
await x.launcher.tick();
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launch-failed');
|
|
assert.equal(x.chiefChecks(), 1, 'the chief launch was attempted once');
|
|
const prerequisite = f.bus.requestLaunches('demo')[0].prerequisite;
|
|
const haltedAgain = await f.call('human', 'processing.halt');
|
|
await x.launcher.tick();
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launch-failed', 'the identical earlier halt receipt is deduplicated');
|
|
await f.call('human', 'processing.resume', { generation: haltedAgain.generation });
|
|
await x.replace(false);
|
|
await x.launcher.tick();
|
|
assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'launches-disabled');
|
|
await x.replace();
|
|
assert.equal(f.bus.requestLaunches('demo')[0].prerequisite, prerequisite);
|
|
const effects = f.journal.effects('demo', f.request).length;
|
|
for (let i = 0; i < 5; i++) await x.launcher.tick();
|
|
assert.equal(x.chiefChecks(), 1, 'a real failure does not retry at the same prerequisite');
|
|
assert.equal(f.journal.effects('demo', f.request).length, effects);
|
|
assert.equal(x.launcher.registry.sessions.length, 0);
|
|
});
|