/**
 * Worker registry, reconciliation on startup (§10.7, R-CTRL-31/32), and the
 * concurrency queue (R-CONC-1…7).
 *
 * The registry, not the process, is the unit of control (R-CONC-3): a queued run
 * has a directory and a status and is steerable and stoppable before it ever
 * spawns.
 */

import * as fs from "node:fs";
import * as path from "node:path";
import { getProcessStartIdentity, isProcessAlive, proveDeath } from "../proof-of-death.ts";
import * as os from "node:os";
import {
	type RunPaths,
	type RunStatus,
	isCompletionOutcomeState,
	isTerminalState,
	listRunIds,
	readFileIfExists,
	readStatus,
	runPaths,
	validateRegistryRunId,
	writeClosed,
	writeStatus,
} from "./status.ts";

export interface RegistryEntry {
	runId: string;
	paths: RunPaths;
	status: RunStatus;
	/** Set when this session drives the child directly (not adopted). */
	live?: boolean;
}

export function runsDir(cwd: string): string {
	return path.join(cwd, ".pi", "agi", ".runtime", "runs");
}

export function agentsDir(cwd: string): string {
	return path.join(cwd, ".pi", "agi", ".runtime", "agents");
}

function agentTarget(cwd: string, runId: string): string {
	return path.relative(agentsDir(cwd), path.join(runsDir(cwd), runId));
}

/** Atomically reserves a permanent public name for a new lineage. */
export function claimAgentName(cwd: string, name: string, runId: string): void {
	const dir = agentsDir(cwd);
	fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
	try {
		fs.symlinkSync(agentTarget(cwd, runId), path.join(dir, name), "dir");
	} catch (error) {
		if ((error as NodeJS.ErrnoException).code === "EEXIST") {
			throw new Error(`Agent '${name}' already exists. Resume it, or choose a different name for a fresh agent.`);
		}
		throw error;
	}
}

/** Point a lineage at its newest attempt without exposing the attempt id. */
export function updateAgentName(cwd: string, name: string, runId: string): void {
	const dir = agentsDir(cwd);
	fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
	const link = path.join(dir, name);
	const temporary = path.join(dir, `.${name}.${process.pid}.${Date.now()}`);
	fs.symlinkSync(agentTarget(cwd, runId), temporary, "dir");
	try {
		fs.renameSync(temporary, link);
	} catch (error) {
		try { fs.unlinkSync(temporary); } catch {}
		throw error;
	}
}

/** Carry the readable lineage history into a resumed private attempt. */
export function inheritAgentHistory(previous: RunPaths, next: RunPaths): void {
	for (const source of [`${previous.trace}.1`, previous.trace]) {
		try {
			const content = fs.readFileSync(source);
			if (content.length > 0) fs.appendFileSync(next.trace, content, { mode: 0o600 });
		} catch {}
	}
	try {
		for (const file of fs.readdirSync(previous.commands)) {
			if (!/^\d+\.log$/u.test(file)) continue;
			fs.copyFileSync(path.join(previous.commands, file), path.join(next.commands, file), fs.constants.COPYFILE_EXCL);
		}
	} catch {}
	try {
		fs.copyFileSync(previous.result, next.result);
	} catch {}
}

export function loadAllRuns(cwd: string): { entries: RegistryEntry[]; problems: Array<{ runId: string; reason: string }> } {
	const dir = runsDir(cwd);
	const entries: RegistryEntry[] = [];
	const problems: Array<{ runId: string; reason: string }> = [];
	for (const runId of listRunIds(dir)) {
		try {
			validateRegistryRunId(runId);
		} catch (error) {
			problems.push({ runId, reason: `invalid registry run directory: ${(error as Error).message}` });
			continue;
		}
		const paths = runPaths(dir, runId);
		const result = readStatus(paths);
		if (result.status === undefined) {
			// R-EXEC-9: unreadable or future-schema is surfaced as `unknown` and left
			// alone. Guessing at a run we cannot read is how a live worker gets killed.
			if (result.problem !== undefined) problems.push({ runId, reason: result.problem });
			continue;
		}
		entries.push({ runId, paths, status: result.status });
	}
	return { entries, problems };
}

