stack/packages/cli/tests/federation-acceptance.test.mjs

// FED R7 / #1802. Run through scripts/test-federation-acceptance.sh: it owns
// the child process groups and scratch lifetime. Ordinary CLI suites skip it.
// sage-3625: federation-host.mjs supplies a labelled readProcess ancestry
// stand-in for setup, grant/revoke and both Console transports. This is NOT
// a genuine outside-agent human proof. Product transports/grant checks stay real.
import test from 'node:test';
import assert from 'node:assert/strict';
import { spawn, spawnSync } from 'node:child_process';
import { closeSync, cpSync, existsSync, mkdirSync, openSync, readFileSync, writeFileSync } from 'node:fs';
import { request } from 'node:http';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { businessDoc, launchDoc, writeJson } from '../../organization/tests/helpers.mjs';
import { REPO_SKILLS } from '../../harness/tests/helpers.mjs';
import { DatabaseSync } from 'node:sqlite';
import { SESSION_COOKIE } from '../../webui/src/sso.mjs';
import { fakeOidc } from '../../webui/tests/helpers/fake-oidc.mjs';
import { socketPath, REPO } from '../src/config.mjs';
import { startTimeOf } from '../src/host.mjs';
import { launchLogFile, readSessions } from '../../seat/src/session.mjs';

const root = process.env.MOSAIC_R7_ROOT;
const helper = fileURLToPath(new URL('./fixtures/federation-human.mjs', import.meta.url));
const adapterModule = fileURLToPath(new URL('./fixtures/thread-adapter.mjs', import.meta.url));
const webCli = join(REPO, 'packages/webui/src/cli.mjs');
const cli = join(REPO, 'packages/cli/src/cli.mjs');
const pause = ms => new Promise(r => setTimeout(r, ms));
const json = path => JSON.parse(readFileSync(path, 'utf8'));

function run(cmd, args, options = {}) {
  const r = spawnSync(cmd, args, { encoding: 'utf8', timeout: 30000, ...options });
  assert.equal(r.status, 0, `${cmd} ${args.join(' ')}: ${r.stderr ?? ''} ${r.error?.message ?? ''}`);
  return r.stdout;
}
async function until(read, label, ms = 20000) {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    const value = await read();
    if (value) return value;
    await pause(50);
  }
  assert.fail(`R7 assertion: ${label}`);
}

function http(instance, path, { cookie, body, method = body === undefined ? 'GET' : 'POST' } = {}) {
  return new Promise((resolve, reject) => {
    const text = body === undefined ? undefined : JSON.stringify(body);
    const headers = { host: new URL(instance.origin).host };
    if (cookie) headers.cookie = cookie;
    if (text !== undefined) Object.assign(headers, { origin: instance.origin, 'sec-fetch-site': 'same-origin', 'content-type': 'application/json', 'content-length': Buffer.byteLength(text) });
    const req = request({ host: '127.0.0.1', port: instance.port, path, method, headers, timeout: 5000 }, res => {
      let data = '';
      res.on('data', c => data += c);
      res.on('end', () => resolve({ status: res.statusCode, headers: res.headers, body: data && String(res.headers['content-type']).startsWith('application/json') ? JSON.parse(data) : data }));
    });
    req.on('error', reject);
    req.on('timeout', () => req.destroy(new Error('R7 HTTP timeout')));
    req.end(text);
  });
}
let serial = 0;
function child(instance, label, command) {
  const log = join(root, `${instance.name}-${label}.log`);
  const fd = openSync(log, 'a', 0o600);
  let proc;
  try { proc = spawn(command[0], command.slice(1), { detached: true, env: instance.env, stdio: ['ignore', fd, fd] }); }
  finally { closeSync(fd); }
  proc.on('error', e => { console.error(`R7 child ${label}: ${e.message}`); });
  assert.ok(proc.pid, `R7 ${label} spawned`);
  const start = startTimeOf(proc.pid);
  assert.ok(start, `R7 ${label} start time recorded`);
  // EXIT trap owns the detached group even if Node's test driver is killed.
  writeFileSync(join(root, 'groups'), `${proc.pid} ${start}\n`, { flag: 'a' });
  proc.unref();
  return { log, proc };
}
// Jason-shell ancestry stand-in is in the fixture host; transport/path are real.
function human(instance, verb, args = {}) {
  const out = join(root, `human-${++serial}.json`);
  run(process.execPath, [helper, socketPath(instance.dataRoot), instance.name, verb, JSON.stringify(args), out], { env: instance.env });
  return json(out);
}
// R5/R4 CLI with that same labelled fixture proof, never a live human shell.
function access(instance, args) {
  return run(process.execPath, [cli, 'access', ...args, '--business', instance.name], { env: instance.env });
}

