#!/usr/bin/env node
/**
 * A deliberately small FIFO queue for slow, free advice requests.
 *
 * Each queue item is one agentic Pi run. Every assistant message in that run is
 * one provider request, so the runner records a turn event as it happens and
 * uses today's total to decide whether launching the next item is safe. The
 * budget gate runs before each item; it does not interrupt a running item.
 *
 * Usage:
 *   node scripts/advice-queue.ts add "prompt..."
 *   node scripts/advice-queue.ts run
 *   node scripts/advice-queue.ts status
 */

import { spawn } from "node:child_process";
import {
	appendFileSync,
	existsSync,
	linkSync,
	mkdirSync,
	readFileSync,
	renameSync,
	unlinkSync,
	writeFileSync,
} from "node:fs";
import { dirname, join, resolve } from "node:path";
import { createInterface } from "node:readline";
import { fileURLToPath, pathToFileURL } from "node:url";

const MODEL = process.env.ADVICE_QUEUE_MODEL ?? "nvidia/z-ai/glm-5.3-flash";
const DAILY_LIMIT = 1000;
// Used only until the first item finishes; a real sample replaces it.
const DEFAULT_EXPECTED_TURNS = 10;

const agentDir = dirname(dirname(fileURLToPath(import.meta.url)));
const stateDir = process.env.ADVICE_QUEUE_DIR ?? join(agentDir, "advice-queue");
const piBin = process.env.PI_BIN ?? join(agentDir, "bin", "pi");

/**
 * Where the queue keeps its files. Every state function takes one of these, defaulting to the live
 * agent directory, so a test can point the whole state layer at a temporary directory.
 */
export interface AdviceQueuePaths {
	readonly stateDir: string;
	readonly queuePath: string;
	readonly turnsPath: string;
	readonly outputDir: string;
	readonly lockPath: string;
}

export function adviceQueuePaths(stateDir: string): AdviceQueuePaths {
	return {
		stateDir,
		queuePath: join(stateDir, "queue.json"),
		turnsPath: join(stateDir, "turns.jsonl"),
		outputDir: join(stateDir, "out"),
		lockPath: join(stateDir, "lock"),
	};
}

const currentPaths = (): AdviceQueuePaths => adviceQueuePaths(stateDir);

type ItemStatus = "pending" | "running" | "done" | "failed";

export interface QueueItem {
	readonly id: string;
	readonly prompt: string;
	readonly status: ItemStatus;
	readonly addedAt: string;
	readonly turns?: number;
	readonly outputFile?: string;
	readonly error?: string;
	readonly startedAt?: string;
	readonly finishedAt?: string;
}

interface TurnEvent {
	readonly ts: string;
	readonly date: string;
	readonly itemId: string;
}

interface TurnStats {
	readonly count: number;
	readonly avg: number;
	readonly median: number;
	readonly max: number;
}

interface Budget {
	readonly used: number;
	readonly remaining: number;
	readonly expected: number;
	readonly canStart: boolean;
	readonly atRisk: boolean;
}

interface AssistantText {
	readonly text: string;
	readonly hasToolCall: boolean;
}

interface ItemResult {
	readonly turns: number;
	readonly text: string;
	readonly error?: string;
}

// ---------------------------------------------------------------------------
// Pure helpers
// ---------------------------------------------------------------------------

function isRecord(value: unknown): value is Record<string, unknown> {
	return typeof value === "object" && value !== null && !Array.isArray(value);
}

function isErrno(error: unknown, code: string): boolean {
	return isRecord(error) && error.code === code;
}

function dateKey(date: Date): string {
	return date.toISOString().slice(0, 10);
}

export function computeStats(turns: readonly number[]): TurnStats {
	const values = turns.filter((value) => Number.isFinite(value) && value >= 0);
	if (values.length === 0) {
		return { count: 0, avg: 0, median: 0, max: 0 };
	}
	const sorted = [...values].sort((a, b) => a - b);
	const middle = Math.floor(sorted.length / 2);
	const median = sorted.length % 2 === 0 ? (sorted[middle - 1] + sorted[middle]) / 2 : sorted[middle];
	return {
		count: values.length,
		avg: values.reduce((sum, value) => sum + value, 0) / values.length,
		median,
		max: sorted[sorted.length - 1],
	};
}