function compareAttempts(left: RegistryEntry, right: RegistryEntry): number {
	return right.status.runIndex - left.status.runIndex || right.status.createdAt.localeCompare(left.status.createdAt) || right.runId.localeCompare(left.runId);
}

export function runsForName(entries: RegistryEntry[], name: string): RegistryEntry[] {
	return entries.filter((entry) => entry.status.name === name).sort(compareAttempts);
}

/** Resolve the active attempt, otherwise the newest durable attempt, for one named lineage. */
export function latestRunForName(entries: RegistryEntry[], name: string): RegistryEntry {
	const matches = runsForName(entries, name);
	if (matches.length === 0) {
		const known = [...new Set(entries.map((entry) => entry.status.name))].sort();
		throw new Error(`No agent named '${name}'. Known agents: ${known.length === 0 ? "(none)" : known.join(", ")}.`);
	}
	const active = activeRunEntries(matches);
	if (active.length > 1) throw new Error(`Agent lineage '${name}' is corrupt: multiple live attempts exist.`);
	return active[0] ?? (matches[0] as RegistryEntry);
}

/** One model/UI row per named lineage. */
export function latestRunEntries(entries: RegistryEntry[]): RegistryEntry[] {
	const byName = new Map<string, RegistryEntry[]>();
	for (const entry of entries) byName.set(entry.status.name, [...(byName.get(entry.status.name) ?? []), entry]);
	return [...byName.values()].map((group) => latestRunForName(group, group[0]?.status.name ?? "")).sort((a, b) => a.status.name.localeCompare(b.status.name));
}

export interface ReconcileOutcome {
	adopted: RegistryEntry[];
	orphaned: RegistryEntry[];
	unconsumed: RegistryEntry[];
	unknown: Array<{ runId: string; reason: string }>;
}

/**
 * §10.7. For every non-terminal run directory, decide adopt vs orphan by positive
 * proof of death, then collect the unconsumed terminal set.
 *
 * R-CTRL-31: alive means adopt. A detached worker that survived the restart is
 * still doing useful work, and killing it discards that work; the price is the
 * loss of the stdin fast path, which is recorded as `detached: true`.
 */
