// Spec-compliance smoke test for extensions/nats-channel.ts.
//
// Drives the extension with a minimal mock of PI's ExtensionAPI, connects
// to a local nats-server, and asserts protocol v0.3 behaviour:
//
//   1. Service registers under `agents` with spec metadata.
//   2. The `prompt` endpoint has `max_payload` / `attachments_ok` metadata
//      and lives at `agents.prompt.pi.{owner}.{name}` (verb-first §2 v0.3).
//   3. Heartbeats arrive on `agents.hb.pi.{owner}.{name}` (§8.1 v0.3).
//   4. The `status` endpoint at `agents.status.pi.{owner}.{name}` replies
//      with a heartbeat-shaped payload (§8.7 (v0.3)).
//   5. Empty payload → 400 + terminator.
//   6. Plain-text prompt yields a `status: ack` chunk, text chunks via the
//      mock emitter, and an empty-body no-headers terminator.
//   7. Request with attachments → 400 + terminator.
//
// Run with:
//   bun test/smoke.mjs
// Server: this test always spawns its own private nats-server (via
// `test/nats-server.conf`, which pins `port: -1` so the kernel picks an
// ephemeral port and `max_payload: 8MB` so the dynamic max_payload path is
// exercised). The chosen port is read back from `--ports_file_dir` and the
// server is killed on exit. Prereq: `nats-server` on PATH.
//
// Exit code is non-zero on any failed assertion.