/**
 * Start when the expected item still fits. Warn when a worst-case item would
 * not, so the operator can decide rather than the runner guessing.
 */
export function evaluateBudget(used: number, stats: TurnStats, limit: number): Budget {
	const remaining = limit - used;
	const expected = stats.count === 0 ? DEFAULT_EXPECTED_TURNS : Math.ceil(stats.avg);
	return {
		used,
		remaining,
		expected,
		canStart: remaining >= Math.max(expected, 1),
		atRisk: stats.count > 0 && remaining < stats.max,
	};
}

export function isAssistantMessage(
	message: unknown,
): message is { readonly role: "assistant"; readonly content?: unknown } {
	return isRecord(message) && message.role === "assistant";
}

export function isAssistantTurnEvent(
	event: unknown,
): event is { readonly type: "message_end"; readonly message: unknown } {
	return isRecord(event) && event.type === "message_end" && isAssistantMessage(event.message);
}

export function parseAssistantText(message: unknown): AssistantText | undefined {
	if (!isAssistantMessage(message)) {
		return undefined;
	}
	const content = message.content;
	if (!Array.isArray(content)) {
		return undefined;
	}
	const textParts = content.flatMap((part) =>
		isRecord(part) && part.type === "text" && typeof part.text === "string" ? [part.text] : [],
	);
	if (textParts.length === 0) {
		return undefined;
	}
	const hasToolCall = content.some(
		(part) => isRecord(part) && (part.type === "toolCall" || part.type === "tool_call"),
	);
	const text = textParts.join("\n").trim();
	return text.length > 0 ? { text, hasToolCall } : undefined;
}

export function parseQueue(value: unknown): readonly QueueItem[] {
	if (!Array.isArray(value)) {
		throw new Error("queue.json must contain an array");
	}
	return value.map((entry) => {
		if (
			!isRecord(entry) ||
			typeof entry.id !== "string" ||
			typeof entry.prompt !== "string" ||
			typeof entry.status !== "string" ||
			typeof entry.addedAt !== "string"
		) {
			throw new Error("queue.json contains an invalid item");
		}
		const status = entry.status;
		if (status !== "pending" && status !== "running" && status !== "done" && status !== "failed") {
			throw new Error(`queue.json item ${entry.id} has an unknown status`);
		}
		return {
			id: entry.id,
			prompt: entry.prompt,
			status,
			addedAt: entry.addedAt,
			turns: typeof entry.turns === "number" ? entry.turns : undefined,
			outputFile: typeof entry.outputFile === "string" ? entry.outputFile : undefined,
			error: typeof entry.error === "string" ? entry.error : undefined,
			startedAt: typeof entry.startedAt === "string" ? entry.startedAt : undefined,
			finishedAt: typeof entry.finishedAt === "string" ? entry.finishedAt : undefined,
		};
	});
}

// ---------------------------------------------------------------------------
// State
// ---------------------------------------------------------------------------

function ensureStateDir(paths: AdviceQueuePaths = currentPaths()): void {
	mkdirSync(paths.outputDir, { recursive: true });
}

export function loadQueue(paths: AdviceQueuePaths = currentPaths()): readonly QueueItem[] {
	if (!existsSync(paths.queuePath)) {
		return [];
	}
	const parsed: unknown = JSON.parse(readFileSync(paths.queuePath, "utf8"));
	return parseQueue(parsed);
}

export function saveQueue(items: readonly QueueItem[], paths: AdviceQueuePaths = currentPaths()): void {
	const temporary = `${paths.queuePath}.${process.pid}.${Date.now()}.tmp`;
	try {
		writeFileSync(temporary, `${JSON.stringify(items, null, 2)}\n`);
		renameSync(temporary, paths.queuePath);
	} finally {
		try {
			unlinkSync(temporary);
		} catch {
			// The rename already removed it, or the write never created it.
		}
	}
}

export function appendTurn(itemId: string, paths: AdviceQueuePaths = currentPaths()): void {
	const now = new Date();
	const event: TurnEvent = { ts: now.toISOString(), date: dateKey(now), itemId };
	appendFileSync(paths.turnsPath, `${JSON.stringify(event)}\n`);
}

