stack/packages/bus/tests/model-historical-ipc.test.mjs

// M2/#649 actual trusted host -> forked bus/process.mjs -> runtime -> Broker route.
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { createHash } from 'node:crypto';
import { startHost } from '../../cli/src/host.mjs';
import { createLauncher } from '../../cli/src/launcher.mjs';
import { loadBusiness } from '../../organization/src/organization-config.mjs';
import { resolveInstance } from '../../organization/src/resolve.mjs';
import { busOrganization } from '../src/organization-context.mjs';
import { sessionsFile, launchLogFile } from '../../seat/src/session.mjs';
import { reviewFixture } from './model-review-fixtures.mjs';
const id = key => createHash('sha256').update(key).digest('hex').slice(0, 12);

async function ipcFixture(t, { down = true, ended = false, live = false } = {}) {
  let host, db;
  const owned = [];
  t.after(async () => {
    const errors = [];
    for (const close of [() => host?.close(), () => db?.close(), ...owned]) {
      try { await close(); } catch (error) { errors.push(error); }
    }
    if (errors.length) throw new AggregateError(errors, 'owned IPC fixture cleanup failed');
  });
  // Capture helper cleanup so real fork exits BEFORE the fake proxy/root disappear.
  const f = await reviewFixture({ after: close => owned.push(close) });
  const first = await f.boot();
  const proxy = await first.modelPreflight({ business: 'demo', role: 'chief' });
  const record = { business: 'demo', role: 'chief', run: f.entry.run, harness: 'pi', pid: f.entry.pid, startTime: f.entry.startTime };
  const launch = first.bindLaunch({ ...record, modelKeyId: f.entry.modelKeyId });
  first.broker.request(launch.cap, { verb: 'role.claim' });
  if (ended) first.endLaunch({ business: 'demo', run: record.run, reason: 'host-lost', exitCode: null });
  await f.closeBus(); if (!live) await f.exit(); f.writeSessions([f.entry]);
  db = new DatabaseSync(join(f.dataRoot, 'bus', 'bus.sqlite'), { readOnly: true });
  const count = (kind, run = record.run) => db.prepare('SELECT count(*) n FROM events WHERE business=? AND kind=? AND subject=?').get('demo', kind, run).n;
  const claim = () => db.prepare('SELECT op,holder_run FROM role_claims WHERE business=? AND role=? ORDER BY seq DESC LIMIT 1').get('demo', 'chief');
  assert.equal(count('session.launched'), 1); assert.equal(claim().op, ended ? 'release' : 'claim');
  f.state.failGet = down;
  const business = loadBusiness('demo', { dir: f.configDir, rolesDir: f.rolesDir }), resolved = {};
  for (const instance of Object.keys(business.roles)) resolved[instance] = resolveInstance({ system: f.system, business, instance });
  const logs = [];
  host = await startHost({ business: 'demo', log: line => logs.push(line), boot: {
    dataRoot: f.dataRoot, readers: ['demo'], businesses: { demo: busOrganization(business, resolved) },
    organization: { dir: f.configDir, rolesDir: f.rolesDir, system: f.system },
    proxy: { port: Number(new URL(proxy.baseUrl).port), binary: '/fixture-unused' },
  } });
  assert.equal(Number.isSafeInteger(host.pids.broker), true); assert.notEqual(host.pids.broker, process.pid, 'actual broker is a forked process');
  const cleanup = input => host.op({ op: 'recoverLaunch', record: input });
  const channelSurvives = async () => assert.equal(Array.isArray(await host.op({ op: 'credentialStatus', business: 'demo', role: 'chief' })), true);
  return { f, host, record, logs, cleanup, channelSurvives, count, claim,
    actualLauncher: async (launches = true) => {
      const launcher = createLauncher({ host, launches, system: JSON.parse(fs.readFileSync(f.config, 'utf8')),
        env: { MOSAIC_CONFIG: f.config, MOSAIC_ROLES_DIR: f.rolesDir, PATH: '/usr/bin' }, log: line => logs.push(line) });
      await launcher.listen(); return launcher; // real host beforeClose owns its cleanup
    } };
}