export function reconcile(
	cwd: string,
	options: { now?: number; hostname?: string; owner?: RunStatus["owner"] } = {},
): ReconcileOutcome {
	const { entries, problems } = loadAllRuns(cwd);
	const outcome: ReconcileOutcome = { adopted: [], orphaned: [], unconsumed: [], unknown: problems };
	const hostname = options.hostname ?? os.hostname();

	for (const entry of entries) {
		const status = entry.status;
		if (isTerminalState(status.state)) {
			// Ordinary worker outcomes remain unread until their result/event verdict is
			// inspected. Orphaned and unreadable records use the separate recovery delta.
			if (isCompletionOutcomeState(status.state) && !status.resultConsumed) outcome.unconsumed.push(entry);
			continue;
		}
		if (status.state === "paused") {
			// Interrupt deliberately ends the process while preserving its session.
			// Reclassifying the dead pid as orphaned on restart destroys resumability.
			continue;
		}
		if (status.state === "queued") {
			// A run that never spawned cannot be adopted and did not die: its owner is
			// gone, so it is failed rather than left in a queue nobody drains.
			const next: RunStatus = {
				...status,
				state: "failed",
				endedAt: new Date(options.now ?? Date.now()).toISOString(),
				error: "queued when the orchestrator restarted; never spawned",
			};
			persist(entry, next);
			outcome.orphaned.push({ ...entry, status: next });
			continue;
		}

		const pid = status.pid;
		if (pid === null || pid <= 0) {
			const next: RunStatus = {
				...status,
				state: "orphaned",
				endedAt: new Date(options.now ?? Date.now()).toISOString(),
				error: "orchestrator restarted while this run was in flight",
			};
			persist(entry, next);
			writeClosed(entry.paths, next);
			outcome.orphaned.push({ ...entry, status: next });
			continue;
		}

		// P12, reused rather than rewritten: `proveDeath` is Phase 2's tested
		// implementation of R-CTRL-27 (hostname match, ESRCH or a differing start
		// identity, identity availability). A second liveness check here would be a
		// second chance to get pid reuse wrong.
		//
		// BUG (found by a reconciliation test): the owner passed here previously had no
		// real hostname, so R-CTRL-27's cross-host guard could never fire and a run
		// recorded on another machine was declared dead — its pid is simply absent from
		// this kernel. `status.hostname` now carries it. A status written before that
		// field existed falls back to this host, which keeps the old (local) behaviour
		// rather than silently adopting every legacy run forever.
		const alive = isProcessAlive(pid);
		const owner = {
			pid,
			processStartIdentity: status.processStartIdentity ?? undefined,
			sessionId: status.runId,
			hostname: typeof status.hostname === "string" && status.hostname.length > 0 ? status.hostname : hostname,
			claimedAt: status.startedAt ?? status.createdAt,
		};
		const verdict = proveDeath(owner, {
			hostname,
			alive,
			currentIdentity: alive ? getProcessStartIdentity(pid) : undefined,
			now: options.now,
		});

		if (!verdict.reclaimable) {
			// Alive (or not provably dead): adopt. stdin is gone with the old pipe, so
			// control is filesystem-only for this run.
			//
			// Adoption also transfers R-CTRL-18 ownership: the previous orchestrator is
			// not coming back for this run, and whoever adopts it has to be able to stop
			// it on shutdown. Left with its old owner it would be permanently out of
			// scope for every bulk control path.
			const next: RunStatus = {
				...status,
				detached: true,
				...(options.owner === undefined ? {} : { owner: options.owner }),
			};
			persist(entry, next);
			outcome.adopted.push({ ...entry, status: next });
			continue;
		}

		// Provably dead. New runs project the provider verdict and final-message
		// semantics into status.json on every assistant message, so a stale partial
		// result.md cannot rescue a provider failure after restart. Old status records
		// without those additive fields retain the legacy result-file fallback.
		const resultText = readFileIfExists(entry.paths.result);
		const providerFailed = status.providerStopReason === "error" || status.providerStopReason === "aborted";
		const durableFinal = status.finalAssistantTextPresent;
		const legacyFinal = durableFinal === undefined && resultText !== undefined && resultText.trim().length > 0;
		const hasFinal = durableFinal === true || legacyFinal;
		const endedAt = new Date(options.now ?? Date.now()).toISOString();
		const next: RunStatus =
			status.settled === true && providerFailed
				? {
						...status,
						state: "failed",
						endedAt,
						error: `worker model ended with ${status.providerStopReason}${status.providerError === null || status.providerError === undefined ? "" : `: ${status.providerError}`}`,
					}
				: status.settled === true && hasFinal
					? {
							...status,
							state: "complete",
							endedAt,
							resultPath: resultText === undefined ? status.resultPath : entry.paths.result,
							resultConsumed: false,
						}
					: status.settled === true && durableFinal === false
						? {
								...status,
								state: "failed",
								endedAt,
								error: "worker completed without producing a final report",
							}
				: {
						...status,
						state: "orphaned",
						endedAt,
						error: "orchestrator restarted while this run was in flight",
					};
		persist(entry, next);
		writeClosed(entry.paths, next);
		if (isCompletionOutcomeState(next.state)) outcome.unconsumed.push({ ...entry, status: next });
		else outcome.orphaned.push({ ...entry, status: next });
	}

	// Keep unread terminal outcomes deterministic for completion-wake projection.
	outcome.unconsumed.sort((a, b) => a.runId.localeCompare(b.runId));
	return outcome;
}

function persist(entry: RegistryEntry, status: RunStatus): void {
	try {
		writeStatus(entry.paths, status);
	} catch {
		// E37: a status write we cannot complete must not abort reconciliation of the
		// other runs.
	}
	entry.status = status;
}