function prepare(name, iss, ca) {
  const home = process.env.HOME;
  const env = { HOME: home, PATH: process.env.PATH, TMPDIR: process.env.TMPDIR };
  run(join(REPO, 'scripts/bootstrap.sh'), ['--instance', name], { env });
  const config = join(home, '.config', `mosaic-${name}`, 'config.json');
  const system = json(config), dataRoot = system.dataRoot, configDir = dirname(config);
  const fixtureRoot = join(root, name);
  mkdirSync(fixtureRoot, { recursive: true });
  const doc = businessDoc(fixtureRoot, name);
  doc.human = 'melanie';
  doc.plugins = { harness: ['pi'], channel: ['webui'] };
  doc.roles.chief.vars = { harness: 'pi', model: 'claude-opus-5-5', thinking: 'high' };
  doc.roles.assistant = { definition: 'assistant', vars: { harness: 'pi', model: 'claude-sonnet-5-5', thinking: 'high' } };
  doc.launch = launchDoc('chief', ['coder', 'assistant'], { opus: 1, sonnet: 2 });
  doc.launch.threadIdleSeconds = 3600;
  for (const role of Object.values(doc.roles)) role.vars = { harness: 'pi', model: 'glm-5.3-flash', thinking: 'high', ...role.vars };
  writeJson(join(configDir, 'businesses', `${name}.json`), doc);
  const origin = `https://${name}.example.test`;
  system.console = { sso: { publicOrigin: origin, providers: [{ id: 'test-op', issuer: iss.issuer, clientId: iss.clientId,
    clientSecret: { slot: name }, algs: ['RS256'], subjects: [{ sub: 'owner-sub', name: 'melanie' }] }] } };
  system.federation = { self: name, peers: [{ id: name === 'hub' ? 'factory' : 'hub', origin: `https://${name === 'hub' ? 'factory' : 'hub'}.example.test`, label: name === 'hub' ? 'Factory' : 'Hub' }] };
  writeJson(config, system);
  writeJson(join(configDir, 'access.json'), { accessVersion: 1, revision: 1, users: [], grants: [] });
  // Generated fake OP material only; never copies a host slot or credential.
  writeJson(join(configDir, 'secrets/mosaic-stack', name), iss.clientSecret);
  cpSync(REPO_SKILLS, join(dataRoot, 'skills-enabled'), { recursive: true });
  const projectRoot = doc.projects.stack.root;
  const gitEnv = { ...env, GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: '/dev/null' };
  const git = (...args) => run('git', ['-c', 'user.name=R7 fixture', '-c', '[email protected]', '-C', projectRoot, ...args], { env: gitEnv });
  git('init', '-q', '-b', 'main');
  writeFileSync(join(projectRoot, 'file.txt'), `${name} fixture\n`);
  git('add', 'file.txt'); git('commit', '-q', '-m', 'scratch fixture');
  const adapter = join(fixtureRoot, 'adapter.sh');
  writeFileSync(adapter, `exec '${process.execPath}' '${adapterModule}'\n`);
  return { name, origin, configDir, dataRoot, adapter, env: { ...env, MOSAIC_CONFIG: config, NODE_EXTRA_CA_CERTS: ca, ZAI_API_KEY: 'fx332-fake-pi' }, port: null };
}

async function signIn(instance, op, iss, sub) {
  const start = await http(instance, '/auth/login');
  assert.equal(start.status, 303, JSON.stringify(start.body));
  const location = new URL(start.headers.location);
  const { code } = op.grant(iss, location.href, { sub });
  const cookie = start.headers['set-cookie'][0].split(';')[0];
  const result = await http(instance, `/auth/callback?${new URLSearchParams({ code, state: location.searchParams.get('state') })}`, { cookie });
  const session = (result.headers['set-cookie'] ?? []).find(c => c.startsWith(`${SESSION_COOKIE}=`) && !c.includes('Max-Age=0'))?.split(';')[0];
  return { ...result, cookie: session };
}
const logs = i => existsSync(launchLogFile(i.dataRoot, i.name)) ? readFileSync(launchLogFile(i.dataRoot, i.name), 'utf8').trim().split('\n').filter(Boolean).map(JSON.parse) : [];