test('M2 IPC exact historical cleanup releases persisted claim with no cap or model-stall binding', async t => {
  const h = await ipcFixture(t); const beforePut = h.f.state.put;
  let reply;
  await assert.doesNotReject(async () => { reply = await h.cleanup(h.record); }, 'actual trusted historical IPC cleanup must dispatch');
  assert.deepEqual(Object.keys(reply), ['released'], 'historical IPC cleanup returns end metadata only');
  assert.equal(reply.released, true); assert.equal(h.claim().op, 'release'); assert.equal(h.count('session.ended'), 1);
  assert.equal(h.f.state.put, beforePut, 'cleanup does not mutate proxy keys');
  await assert.rejects(h.host.op({ op: 'modelStall', business: 'demo', role: 'chief', run: h.record.run, keyId: h.f.entry.modelKeyId }), error => error.code === 'proxy-stop-failed');
  await h.channelSurvives();
});
test('M2 IPC actual launcher recovery dispatches cleanup despite proxy-down admission', async t => {
  const h = await ipcFixture(t); await h.actualLauncher();
  assert.equal(h.count('session.ended'), 1, 'real fork/launcher recovery appends historical end');
  assert.equal(h.claim().op, 'release');
  const lines = fs.readFileSync(launchLogFile(h.f.dataRoot, 'demo'), 'utf8').trim().split('\n').map(line => JSON.parse(line));
  assert.equal(lines.filter(line => line.event === 'end' && line.run === h.record.run).length, 1);
  await h.channelSurvives();
});
for (const [label, patch, code] of [
  ['fresh', { run: 'ipc-fresh' }, 'unknown-run'],
  ['forged role', { role: 'coder' }, 'launch-record-mismatch'],
  ['mismatched harness', { harness: 'claude-code' }, 'launch-record-mismatch'],
  ['mismatched start', { startTime: '0' }, 'launch-record-mismatch'],
]) test(`M2 IPC ${label} cannot use historical cleanup and leaves channel usable`, async t => {
  const h = await ipcFixture(t);
  await assert.rejects(h.cleanup({ ...h.record, ...patch }), error => error.code === code, `trusted historical IPC ${label} refusal uses ${code}`);
  assert.equal(h.count('session.ended'), 0); assert.equal(h.claim().op, 'claim');
  if (patch.run) assert.equal(h.count('session.launched', patch.run), 0);
  await h.channelSurvives();
});
test('M2 IPC already-ended history refuses without duplicate release and preserves channel', async t => {
  const h = await ipcFixture(t, { ended: true });
  await assert.rejects(h.cleanup(h.record), error => error.code === 'run-ended', 'trusted historical IPC ended refusal uses run-ended');
  assert.equal(h.count('session.ended'), 1); assert.equal(h.claim().op, 'release'); await h.channelSurvives();
});
test('M2 IPC ordinary fresh bind remains proxy-admitted and capable only through normal bind', async t => {
  const h = await ipcFixture(t, { down: false });
  const proxy = await h.host.op({ op: 'modelPreflight', business: 'demo', role: 'chief' });
  const launch = await h.host.bindLaunch({ ...h.record, run: 'ipc-normal', modelKeyId: id(proxy.key) });
  assert.equal(typeof launch.cap, 'string');
  assert.equal(h.count('session.launched', 'ipc-normal'), 1);
  await h.host.op({ op: 'endLaunch', record: { business: 'demo', run: 'ipc-normal', reason: 'stopped', exitCode: null } });
  await h.channelSurvives();
});
test('M2 IPC ordinary fresh bind still refuses proxy-down and no-launch recovery remains inert', async t => {
  const h = await ipcFixture(t, { live: true }); const before = fs.readFileSync(sessionsFile(h.f.dataRoot), 'utf8');
  await assert.rejects(h.host.bindLaunch({ ...h.record, run: 'ipc-denied', modelKeyId: h.f.entry.modelKeyId }), error => error.code === 'proxy-down');
  await h.actualLauncher(false);
  assert.equal(h.count('session.ended'), 0); assert.equal(h.claim().op, 'claim');
  assert.equal(fs.readFileSync(sessionsFile(h.f.dataRoot), 'utf8'), before); await h.channelSurvives();
});