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

// M2/#649: actual Store reopen and createLauncher recovery, no real proxy/config.
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 { createLauncher } from '../../cli/src/launcher.mjs';
import { sessionsFile, launchLogFile } from '../../seat/src/session.mjs';
import { startTimeOf } from '../../seat/src/proc.mjs';
import { reviewFixture } from './model-review-fixtures.mjs';

async function historical(t, mode, { ended = false, live = false } = {}) {
  const f = await reviewFixture(t);
  const original = { ...f.entry };
  if (mode === 'legacy') delete f.entry.modelKeyId;
  const first = await f.boot({ proxyMode: false });
  const record = { business: 'demo', role: 'chief', run: original.run, harness: 'pi', pid: original.pid, startTime: original.startTime };
  const launched = first.bindLaunch(record);
  first.broker.request(launched.cap, { verb: 'role.claim' });
  const db = new DatabaseSync(join(f.dataRoot, 'bus', 'bus.sqlite'), { readOnly: true });
  t.after(() => db.close());
  const count = (kind, run = original.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, 'claim'); assert.equal(claim().holder_run, original.run);
  if (ended) first.endLaunch({ business: 'demo', run: original.run, reason: 'host-lost', exitCode: null });
  await f.closeBus();
  // SQLite and immutable launch/claim history persist through the real runtime Store reopen.
  if (!live) await f.exit();
  f.writeSessions([f.entry]);
  if (mode === 'down') f.state.failGet = true;
  if (mode === 'open') f.state.hideSentinel = true;
  if (mode === 'revoking' || mode === 'revoked') {
    f.setModel({ state: 'revoking', revokingAt: new Date().toISOString() });
    if (mode === 'revoking') f.state.failPut = true;
  }
  const bus = await f.boot();
  assert.equal(count('session.launched'), 1);
  assert.equal(claim().op, ended ? 'release' : 'claim', 'persisted claim state survives Store reopen');
  const logs = [], binds = [], cleanupReplies = [];
  const host = {
    business: 'demo', beforeClose() {},
    bindLaunch: async r => { const result = bus.bindLaunch(r); binds.push(true); return result; },
    op: async m => {
      if (m.op === 'endLaunch') return bus.endLaunch(m.record);
      // Proposed trusted method/caller contract ONLY; old launcher never invokes
      // this branch. No product implementation or admission bypass in this fixture.
      if (m.op === 'recoverLaunch') { const result = bus.recoverLaunch(m.record); cleanupReplies.push(result); return result; }
      if (m.op === 'lifecycleRequests') return bus.lifecycleRequests(m.business);
      if (m.op === 'threadsState') return bus.threadsState(m.business);
      throw Error('fixture-unrequested-op');
    },
  };
  const recover = async ({ launches = true } = {}) => {
    const launcher = createLauncher({ host, system: JSON.parse(fs.readFileSync(f.config, 'utf8')),
      env: { MOSAIC_CONFIG: f.config, MOSAIC_ROLES_DIR: f.rolesDir, PATH: '/usr/bin' }, launches, log: line => logs.push(line) });
    f.attachLauncher(launcher); await launcher.listen();
  };
  const endedOnce = () => {
    assert.equal(count('session.ended'), 1, 'historical cleanup appends one end despite proxy admission refusal');
    assert.equal(claim().op, 'release', 'historical cleanup releases persisted claim');
    const body = JSON.parse(db.prepare("SELECT body FROM events WHERE kind='session.ended' AND subject=? ORDER BY seq DESC LIMIT 1").get(original.run).body);
    assert.equal(body.reason, 'host-lost'); assert.equal(body.released, true);
    const lines = fs.readFileSync(launchLogFile(f.dataRoot, 'demo'), 'utf8').trim().split('\n').map(line => JSON.parse(line));
    assert.equal(lines.filter(line => line.event === 'end' && line.run === original.run && line.reason === 'host-lost').length, 1);
  };
  const noCapability = () => {
    assert.equal(binds.length, 0, 'historical cleanup never calls capability-minting bindLaunch');
    assert.ok(cleanupReplies.every(reply => !Object.hasOwn(reply, 'cap')), 'cleanup result contains no capability');
  };
  return { f, bus, original, record, count, claim, logs, recover, endedOnce, noCapability };
}

