stack/packages/cli/tests/org-recovery.test.mjs

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, readdirSync, existsSync, symlinkSync, unlinkSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { reserveOrg, activateOrg } from '../../organization/src/org-registry.mjs';
import { startTimeOf } from '../src/host.mjs';
import { humanEnv } from './helpers.mjs';
import { execFile, spawn } from 'node:child_process';
import { once } from 'node:events';
import { consoleStart, consoleStop } from '../src/org-console.mjs';
import { promisify } from 'node:util';
const exec = promisify(execFile);

async function fixture() {
  const root = mkdtempSync(join(tmpdir(), 'recover-'));
  const configDir = join(root, '.config', 'mosaic-test'), dataRoot = join(root, 'data');
  mkdirSync(configDir, { mode: 0o700, recursive: true }); mkdirSync(dataRoot, { mode: 0o700 });
  const ctx = { configDir, dataRoot, instance: 'test' };
  const org = await reserveOrg(ctx, 'alpha'); await activateOrg(ctx, 'alpha');
  const env = humanEnv({ HOME: root, MOSAIC_CONFIG: join(configDir, 'config.json') });
  const moduleUrl = new URL('../src/cli.mjs', import.meta.url).href;
  const code = `import { run } from ${JSON.stringify(moduleUrl)}; process.exitCode = await run(JSON.parse(process.argv[1]), undefined, {system: {dataRoot: process.argv[2]}});`;
  const call = async (...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env }); return { out: { text: x.stdout }, err: { text: x.stderr } }; };
  return { root, ctx, org, call };
}
const dead = 2147483647;
const json = (path, value) => writeFileSync(path, JSON.stringify(value), { mode: 0o600 });

test('console stop quarantines dead and wrong-process states without signaling', async () => {
  const f = await fixture(), dir = join(f.ctx.dataRoot, 'console'); mkdirSync(dir, { mode: 0o700 });
  for (const pid of [dead, process.pid]) {
    json(join(dir, 'console.json'), { pid, startTime: '1', url: 'http://127.0.0.1:1/' });
    const x = await f.call('console', 'stop');
    assert.equal(JSON.parse(x.out.text).stopped, false);
    assert.match(x.err.text, /stale/i);
    assert.equal(existsSync(join(dir, 'console.json')), false);
  }
  assert.equal(readdirSync(dir).filter(x => x.startsWith('console.json.stale-')).length, 2);
});

test('live console from another entry path remains recorded and is not signalled', { timeout: 30000 }, async t => {
  const f = await fixture(), dir = join(f.ctx.dataRoot, 'console'), alias = join(f.root, 'console.mjs');
  mkdirSync(dir, { mode: 0o700 });
  symlinkSync(new URL('../../webui/src/cli.mjs', import.meta.url).pathname, alias);
  const child = spawn(process.execPath, [alias, 'serve', '--port', '0'], { env: humanEnv({ HOME: f.root, MOSAIC_CONFIG: join(f.ctx.configDir, 'config.json') }), stdio: ['ignore', 'pipe', 'pipe'] });
  t.after(async () => { if (child.exitCode === null && child.signalCode === null) { const ended = once(child, 'exit'); child.kill('SIGTERM'); await ended; } });
  const url = await new Promise((resolve, reject) => {
    let output = ''; const timer = setTimeout(() => reject(Error('startup timeout')), 10000);
    child.once('exit', () => { clearTimeout(timer); reject(Error('console exited')); });
    child.stdout.on('data', b => { output += b; const m = /Mosaic Console: (http:\/\/127\.0\.0\.1:\d+\/)/.exec(output); if (m) { clearTimeout(timer); resolve(m[1]); } });
  });
  const record = { pid: child.pid, startTime: startTimeOf(child.pid), url };
  json(join(dir, 'console.json'), record);
  await assert.rejects(consoleStart(f.ctx), /console state exists/);
  await assert.rejects(consoleStop(f.ctx), e => /another path/.test(e.message) && e.message.includes(alias));
  assert.deepEqual(JSON.parse(readFileSync(join(dir, 'console.json'), 'utf8')), record);
  assert.equal(startTimeOf(child.pid), record.startTime);
  assert.equal(readdirSync(dir).filter(x => x.startsWith('console.json.stale-')).length, 0);
});

