import { spawn, spawnSync } from "node:child_process";
import { EventEmitter } from "node:events";
import { chmodSync, existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { homedir, tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type * as kernelBootstrap from "../src/core/kernel/bootstrap.js";
import { ReplKernelManager } from "../src/core/kernel/index.js";
import { ORPHAN_PROCESS_JOURNAL_ENV } from "../src/core/orphan-process-journal.js";

const ensureKernelPythonMock = vi.hoisted(() => vi.fn());

vi.mock("../src/core/kernel/bootstrap.js", async (importOriginal) => {
	const original = await importOriginal<typeof kernelBootstrap>();
	return { ...original, ensureKernelPython: ensureKernelPythonMock };
});

let tempDir = "";
const savedJournalPath = process.env[ORPHAN_PROCESS_JOURNAL_ENV];

function writeFakePython(script: string[]): string {
	const python = join(tempDir, "python");
	writeFileSync(python, script.join("\n"));
	chmodSync(python, 0o755);
	return python;
}

interface JournalRecord {
	pid: number;
	ownerPid: number;
	active: boolean;
}

function readJournalRecords(path: string): JournalRecord[] {
	return readFileSync(path, "utf8")
		.split("\n")
		.filter(Boolean)
		.map((line) => JSON.parse(line) as JournalRecord);
}

describe("repl kernel parent watchdog", () => {
	beforeEach(() => {
		tempDir = mkdtempSync(join(tmpdir(), "prime-agent-repl-watchdog-"));
	});

	afterEach(() => {
		ensureKernelPythonMock.mockReset();
		if (savedJournalPath === undefined) delete process.env[ORPHAN_PROCESS_JOURNAL_ENV];
		else process.env[ORPHAN_PROCESS_JOURNAL_ENV] = savedJournalPath;
		if (tempDir) {
			rmSync(tempDir, { recursive: true, force: true });
			tempDir = "";
		}
	});

	it("spawn sets PRIME_AGENT_KERNEL_OWNER_PID and journals the kernel pid", async () => {
		const envDump = join(tempDir, "kernel-env");
		const python = writeFakePython(["#!/bin/sh", `env > "${envDump}"`, "exit 42", ""]);
		const journalPath = join(tempDir, "orphans.jsonl");
		process.env[ORPHAN_PROCESS_JOURNAL_ENV] = journalPath;
		const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
		const manager = new ReplKernelManager({ python, cwd: tempDir });

		try {
			await expect(manager.execute("x")).rejects.toThrow(/Kernel exited before ready/);
		} finally {
			errorSpy.mockRestore();
			await manager.shutdown({ snapshot: true, drainHostRequests: true });
		}

		expect(readFileSync(envDump, "utf8")).toMatch(new RegExp(`^PRIME_AGENT_KERNEL_OWNER_PID=${process.pid}$`, "m"));

		// Self-exited child: the kill signals nothing, so the record must stay active.
		await vi.waitFor(() => {
			const records = readJournalRecords(journalPath);
			expect(records).toHaveLength(1);
			expect(records[0]?.ownerPid).toBe(process.pid);
			expect(records[0]?.active).toBe(true);
		});
	});

	it("leaves the journal record active when the kill signals nothing", async () => {
		const journalPath = join(tempDir, "orphans.jsonl");
		process.env[ORPHAN_PROCESS_JOURNAL_ENV] = journalPath;
		const manager = new ReplKernelManager({ python: "/nonexistent/python", cwd: tempDir });
		const internals = manager as unknown as {
			child?: { pid?: number; kill(signal: string): boolean };
			cleanupResources(): void;
		};

		try {
			internals.child = { pid: 999999, kill: () => false };
			internals.cleanupResources();

			const records = existsSync(journalPath) ? readJournalRecords(journalPath) : [];
			expect(records.some((r) => r.pid === 999999 && !r.active)).toBe(false);
		} finally {
			await manager.shutdown({ snapshot: true, drainHostRequests: true });
		}
	});

	it("writes the inactive journal record when a signaled kill proves ownership", async () => {
		const journalPath = join(tempDir, "orphans.jsonl");
		process.env[ORPHAN_PROCESS_JOURNAL_ENV] = journalPath;
		const manager = new ReplKernelManager({ python: "/nonexistent/python", cwd: tempDir });
		const internals = manager as unknown as {
			child?: { pid?: number; kill(signal: string): boolean };
			cleanupResources(signal?: NodeJS.Signals): void;
		};

		try {
			internals.child = { pid: 999999, kill: () => true };
			internals.cleanupResources("SIGKILL");

			const records = readJournalRecords(journalPath);
			expect(records.some((r) => r.pid === 999999 && !r.active)).toBe(true);
		} finally {
			await manager.shutdown({ snapshot: true, drainHostRequests: true });
		}
	});

	it("a stale doStart resumed after a concurrent teardown never touches the new kernel", async () => {
		const journalPath = join(tempDir, "orphans.jsonl");
		process.env[ORPHAN_PROCESS_JOURNAL_ENV] = journalPath;
		let resolvePython: (python: string) => void = () => {};
		ensureKernelPythonMock.mockReturnValue(
			new Promise<string>((resolve) => {
				resolvePython = resolve;
			}),
		);
		const manager = new ReplKernelManager({ cwd: tempDir });
		const internals = manager as unknown as {
			state: string;
			child?: { pid?: number; kill(signal: string): boolean };
		};
		const killB = vi.fn(() => true);
		const childB = { pid: 222222, kill: killB };

		try {
			// Start A blocks inside `await ensureKernelPython(...)`.
			const staleStart = manager.start();
			staleStart.catch(() => {});
			await vi.waitFor(() => expect(ensureKernelPythonMock).toHaveBeenCalledTimes(1));

			// A concurrent teardown bumps the start generation and brings up a new
			// running kernel B.
			await manager.kill();
			internals.state = "running";
			internals.child = childB;

			// The stale doStart resumes: it must fail without touching B or spawning.
			resolvePython("/nonexistent/python");
			await expect(staleStart).rejects.toThrow(/Kernel start superseded/);

			expect(internals.state).toBe("running");
			expect(internals.child).toBe(childB);
			expect(killB).not.toHaveBeenCalled();
			const records = existsSync(journalPath) ? readJournalRecords(journalPath) : [];
			expect(records.some((r) => r.pid === childB.pid)).toBe(false);
		} finally {
			internals.child = undefined;
			internals.state = "idle";
			await manager.shutdown({ snapshot: true, drainHostRequests: true });
		}
	});

	it("a shutdown superseded by a concurrent kill reports not-owner so recovery cannot resurrect to idle", async () => {
		const manager = new ReplKernelManager({ python: "/nonexistent-python", cwd: tmpdir() });
		const internals = manager as unknown as {
			state: string;
			child: unknown;
			writeLine: (request: Record<string, unknown>) => Promise<void>;
			shutdown(): Promise<boolean>;
			kill(): Promise<void>;
		};
		internals.state = "running";
		// A live child handle keeps waitForKernelExit parked so the send actually blocks the shutdown.
		const child = Object.assign(new EventEmitter(), {
			exitCode: null,
			signalCode: null,
			kill: () => true,
			pid: undefined,
			stdin: { destroyed: false, destroy: () => {} },
			stdout: { destroy: () => {} },
			stderr: { destroy: () => {} },
		});
		internals.child = child;
		let releaseSend: () => void = () => {};
		internals.writeLine = () =>
			new Promise<void>((resolve) => {
				releaseSend = resolve;
			});
		const shutdownResult = internals.shutdown();
		await new Promise((resolve) => setTimeout(resolve, 10)); // park in the stdin write
		await internals.kill(); // concurrent teardown wins ownership
		releaseSend();
		child.emit("exit", 0, null);
		expect(await shutdownResult).toBe(false); // recovery must not set idle
		expect(internals.state).toBe("shutdown");
	});

	it("a stale shutdown parked in its stdin-write await never cleans up a successor kernel", async () => {
		const journalPath = join(tempDir, "orphans.jsonl");
		process.env[ORPHAN_PROCESS_JOURNAL_ENV] = journalPath;
		let releaseSend: () => void = () => {};
		const parkedSend = new Promise<void>((resolve) => {
			releaseSend = resolve;
		});
		const killA = vi.fn(() => true);
		const killB = vi.fn(() => true);
		const childA = Object.assign(new EventEmitter(), {
			exitCode: null,
			signalCode: null,
			kill: killA,
			pid: 111111,
			stdin: { destroyed: false, destroy: () => {} },
			stdout: { destroy: () => {} },
			stderr: { destroy: () => {} },
		});
		const childB = Object.assign(new EventEmitter(), {
			exitCode: null,
			signalCode: null,
			kill: killB,
			pid: 222222,
			stdin: { destroyed: false, destroy: () => {} },
			stdout: { destroy: () => {} },
			stderr: { destroy: () => {} },
		});
		const manager = new ReplKernelManager({ python: "/nonexistent/python", cwd: tempDir });
		const internals = manager as unknown as {
			state: string;
			child?: unknown;
			writeLine: (request: Record<string, unknown>) => Promise<void>;
			startPromise?: Promise<void>;
		};

		try {
			// Kernel A is running with a stdin write that parks forever until
			// released — shutdown() will suspend inside its await window after
			// having synchronously set state = "shutdown".
			internals.state = "running";
			internals.child = childA;
			internals.writeLine = () => parkedSend;
			const staleShutdown = manager.shutdown();

			// While A's shutdown is parked, a concurrent teardown reclaims A and a
			// new start brings up kernel B.
			await manager.kill();
			internals.state = "running";
			internals.child = childB;
			const startPromiseB = Promise.resolve();
			internals.startPromise = startPromiseB;

			// A's stale shutdown resumes: it must not clean up B or clear B's start.
			releaseSend();
			await staleShutdown;

			expect(internals.state).toBe("running");
			expect(internals.child).toBe(childB);
			expect(killB).not.toHaveBeenCalled();
			expect(internals.startPromise).toBe(startPromiseB);
			// A was reclaimed by kill(); B must never gain an inactive journal record.
			expect(killA).toHaveBeenCalledWith("SIGKILL");
			const records = existsSync(journalPath) ? readJournalRecords(journalPath) : [];
			expect(records.some((r) => r.pid === childB.pid && !r.active)).toBe(false);
		} finally {
			internals.child = undefined;
			internals.startPromise = undefined;
			internals.state = "idle";
			await manager.shutdown({ snapshot: true, drainHostRequests: true });
		}
	});
});

function resolveReplPython(): string | null {
	const candidates = [
		process.env.PRIME_AGENT_KERNEL_PYTHON,
		resolve(__dirname, "..", "..", "..", "prime-agent-runtime", ".venv", "bin", "python"),
		join(homedir(), ".prime", "agent", "kernel-venv", "bin", "python"),
	].filter((p): p is string => Boolean(p));
	for (const python of candidates) {
		if (!existsSync(python)) continue;
		const check = spawnSync(python, ["-c", "import rlm.repl"], { encoding: "utf8" });
		if (check.status === 0) return python;
	}
	return null;
}

const replPython = resolveReplPython();
const describeIf = replPython && process.platform !== "win32" ? describe : describe.skip;

describeIf("repl runtime outlives-owner watchdog (real runtime)", { tags: ["kernel-heavy"] }, () => {
	it("runtime exits after its owner is SIGKILLed (stdin EOF watchdog)", async () => {
		const dir = mkdtempSync(join(tmpdir(), "prime-agent-repl-watchdog-int-"));
		const pidFile = join(dir, "runtime.pid");
		// The owner must be a separate killable process; it replicates the
		// manager's exact spawn line: piped stdin, so owner death delivers EOF.
		const ownerScript = [
			`const { spawn } = require("node:child_process");`,
			`const { writeFileSync } = require("node:fs");`,
			`const k = spawn(${JSON.stringify(replPython)}, ["-m", "rlm.repl"], {`,
			`  env: { ...process.env, PRIME_AGENT_KERNEL_OWNER_PID: String(process.pid) },`,
			`  stdio: ["pipe", "ignore", "ignore"],`,
			`});`,
			`writeFileSync(${JSON.stringify(pidFile)}, String(k.pid));`,
			`setInterval(() => {}, 1000);`,
		].join("\n");
		const owner = spawn(process.execPath, ["-e", ownerScript], { stdio: ["ignore", "ignore", "inherit"] });
		let runtimePid = 0;

		try {
			await vi.waitFor(
				() => {
					runtimePid = Number(readFileSync(pidFile, "utf8"));
					expect(runtimePid).toBeGreaterThan(0);
					expect(() => process.kill(runtimePid, 0)).not.toThrow();
				},
				{ timeout: 20_000, interval: 500 },
			);

			owner.kill("SIGKILL");

			await vi.waitFor(
				() => {
					expect(() => process.kill(runtimePid, 0)).toThrow();
				},
				{ timeout: 20_000, interval: 500 },
			);
		} finally {
			if (runtimePid > 0) {
				try {
					process.kill(runtimePid, "SIGKILL");
				} catch {
					// Already exited (the expected outcome).
				}
			}
			try {
				owner.kill("SIGKILL");
			} catch {
				// Already exited.
			}
			rmSync(dir, { recursive: true, force: true });
		}
	}, 30_000);

	it("runtime exits after owner death even while a non-yielding cell holds the loop", async () => {
		const dir = mkdtempSync(join(tmpdir(), "prime-agent-repl-watchdog-busy-"));
		const pidFile = join(dir, "runtime.pid");
		const busyFile = join(dir, "busy.marker");
		// The cell marks the file, then spins synchronously: the asyncio loop is
		// monopolized, so the stdin-EOF shutdown can never run — only the
		// watchdog thread can take the runtime down.
		const busyRequest = JSON.stringify({
			type: "execute",
			id: "busy-cell",
			code: `open(${JSON.stringify(busyFile)}, "w").write("busy")\nwhile True: pass`,
		});
		const ownerScript = [
			`const { spawn } = require("node:child_process");`,
			`const { writeFileSync } = require("node:fs");`,
			`const k = spawn(${JSON.stringify(replPython)}, ["-m", "rlm.repl"], {`,
			`  env: { ...process.env, PRIME_AGENT_KERNEL_OWNER_PID: String(process.pid) },`,
			`  stdio: ["pipe", "ignore", "ignore"],`,
			`});`,
			`writeFileSync(${JSON.stringify(pidFile)}, String(k.pid));`,
			`k.stdin.write(${JSON.stringify(`${busyRequest}\n`)});`,
			`setInterval(() => {}, 1000);`,
		].join("\n");
		const owner = spawn(process.execPath, ["-e", ownerScript], { stdio: ["ignore", "ignore", "inherit"] });
		let runtimePid = 0;

		try {
			await vi.waitFor(
				() => {
					runtimePid = Number(readFileSync(pidFile, "utf8"));
					expect(runtimePid).toBeGreaterThan(0);
					expect(() => process.kill(runtimePid, 0)).not.toThrow();
					// The marker proves the busy cell has entered its spin.
					expect(existsSync(busyFile)).toBe(true);
				},
				{ timeout: 20_000, interval: 500 },
			);

			owner.kill("SIGKILL");

			await vi.waitFor(
				() => {
					expect(() => process.kill(runtimePid, 0)).toThrow();
				},
				{ timeout: 20_000, interval: 500 },
			);
		} finally {
			if (runtimePid > 0) {
				try {
					process.kill(runtimePid, "SIGKILL");
				} catch {
					// Already exited (the expected outcome).
				}
			}
			try {
				owner.kill("SIGKILL");
			} catch {
				// Already exited.
			}
			rmSync(dir, { recursive: true, force: true });
		}
	}, 30_000);
});