import { randomUUID } from 'node:crypto';
import { spawnSync, execFile } from 'node:child_process';
import { existsSync, lstatSync, readdirSync, realpathSync } from 'node:fs';
import { isAbsolute, join, relative, resolve } from 'node:path';
import { promisify } from 'node:util';
import { initializeControl, assertProcessing } from './state.mjs';
import { registerRuntime, updateRuntime, runtime, rootTag } from './runtimes.mjs';
const execute = promisify(execFile);
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
const WORKER_KEYS = new Set([
  'kind', 'taskFile', 'missionFile', 'workspaceDir', 'sessionDir',
  'sessionForkFile', 'agentSoulFile', 'skillDirs',
]);
const WORKER_ENV = Object.freeze({
  MOSAIC_SYSTEM_PROMPT_FILE: '/var/lib/mosaic/private/system-prompt.md',
  MOSAIC_TASK_FILE: '',
  MOSAIC_USER_DIR: '',
  MOSAIC_MISSION_FILE: '',
  MOSAIC_AGENT_SOUL_FILE: '',
  MOSAIC_WORKSPACE: '',
  MOSAIC_SESSION_DIR: '',
  MOSAIC_SESSION_FORK: '',
  MOSAIC_SKILLS: '',
});

function workerError(message) {
  const error = new Error(`worker launch: ${message}`);
  error.code = 'processing-worker-launch-invalid';
  return error;
}

function components(root, source, what, type) {
  if (typeof source !== 'string' || !isAbsolute(source)) throw workerError(`${what} must be an absolute path`);
  let stat;
  try { stat = lstatSync(source); }
  catch { throw workerError(`${what} is missing: ${source}`); }
  if (stat.isSymbolicLink() || (type === 'file' ? !stat.isFile() : !stat.isDirectory())) {
    throw workerError(`${what} must be a non-symbolic-link ${type}: ${source}`);
  }
  const canonical = realpathSync(source);
  const rel = relative(root, canonical);
  if (rel === '' || rel === '..' || rel.startsWith('../') || isAbsolute(rel)) {
    throw workerError(`${what} must be below the configured data root: ${source}`);
  }
  return { source: canonical, parts: rel.split('/'), stat };
}

function exactParts(found, expected, what) {
  if (found.parts.length !== expected.length || expected.some((part, i) => part !== null && found.parts[i] !== part)) {
    throw workerError(`${what} is outside its allowed data-root path: ${found.source}`);
  }
  return found;
}

function mount(source, target, readOnly) {
  if (source.includes(':')) throw workerError(`mount source contains an unsupported colon: ${source}`);
  return Object.freeze({ source, target, readOnly });
}