function claimHistory(instance, run) {
  const store = new DatabaseSync(join(instance.dataRoot, 'bus/bus.sqlite'), { readOnly: true });
  try { return store.prepare('SELECT op, reason FROM thread_claims WHERE holder_run=? ORDER BY seq').all(run); }
  finally { store.close(); }
}

test('FED R7 scratch two-instance acceptance', { skip: !root && 'run scripts/test-federation-acceptance.sh', timeout: 180000 }, async t => {
  const op = await fakeOidc();
  t.after(() => op.close());
  const hubIssuer = op.issuer('hub'), factoryIssuer = op.issuer('factory');
  assert.notEqual(hubIssuer.clientId, factoryIssuer.clientId);
  const ca = join(root, 'fake-op-ca.pem'); writeFileSync(ca, op.cert);
  assert.ok(![7330, 7331, 7332].includes(Number(new URL(op.base).port)));
  writeFileSync(join(root, 'ports'), `${new URL(op.base).port}\n`, { flag: 'a' });
  const hub = prepare('hub', hubIssuer, ca), factory = prepare('factory', factoryIssuer, ca);
  assert.notEqual(hub.dataRoot, factory.dataRoot);
  assert.notEqual(hub.env.MOSAIC_CONFIG, factory.env.MOSAIC_CONFIG);
  for (const i of [hub, factory]) {
    const ready = join(root, `${i.name}-host.ready`);
    const host = child(i, 'host', [process.execPath, '--experimental-test-module-mocks',
      fileURLToPath(new URL('./fixtures/federation-host.mjs', import.meta.url)), i.name, i.adapter, ready]);
    await until(() => existsSync(ready), `${i.name} fixture host ready; ${host.log}`);
    writeFileSync(join(root, 'sockets'), `${socketPath(i.dataRoot)}\n`, { flag: 'a' });
  }
  const shared = human(hub, 'project.create', { name: 'hub-shared', repo: 'stack' }).id;
  const privateProject = human(hub, 'project.create', { name: 'hub-private' }).id;
  human(factory, 'project.create', { name: 'factory-only' });
  access(hub, ['user', 'add', 'jason-test', '--home', 'factory', '--subject', 'test-op:jason-test-sub', '--revision', '1']);
  access(hub, ['grant', '--user', 'jason-test', '--projects', shared, '--instances', 'assistant', '--level', 'collaborator', '--revision', '2']);
  for (const i of [hub, factory]) {
    const { log } = child(i, 'console', [process.execPath, webCli, 'serve', '--port', '0', '--business', i.name, '--board', 'http://127.0.0.1:1', '--sso', '--send']);
    i.port = Number(await until(() => existsSync(log) && readFileSync(log, 'utf8').match(/Mosaic Console: http:\/\/127\.0\.0\.1:(\d+)\//)?.[1], `${i.name} Console started`));
    assert.ok(![7330, 7331, 7332].includes(i.port));
    writeFileSync(join(root, 'ports'), `${i.port}\n`, { flag: 'a' });
  }
  assert.notEqual(hub.port, factory.port);
  if (process.env.MOSAIC_R7_INTERRUPT === '1') {
    console.log('R7 interruption: both Consoles started; killing test driver');
    process.kill(process.pid, 'SIGKILL');
    await pause(60000);
  }
  const signed = await signIn(hub, op, hubIssuer, 'jason-test-sub');
  assert.equal(signed.status, 303, JSON.stringify(signed.body)); assert.ok(signed.cookie);
  const cookie = signed.cookie;
  const config = await http(hub, '/api/config', { cookie });
  assert.equal(config.status, 200); assert.equal(config.body.user, true);
  const projects = await http(hub, '/api/chat/projects', { cookie });
  assert.equal(projects.status, 200, JSON.stringify(projects.body));
  assert.deepEqual(projects.body.rows.map(p => p.id), [shared], 'R7 ungranted project must not appear');
  const created = await http(hub, '/api/chat/threads', { cookie, body: { project: shared, instance: 'assistant', name: 'Jason scratch thread' } });
  assert.equal(created.status, 200, JSON.stringify(created.body));
  const thread = created.body.id;
  const sent = await http(hub, `/api/chat/threads/${thread}/messages`, { cookie, body: { body: 'R7 scratch question' } });
  assert.equal(sent.status, 200, JSON.stringify(sent.body));
  const reply = await until(async () => {
    const r = await http(hub, `/api/chat/threads/${thread}/messages`, { cookie });
    assert.equal(r.status, 200, JSON.stringify(r.body));
    return r.body.rows.find(m => m.class === 'RESULT' && m.in_reply_to === sent.body.id);
  }, 'granted instance replies in Jason thread');
  assert.equal(reply.from_role, 'assistant'); assert.equal(reply.to_role, 'user');
  assert.match(reply.body, /R7 scratch question/);
  const refused = await http(hub, '/api/chat/threads', { cookie, body: { project: privateProject, instance: 'assistant', name: 'must refuse' } });
  assert.equal(refused.status, 403); assert.equal(refused.body.error, 'access-refused');
  assert.deepEqual(json(join(hub.configDir, 'access.json')).grants[0].projects, [shared], 'R7 grant must name exactly one project');
  for (const path of ['/api/board', '/api/bus/inbox', '/api/bus/tasks']) {
    const r = await http(hub, path, { cookie });
    assert.equal(r.status, 403, path); assert.equal(r.body.error, 'owner-only', path);
  }
  const factoryLogin = await signIn(factory, op, factoryIssuer, 'jason-test-sub');
  assert.equal(factoryLogin.status, 403); assert.equal(factoryLogin.cookie, undefined);
  const factoryOwner = await signIn(factory, op, factoryIssuer, 'owner-sub');
  assert.equal(factoryOwner.status, 303); assert.ok(factoryOwner.cookie);
  const factoryProjects = await http(factory, '/api/chat/projects', { cookie: factoryOwner.cookie });
  assert.equal(factoryProjects.status, 200);
  assert.deepEqual(factoryProjects.body.rows.map(p => p.name), ['factory-only']);
  const factoryThreads = await http(factory, '/api/chat/threads', { cookie: factoryOwner.cookie });
  assert.equal(factoryThreads.status, 200); assert.deepEqual(factoryThreads.body.rows, []);
  const hubCookieAtFactory = await http(factory, '/api/chat/projects', { cookie });
  assert.equal(hubCookieAtFactory.status, 401);
  const [launch] = logs(hub).filter(e => e.event === 'launch' && e.thread === thread);
  assert.ok(launch, 'R7 sandboxed run launched');
  const liveRun = readSessions(hub.dataRoot).find(s => s.run === launch.run);
  assert.ok(liveRun, 'R7 run remains live before revoke');
  access(hub, ['revoke', '--user', 'jason-test', '--revision', '3']);
  const after = await http(hub, '/api/chat/projects', { cookie });
  assert.equal(after.status, 401); assert.equal(after.body.error, 'sign-in required');
  assert.equal((await signIn(hub, op, hubIssuer, 'jason-test-sub')).status, 403);
  // Pin the claim before waiting for process exit: a missing broker revocation
  // must fail this assertion, even if it also prevents the runner from ending.
  assert.ok(claimHistory(hub, launch.run).some(r => r.op === 'revoke' && r.reason === 'access-revoked'),
    'R7 revoke must end the claim as access-revoked');
  const ended = await until(() => logs(hub).find(e => e.event === 'end' && e.run === launch.run), 'revoked run ended');
  assert.equal(ended.exitCode, 0, JSON.stringify(ended));
  assert.deepEqual(readSessions(hub.dataRoot), []);
  await until(() => startTimeOf(liveRun.pid) !== liveRun.startTime && startTimeOf(liveRun.runnerPid) !== liveRun.runnerStartTime, 'revoked engine and runner processes gone');
  assert.equal(claimHistory(hub, launch.run).at(-1).op, 'revoke', 'R7 shutdown must not replace missing revocation with release');
  assert.deepEqual(human(factory, 'thread.list'), []);
});