/** R-CTRL-32 / Issue 7. Minimal per-run recovery deltas not covered by completion wakes. */
export function recoveryBlock(outcome: ReconcileOutcome): string | undefined {
	const lines: string[] = [];
	for (const entry of outcome.adopted) {
		lines.push(
			`Agent ${entry.status.name} is still active and adopted after restart. ` +
				`Inspect it with agi_worker({name:"${entry.status.name}", view:"status"}).`,
		);
	}
	for (const entry of outcome.orphaned) {
		lines.push(
			`Agent ${entry.status.name} is orphaned after restart. ` +
				`Inspect .pi/agi/.runtime/agents/${entry.status.name}/trace.log and its status.`,
		);
	}
	for (const _problem of outcome.unknown) {
		lines.push("A private worker record could not be reconciled after restart. Inspect runtime diagnostics before deciding what to do.");
	}
	if (lines.length === 0) return undefined;
	return `# AGI recovery delta\n\n${lines.join("\n")}`;
}

export function elapsed(status: RunStatus, now = Date.now()): string {
	const start = Date.parse(status.startedAt ?? status.createdAt);
	if (Number.isNaN(start)) return "?";
	const end = status.endedAt === null ? now : Date.parse(status.endedAt);
	const ms = Math.max(0, (Number.isNaN(end) ? now : end) - start);
	const minutes = Math.floor(ms / 60000);
	if (minutes < 1) return `${Math.floor(ms / 1000)}s`;
	if (minutes < 60) return `${minutes}m`;
	return `${Math.floor(minutes / 60)}h${minutes % 60}m`;
}

/**
 * Return the current non-terminal run records shown to operators and the model.
 *
 * Resume creates a successor record so the previous checkpoint remains
 * inspectable. Once that successor exists, the paused predecessor is history,
 * not another live worker.
 */
export function activeRunEntries(entries: RegistryEntry[]): RegistryEntry[] {
	const resumed = new Set(
		entries.map((entry) => entry.status.previousRunId).filter((runId): runId is string => runId !== undefined),
	);
	return entries.filter(
		(entry) => !isTerminalState(entry.status.state) && !(entry.status.state === "paused" && resumed.has(entry.runId)),
	);
}

/** R-CONC-1. Every live worker consumes the same global worker slot. */
export function countActive(entries: RegistryEntry[]): number {
	let active = 0;
	for (const entry of activeRunEntries(entries)) {
		const state = entry.status.state;
		// `paused` is not terminal and its process is still alive — an interrupted
		// worker still holds its working tree. Counting only spawning/running let a
		// second worker into the same worktree at maxConcurrentWorkers: 1,
		// which is exactly the silent corruption R-CONC-1 exists to prevent.
		if (state !== "spawning" && state !== "running" && state !== "paused") continue;
		active += 1;
	}
	return active;
}

export function hasSlot(
	entries: RegistryEntry[],
	_readOnly: boolean,
	config: { maxConcurrentWorkers: number },
): boolean {
	return countActive(entries) < config.maxConcurrentWorkers;
}

/** Keep predecessor links visible while excluding the run being resumed itself. */
export function accountingEntriesForResume(entries: RegistryEntry[], runId: string): RegistryEntry[] {
	return entries.map((entry) => entry.runId === runId
		? { ...entry, status: { ...entry.status, state: "complete" } }
		: entry);
}

/**
 * R-CONC-14: at most one live run per name. Two workers on one task waste one.
 *
 * `paused` counts as live: the process still exists and can be resumed, so a second
 * run for the same task would put two workers on it.
 */
export function liveRunForName(entries: RegistryEntry[], name: string): RegistryEntry | undefined {
	return activeRunEntries(entries).find((entry) => entry.status.name === name);
}

export function gitignoreRuntime(cwd: string): void {
	// R-EXEC-8: `.runtime/` must be gitignored. Everything else under .pi/agi/ is
	// intentionally committable, so the ignore file lives inside .pi/agi/.
	const dir = path.join(cwd, ".pi", "agi");
	const file = path.join(dir, ".gitignore");
	try {
		if (readFileIfExists(file) === undefined) {
			fs.mkdirSync(dir, { recursive: true });
			fs.writeFileSync(file, ".runtime/\n", "utf8");
		}
	} catch {
		// A missing ignore file is untracked noise, never a failure.
	}
}