stack/packages/bus/tests/model-restart.test.mjs

// 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();
});