test('interrupted quarantine leaves no hard link that blocks retry', async () => {
  const f = await fixture(), dir = join(f.ctx.dataRoot, 'console'), file = join(dir, 'console.json');
  mkdirSync(dir, { mode: 0o700 }); json(file, { pid: dead, startTime: '1', url: 'http://127.0.0.1:1/' });
  const moduleUrl = new URL('../src/org-console.mjs', import.meta.url).href;
  const code = `import fs from 'node:fs'; import { syncBuiltinESMExports } from 'node:module';
    for (const name of ['linkSync', 'renameSync']) { const real = fs[name]; fs[name] = (...args) => { const value = real(...args); if (String(args[1]).includes('console.json.stale-')) process.kill(process.pid, 'SIGKILL'); return value; }; } syncBuiltinESMExports();
    const { consoleStop } = await import(${JSON.stringify(moduleUrl)}); await consoleStop(${JSON.stringify(f.ctx)});`;
  const child = spawn(process.execPath, ['--input-type=module', '-e', code], { stdio: 'ignore' });
  assert.equal((await once(child, 'exit'))[1], 'SIGKILL');
  unlinkSync(join(dir, 'start.lock'));
  assert.deepEqual(await consoleStop(f.ctx), { stopped: false });
  assert.equal(existsSync(file), false);
  assert.equal(readdirSync(dir).filter(x => x.startsWith('console.json.stale-')).length, 1);
});

test('lock recovery reports fences and recovers a dead clearer while refusing live or unreadable fences', async () => {
  const f = await fixture(), lock = join(f.ctx.configDir, 'org-lifecycle-alpha.lock'), fence = `${lock}.clear`;
  json(lock, { pid: dead, at: '2026-01-01T00:00:00Z' });
  json(fence, { pid: process.pid, startTime: startTimeOf(process.pid) });
  let shown = JSON.parse((await f.call('org', 'lock', 'show', 'alpha')).out.text);
  assert.equal(shown.fence.pid, process.pid);
  await assert.rejects(f.call('org', 'lock', 'clear', 'alpha'), new RegExp(`recovery is already in progress.*${process.pid}`));
  writeFileSync(fence, '');
  await assert.rejects(f.call('org', 'lock', 'clear', 'alpha'), e => e.message.includes(fence) && /remove.*hand/i.test(e.message));
  unlinkSync(fence);
  const registry = new URL('../../organization/src/org-registry.mjs', import.meta.url).href;
  const code = `import fs from 'node:fs'; import { syncBuiltinESMExports } from 'node:module';
    const real = fs.writeFileSync; fs.writeFileSync = (...args) => { const value = real(...args); if (typeof args[0] === 'number') process.kill(process.pid, 'SIGKILL'); return value; }; syncBuiltinESMExports();
    const { orgLock } = await import(${JSON.stringify(registry)}); await orgLock(${JSON.stringify(f.ctx)}, 'alpha', { clear: true });`;
  const child = spawn(process.execPath, ['--input-type=module', '-e', code], { stdio: 'ignore' });
  assert.equal((await once(child, 'exit'))[1], 'SIGKILL');
  shown = JSON.parse((await f.call('org', 'lock', 'show', 'alpha')).out.text);
  assert.equal(shown.fence.pid, child.pid);
  assert.equal(shown.fence.alive, false);
  await f.call('org', 'lock', 'clear', 'alpha');
  assert.equal(existsSync(lock), false); assert.equal(existsSync(fence), false);
  assert.equal(readdirSync(f.ctx.configDir).filter(x => x.startsWith('org-lifecycle-alpha.lock.clear.stale-')).length, 1);
});

