import test from 'node:test';
import assert from 'node:assert/strict';
import { existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs';
import { createRequire } from 'node:module';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { assertProcessing, initializeControl, readControl, transition } from '../src/state.mjs';
const api = await import('../src/containers.mjs').catch(e => { if (e.code === 'ERR_MODULE_NOT_FOUND') return {}; throw e; });
const { parse: parseYaml } = createRequire(import.meta.url)('yaml');
function fixture(t) { const root = mkdtempSync(join(tmpdir(), 'processing-container-')); initializeControl(root); t.after(() => rmSync(root, { recursive: true, force: true })); return root; }
const actor = { business: 'demo', human: 'jason', halted: true };

test('static Compose service has no data-root or control-directory mount', () => {
  const compose = parseYaml(readFileSync(new URL('../../../compose.yaml', import.meta.url), 'utf8'));
  const volumes = compose.services['mosaic-agent'].volumes;
  const targets = volumes.map(volume => {
    if (typeof volume !== 'string') return volume.target;
    const searchFrom = volume.startsWith('${') ? volume.indexOf('}') + 1 : 0;
    const sourceEnd = volume.indexOf(':', searchFrom);
    return volume.slice(sourceEnd + 1).split(':')[0];
  });
  assert.deepEqual(targets, ['/home/node/.pi/agent/auth.json']);
});

test('source launch view maps the control directory so actual atomic state replacements stay path-visible', t => {
  const root = fixture(t), initial = readControl(root);
  const launch = api.workerLaunch(root, { kind: 'probe' });
  const control = launch.mounts.find(mount => mount.target === '/var/lib/mosaic/control');
  assert.deepEqual(control, {
    source: join(root, 'control'), target: '/var/lib/mosaic/control', readOnly: true,
  });
  const file = join(control.source, 'processing.json');
  const initialInode = statSync(file).ino;

  const halted = transition(root, { business: 'demo', human: 'jason', halted: true });
  assert.notEqual(statSync(file).ino, initialInode, 'the state writer replaces processing.json with a new inode');
  const { changed: haltChanged, ...publishedHalt } = halted;
  assert.equal(haltChanged, true);
  assert.deepEqual(JSON.parse(readFileSync(file, 'utf8')), publishedHalt);
  assert.throws(() => assertProcessing(root, initial.generation), error => error.code === 'processing-halted');

  const haltedInode = statSync(file).ino;
  const cleared = transition(root, { business: 'demo', human: 'jason', halted: false, generation: halted.generation });
  assert.notEqual(statSync(file).ino, haltedInode, 'clearing also publishes by atomic replacement');
  assert.equal(readControl(root).generation, cleared.generation);
  assert.throws(() => assertProcessing(root, initial.generation), error => error.code === 'processing-generation-changed');
});

function workerTree(t) {
  const root = fixture(t);
  const files = {
    task: join(root, 'runs', 'r-own', 'task.json'),
    mission: join(root, 'runs', 'r-own', 'mission.json'),
    evidence: join(root, 'runs', 'r-own', 'result.json'),
    siblingEvidence: join(root, 'runs', 'r-sibling', 'result.json'),
    workspace: join(root, 'workspaces', 'own'),
    siblingWorkspace: join(root, 'workspaces', 'sibling'),
    session: join(root, 'sessions', 'own'),
    forkSource: join(root, 'sessions', 'base', 'turn.jsonl'),
    forkSnapshot: join(root, 'runs', 'r-own', 'session-fork.jsonl'),
    foreignForkSnapshot: join(root, 'runs', 'r-sibling', 'session-fork.jsonl'),
    siblingSession: join(root, 'sessions', 'sibling'),
    missionInteractive: join(root, 'agent-missions', 'alice.json'),
    soul: join(root, 'agents', 'alice', 'SOUL.md'),
    skill: join(root, 'skills-enabled', 'review'),
    socket: join(root, 'bus', 'broker.sock'),
    unrelatedUserFile: join(root, 'user', 'private.bin'),
  };
  for (const dir of [
    join(root, 'runs', 'r-own'), join(root, 'runs', 'r-sibling'), files.workspace,
    files.siblingWorkspace, files.session, join(root, 'sessions', 'base'),
    files.siblingSession, join(root, 'user'), join(root, 'agent-missions'),
    join(root, 'agents', 'alice'), files.skill, join(root, 'bus'),
  ]) mkdirSync(dir, { recursive: true });
  for (const file of [files.task, files.mission, files.evidence, files.siblingEvidence,
    files.forkSource, files.forkSnapshot, files.foreignForkSnapshot,
    files.missionInteractive, files.soul, files.socket]) writeFileSync(file, '{}\n');
  writeFileSync(join(root, 'user', 'USER.md'), '# User\n');
  writeFileSync(files.unrelatedUserFile, 'not prompt context\n');
  writeFileSync(join(files.skill, 'SKILL.md'), '# Review\n');
  return { root, files };
}

test('task worker launch mounts only selected inputs RO and selected workspace/session RW', t => {
  assert.equal(typeof api.workerLaunch, 'function');
  const { root, files } = workerTree(t);
  const launch = api.workerLaunch(root, {
    kind: 'task', taskFile: files.task, missionFile: files.mission,
    workspaceDir: files.workspace, sessionDir: files.session, sessionForkFile: files.forkSnapshot,
  });
  const byTarget = Object.fromEntries(launch.mounts.map(m => [m.target, m]));
  assert.deepEqual(Object.keys(byTarget).sort(), [
    '/var/lib/mosaic/control', '/var/lib/mosaic/input/mission.json',
    '/var/lib/mosaic/input/session-fork.jsonl', '/var/lib/mosaic/input/task.json',
    '/var/lib/mosaic/input/user/USER.md', '/var/lib/mosaic/session', '/workspace',
  ].sort());
  for (const target of ['/var/lib/mosaic/control', '/var/lib/mosaic/input/mission.json',
    '/var/lib/mosaic/input/session-fork.jsonl', '/var/lib/mosaic/input/task.json', '/var/lib/mosaic/input/user/USER.md']) {
    assert.equal(byTarget[target].readOnly, true, target);
  }
  assert.equal(byTarget['/workspace'].readOnly, false);
  assert.equal(byTarget['/var/lib/mosaic/session'].readOnly, false);
  assert.equal(launch.environment.MOSAIC_TASK_FILE, '/var/lib/mosaic/input/task.json');
  assert.equal(launch.environment.MOSAIC_MISSION_FILE, '/var/lib/mosaic/input/mission.json');
  assert.equal(launch.environment.MOSAIC_WORKSPACE, '/workspace');
  assert.equal(launch.environment.MOSAIC_SESSION_DIR, '/var/lib/mosaic/session');
  assert.equal(launch.environment.MOSAIC_SESSION_FORK, '/var/lib/mosaic/input/session-fork.jsonl');
  assert.equal(byTarget['/var/lib/mosaic/input/session-fork.jsonl'].source, files.forkSnapshot);
  assert.equal(launch.environment.MOSAIC_SYSTEM_PROMPT_FILE, '/var/lib/mosaic/private/system-prompt.md');
  const sources = launch.mounts.map(m => m.source);
  assert.equal(sources.includes(root), false);
  for (const hidden of [files.evidence, files.siblingEvidence, files.siblingWorkspace, files.siblingSession,
    files.forkSource, files.foreignForkSnapshot, files.socket]) {
    assert.equal(sources.includes(hidden), false, hidden);
  }
  assert.equal(sources.includes(files.unrelatedUserFile), false);
});

test('task fork slot refuses a shared session file or another run snapshot', t => {
  const { root, files } = workerTree(t);
  const input = { kind: 'task', taskFile: files.task, sessionDir: files.session };
  assert.throws(() => api.workerLaunch(root, { ...input, sessionForkFile: files.forkSource }), /sessionForkFile/);
  assert.throws(() => api.workerLaunch(root, { ...input, sessionForkFile: files.foreignForkSnapshot }), /sessionForkFile/);
});

test('probe refuses symlinked or non-file user markdown and colon mount sources', t => {
  for (const [name, arrange, expected] of [
    ['symlinked user directory', root => {
      mkdirSync(join(root, 'actual-user')); symlinkSync(join(root, 'actual-user'), join(root, 'user'));
    }, /user context directory.*non-symbolic-link/],
    ['symlinked markdown', root => {
      mkdirSync(join(root, 'user')); writeFileSync(join(root, 'context.md'), '# context\n');
      symlinkSync(join(root, 'context.md'), join(root, 'user', 'USER.md'));
    }, /user context file.*non-symbolic-link/],
    ['markdown-named directory', root => {
      mkdirSync(join(root, 'user', 'USER.md'), { recursive: true });
    }, /user context file.*non-symbolic-link/],
    ['colon source', root => {
      mkdirSync(join(root, 'user')); writeFileSync(join(root, 'user', 'USER:.md'), '# context\n');
    }, /mount source contains an unsupported colon/],
  ]) {
    const root = fixture(t);
    arrange(root);
    assert.throws(() => api.workerLaunch(root, { kind: 'probe' }), expected, name);
  }
});

test('interactive worker launch keeps role inputs RO and only its workspace/session RW', t => {
  assert.equal(typeof api.workerLaunch, 'function');
  const { root, files } = workerTree(t);
  const launch = api.workerLaunch(root, {
    kind: 'interactive', missionFile: files.missionInteractive, agentSoulFile: files.soul,
    skillDirs: [files.skill], workspaceDir: files.workspace, sessionDir: files.session,
  });
  const byTarget = Object.fromEntries(launch.mounts.map(m => [m.target, m]));
  assert.equal(byTarget['/var/lib/mosaic/input/mission.json'].readOnly, true);
  assert.equal(byTarget['/var/lib/mosaic/input/agent-soul.md'].readOnly, true);
  assert.equal(byTarget['/var/lib/mosaic/input/skills/review'].readOnly, true);
  assert.equal(byTarget['/var/lib/mosaic/input/user/USER.md'].readOnly, true);
  assert.equal(byTarget['/workspace'].readOnly, false);
  assert.equal(byTarget['/var/lib/mosaic/session'].readOnly, false);
  assert.equal(launch.environment.MOSAIC_AGENT_SOUL_FILE, '/var/lib/mosaic/input/agent-soul.md');
  assert.equal(launch.environment.MOSAIC_SKILLS, '/var/lib/mosaic/input/skills/review');
  assert.equal(launch.environment.MOSAIC_TASK_FILE, '');
  assert.equal(launch.environment.MOSAIC_SESSION_FORK, '');
});

test('unsupported or incomplete worker launch refuses before runtime registration', t => {
  assert.equal(typeof api.workerLaunch, 'function');
  const { root, files } = workerTree(t);
  assert.throws(() => api.workerLaunch(root, { kind: 'task' }), /taskFile/);
  assert.throws(() => api.workerLaunch(root, { kind: 'probe', brokerSocket: files.socket }), /unsupported worker launch key/);
  assert.throws(() => api.workerLaunch(root, { kind: 'interactive', workspaceDir: files.workspace }), /sessionDir/);
  const runtimes = join(root, 'control', 'runtimes');
  assert.throws(() => api.runContainer(root, { kind: 'task', business: 'instance', run: 'missing' }, {}), /worker launch/);
  assert.equal(existsSync(runtimes) ? readdirSync(runtimes).length : 0, 0);
});
test('container admission publishes its generated name and root/runtime labels before dispatch', t => {
  assert.equal(typeof api.prepareContainer, 'function');
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'run-1' });
  const args = api.containerArgs(r, { args: ['request'], worker: api.workerLaunch(root, { kind: 'probe' }) });
  assert.match(r.containerName, /^mosaic-[a-f0-9]{12}-[a-f0-9-]{36}$/);
  assert.ok(args.includes(r.containerName)); assert.ok(args.includes(`mosaic.root=${r.root}`));
  assert.ok(args.includes(`mosaic.runtime=${r.id}`)); assert.ok(args.includes(`MOSAIC_PROCESSING_GENERATION=${r.generation}`));
  assert.deepEqual(args.slice(-2), ['mosaic-agent', 'request']);
  transition(root, actor); assert.throws(() => api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'new' }), e => e.code === 'processing-halted');
});
function inspected(r, running = true, id = 'a'.repeat(64)) {
  return { Id: id, Name: '/' + r.containerName, Config: { Labels: { 'mosaic.root': r.root, 'mosaic.runtime': r.id } }, State: { Running: running } };
}
test('container sweep sends TERM then KILL only after verifying exact id and both labels', async t => {
  assert.equal(typeof api.createContainerStopper, 'function');
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
  const calls = []; let running = true;
  const stop = api.createContainerStopper({ docker: async args => {
    calls.push(args);
    if (args[0] === 'inspect') return JSON.stringify(inspected(r, running));
    if (args.includes('KILL')) running = false;
    return '';
  } });
  const result = await stop(r, { graceMs: 30, killWaitMs: 100 });
  assert.equal(result.outcome, 'killed'); assert.deepEqual(result.signals, ['SIGTERM', 'SIGKILL']);
  assert.deepEqual(calls.filter(c => c[0] === 'kill'), [['kill', '--signal', 'TERM', 'a'.repeat(64)], ['kill', '--signal', 'KILL', 'a'.repeat(64)]]);
});
test('foreign labels and a replaced container id stay unknown and are never signalled', async t => {
  assert.equal(typeof api.createContainerStopper, 'function');
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
  const calls = [];
  const stop = api.createContainerStopper({ docker: async args => {
    calls.push(args); const doc = inspected(r); doc.Config.Labels['mosaic.root'] = 'b'.repeat(64); return JSON.stringify(doc);
  } });
  assert.equal((await stop(r, { graceMs: 1, killWaitMs: 1 })).outcome, 'unknown');
  assert.equal(calls.filter(c => c[0] === 'kill').length, 0);
  let reads = 0; calls.length = 0;
  const replacement = api.createContainerStopper({ docker: async args => {
    calls.push(args); if (args[0] !== 'inspect') return '';
    return JSON.stringify(inspected(r, true, (++reads > 1 ? 'b' : 'a').repeat(64)));
  } });
  assert.equal((await replacement(r, { graceMs: 1, killWaitMs: 1 })).outcome, 'unknown');
  assert.equal(calls.filter(c => c[0] === 'kill').length, 0);
});
test('an unverified missing container remains unknown rather than claiming a pending start ended', async t => {
  assert.equal(typeof api.createContainerStopper, 'function');
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'container', business: 'instance', run: 'one' });
  const stop = api.createContainerStopper({ docker: async () => { throw new Error('no container'); } });
  assert.equal((await stop(r, { graceMs: 5, killWaitMs: 5 })).outcome, 'unknown');
});

