pi-ext

import { spawn, spawnSync } from "node:child_process";
import {
	createWriteStream,
	existsSync,
	lstatSync,
	mkdirSync,
	readdirSync,
	readFileSync,
	statSync,
	writeFileSync,
} from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { performance } from "node:perf_hooks";
import { finished } from "node:stream/promises";
import { isDeepStrictEqual } from "node:util";

const root = resolve(import.meta.dirname, "../../..");
const cases: Record<string, string> = {
	proquint:
		"Implement a tiny reusable Nushell module proquint.nu with IPv4 dotted-quad ↔ proquint functions. Use the paper's alphabet bdfghjklmnprstvz / aiou and verify 127.0.0.1 ↔ lusab-babad, 0.0.0.0, 255.255.255.255, malformed input, and round trips. Use only your Nushell tools for shell work. Produce the module inside the current working directory; run positive and negative checks. No network, no writes outside cwd.",
	base32:
		"Implement a small reusable Nushell module crockford.nu for fixed-width unsigned 32-bit Crockford Base32 IDs, encoding and decoding with canonical uppercase alphabet and human aliases O/0 and I/L/1. Include explicit leading-zero width and reject overflow or invalid symbols. Use only your Nushell tools for shell work. Produce the module inside the current working directory; run positive and negative round-trip checks. No network, no writes outside cwd.",
	csv: "Implement a tiny reusable Nushell module csv-headers.nu reporting added, removed, and reordered CSV header columns between two files. Handle quoted commas in headers, empty files, and duplicate headers explicitly. Make two small fixture files in cwd and run positive and negative checks. Use only your Nushell tools for shell work. No network, no writes outside cwd.",
	manifest:
		"Implement a small reusable Nushell module manifest.nu which snapshots regular files in a given directory into records of relative path, size, and SHA-256; compare two snapshots to report added, removed, changed. Explicitly skip symlinks. Make small fixture directories in cwd and run positive and negative checks. Use only your Nushell tools for shell work. No network, no writes outside cwd.",
	trace:
		'Implement trace.nu: summarize PI_NUSHELL_DEBUG JSONL in actual generated logger format. A fixture.jsonl is already in cwd; inspect it, do not invent another schema. Pair fresh.start with fresh.finish or fresh.error, and session.request.start with session.request.finish or session.request.error, by spanId, including interleaved starts. Ignore session.connect spans and orphan terminals. Use durationMs from paired terminal records; nearest-rank p50/p95 per mode, empty group null. Count safe fresh terminal status classifications and session.request.error kind classifications. Detect a malformed or truncated final JSONL line; malformed earlier lines must fail. For fixture expect fresh {count:2,p50:10,p95:30,statusCounts:{nonzero:1,success:1}}, session {count:2,p50:40,p95:60,errorKindCounts:{nu:1}}, unmatchedStartSpanIds:["1234-5"], malformedFinalLineCount:1. Return exactly fresh {count,p50,p95,statusCounts}, session {count,p50,p95,errorKindCounts}, unmatchedStartSpanIds, malformedFinalLineCount. Run positive and negative assertions using actual fixture. Do not read process env or actual live debug log. Use only Nushell tools; no network, no writes outside cwd.',
	state:
		"Exercise the persistent Nushell tool contract within this one Pi invocation. First nu_session call defines a uniquely named function pq-harness-encode and a variable pq_harness_alphabet; a separate later nu_session call must use both stored bindings and verify a value. Run a deliberate Nushell evaluation error, then verify a subsequent nu_session call still sees both bindings. Call the fresh nu tool once to check it does not inherit either binding. Finally report observed outputs and whether fresh and persistent state were isolated. No file writes, no network, do not run Nu inside Nu.",
};
function traceOracle(file: string) {
	type TraceRecord = {
		event?: unknown;
		spanId?: unknown;
		durationMs?: unknown;
		status?: unknown;
		kind?: unknown;
	};
	const lines = readFileSync(file, "utf8").split("\n");
	while (lines.at(-1) === "") lines.pop();
	let malformedFinalLineCount = 0;
	const records = lines.flatMap((line, index) => {
		try {
			return [JSON.parse(line) as TraceRecord];
		} catch (error) {
			if (index === lines.length - 1) {
				malformedFinalLineCount++;
				return [];
			}
			throw error;
		}
	});
	const pending = new Map<string, "fresh" | "session">();
	const fresh: number[] = [];
	const session: number[] = [];
	const statusCounts: Record<string, number> = {};
	const errorKindCounts: Record<string, number> = {};
	for (const record of records) {
		const mode =
			record.event === "fresh.start"
				? "fresh"
				: record.event === "session.request.start"
					? "session"
					: undefined;
		if (mode && typeof record.spanId === "string")
			pending.set(record.spanId, mode);
		const endMode =
			record.event === "fresh.finish" || record.event === "fresh.error"
				? "fresh"
				: record.event === "session.request.finish" ||
						record.event === "session.request.error"
					? "session"
					: undefined;
		if (
			!endMode ||
			typeof record.spanId !== "string" ||
			pending.get(record.spanId) !== endMode
		)
			continue;
		pending.delete(record.spanId);
		if (typeof record.durationMs !== "number") continue;
		if (endMode === "fresh") {
			fresh.push(record.durationMs);
			if (typeof record.status === "string")
				statusCounts[record.status] = (statusCounts[record.status] ?? 0) + 1;
		} else {
			session.push(record.durationMs);
			if (
				record.event === "session.request.error" &&
				typeof record.kind === "string"
			)
				errorKindCounts[record.kind] = (errorKindCounts[record.kind] ?? 0) + 1;
		}
	}
	const summary = (values: number[]) => {
		const ordered = values.toSorted((a, b) => a - b);
		return {
			count: ordered.length,
			p50: ordered[Math.ceil(ordered.length * 0.5) - 1] ?? null,
			p95: ordered[Math.ceil(ordered.length * 0.95) - 1] ?? null,
		};
	};
	return {
		fresh: { ...summary(fresh), statusCounts },
		session: { ...summary(session), errorKindCounts },
		unmatchedStartSpanIds: [...pending.keys()],
		malformedFinalLineCount,
	};
}