test('org lock show and clear require a provably dead holder', async () => {
  const f = await fixture(), path = join(f.ctx.configDir, 'org-lifecycle-alpha.lock');
  let x = await f.call('org', 'lock', 'show', 'alpha');
  assert.deepEqual(JSON.parse(x.out.text), { held: false });
  x = await f.call('org', 'lock', 'clear', 'alpha');
  assert.deepEqual(JSON.parse(x.out.text), { held: false, cleared: false });
  json(path, { pid: process.pid, at: '2026-01-01T00:00:00Z' });
  x = await f.call('org', 'lock', 'show', 'alpha');
  assert.equal(JSON.parse(x.out.text).alive, true);
  await assert.rejects(f.call('org', 'lock', 'clear', 'alpha'), /held|alive/i);
  assert.equal(existsSync(path), true);
  json(path, { pid: dead, at: '2026-01-01T00:00:00Z' });
  x = await f.call('org', 'lock', 'show', 'alpha');
  assert.equal(JSON.parse(x.out.text).alive, false);
  await f.call('org', 'lock', 'clear', 'alpha');
  assert.equal(existsSync(path), false);
});

test('bus recover preserves evidence for dead and wrong-process hosts without signals', async () => {
  const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 });
  for (const pid of [dead, process.pid]) {
    const state = { pid, startTime: startTimeOf(pid) ?? '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
    json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
    const x = await f.call('bus', 'recover', 'alpha');
    const result = JSON.parse(x.out.text);
    assert.equal(result.recovered, true);
    const receipt = JSON.parse(readFileSync(result.receipt, 'utf8'));
    assert.equal(receipt.action, 'bus.recover');
    assert.equal(receipt.stopped, true);
    assert.equal(receipt.host.pid, pid);
    assert.equal(existsSync(join(dir, 'host.json')), false);
    assert.equal(existsSync(join(dir, 'host.lock')), false);
    assert.ok(readdirSync(dir).some(x => x.startsWith('recovery-')));
  }
  assert.equal(process.pid > 0, true);
});

test('lock recovery refuses malformed or unreadable holders', async () => {
  const f = await fixture(), path = join(f.ctx.configDir, 'org-lifecycle-alpha.lock');
  for (const value of [{ pid: 0, at: '2026-01-01T00:00:00Z' }, { pid: dead }, null]) {
    json(path, value);
    await assert.rejects(f.call('org', 'lock', 'clear', 'alpha'));
    assert.equal(existsSync(path), true);
  }
});

test('bus recovery refuses a live writer or listening socket', async t => {
  const { createServer } = await import('node:net');
  const f = await fixture(), dir = join(f.org.root, 'bus-host'), bus = join(f.org.root, 'bus');
  mkdirSync(dir, { mode: 0o700 }); mkdirSync(bus, { mode: 0o700 });
  const state = { pid: dead, startTime: '1', business: 'alpha', startedAt: '2026-01-01T00:00:00Z' };
  json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state);
  json(join(bus, 'writer.lock'), { pid: process.pid, at: '2026-01-01T00:00:00Z' });
  await assert.rejects(f.call('bus', 'recover', 'alpha'), /writer.*alive/);
  assert.equal(existsSync(join(dir, 'host.json')), true);
  const { unlinkSync } = await import('node:fs'); unlinkSync(join(bus, 'writer.lock'));
  const server = createServer(s => s.end());
  await new Promise((resolve, reject) => server.once('error', reject).listen(join(bus, 'broker.sock'), resolve));
  t.after(() => new Promise(resolve => server.close(resolve)));
  await assert.rejects(f.call('bus', 'recover', 'alpha'), /socket/);
  assert.equal(existsSync(join(dir, 'host.json')), true);
});