export function readTurnsForToday(paths: AdviceQueuePaths = currentPaths()): number {
	if (!existsSync(paths.turnsPath)) {
		return 0;
	}
	const today = dateKey(new Date());
	return readFileSync(paths.turnsPath, "utf8")
		.split("\n")
		.filter((line) => line.length > 0)
		.reduce((count, line) => {
			try {
				const parsed: unknown = JSON.parse(line);
				return isRecord(parsed) && parsed.date === today ? count + 1 : count;
			} catch {
				return count;
			}
		}, 0);
}

export function itemTurns(items: readonly QueueItem[]): readonly number[] {
	return items.flatMap((item) => (item.turns === undefined || item.turns <= 0 ? [] : [item.turns]));
}

export function writeOutput(
	item: QueueItem,
	result: ItemResult,
	startedAt: string,
	paths: AdviceQueuePaths = currentPaths(),
): string {
	const safeId = item.id.replace(/[^A-Za-z0-9._-]/g, "_");
	const outputFile = join(paths.outputDir, `${safeId}.md`);
	const prompt = item.prompt.replace(/\n/g, "\n> ");
	const body = [
		`# ${item.id}`,
		"",
		`- Model: \`${MODEL}\``,
		`- Started: ${startedAt}`,
		`- Finished: ${new Date().toISOString()}`,
		`- Turns: ${result.turns}`,
		`- Status: ${result.error ? "failed (partial output)" : "done"}`,
		"",
		"## Prompt",
		"",
		`> ${prompt}`,
		"",
		"## Advice",
		"",
		result.text,
		"",
	].join("\n");
	writeFileSync(outputFile, body);
	return outputFile;
}

// ---------------------------------------------------------------------------
// Lock
// ---------------------------------------------------------------------------

function processIsAlive(pid: number): boolean {
	try {
		process.kill(pid, 0);
		return true;
	} catch (error) {
		return !isErrno(error, "ESRCH");
	}
}

function tryCreateLock(path: string): boolean {
	const temporary = `${path}.${process.pid}.${Date.now()}.tmp`;
	writeFileSync(temporary, String(process.pid));
	try {
		linkSync(temporary, path);
		return true;
	} catch (error) {
		if (isErrno(error, "EEXIST")) {
			return false;
		}
		throw error;
	} finally {
		try {
			unlinkSync(temporary);
		} catch {
			// The temporary file is best-effort cleanup.
		}
	}
}

function readLockPid(path: string): number | undefined {
	try {
		const value = Number(readFileSync(path, "utf8").trim());
		return Number.isInteger(value) && value > 0 ? value : undefined;
	} catch {
		return undefined;
	}
}

function releaseOwnLock(path: string): void {
	if (readLockPid(path) !== process.pid) {
		return;
	}
	try {
		unlinkSync(path);
	} catch {
		// A missing lock is already released.
	}
}

export function acquireRunLock(paths: AdviceQueuePaths = currentPaths()): () => void {
	ensureStateDir(paths);
	for (let attempt = 0; attempt < 3; attempt += 1) {
		if (tryCreateLock(paths.lockPath)) {
			return () => releaseOwnLock(paths.lockPath);
		}
		const existingPid = readLockPid(paths.lockPath);
		if (existingPid !== undefined && processIsAlive(existingPid)) {
			throw new Error(`another queue runner is active (pid ${existingPid})`);
		}
		try {
			unlinkSync(paths.lockPath);
		} catch (error) {
			if (!isErrno(error, "ENOENT")) {
				throw error;
			}
		}
	}
	throw new Error("could not acquire the queue lock");
}

async function withQueueLock<T>(
	update: () => T | Promise<T>,
	paths: AdviceQueuePaths = currentPaths(),
): Promise<T> {
	const path = `${paths.queuePath}.lock`;
	const deadline = Date.now() + 5000;
	for (;;) {
		if (tryCreateLock(path)) {
			break;
		}
		const pid = readLockPid(path);
		if (pid !== undefined && !processIsAlive(pid)) {
			try {
				unlinkSync(path);
			} catch {
				// Someone else already removed the stale lock.
			}
			continue;
		}
		if (Date.now() > deadline) {
			throw new Error("timed out waiting for the queue lock");
		}
		await new Promise((resolveWait) => setTimeout(resolveWait, 50));
	}
	try {
		return await update();
	} finally {
		releaseOwnLock(path);
	}
}