const verifying = process.argv[2] === "--verify";
const names = verifying ? [] : process.argv.slice(2);
for (const name of names)
	if (!cases[name]) throw new Error(`Unknown case: ${name}`);
const chosen = verifying ? [] : names.length ? names : Object.keys(cases);
const evidenceRoot = join(tmpdir(), "pi-ext-nushell-json");
mkdirSync(evidenceRoot, { recursive: true, mode: 0o700 });
const evidenceStat = lstatSync(evidenceRoot);
if (
	!evidenceStat.isDirectory() ||
	evidenceStat.isSymbolicLink() ||
	(process.platform !== "win32" && (evidenceStat.mode & 0o077) !== 0)
)
	throw new Error("Nushell JSON evidence directory is not private");
const runRoot = verifying
	? resolve(process.argv[3] ?? "")
	: join(evidenceRoot, new Date().toISOString().replace(/[:.]/g, "-"));
if (verifying && !runRoot.startsWith(`${evidenceRoot}/`))
	throw new Error(
		"Verification path must be a retained Nushell JSON evidence run",
	);
const model = process.env.PI_NUSHELL_E2E_MODEL ?? "openai-codex/gpt-5.6-terra";
const binary = resolve(root, "extensions/nushell/index.ts");
const results: Record<string, unknown>[] = verifying
	? JSON.parse(readFileSync(join(runRoot, "results.json"), "utf8"))
	: [];