for (const mode of ['down', 'open', 'revoking', 'revoked', 'legacy']) test(`M2 historical cleanup survives ${mode} proxy admission with persisted end and release`, async t => {
  const h = await historical(t, mode); await h.recover(); h.endedOnce(); h.noCapability();
});
test('M2 healthy historical cleanup also mints no capability or model-run binding', async t => {
  const h = await historical(t, 'healthy'); await h.recover(); h.endedOnce(); h.noCapability();
});
test('M2 fresh ordinary launch still requires current proxy admission', async t => {
  const h = await historical(t, 'down');
  await assert.rejects(async () => h.bus.bindLaunch({ ...h.record, run: 'fresh-m2' }), error => error.code === 'proxy-down');
  assert.equal(h.count('session.launched', 'fresh-m2'), 0); assert.equal(h.count('session.ended', 'fresh-m2'), 0);
});
test('M2 forged historical recovery without immutable launch cannot create or end a run', async t => {
  const h = await historical(t, 'healthy'); h.f.writeSessions([{ ...h.f.entry, run: 'forged-m2' }]);
  await h.recover();
  assert.equal(h.count('session.launched', 'forged-m2'), 0, 'cleanup cannot synthesize a fresh immutable launch');
  assert.equal(h.count('session.ended', 'forged-m2'), 0); assert.equal(h.claim().op, 'claim');
});
for (const [field, value] of [['instance', 'coder'], ['harness', 'claude-code'], ['pid', 'runner'], ['startTime', '0'], ['thread', 'thread-m2'], ['session', 'session-m2']]) {
  test(`M2 mismatched historical ${field} never releases or ends the immutable run`, async t => {
    const h = await historical(t, 'healthy');
    h.f.writeSessions([{ ...h.f.entry, [field]: value === 'runner' ? h.f.entry.runnerPid : value }]);
    await h.recover(); assert.equal(h.count('session.ended'), 0); assert.equal(h.claim().op, 'claim');
  });
}
test('M2 already-ended historical cleanup is classified before proxy admission without duplicate release', async t => {
  const h = await historical(t, 'down', { ended: true }); await h.recover();
  assert.equal(h.count('session.ended'), 1); assert.equal(h.claim().op, 'release'); h.noCapability();
  assert.ok(h.logs.some(line => line.includes('already ended')), 'ended history is recognized despite proxy down');
});
test('M2 no-launch preserves prior live sessions and claim with no cleanup or binding', async t => {
  const h = await historical(t, 'down', { live: true }); const before = fs.readFileSync(sessionsFile(h.f.dataRoot), 'utf8');
  await h.recover({ launches: false });
  assert.equal(h.count('session.ended'), 0); assert.equal(h.claim().op, 'claim'); h.noCapability();
  assert.equal(startTimeOf(h.original.pid), h.original.startTime); assert.equal(startTimeOf(h.original.runnerPid), h.original.runnerStartTime);
  assert.equal(fs.readFileSync(sessionsFile(h.f.dataRoot), 'utf8'), before);
});

test('M2 persisted claim blocks a new claim until actual historical cleanup releases it', async t => {
  const h = await historical(t, 'healthy');
  const model = h.bus.credentials.status().find(row => row.instance === 'demo/chief' && row.service === 'model');
  const fresh = h.bus.bindLaunch({ ...h.record, run: 'claim-m2', modelKeyId: model.keyId });
  assert.throws(() => h.bus.broker.request(fresh.cap, { verb: 'role.claim' }), error => error.code === 'role-already-held');
  await h.recover();
  assert.equal(h.claim().op, 'release');
  h.bus.broker.request(fresh.cap, { verb: 'role.claim' });
  assert.equal(h.claim().holder_run, 'claim-m2');
  h.bus.endLaunch({ business: 'demo', run: 'claim-m2', reason: 'stopped', exitCode: null });
});