export async function mutateQueue(
	update: (items: readonly QueueItem[]) => readonly QueueItem[],
	paths: AdviceQueuePaths = currentPaths(),
): Promise<void> {
	await withQueueLock(() => saveQueue(update(loadQueue(paths)), paths), paths);
}

// ---------------------------------------------------------------------------
// Runner
// ---------------------------------------------------------------------------

let activeChild: ReturnType<typeof spawn> | undefined;

function runItem(item: QueueItem, paths: AdviceQueuePaths = currentPaths()): Promise<ItemResult> {
	return new Promise((resolveResult) => {
		let child: ReturnType<typeof spawn>;
		try {
			child = spawn(
				piBin,
				[
					"--mode",
					"json",
					"--no-session",
					"--no-extensions",
					"--no-skills",
					"--model",
					MODEL,
					"--tools",
					"read,grep,find,ls",
					"-p",
					"--",
					item.prompt,
				],
				{ cwd: process.cwd(), env: process.env, stdio: ["ignore", "pipe", "pipe"] },
			);
		} catch (error) {
			resolveResult({
				turns: 0,
				text: "",
				error: error instanceof Error ? error.message : String(error),
			});
			return;
		}
		activeChild = child;
		let turns = 0;
		let finalText = "";
		let fallbackText = "";
		let stderr = "";
		let settled = false;
		const finish = (result: ItemResult): void => {
			if (!settled) {
				settled = true;
				activeChild = undefined;
				resolveResult(result);
			}
		};

		if (!child.stdout || !child.stderr) {
			child.kill();
			finish({ turns: 0, text: "", error: "could not capture pi output" });
			return;
		}

		const lines = createInterface({ input: child.stdout });
		lines.on("line", (line) => {
			let parsed: unknown;
			try {
				parsed = JSON.parse(line);
			} catch {
				return;
			}
			if (!isAssistantTurnEvent(parsed)) {
				return;
			}
			turns += 1;
			appendTurn(item.id, paths);
			process.stderr.write(`  ${item.id}: turn ${turns}\n`);
			const assistant = parseAssistantText(parsed.message);
			if (!assistant) {
				return;
			}
			fallbackText = assistant.text;
			if (!assistant.hasToolCall) {
				finalText = assistant.text;
			}
		});
		child.stderr.on("data", (chunk: Buffer) => {
			stderr = `${stderr}${chunk.toString()}`.slice(-2000);
		});
		child.on("error", (error) => {
			lines.close();
			child.kill();
			finish({ turns, text: finalText || fallbackText, error: error.message });
		});
		child.on("close", (code) => {
			lines.close();
			const text = finalText || fallbackText;
			if (code === 0 && text.length > 0) {
				finish({ turns, text });
				return;
			}
			const detail = stderr.trim().split("\n").at(-1) ?? "no stderr";
			const error = code === 0 ? "pi produced no final answer" : `pi exited ${code}: ${detail}`;
			finish({ turns, text, error });
		});
	});
}

function printBudget(used: number, stats: TurnStats): Budget {
	const budget = evaluateBudget(used, stats, DAILY_LIMIT);
	const history =
		stats.count === 0
			? "no completed items yet"
			: `avg ${stats.avg.toFixed(1)}, median ${stats.median}, max ${stats.max} (n=${stats.count})`;
	console.log(`turns: ${history}`);
	console.log(
		`today: ${used}/${DAILY_LIMIT} requests, ${budget.remaining} remaining, next expected ~${budget.expected}`,
	);
	if (!budget.canStart) {
		console.log("stop: the expected next item does not fit in today's budget");
	} else if (budget.atRisk) {
		console.log(`warning: an above-average item (max ${stats.max}) would exceed today's limit`);
	}
	return budget;
}