function debugFile(directory: string): string | undefined {
	const path = join(directory, "state", "pi-ext", "debug", "nushell");
	if (!existsSync(path)) return undefined;
	return readdirSync(path)
		.filter((name) => name.endsWith(".jsonl"))
		.map((name) => join(path, name))
		.at(-1);
}
for (const name of chosen) {
	const directory = join(runRoot, name);
	const workspace = join(directory, "workspace");
	mkdirSync(workspace, { recursive: true, mode: 0o700 });
	if (name === "trace") {
		const fixture = [
			{ event: "fresh.start", spanId: "1234-1", timeoutMs: 1_000 },
			{ event: "fresh.start", spanId: "1234-2", timeoutMs: 1_000 },
			{
				event: "fresh.finish",
				spanId: "1234-2",
				durationMs: 30,
				status: "nonzero",
				count: 12,
			},
			{
				event: "fresh.finish",
				spanId: "1234-1",
				durationMs: 10,
				status: "success",
				count: 4,
			},
			{ event: "session.request.start", spanId: "1234-3" },
			{
				event: "session.connect.finish",
				spanId: "1234-connect",
				durationMs: 15,
			},
			{
				event: "session.request.finish",
				spanId: "1234-3",
				durationMs: 40,
				count: 25,
			},
			{ event: "session.request.start", spanId: "1234-4" },
			{
				event: "session.request.error",
				spanId: "1234-4",
				durationMs: 60,
				kind: "nu",
			},
			{ event: "fresh.start", spanId: "1234-5" },
			{
				event: "fresh.error",
				spanId: "1234-6",
				durationMs: 99,
				status: "failed",
			},
		];
		writeFileSync(
			join(workspace, "fixture.jsonl"),
			`${fixture.map((record) => JSON.stringify({ timestamp: "2026-01-01T00:00:00.000Z", extension: "nushell", pid: 1234, ...record })).join("\n")}\n{"event":"fresh.finish"`,
		);
	}
	const prompt = `${cases[name]} Prefer nu for independent commands; use nu_session only if later calls need session variables/functions/cwd. Report what worked, failed, and which module checks passed.`;
	const args = [
		"--print",
		"--mode",
		"json",
		"--no-session",
		"--nushell",
		"--offline",
		"--no-approve",
		"--no-context-files",
		"--no-extensions",
		"--extension",
		binary,
		"--no-skills",
		"--no-prompt-templates",
		"--no-themes",
		"--model",
		model,
		"--thinking",
		"low",
		"--tools",
		process.env.PI_NUSHELL_E2E_TOOLS ?? "bash,read,write,edit,nu,nu_session",
		prompt,
	];
	writeFileSync(
		join(directory, "invocation.json"),
		`${JSON.stringify({ cwd: workspace, command: "pi", args, timeoutMs: 180_000, debug: join(directory, "state", "pi-ext", "debug", "nushell") }, null, 2)}\n`,
	);
	const out = createWriteStream(join(directory, "events.jsonl"), {
		mode: 0o600,
	});
	const err = createWriteStream(join(directory, "stderr.txt"), { mode: 0o600 });
	const started = performance.now();
	const result = await new Promise<{
		exitCode: number | null;
		signal: string | null;
		timedOut: boolean;
	}>((done, reject) => {
		const child = spawn("pi", args, {
			cwd: workspace,
			env: {
				...process.env,
				PI_NUSHELL_DEBUG: "1",
				XDG_STATE_HOME: join(directory, "state"),
			},
			stdio: ["ignore", "pipe", "pipe"],
			detached: process.platform !== "win32",
		});
		child.stdout.pipe(out);
		child.stderr.pipe(err);
		let timedOut = false;
		const timer = setTimeout(() => {
			timedOut = true;
			if (child.pid) {
				if (process.platform === "win32") child.kill("SIGKILL");
				else process.kill(-child.pid, "SIGKILL");
			}
		}, 180_000);
		child.once("error", (error) => {
			clearTimeout(timer);
			reject(error);
		});
		child.once("close", (exitCode, signal) => {
			clearTimeout(timer);
			done({ exitCode, signal, timedOut });
		});
	});
	await Promise.all([finished(out), finished(err)]);
	const events = readFileSync(join(directory, "events.jsonl"), "utf8")
		.split("\n")
		.filter(Boolean)
		.map(
			(line) =>
				JSON.parse(line) as {
					type: string;
					toolName?: string;
					isError?: boolean;
					args?: { command?: string };
					message?: {
						role?: string;
						stopReason?: string;
						content?: { type: string; text?: string }[];
						usage?: { cost?: { total?: number } };
					};
				},
		);
	const calls = events.filter((event) => event.type === "tool_execution_start");
	const lastAssistant = events
		.filter(
			(event) =>
				event.type === "message_end" && event.message?.role === "assistant",
		)
		.at(-1);
	const trace = debugFile(directory);
	const debugEvents = (trace ? readFileSync(trace, "utf8") : "")
		.split("\n")
		.filter(Boolean)
		.map(
			(line) =>
				JSON.parse(line) as {
					event: string;
					kind?: string;
					status?: string;
				},
		);
	const finalText =
		lastAssistant?.message?.content
			?.filter((part) => part.type === "text")
			.map((part) => part.text ?? "")
			.join("\n") ?? "";
	writeFileSync(join(directory, "final.txt"), `${finalText}\n`);
	const analysis = {
		settled: events.some((event) => event.type === "agent_settled"),
		stopReason: lastAssistant?.message?.stopReason,
		toolCalls: Object.fromEntries(
			[...new Set(calls.map((event) => event.toolName))].map((name) => [
				name,
				calls.filter((call) => call.toolName === name).length,
			]),
		),
		toolErrors: events.filter(
			(event) => event.type === "tool_execution_end" && event.isError,
		).length,
		debugErrors: debugEvents
			.filter(
				(event) => event.event.endsWith(".error") || event.status === "nonzero",
			)
			.map((event) => ({
				event: event.event,
				kind: event.kind,
				status: event.status,
			})),
		stderrBytes: statSync(join(directory, "stderr.txt")).size,
		hasFinal: finalText.length > 0,
		nestedNu: calls.filter((event) =>
			/(?:\^?nu\s+-c\b)/.test(event.args?.command ?? ""),
		).length,
		debugEnabled: debugEvents.some((event) => event.event === "tools.enabled"),
		projectedCostUSD: events
			.filter(
				(event) =>
					event.type === "message_end" && event.message?.role === "assistant",
			)
			.reduce(
				(sum, event) => sum + (event.message?.usage?.cost?.total ?? 0),
				0,
			),
	};
	writeFileSync(
		join(directory, "analysis.json"),
		`${JSON.stringify(analysis, null, 2)}\n`,
	);
	const row = {
		name,
		...result,
		...analysis,
		durationMs: Math.round(performance.now() - started),
		directory,
	};
	results.push(row);
	writeFileSync(
		join(directory, "result.json"),
		`${JSON.stringify(row, null, 2)}\n`,
	);
	console.log(JSON.stringify(row));
}
const trace = results.find((row) => row.name === "trace");
if (trace) {
	const traceWorkspace = join(runRoot, "trace", "workspace");
	const validations = results.map((row) => {
		const file = debugFile(String(row.directory));
		if (!file)
			throw new Error(`Missing debug trace for ${String(row.directory)}`);
		const check = spawnSync(
			"nu",
			[
				"--no-config-file",
				"--no-history",
				"-c",
				"use trace.nu *; summarize $env.PI_NUSHELL_TRACE_INPUT | to json -r",
			],
			{
				cwd: traceWorkspace,
				env: {
					PATH: process.env.PATH ?? "",
					HOME: process.env.HOME ?? traceWorkspace,
					XDG_CONFIG_HOME: process.env.XDG_CONFIG_HOME,
					MISE_SHELL: process.env.MISE_SHELL,
					PI_NUSHELL_TRACE_INPUT: file,
				},
				encoding: "utf8",
				timeout: 15_000,
				maxBuffer: 1_000_000,
			},
		);
		const expected = traceOracle(file);
		let observed: unknown;
		try {
			observed = JSON.parse(check.stdout);
		} catch {
			observed = null;
		}
		return {
			name: row.name,
			passed: check.status === 0 && isDeepStrictEqual(observed, expected),
			expected,
			observed,
			error: check.status === 0 ? undefined : check.stderr.slice(0, 1000),
		};
	});
	trace.traceValidation = validations.every((item) => item.passed);
	writeFileSync(
		join(runRoot, "trace", "validation.json"),
		`${JSON.stringify(validations, null, 2)}\n`,
	);
	writeFileSync(
		join(runRoot, "trace", "result.json"),
		`${JSON.stringify(trace, null, 2)}\n`,
	);
	console.log(
		JSON.stringify({
			traceValidation: trace.traceValidation,
			cases: validations.map(({ name, passed }) => ({ name, passed })),
		}),
	);
}
writeFileSync(
	join(runRoot, "results.json"),
	`${JSON.stringify(results, null, 2)}\n`,
);
console.log(`Evidence: ${runRoot}`);
if (
	results.some(
		(result) =>
			result.exitCode !== 0 ||
			result.timedOut ||
			result.stopReason !== "stop" ||
			!result.settled ||
			!result.debugEnabled ||
			result.traceValidation === false ||
			(!(result.toolCalls as Record<string, number>).nu &&
				!(result.toolCalls as Record<string, number>).nu_session),
	)
)
	throw new Error(
		`One or more runs lacked a successful Nu-tool session; inspect ${runRoot}`,
	);