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 callWith = async (extra, ...args) => { const x = await exec(process.execPath, ['--input-type=module', '-e', code, JSON.stringify(args), dataRoot], { env: { ...env, ...extra } }); return { out: { text: x.stdout }, err: { text: x.stderr } }; }; const call = (...args) => callWith({}, ...args); return { root, ctx, org, call, callWith }; } 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 => { // The other entry sits under the sentinel directory: the refusal must name neither (PR #1916). const SENTINEL = 'sk-test-NOT-A-SECRET-1234'; const f = await fixture(), dir = join(f.ctx.dataRoot, 'console'), alias = join(f.root, SENTINEL, 'console.mjs'); mkdirSync(dir, { mode: 0o700 }); mkdirSync(join(f.root, SENTINEL), { 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 => { assert.equal(e.exitCode, 3); assert.equal(e.message, 'console was started from another checkout; stop it there'); assert.ok(!e.message.includes(SENTINEL) && !e.message.includes(alias), e.message); return true; }); 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, ''); // #1873: the fence is named by label in the config directory, never by its path. await assert.rejects(f.call('org', 'lock', 'clear', 'alpha'), e => !e.message.includes(f.ctx.configDir) && e.message.includes('recovery fence org-lifecycle-alpha.lock.clear in the config directory') && /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 }); // A wrong process is a live pid whose start time is not the recorded one (the pid was reused). for (const pid of [dead, process.pid]) { const state = { pid, startTime: '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('D2: bus recover refuses a live pid with the recorded start time, whatever its argv', async () => { const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { mode: 0o700 }); const state = { pid: process.pid, startTime: startTimeOf(process.pid), business: 'alpha', startedAt: '2026-01-01T00:00:00Z' }; assert.match(state.startTime, /^\d+$/); json(join(dir, 'host.json'), state); json(join(dir, 'host.lock'), state); await assert.rejects(f.call('bus', 'recover', 'alpha'), /bus host identity is uncertain/); assert.deepEqual(JSON.parse(readFileSync(join(dir, 'host.json'), 'utf8')), state); assert.equal(existsSync(join(dir, 'host.lock')), true); assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-'))); }); test('D2: bus recover refuses inside an agent run before it reads anything', async () => { const f = await fixture(), dir = join(f.org.root, 'bus-host'); mkdirSync(dir, { 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); await assert.rejects(f.callWith({ MOSAIC_AGENT_RUN: '1' }, 'bus', 'recover', 'alpha'), e => e.code === 3 && /outside any agent run/.test(e.stderr)); assert.equal(existsSync(join(dir, 'host.json')), true); assert.ok(!readdirSync(dir).some(x => x.startsWith('recovery-'))); }); 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); });