import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import {
	existsSync,
	mkdirSync,
	mkdtempSync,
	readFileSync,
	readdirSync,
	rmSync,
	writeFileSync,
} from "node:fs";
import { homedir, tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { setTimeout as delay } from "node:timers/promises";
import { fileURLToPath } from "node:url";
import { connect } from "@nats-io/transport-node";
import { createInbox } from "@nats-io/nats-core";
import { Svcm } from "@nats-io/services";

// ── Spawn a private nats-server on an ephemeral port ──────────────────────
// Avoids touching whatever the developer has running on 4222 — every smoke
// run is fully isolated. We must do this BEFORE importing the channel
// factory below: the factory reads `NATS_URL` at activation time, and we
// want it pointing at our private server rather than at the user's
// `localhost` context.
const __dirname = dirname(fileURLToPath(import.meta.url));
const PORTS_DIR = mkdtempSync(join(tmpdir(), "nats-pi-smoke-ports-"));
const NATS_CONF = join(__dirname, "nats-server.conf");
const spawnedServer = spawn(
	"nats-server",
	["-c", NATS_CONF, "--ports_file_dir", PORTS_DIR],
	{ stdio: "ignore" },
);
spawnedServer.on("error", (err) => {
	console.error(`could not spawn nats-server: ${err.message}`);
	console.error("install nats-server (https://nats.io) before re-running this smoke");
	process.exit(2);
});
process.on("exit", () => {
	if (spawnedServer && !spawnedServer.killed) spawnedServer.kill();
	try {
		rmSync(PORTS_DIR, { recursive: true, force: true });
	} catch {}
});

async function readBoundUrl() {
	const deadline = Date.now() + 5000;
	while (Date.now() < deadline) {
		try {
			const entries = readdirSync(PORTS_DIR).filter((f) => f.endsWith(".ports"));
			if (entries.length > 0) {
				const ports = JSON.parse(readFileSync(join(PORTS_DIR, entries[0]), "utf8"));
				if (Array.isArray(ports.nats) && ports.nats.length > 0) return ports.nats[0];
			}
		} catch {}
		await delay(50);
	}
	throw new Error("nats-server never wrote a ports file");
}
const SERVER_URL = await readBoundUrl();
console.log(`  (spawned nats-server at ${SERVER_URL})`);

// The harness's resolution chain is `NATS_CONTEXT > config.context >
// NATS_URL > default`. The PI agent's persisted config (e.g. a developer's
// `~/.pi/agent/nats-channel.json` pointing at NGS) beats `NATS_URL`, so
// `NATS_URL` alone isn't enough to redirect the harness here. Write a
// throwaway NATS context file with the spawned server's URL and point
// `NATS_CONTEXT` at it — that wins unconditionally.
const NATS_CONTEXT_DIR = join(homedir(), ".config", "nats", "context");
mkdirSync(NATS_CONTEXT_DIR, { recursive: true });
const SMOKE_CONTEXT_NAME = `pi-smoke-${process.pid}`;
const SMOKE_CONTEXT_PATH = join(NATS_CONTEXT_DIR, `${SMOKE_CONTEXT_NAME}.json`);
writeFileSync(
	SMOKE_CONTEXT_PATH,
	JSON.stringify({ url: SERVER_URL, description: "pi smoke ephemeral" }),
);
process.on("exit", () => {
	try {
		rmSync(SMOKE_CONTEXT_PATH, { force: true });
	} catch {}
});

process.env.NATS_CONTEXT = SMOKE_CONTEXT_NAME;
process.env.NATS_SESSION_NAME = `smoke-${process.pid}`;
process.env.USER = process.env.USER || "smoke";

const { default: channelFactory } = await import("../extensions/nats-channel.ts");
const { formatHumanBytes } = await import("@synadia-ai/agents");
const { DEFAULT_MAX_PAYLOAD } = await import("@synadia-ai/agent-service");

let ok = 0;
let fail = 0;
function step(name, fn) {
	return async () => {
		try {
			await fn();
			console.log(`  ✓ ${name}`);
			ok++;
		} catch (e) {
			console.error(`  ✗ ${name}\n      ${e.message}`);
			fail++;
		}
	};
}

// ── Mock ExtensionAPI ──────────────────────────────────────────────────────
const listeners = new Map();
let pendingSendUserMessage = null;
const mockCtx = {
	cwd: process.cwd(),
	isIdle: () => true,
	ui: {
		notify: (line, _level) => console.log(`    [notify] ${line}`),
		setStatus: (_key, _val) => {},
	},
};
const mockPi = {
	on(event, cb) {
		if (!listeners.has(event)) listeners.set(event, []);
		listeners.get(event).push(cb);
	},
	sendUserMessage(text) {
		pendingSendUserMessage = text;
	},
	registerCommand(_name, _spec) {},
};

function emit(event, ...args) {
	const cbs = listeners.get(event) ?? [];
	return Promise.all(cbs.map((cb) => cb(...args, mockCtx)));
}

// ── Observer NATS connection (separate from the one the extension opens) ──
const obs = await connect({ servers: SERVER_URL });

// Subscribe to heartbeats BEFORE the extension registers — §8.5.
// v0.3 wildcard: `agents.hb.<agent>.<owner>.<name>`.
const hbSub = obs.subscribe("agents.hb.*.*.*");
const heartbeats = [];
(async () => {
	for await (const m of hbSub) {
		try {
			heartbeats.push(JSON.parse(new TextDecoder().decode(m.data)));
		} catch {}
	}
})();

// ── Boot the extension ────────────────────────────────────────────────────
channelFactory(mockPi);
await emit("session_start", {});

// Let registration + first heartbeat settle.
await delay(500);

const owner = (process.env.USER ?? "smoke")
	.replace(/[^a-zA-Z0-9_-]/g, "-")
	.toLowerCase()
	.replace(/^-+|-+$/g, "");
const session = process.env.NATS_SESSION_NAME;
const expectedSubject = `agents.prompt.pi.${owner}.${session}`;
const expectedStatusSubject = `agents.status.pi.${owner}.${session}`;

// ── Tests ──────────────────────────────────────────────────────────────────

await step("$SRV.INFO returns spec-shaped service info", async () => {
	const svcm = new Svcm(obs);
	const client = svcm.client({ strategy: "stall", maxWait: 1000, maxMessages: 20 });
	const iter = await client.info("agents");
	const infos = [];
	for await (const si of iter) infos.push(si);
	const mine = infos.find(
		(si) => si.metadata?.agent === "pi" && si.metadata?.session === session,
	);
	assert.ok(mine, `no agents-service instance with session=${session} found`);
	assert.equal(mine.metadata.agent, "pi");
	assert.equal(mine.metadata.owner, owner);
	assert.equal(mine.metadata.protocol_version, "0.3");
	assert.ok(mine.metadata.session.length > 0);

	const ep = mine.endpoints?.find((e) => e.name === "prompt");
	assert.ok(ep, "prompt endpoint missing");
	assert.equal(ep.subject, expectedSubject);
	assert.equal(ep.queue_group, "agents", "prompt endpoint must register queue_group=agents (spec §3.3)");
	// max_payload is server-driven (`nc.info.max_payload`), so verify it
	// matches the §2.1 grammar AND the value our server is advertising.
	const expectedMaxPayload = obs.info?.max_payload
		? formatHumanBytes(obs.info.max_payload)
		: DEFAULT_MAX_PAYLOAD;
	assert.match(ep.metadata?.max_payload ?? "", /^\d+(B|KB|MB|GB)$/);
	assert.equal(ep.metadata?.max_payload, expectedMaxPayload);
	assert.equal(ep.metadata?.attachments_ok, "true");
})();

await step("heartbeat published on agents.hb.pi.{owner}.{session}", async () => {
	const mine = heartbeats.find(
		(hb) => hb.agent === "pi" && hb.session === session && hb.owner === owner,
	);
	assert.ok(mine, "no matching heartbeat received");
	assert.equal(typeof mine.instance_id, "string");
	assert.equal(typeof mine.ts, "string");
	assert.equal(mine.interval_s, 30);
})();

await step("status endpoint replies with a heartbeat-shaped payload", async () => {
	const reply = await obs.request(expectedStatusSubject, "", { timeout: 2000 });
	const body = JSON.parse(new TextDecoder().decode(reply.data));
	assert.equal(body.agent, "pi");
	assert.equal(body.owner, owner);
	assert.equal(body.session, session);
	assert.equal(typeof body.instance_id, "string");
	assert.equal(typeof body.ts, "string");
	assert.equal(body.interval_s, 30);
})();

// Helper — consume a stream until terminator, capturing chunks + error.
async function collectStream(requestSubject, payload) {
	const inbox = createInbox();
	const sub = obs.subscribe(inbox);
	const chunks = [];
	let error = null;
	let terminator = null;
	const done = (async () => {
		for await (const msg of sub) {
			const code = msg.headers?.get("Nats-Service-Error-Code");
			const hasHeaders = msg.headers && [...msg.headers].length > 0;
			if (code) {
				error = {
					code: Number(code),
					description: msg.headers?.get("Nats-Service-Error") ?? "",
				};
				continue;
			}
			if (msg.data.byteLength === 0 && !hasHeaders) {
				terminator = { hasHeaders: false };
				sub.unsubscribe();
				return;
			}
			chunks.push(new TextDecoder().decode(msg.data));
		}
	})();
	obs.publish(requestSubject, payload, { reply: inbox });
	await Promise.race([done, delay(3000)]);
	sub.unsubscribe();
	return { chunks, error, terminator };
}

await step("empty payload → 400 + terminator", async () => {
	const { chunks, error, terminator } = await collectStream(expectedSubject, "");
	assert.ok(error, "expected error response");
	assert.equal(error.code, 400);
	assert.ok(terminator, "expected empty terminator after error");
	assert.equal(chunks.length, 0);
})();

await step("invalid base64 in attachment → 400", async () => {
	// Contains non-base64 characters.
	const env = JSON.stringify({
		prompt: "hi",
		attachments: [{ filename: "x.txt", content: "not base64!!" }],
	});
	const { error } = await collectStream(expectedSubject, env);
	assert.ok(error);
	assert.equal(error.code, 400);
})();

await step("URL-safe base64 in attachment → 400 (strict RFC 4648 §4)", async () => {
	// `-` and `_` are URL-safe alphabet, forbidden by spec §5.2.
	const env = JSON.stringify({
		prompt: "hi",
		attachments: [{ filename: "x.txt", content: "aGVsbG8-_w==" }],
	});
	const { error } = await collectStream(expectedSubject, env);
	assert.ok(error);
	assert.equal(error.code, 400);
})();

await step("path-traversal filename → 400", async () => {
	const env = JSON.stringify({
		prompt: "hi",
		attachments: [{ filename: "../../etc/passwd", content: "aGVsbG8=" }],
	});
	const { error } = await collectStream(expectedSubject, env);
	assert.ok(error);
	assert.equal(error.code, 400);
})();

await step("absolute-path filename → 400", async () => {
	const env = JSON.stringify({
		prompt: "hi",
		attachments: [{ filename: "/etc/passwd", content: "aGVsbG8=" }],
	});
	const { error } = await collectStream(expectedSubject, env);
	assert.ok(error);
	assert.equal(error.code, 400);
})();

await step("malformed JSON → 400 + terminator", async () => {
	const { error, terminator } = await collectStream(expectedSubject, "{not json");
	assert.ok(error);
	assert.equal(error.code, 400);
	assert.ok(terminator);
})();

await step("JSON envelope missing prompt → 400", async () => {
	const { error } = await collectStream(expectedSubject, '{"hello":"world"}');
	assert.ok(error);
	assert.equal(error.code, 400);
})();

await step("plain-text prompt → ack → response chunks → terminator", async () => {
	const env = "Hello, world.";
	pendingSendUserMessage = null;
	const inbox = createInbox();
	const sub = obs.subscribe(inbox);
	const observed = [];
	let terminatorNoHeaders = false;
	const done = (async () => {
		for await (const msg of sub) {
			const code = msg.headers?.get("Nats-Service-Error-Code");
			const hasHeaders = msg.headers && [...msg.headers].length > 0;
			if (code) {
				observed.push({ kind: "error", code: Number(code) });
				continue;
			}
			if (msg.data.byteLength === 0 && !hasHeaders) {
				terminatorNoHeaders = true;
				sub.unsubscribe();
				return;
			}
			const parsed = JSON.parse(new TextDecoder().decode(msg.data));
			observed.push({ kind: "chunk", ...parsed });
		}
	})();
	obs.publish(expectedSubject, env, { reply: inbox });

	// Wait for the extension to inject into our mock PI.
	for (let i = 0; i < 50 && pendingSendUserMessage === null; i++) await delay(20);
	assert.equal(pendingSendUserMessage, env);

	// Simulate PI producing text_delta events, then agent_end.
	await emit("message_update", {
		assistantMessageEvent: { type: "text_delta", delta: "Hi " },
	});
	await emit("message_update", {
		assistantMessageEvent: { type: "text_delta", delta: "there!" },
	});
	await emit("agent_end", {});

	await Promise.race([done, delay(2000)]);
	sub.unsubscribe();

	assert.ok(terminatorNoHeaders, "stream did not end with empty-no-headers terminator");

	// First observed chunk MUST be status:ack per §6.4.
	assert.deepEqual(
		{ type: observed[0]?.type, data: observed[0]?.data },
		{ type: "status", data: "ack" },
	);
	// Subsequent response chunks concatenate to the emitted deltas.
	const text = observed
		.filter((o) => o.kind === "chunk" && o.type === "response")
		.map((o) => o.data)
		.join("");
	assert.equal(text, "Hi there!");
})();

// Track where the agent stages attachments so we can verify cleanup later.
const attachmentsSessionDir = join(homedir(), ".pi", "agent", "attachments", session);
let stagedPathSeen = null;

await step("valid attachment → file on disk, prompt augmented, stream ok", async () => {
	const payloadBytes = Buffer.from("hello attachment", "utf8");
	const content = payloadBytes.toString("base64"); // standard, padded
	const env = JSON.stringify({
		prompt: "describe the file",
		attachments: [{ filename: "hello.txt", content }],
	});
	pendingSendUserMessage = null;

	const inbox = createInbox();
	const sub = obs.subscribe(inbox);
	const observed = [];
	let terminatorNoHeaders = false;
	const done = (async () => {
		for await (const msg of sub) {
			const hasHeaders = msg.headers && [...msg.headers].length > 0;
			if (msg.data.byteLength === 0 && !hasHeaders) {
				terminatorNoHeaders = true;
				sub.unsubscribe();
				return;
			}
			if (hasHeaders) continue;
			observed.push(JSON.parse(new TextDecoder().decode(msg.data)));
		}
	})();
	obs.publish(expectedSubject, env, { reply: inbox });

	for (let i = 0; i < 100 && pendingSendUserMessage === null; i++) await delay(20);
	assert.ok(pendingSendUserMessage, "pi.sendUserMessage was not called");

	// The prompt handed to PI should start with the [Attachments] block and
	// contain the original prompt text.
	assert.match(pendingSendUserMessage, /^\[Attachments available at the following absolute paths\]/);
	assert.ok(
		pendingSendUserMessage.endsWith("describe the file"),
		"original prompt text missing from augmented prompt",
	);

	// Extract the staged path from the augmented prompt and verify the bytes
	// landed correctly on disk.
	const match = pendingSendUserMessage.match(/^- (\S.*)$/m);
	assert.ok(match, "no staged path found in augmented prompt");
	const stagedPath = match[1];
	stagedPathSeen = stagedPath;
	assert.ok(existsSync(stagedPath), `staged file missing at ${stagedPath}`);
	assert.deepEqual(readFileSync(stagedPath), payloadBytes);
	assert.ok(stagedPath.startsWith(attachmentsSessionDir + "/"), "staged path outside session dir");
	assert.ok(stagedPath.endsWith("/hello.txt"), "staged filename mismatched");

	// Drive a trivial response stream so the cycle completes cleanly.
	await emit("message_update", {
		assistantMessageEvent: { type: "text_delta", delta: "ok" },
	});
	await emit("agent_end", {});

	await Promise.race([done, delay(2000)]);
	sub.unsubscribe();
	assert.ok(terminatorNoHeaders, "terminator not observed");
	// First observed JSON chunk should be the ack.
	assert.deepEqual(
		{ type: observed[0]?.type, data: observed[0]?.data },
		{ type: "status", data: "ack" },
	);
})();

// ── Teardown ───────────────────────────────────────────────────────────────
await emit("session_shutdown", {});
await delay(200);
hbSub.unsubscribe();
await obs.drain();

await step("session_shutdown removes the staged attachments directory", async () => {
	assert.ok(stagedPathSeen, "no staged path captured in earlier test");
	assert.equal(
		existsSync(attachmentsSessionDir),
		false,
		`attachments dir ${attachmentsSessionDir} was not cleaned up on session_shutdown`,
	);
})();

console.log(`\n${ok} passed, ${fail} failed`);
process.exit(fail === 0 ? 0 : 1);