// Pure legacy-worker launch contract. Every host source occupies a typed slot
// with one fixed container target; arbitrary mounts and the data-root root are
// not accepted. The returned prompt path has no host source and is private to
// this one disposable container filesystem.
export function workerLaunch(dataRoot, input) {
  if (!input || typeof input !== 'object' || Array.isArray(input)) throw workerError('worker launch input is required');
  for (const key of Object.keys(input)) if (!WORKER_KEYS.has(key)) throw workerError(`unsupported worker launch key: ${key}`);
  if (!['task', 'interactive', 'probe'].includes(input.kind)) throw workerError(`unsupported worker launch kind: ${JSON.stringify(input.kind)}`);
  const root = realpathSync(resolve(dataRoot));
  const environment = { ...WORKER_ENV };
  const mounts = [];
  const add = (source, target, readOnly = true) => mounts.push(mount(source, target, readOnly));

  // Keep the host-owned directory as the read-only view: state publication
  // renames a fresh inode over processing.json, which a file bind would pin.
  // This package produces only processing.json, transient .state-* files, and
  // runtimes/, stops/, run-stops/ metadata here; sockets live outside control/.
  const control = exactParts(components(root, join(root, 'control'), 'control directory', 'directory'),
    ['control'], 'control directory');
  add(control.source, '/var/lib/mosaic/control');

  const userDir = join(root, 'user');
  if (existsSync(userDir)) {
    const user = exactParts(components(root, userDir, 'user context directory', 'directory'), ['user'], 'user context directory');
    for (const entry of readdirSync(user.source, { withFileTypes: true }).filter(e => e.name.endsWith('.md')).sort((a, b) => a.name.localeCompare(b.name))) {
      const file = exactParts(components(root, join(user.source, entry.name), 'user context file', 'file'), ['user', entry.name], 'user context file');
      add(file.source, `/var/lib/mosaic/input/user/${entry.name}`);
    }
    environment.MOSAIC_USER_DIR = '/var/lib/mosaic/input/user';
  }

  if (input.kind === 'probe') {
    for (const key of WORKER_KEYS) if (key !== 'kind' && input[key] !== undefined) throw workerError(`${key} is unsupported for probe launches`);
    return Object.freeze({ mounts: Object.freeze(mounts), environment: Object.freeze(environment) });
  }

  if (input.kind === 'task') {
    if (!input.taskFile) throw workerError('taskFile is required for task launches');
    if (input.agentSoulFile !== undefined || input.skillDirs !== undefined) throw workerError('role inputs are unsupported for task launches');
    const task = exactParts(components(root, input.taskFile, 'taskFile', 'file'), ['runs', null, 'task.json'], 'taskFile');
    add(task.source, '/var/lib/mosaic/input/task.json');
    environment.MOSAIC_TASK_FILE = '/var/lib/mosaic/input/task.json';
    if (input.missionFile) {
      const mission = exactParts(components(root, input.missionFile, 'missionFile', 'file'), ['runs', task.parts[1], 'mission.json'], 'missionFile');
      add(mission.source, '/var/lib/mosaic/input/mission.json');
      environment.MOSAIC_MISSION_FILE = '/var/lib/mosaic/input/mission.json';
    }
    if (input.workspaceDir) {
      const workspace = components(root, input.workspaceDir, 'workspaceDir', 'directory');
      const persistent = workspace.parts.length === 2 && workspace.parts[0] === 'workspaces';
      const perRun = workspace.parts.length === 3 && workspace.parts[0] === 'runs' && workspace.parts[1] === task.parts[1] && workspace.parts[2] === 'workspace';
      if (!persistent && !perRun) throw workerError(`workspaceDir is outside its allowed data-root path: ${workspace.source}`);
      add(workspace.source, '/workspace', false);
      environment.MOSAIC_WORKSPACE = '/workspace';
    }
    if (input.sessionDir) {
      const session = exactParts(components(root, input.sessionDir, 'sessionDir', 'directory'), ['sessions', null], 'sessionDir');
      add(session.source, '/var/lib/mosaic/session', false);
      environment.MOSAIC_SESSION_DIR = '/var/lib/mosaic/session';
    }
    if (input.sessionForkFile) {
      if (!input.sessionDir) throw workerError('sessionForkFile requires sessionDir');
      const fork = exactParts(components(root, input.sessionForkFile, 'sessionForkFile', 'file'),
        ['runs', task.parts[1], 'session-fork.jsonl'], 'sessionForkFile');
      add(fork.source, '/var/lib/mosaic/input/session-fork.jsonl');
      environment.MOSAIC_SESSION_FORK = '/var/lib/mosaic/input/session-fork.jsonl';
    }
  } else {
    if (input.taskFile !== undefined || input.sessionForkFile !== undefined) throw workerError('task inputs are unsupported for interactive launches');
    if (!input.workspaceDir) throw workerError('workspaceDir is required for interactive launches');
    if (!input.sessionDir) throw workerError('sessionDir is required for interactive launches');
    const workspace = exactParts(components(root, input.workspaceDir, 'workspaceDir', 'directory'), ['workspaces', null], 'workspaceDir');
    const session = exactParts(components(root, input.sessionDir, 'sessionDir', 'directory'), ['sessions', null], 'sessionDir');
    add(workspace.source, '/workspace', false);
    add(session.source, '/var/lib/mosaic/session', false);
    environment.MOSAIC_WORKSPACE = '/workspace';
    environment.MOSAIC_SESSION_DIR = '/var/lib/mosaic/session';
    if (input.missionFile) {
      const mission = exactParts(components(root, input.missionFile, 'missionFile', 'file'), ['agent-missions', null], 'missionFile');
      if (!mission.parts[1].endsWith('.json')) throw workerError(`missionFile must be an agent mission JSON file: ${mission.source}`);
      add(mission.source, '/var/lib/mosaic/input/mission.json');
      environment.MOSAIC_MISSION_FILE = '/var/lib/mosaic/input/mission.json';
    }
    if (input.agentSoulFile) {
      const soul = exactParts(components(root, input.agentSoulFile, 'agentSoulFile', 'file'), ['agents', null, 'SOUL.md'], 'agentSoulFile');
      add(soul.source, '/var/lib/mosaic/input/agent-soul.md');
      environment.MOSAIC_AGENT_SOUL_FILE = '/var/lib/mosaic/input/agent-soul.md';
    }
    const skills = input.skillDirs ?? [];
    if (!Array.isArray(skills)) throw workerError('skillDirs must be an array');
    const targets = [];
    for (const skillDir of skills) {
      const skill = exactParts(components(root, skillDir, 'skillDirs entry', 'directory'), ['skills-enabled', null], 'skillDirs entry');
      const target = `/var/lib/mosaic/input/skills/${skill.parts[1]}`;
      if (targets.includes(target)) throw workerError(`duplicate skill directory: ${skill.source}`);
      targets.push(target);
      add(skill.source, target);
    }
    environment.MOSAIC_SKILLS = targets.join(',');
  }
  return Object.freeze({ mounts: Object.freeze(mounts), environment: Object.freeze(environment) });
}
export function prepareContainer(dataRoot, record) {
  initializeControl(dataRoot);
  const state = assertProcessing(dataRoot);
  const containerName = `mosaic-${rootTag(dataRoot).slice(0, 12)}-${randomUUID()}`;
  const managed = registerRuntime(dataRoot, { ...record, containerName });
  assertProcessing(dataRoot, state.generation);
  return managed;
}
export function containerArgs(record, { tty = false, args = [], worker } = {}) {
  if (!worker || !Array.isArray(worker.mounts) || !worker.environment) throw workerError('validated worker launch is required');
  const volumes = worker.mounts.flatMap(({ source, target, readOnly }) => ['--volume', `${source}:${target}:${readOnly ? 'ro' : 'rw'}`]);
  const environment = Object.entries(worker.environment).flatMap(([key, value]) => ['-e', `${key}=${value}`]);
  return ['compose', 'run', '--rm', ...(!tty ? ['-T'] : []), '--name', record.containerName,
    '--label', `mosaic.root=${record.root}`, '--label', `mosaic.runtime=${record.id}`,
    '-e', `MOSAIC_PROCESSING_GENERATION=${record.generation}`, ...environment, ...volumes, 'mosaic-agent', ...args];
}
export function runContainer(dataRoot, record, { tty = false, args = [], worker, ...options } = {}) {
  const launch = workerLaunch(dataRoot, worker);
  const managed = prepareContainer(dataRoot, record);
  assertProcessing(dataRoot, managed.generation);
  // Keep this synchronous. The independent broker sweeps the prepublished name while this waits.
  const proc = spawnSync('docker', containerArgs(managed, { tty, args, worker: launch }), options);
  const current = runtime(dataRoot, managed.id);
  // An interrupted Docker client is not proof its container exited. Keep that receipt for inspection.
  if (!proc.error && proc.status === 0) updateRuntime(dataRoot, managed.id, { phase: 'ended' });
  return { proc, managed, haltGeneration: current.haltGeneration ?? null };
}
export function createContainerStopper({ dataRoot = null, docker = async args => {
  const r = await execute('docker', args, { encoding: 'utf8', timeout: 2000, maxBuffer: 131072 });
  return r.stdout;
} } = {}) {
  return async (record, { graceMs = 5000, killWaitMs = 2000 } = {}) => {
    const signals = [];
    let id = record.containerId ?? null, seen = false;
    const start = Date.now(), termEnd = start + graceMs, end = termEnd + killWaitMs;
    async function inspect(target) {
      const doc = JSON.parse(await docker(['inspect', '--format', '{{json .}}', target]));
      if (!doc || !/^[a-f0-9]{64}$/.test(doc.Id) || doc.Name !== '/' + record.containerName ||
          doc.Config?.Labels?.['mosaic.root'] !== record.root || doc.Config?.Labels?.['mosaic.runtime'] !== record.id ||
          typeof doc.State?.Running !== 'boolean' || (id && doc.Id !== id)) {
        const error = new Error('container identity mismatch'); error.code = 'identity-mismatch'; throw error;
      }
      if (!id) {
        id = doc.Id;
        if (dataRoot) updateRuntime(dataRoot, record.id, { containerId: id });
      }
      seen = true;
      return doc;
    }
    while (Date.now() <= end) {
      try {
        const doc = await inspect(id ?? record.containerName);
        if (!doc.State.Running) return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
        const next = !signals.length ? 'TERM' : Date.now() >= termEnd && !signals.includes('SIGKILL') ? 'KILL' : null;
        if (next) {
          // Revalidate immediately before each signal, targeting the immutable inspected id.
          const checked = await inspect(id);
          if (!checked.State.Running) continue;
          await docker(['kill', '--signal', next, id]);
          signals.push('SIG' + next);
        }
      } catch (e) {
        // A missing name can still be a queued create. Only a container observed before a signal
        // and then removed by --rm is known to have finished. Never signal on an inspect failure.
        if (e.code === 'identity-mismatch') return { outcome: 'unknown', signals };
        if (id && seen && signals.length) {
          try {
            const listed = await docker(['ps', '--all', '--no-trunc', '--filter', `id=${id}`, '--format', '{{.ID}}']);
            if (listed.trim() === '') return { outcome: signals.includes('SIGKILL') ? 'killed' : 'terminated', signals };
          } catch { /* a failed Docker request proves nothing about the workload */ }
        }
        if (dataRoot) {
          try {
            if (runtime(dataRoot, record.id).phase === 'ended')
              return { outcome: signals.includes('SIGKILL') ? 'killed' : signals.length ? 'terminated' : 'already-exited', signals };
          } catch { return { outcome: 'unknown', signals }; }
        }
      }
      await pause(Math.min(20, Math.max(1, end - Date.now())));
    }
    return { outcome: 'unknown', signals };
  };
}