// B1 / 647: real runtime timer, scratch proxy and persisted sessions, never the host instance. import test from 'node:test'; import assert from 'node:assert/strict'; import fs from 'node:fs'; import http from 'node:http'; import { spawn } from 'node:child_process'; import { once } from 'node:events'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { createHash, randomBytes } from 'node:crypto'; import { setTimeout as wait } from 'node:timers/promises'; import { startBroker } from '../src/runtime.mjs'; import { Credentials } from '../src/credentials.mjs'; import { ModelKeys } from '../src/model-keys.mjs'; import { ManagementClient } from '../../model-proxy/src/client.mjs'; import { busOrganization } from '../src/organization-context.mjs'; import { loadBusiness } from '../../organization/src/organization-config.mjs'; import { resolveInstance } from '../../organization/src/resolve.mjs'; import { businessDoc, rolesCopy, systemFor, systemConfig, writeJson } from '../../organization/tests/helpers.mjs'; import { startTimeOf } from '../../seat/src/proc.mjs'; import { Registry, SESSIONS_VERSION, sessionsFile } from '../../seat/src/session.mjs'; import { createLauncher } from '../../cli/src/launcher.mjs'; const id = value => createHash('sha256').update(value).digest('hex').slice(0, 12); async function fixture(t, { overdue = false, exited = false } = {}) { const root = fs.mkdtempSync(join(tmpdir(), 'b1-')); let server, credentials, child, childClosed, bus, launcher; let completed = 0; const originalSweep = ModelKeys.prototype.sweep; const exit = async () => { // Only the ChildProcess this fixture spawned: never use a registry-supplied PID. if (child?.pid && child.exitCode === null && child.signalCode === null) child.kill('SIGTERM'); if (childClosed) await childClosed; }; t.after(async () => { const errors = []; for (const close of [() => launcher?.close(), () => bus?.close(), () => credentials?.close(), exit, async () => { server?.closeAllConnections(); if (server?.listening) await new Promise(resolve => server.close(resolve)); }]) { try { await close(); } catch (error) { errors.push(error); } } ModelKeys.prototype.sweep = originalSweep; fs.rmSync(root, { recursive: true, force: true }); if (errors.length) throw new AggregateError(errors, 'scratch fixture cleanup failed'); }); // Await the ORIGINAL sweep, including its real HTTP/filesystem work. Count both // fulfilled and rejected completions; the runtime still handles its own refusal. ModelKeys.prototype.sweep = async function (...args) { try { return await originalSweep.apply(this, args); } finally { completed++; } }; const observeSweep = async () => { const before = completed, deadline = Date.now() + 5000; // Same five-second bounded wait budget as bus/tests/host-run.test.mjs:47. // Production interval remains 1000ms; no clock or scheduler is replaced. while (completed <= before && Date.now() < deadline) await wait(20); assert.ok(completed > before, 'original runtime sweep completed within five seconds'); }; const configDir = join(root, 'config'), dataRoot = join(root, 'data'); fs.mkdirSync(dataRoot, { mode: 0o700 }); const config = systemConfig(root), system = systemFor(root), rolesDir = rolesCopy(root); const doc = businessDoc(root, 'demo'); doc.plugins = { ...doc.plugins, harness: ['pi'] }; writeJson(join(configDir, 'businesses', 'demo.json'), doc); const business = loadBusiness('demo', { dir: configDir, rolesDir }), resolved = {}; for (const instance of Object.keys(business.roles)) resolved[instance] = resolveInstance({ system, business, instance }); const storeRoot = join(configDir, 'secrets', 'agents'); fs.mkdirSync(storeRoot, { recursive: true, mode: 0o700 }); fs.writeFileSync(join(configDir, 'secrets', 'proxy-management.key'), 'fx647-manage', { mode: 0o600 }); const prefix = `msk-${id(configDir)}-`; const state = { keys: [prefix + 'sentinel-' + randomBytes(32).toString('base64url')], put: 0, removed: 0 }; server = http.createServer(async (req, res) => { if (req.url === '/v8/management/config/access/api-keys') { if (req.headers.authorization !== 'Bearer fx647-manage') { res.writeHead(401); res.end('{}'); return; } if (req.method === 'GET') { res.end(JSON.stringify(state.keys)); return; } const chunks = []; for await (const chunk of req) chunks.push(chunk); const next = JSON.parse(Buffer.concat(chunks)); state.removed += state.keys.filter(key => !next.includes(key)).length; state.keys = next; state.put++; res.end(JSON.stringify({ status: 'ok' })); return; } res.writeHead(state.keys.includes(req.headers.authorization?.slice(7)) ? 200 : 401); res.end('{}'); }); await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); const baseUrl = `http://127.0.0.1:${server.address().port}`; credentials = new Credentials({ storeRoot, dataRoot, modelAgents: ['demo/chief'] }); const manager = new ModelKeys({ credentials, configDir, baseUrl, client: new ManagementClient({ baseUrl, key: 'fx647-manage' }) }); await manager.reconcile(); const first = await manager.ensure('demo/chief'); await manager.rotate('demo/chief'); const record = credentials.modelRecord('demo/chief'); credentials.writeModel('demo/chief', { ...record, retiring: record.retiring.map(r => ({ ...r, retireAt: new Date(Date.now() + (overdue ? -60_000 : 3_600_000)).toISOString() })) }); credentials.close(); credentials = null; state.put = 0; state.removed = 0; // A harmless scratch-owned process supplies real /proc identity. It is deliberately // not a session runner, so stopSession must refuse rather than signal this stand-in. child = spawn('/usr/bin/sleep', ['60'], { stdio: 'ignore' }); childClosed = new Promise(resolve => child.once('close', resolve)); await once(child, 'spawn'); const ticks = startTimeOf(child.pid); assert.equal(typeof ticks, 'string'); const entry = { business: 'demo', instance: 'chief', run: 'r647-old', harness: 'pi', provider: 'fake', model: 'fake', pid: child.pid, startTime: ticks, runnerPid: child.pid, runnerStartTime: ticks, modelKeyId: id(first.key), launchedBy: 'chief' }; const writeSessions = entries => writeJson(sessionsFile(dataRoot), { sessionsVersion: SESSIONS_VERSION, sessions: entries }); writeSessions([entry]); if (exited) await exit(); const boot = async () => { bus = await startBroker({ dataRoot, businesses: { demo: busOrganization(business, resolved) }, organization: { dir: configDir, rolesDir, system }, proxy: { port: server.address().port, binary: '/fixture-unused' } }); return bus; }; const noLaunch = async () => { const ops = []; const host = { business: 'demo', beforeClose() {}, bindLaunch: async r => { ops.push('bindLaunch'); return bus.bindLaunch(r); }, op: async r => { ops.push(r.op); throw Error('fixture-unrequested-op'); } }; launcher = createLauncher({ host, system: JSON.parse(fs.readFileSync(config, 'utf8')), launches: false, env: { MOSAIC_CONFIG: config, MOSAIC_ROLES_DIR: rolesDir, PATH: '/usr/bin' }, log: () => {} }); await launcher.listen(); return ops; }; const retained = () => { assert.equal(state.keys.includes(first.key), true, 'old key stays remotely installed'); assert.equal(bus.credentials.modelRecord('demo/chief').retiring.some(r => r.keyId === entry.modelKeyId), true, 'retiring id stays locally recorded'); assert.equal(state.removed, 0, 'no remote key removal'); }; return { root, dataRoot, state, entry, writeSessions, boot, noLaunch, exit, retained, observeSweep, alive: () => startTimeOf(child.pid) === ticks, absent: () => { assert.equal(state.keys.includes(first.key), false); assert.equal(bus.credentials.modelRecord('demo/chief').retiring.length, 0); }, bind: () => bus.bindLaunch({ business: entry.business, role: entry.instance, run: entry.run, harness: entry.harness, pid: entry.pid, startTime: entry.startTime, modelKeyId: entry.modelKeyId }) }; } test('B1 restart before grace discovers persisted old-key run without bindLaunch', async t => { const f = await fixture(t); await f.boot(); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); test('B1 no-launch restart past grace neither signals nor ends a prior session', async t => { const f = await fixture(t, { overdue: true }); const before = fs.readFileSync(sessionsFile(f.dataRoot), 'utf8'); await f.boot(); const ops = await f.noLaunch(); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); assert.equal(ops.includes('bindLaunch') || ops.includes('endLaunch'), false); assert.equal(fs.readFileSync(sessionsFile(f.dataRoot), 'utf8'), before); }); test('B1 normal recovery gap retains boot inventory when Registry clears before rebinding', async t => { const f = await fixture(t); await f.boot(); new Registry(f.dataRoot); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); for (const corruption of ['malformed', 'missing']) test(`B1 ${corruption} persisted discovery never authorizes retirement`, async t => { const f = await fixture(t, { overdue: true }); if (corruption === 'malformed') fs.writeFileSync(sessionsFile(f.dataRoot), '{'); else fs.unlinkSync(sessionsFile(f.dataRoot)); await f.boot(); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); for (const field of ['startTime', 'runnerStartTime', 'modelKeyId']) test(`B1 persisted identity disagreement in ${field} retains key and record`, async t => { const f = await fixture(t, { overdue: true }); f.writeSessions([{ ...f.entry, [field]: field === 'modelKeyId' ? 'fedcba987654' : '0' }]); await f.boot(); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); test('B1 discovery failure after boot retains the cached live old-key inventory', async t => { const f = await fixture(t, { overdue: true }); await f.boot(); fs.writeFileSync(sessionsFile(f.dataRoot), '{'); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); test('B1 bound run with unauthorized runner stop retains key and retiring record', async t => { const f = await fixture(t, { overdue: true }); await f.boot(); f.bind(); await f.observeSweep(); f.retained(); assert.equal(f.alive(), true); }); test('B1 confirmed persisted runner exit permits old-key retirement after grace', async t => { const f = await fixture(t, { overdue: true, exited: true }); await f.boot(); await f.observeSweep(); f.absent(); }); test('B1 an observed persisted exit after restart permits retirement, never an empty registry alone', async t => { const f = await fixture(t, { overdue: true }); await f.boot(); await f.observeSweep(); f.retained(); await f.exit(); await f.observeSweep(); f.absent(); });