export async function runQueue(paths: AdviceQueuePaths = currentPaths()): Promise<void> {
	ensureStateDir(paths);
	const release = acquireRunLock(paths);
	const stop = (): void => {
		process.removeListener("SIGINT", stop);
		process.removeListener("SIGTERM", stop);
		const child = activeChild;
		let stopped = false;
		let timer: ReturnType<typeof setTimeout> | undefined;
		const finishStop = (): void => {
			if (stopped) {
				return;
			}
			stopped = true;
			if (timer !== undefined) {
				clearTimeout(timer);
			}
			release();
			process.exit(1);
		};
		if (!child || child.exitCode !== null || child.signalCode !== null) {
			finishStop();
			return;
		}
		timer = setTimeout(() => {
			child.kill("SIGKILL");
			finishStop();
		}, 2000);
		child.once("close", finishStop);
		try {
			child.kill("SIGTERM");
		} catch {
			finishStop();
		}
	};
	process.once("SIGINT", stop);
	process.once("SIGTERM", stop);
	try {
		await mutateQueue(
			(items) =>
				items.map((item) =>
					item.status === "running"
						? { ...item, status: "pending" as const, error: "interrupted; requeued" }
						: item,
				),
			paths,
		);

		for (;;) {
			const items = loadQueue(paths);
			const item = items.find((candidate) => candidate.status === "pending");
			if (!item) {
				console.log("queue empty");
				break;
			}
			const budget = printBudget(readTurnsForToday(paths), computeStats(itemTurns(items)));
			if (!budget.canStart) {
				break;
			}

			const startedAt = new Date().toISOString();
			await mutateQueue(
				(current) =>
					current.map((candidate) =>
						candidate.id === item.id
							? { ...candidate, status: "running" as const, startedAt }
							: candidate,
					),
				paths,
			);
			console.log(`running ${item.id}`);
			const result = await runItem(item, paths);
			const finishedAt = new Date().toISOString();
			const outputFile =
				!result.error || result.text.length > 0
					? writeOutput(item, result, startedAt, paths)
					: undefined;

			await mutateQueue(
				(current) =>
					current.map((candidate) =>
						candidate.id === item.id
							? {
									...candidate,
									status: result.error ? ("failed" as const) : ("done" as const),
									turns: result.turns,
									finishedAt,
									error: result.error,
									outputFile,
								}
							: candidate,
					),
				paths,
			);
			if (result.error) {
				console.error(`${item.id} failed after ${result.turns} turns: ${result.error}`);
			} else {
				console.log(`${item.id} done in ${result.turns} turns -> ${outputFile}`);
			}
		}

		const stats = computeStats(itemTurns(loadQueue(paths)));
		console.log(
			`stats: n=${stats.count}, avg=${stats.avg.toFixed(1)}, median=${stats.median}, max=${stats.max}`,
		);
	} finally {
		process.removeListener("SIGINT", stop);
		process.removeListener("SIGTERM", stop);
		release();
	}
}

function showStatus(): void {
	ensureStateDir();
	const items = loadQueue();
	const stats = computeStats(itemTurns(items));
	const count = (status: ItemStatus): number => items.filter((item) => item.status === status).length;
	console.log(
		`queue: ${count("pending")} pending, ${count("running")} running, ${count("done")} done, ${count("failed")} failed`,
	);
	printBudget(readTurnsForToday(), stats);
	const pending = items.filter((item) => item.status === "pending");
	if (pending.length > 0) {
		console.log("pending:");
		for (const item of pending) {
			console.log(`  ${item.id}  ${item.prompt.slice(0, 100)}`);
		}
	}
}

async function addItem(prompt: string): Promise<void> {
	ensureStateDir();
	const now = new Date();
	const item: QueueItem = {
		id: `q-${now.getTime().toString(36)}-${Math.random().toString(36).slice(2, 6)}`,
		prompt,
		status: "pending",
		addedAt: now.toISOString(),
	};
	await mutateQueue((items) => [...items, item]);
	console.log(`queued ${item.id}`);
}

// ---------------------------------------------------------------------------
// CLI
// ---------------------------------------------------------------------------

function usage(): void {
	console.log("usage: node scripts/advice-queue.ts <add|run|status> [prompt]");
}

async function main(): Promise<void> {
	const [command, ...rest] = process.argv.slice(2);
	switch (command) {
		case "add": {
			const prompt = rest.join(" ").trim();
			if (prompt.length === 0) {
				usage();
				process.exitCode = 1;
				return;
			}
			await addItem(prompt);
			return;
		}
		case "run":
			await runQueue();
			return;
		case "status":
			showStatus();
			return;
		default:
			usage();
			process.exitCode = 1;
	}
}

const invokedDirectly =
	process.argv[1] !== undefined && pathToFileURL(resolve(process.argv[1])).href === import.meta.url;

if (invokedDirectly) {
	main().catch((error: unknown) => {
		console.error(error instanceof Error ? error.message : String(error));
		process.exitCode = 1;
	});
}