test('a removed --rm container is settled only by its trusted owner completion receipt', async t => {
  const { updateRuntime } = await import('../src/runtimes.mjs');
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
  let removed = false;
  const stop = api.createContainerStopper({ dataRoot: root, docker: async args => {
    if (args[0] === 'kill') { removed = true; updateRuntime(root, r.id, { phase: 'ended' }); return ''; }
    if (removed) throw Error('removed by --rm');
    return JSON.stringify(inspected(r));
  } });
  const result = await stop(r, { graceMs: 30, killWaitMs: 100 });
  assert.equal(result.outcome, 'terminated'); assert.deepEqual(result.signals, ['SIGTERM']);
});

test('after signalling a verified container, successful empty Docker listing proves --rm removed that id', async t => {
  const root = fixture(t), r = api.prepareContainer(root, { kind: 'task', business: 'instance', run: 'one' });
  let removed = false;
  const stop = api.createContainerStopper({ docker: async args => {
    if (args[0] === 'kill') { removed = true; return ''; }
    if (args[0] === 'ps') return '';
    if (removed) throw Error('removed');
    return JSON.stringify(inspected(r));
  } });
  const result = await stop(r, { graceMs: 20, killWaitMs: 100 });
  assert.equal(result.outcome, 'terminated'); assert.deepEqual(result.signals, ['SIGTERM']);
});