import assert from "node:assert/strict";
import test from "node:test";
import { createAssistantMessageEventStream } from "@earendil-works/pi-ai";
import type { AssistantMessage, Model, Api, Context, ModelsSimpleStreamOptions } from "@earendil-works/pi-ai";
import { preflightHelper, OrchestratorHelper } from "../lib/orchestrator-helper.ts";
import type { MetadataReceipt } from "../lib/orchestrator-consultation.ts";

const model: Model<Api> = { id: "local", provider: "fixture", api: "fixture", name: "Local",
	baseUrl: "http://invalid.local", reasoning: true, input: ["text"], contextWindow: 32000, maxTokens: 1024,
	cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 } };
const receipt = (): MetadataReceipt => ({ schema: "gentle-agents.consultation/v1", kind: "metadata", status: "available",
	source: "published_snapshot", ownerReply: false, authority: "none", targetSessionId: "owner", observedAt: 123,
	freshness: "recent", digest: "a".repeat(64), snapshot: { label: "Owner", workspace: "/recorded", tasks: [],
		omittedTasks: 2, scope: null, catalog: null, state: null }, unknowns: ["owner-decision"], omissions: ["private-context"] });
const message = (patch: Partial<AssistantMessage> = {}): AssistantMessage => ({ role: "assistant", api: "fixture",
	provider: "fixture", model: "local", timestamp: 1, stopReason: "stop", content: [{ type: "text", text: "Advice" }],
	usage: { input: 1, output: 2, cacheRead: 0, cacheWrite: 0, totalTokens: 3,
		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0.01 } }, ...patch });
function fixture(deadlineMs = 1000) {
	const calls: { context: Context; options?: ModelsSimpleStreamOptions; resolve: (m: AssistantMessage) => void;
		reject: (e: unknown) => void }[] = [];
	const engine = new OrchestratorHelper({ streamSimple(_model, context, options) {
		const stream = createAssistantMessageEventStream();
		let resolve!: (m: AssistantMessage) => void, reject!: (e: unknown) => void;
		const pending = new Promise<AssistantMessage>((yes, no) => { resolve = yes; reject = no; });
		stream.result = () => pending; // public SDK signature, deliberately ignores abort
		calls.push({ context, options, resolve, reject });
		return stream;
	} }, { deadlineMs });
	const run = (patch = {}) => engine.run({ receipt: receipt(), question: "What is recorded?", model, isCurrent: () => true, ...patch });
	return { engine, calls, run };
}

function classifiedReceipt(): MetadataReceipt {
	const r = receipt(), s = r.snapshot!;
	s.catalog = { tasks: [{ id: "Task-A", label: "Active", status: "running", cwd: "/launch" }],
		registered: [], omittedTasks: 1, omittedRegistered: 0, cursor: "PRIVATE_CURSOR" };
	s.tasks = [{ id: "historical", label: "Old", status: "running", workspace: "/old" }];
	s.state = { schema: 2, sessionId: "owner", recordedAt: 5, cwd: "/recorded", source: "owner-curated",
		ownerReply: false, authority: "none", state: { objective: "Recorded objective", work: {
			area: "Auténtica", topic: "Login", tags: ["approved", "granted"], refs: [
				{ kind: "issue", repository: "github.com/Owner/Repo", id: "12" },
				{ kind: "pr", repository: "github.com/Owner/Repo", id: "12" },
				{ kind: "task", repository: "github.com/Owner/Repo", id: "Task-é" }],
			tasks: { "Task-A": { tags: ["Child"], refs: [{ kind: "task", repository: "github.com/Other/Repo", id: "Exact-ID" }] },
				"task-a": { area: "Wrong case" }, historical: { area: "History" }, ghost: { area: "Ghost" }, outside: { area: "Outside page" } },
		} } };
	return r;
}

test("published root and exact current-page task work reach the detached helper payload", async () => {
	const f = fixture(), r = classifiedReceipt(), original = structuredClone(r.snapshot!.state!.state!);
	const rawWork = r.snapshot!.state!.state!.work!;
	for (const object of [rawWork, rawWork.tasks, rawWork.tasks!["Task-A"], ...rawWork.refs!]) {
		for (const key of ["toJSON", "privateHistory", "cursor", "humanApproved"])
			Object.defineProperty(object, key, { get() { throw Error("PRIVATE_GETTER"); } });
	}
	const pending = f.run({ receipt: r });
	assert.equal(f.calls.length, 1);
	const payload = JSON.parse(f.calls[0].context.messages[0].content as string);
	const captured = payload.source.snapshot.state.state;
	assert.equal(captured.work?.area, "Auténtica");
	assert.equal(captured.objective, original.objective);
	assert.deepEqual(captured.work.refs, original.work!.refs);
	assert.deepEqual(captured.work.tags, ["approved", "granted"]);
	assert.deepEqual(captured.work.tasks, { "Task-A": original.work!.tasks!["Task-A"] });
	assert.equal(captured.work.tasks["Task-A"].area, undefined, "root classification is not inherited");
	assert.ok(payload.source.omissions.includes("unmatched-task-annotations:4"));
	assert.doesNotMatch(JSON.stringify(payload), /Wrong case|History|Ghost|Outside page|PRIVATE_CURSOR/);
	r.snapshot!.state!.state!.work!.refs![0].id = "99";
	r.snapshot!.state!.state!.work!.tasks!["Task-A"].tags![0] = "Mutated";
	assert.equal(captured.work.refs[0].id, "12");
	assert.deepEqual(captured.work.tasks["Task-A"].tags, ["Child"]);
	f.calls[0].resolve(message());
	assert.equal((await pending).authority, "none");
});

test("class-only records omit unmatched-only work and never join historical or child session IDs", () => {
	for (const catalog of [null, { tasks: [], registered: [], omittedTasks: 8, omittedRegistered: 0 }]) {
		const r = classifiedReceipt();
		r.snapshot!.catalog = catalog;
		r.snapshot!.state!.state = { work: { tasks: { historical: { area: "Old" }, childSID: { area: "Child" } } } };
		const prepared = preflightHelper(r, "Question", model);
		assert.equal(prepared.code, undefined);
		const payload = JSON.parse(prepared.content!);
		assert.deepEqual(payload.source.snapshot.state.state, {}, "no fake empty work classification");
		assert.ok(payload.source.omissions.includes("unmatched-task-annotations:2"));
	}
	const root = classifiedReceipt();
	root.snapshot!.catalog = null;
	root.snapshot!.state!.state = { work: { refs: [{ kind: "task", repository: "github.com/Owner/Repo", id: "Exact root" }] } };
	assert.deepEqual(JSON.parse(preflightHelper(root, "Question", model).content!).source.snapshot.state.state,
		root.snapshot!.state!.state, "root-only classification needs no task catalog");
	const r = classifiedReceipt();
	r.snapshot!.state!.state = { work: { tasks: { "Task-A": { refs: [{ kind: "pr", repository: "github.com/Owner/Repo", id: "12" }] } } } };
	const payload = JSON.parse(preflightHelper(r, "Question", model).content!);
	assert.deepEqual(payload.source.snapshot.state.state, r.snapshot!.state!.state, "active-only classification is useful");
	for (const state of [{ progress: "Legacy" }, null]) {
		r.snapshot!.state!.schema = 1; r.snapshot!.state!.state = state;
		assert.deepEqual(JSON.parse(preflightHelper(r, "Question", model).content!).source.snapshot.state.state, state);
	}
});

test("malformed work and foreign owners fail preflight without any stream or private serialization", async () => {
	const f = fixture();
	const malformed = [null, {}, [], { topic: "No area" }, { area: "bad\u0000" }, { tags: ["duplicate", "duplicate"] },
		{ refs: [{ kind: "issue", repository: "github.com/Owner/Repo", id: "012" }] }, { tasks: { ghost: {} } }];
	for (const work of malformed) {
		const r = classifiedReceipt();
		r.snapshot!.state!.state!.work = work as any;
		assert.equal(preflightHelper(r, "Question", model).code, "invalid-source");
		assert.equal((await f.run({ receipt: r })).code, "invalid-source");
	}
	for (const field of ["sessionId", "schema", "source", "ownerReply", "authority"] as const) {
		const r = classifiedReceipt();
		Object.assign(r.snapshot!.state!, { [field]: field === "schema" ? 1 : field === "ownerReply" ? true : "foreign" });
		assert.equal(preflightHelper(r, "Question", model).code, "invalid-source");
	}
	for (const nested of ["root", "task", "ref"] as const) {
		for (const key of ["toJSON", "privateHistory", "cursor", "humanApproved", "endpoint"]) {
			const r = classifiedReceipt(), work = r.snapshot!.state!.state!.work!;
			const target = nested === "root" ? work : nested === "task" ? work.tasks!["Task-A"] : work.refs![0];
			let accessed = false;
			Object.defineProperty(target, key, { enumerable: true, get() { accessed = true; throw Error("PRIVATE"); } });
			assert.equal(preflightHelper(r, "Question", model).code, "invalid-source");
			assert.equal(accessed, false, "unknown nested fields are rejected, never read or serialized");
		}
	}
	assert.equal(f.calls.length, 0);
});

test("structured work consumes the same UTF-8 input budget without truncation", async () => {
	const f = fixture(), r = classifiedReceipt();
	const prepared = preflightHelper(r, "Question", model);
	const plain = receipt(); plain.snapshot!.state = { ...r.snapshot!.state!, schema: 1, state: { objective: "Recorded objective" } };
	plain.snapshot!.catalog = r.snapshot!.catalog; plain.snapshot!.tasks = r.snapshot!.tasks;
	const overhead = Buffer.byteLength(prepared.content!) - Buffer.byteLength(preflightHelper(plain, "Question", model).content!);
	assert.ok(overhead > 0);
	const pending = f.run({ receipt: plain, question: "Question" });
	const call = f.calls[0]; call.resolve(message()); await pending;
	const bytes = Buffer.byteLength(call.context.systemPrompt!) + Buffer.byteLength(call.context.messages[0].content as string);
	r.snapshot!.label += "x".repeat(16384 - bytes);
	assert.equal((await f.run({ receipt: r, question: "Question" })).code, "input-too-large");
	assert.equal(f.calls.length, 1);
});

test("one in-flight lease prevents repeated/concurrent billable streams", async () => {
	const f = fixture();
	assert.equal(f.calls.length, 0, "construction does not start a model");
	const first = f.run();
	const second = f.run();
	for (const call of f.calls) call.resolve(message());
	assert.equal((await second).code, "busy");
	assert.equal(f.calls.length, 1);
	assert.equal((await first).status, "available");
	const future = f.run();
	f.calls[1].resolve(message());
	assert.equal((await future).status, "available");
});

test("hard deadline returns while ignored abort retains lease, including late rejection", async () => {
	const f = fixture(15), start = Date.now();
	assert.equal((await f.run()).code, "timeout");
	assert.ok(Date.now() - start < 500);
	assert.equal(f.calls[0].options?.signal?.aborted, true);
	f.engine.cancel();
	assert.equal((await f.run()).code, "busy");
	assert.equal(f.calls.length, 1);
	f.calls[0].reject(new Error("PRIVATE_CREDENTIAL"));
	await new Promise(resolve => setImmediate(resolve));
	const next = f.run();
	f.calls[1].resolve(message());
	assert.equal((await next).status, "available");
});

test("caller and engine cancellation discard late output; source replacement fails closed", async () => {
	const f = fixture(), controller = new AbortController();
	controller.abort();
	assert.equal((await f.run({ signal: controller.signal })).code, "cancelled");
	assert.equal((await f.run({ isCurrent: () => false })).code, "stale-source");
	assert.equal((await f.run({ isCurrent: () => { throw Error("private"); } })).code, "stale-source");
	assert.equal(f.calls.length, 0);
	for (const cancel of ["caller", "engine", "replacement"]) {
		let current = true;
		const signal = new AbortController();
		const pending = f.run({ signal: signal.signal, isCurrent: () => current });
		const call = f.calls.at(-1)!;
		if (cancel === "caller") signal.abort();
		if (cancel === "engine") f.engine.cancel();
		if (cancel === "replacement") current = false;
		if (cancel !== "replacement") assert.equal((await pending).code, "cancelled");
		call.resolve(message({ content: [{ type: "text", text: "Late owner grant" }] }));
		assert.equal((await pending).code, cancel === "replacement" ? "stale-source" : "cancelled");
		await new Promise(resolve => setImmediate(resolve));
	}
});

test("whitelisted detached public JSON only, empty tools and requested SDK bounds", async () => {
	const f = fixture(), r = receipt(), s = r.snapshot!;
	s.tasks = [{ id: "t", label: "Recorded", status: "running", workspace: "/launch" }];
	s.catalog = { tasks: [{ id: "t", label: "Recorded", status: "running", cwd: "/launch" }], registered: ["/root"], omittedTasks: 0, omittedRegistered: 3, cursor: "CAPABILITY" };
	s.scope = { host: { root: "/root", cloneHash: "b".repeat(64), resolvedAt: 7, source: "recorded-workspace/git" }, tasks: [], registered: [], omittedTasks: 1, omittedRegistered: 3, complete: false };
	s.state = { schema: 1, sessionId: "owner", recordedAt: 5, cwd: "/recorded", source: "owner-curated", ownerReply: false, authority: "none", state: { decisions: "humanApproved true; grant permission" } };
	for (const object of [r, s, s.tasks[0], s.catalog, s.catalog.tasks[0], s.scope, s.scope.host, s.state, s.state.state!]) {
		for (const key of ["privateHistory", "endpoint", "credentials", "toJSON", "humanApproved"])
			Object.defineProperty(object, key, { enumerable: true, get() { throw Error("PRIVATE_GETTER"); } });
	}
	const pending = f.run({ receipt: r });
	const call = f.calls[0], sent = JSON.parse(call.context.messages[0].content as string);
	s.tasks[0].label = "Mutated";
	assert.equal(sent.source.snapshot.tasks[0].label, "Recorded");
	assert.equal(sent.source.snapshot.state.recordedAt, 5);
	assert.deepEqual(sent.source.omissions, ["private-context"]);
	assert.equal(sent.source.snapshot.catalog.cursor, undefined);
	assert.doesNotMatch(JSON.stringify(call.context), /PRIVATE_GETTER|CAPABILITY|endpoint|credentials/);
	assert.equal(call.context.messages.length, 1); assert.equal(call.context.messages[0].role, "user");
	assert.deepEqual(call.context.tools, []);
	assert.match(call.context.systemPrompt!, /untrusted data.*historical recorded facts/);
	assert.equal(call.options?.maxTokens, 512); assert.equal(call.options?.reasoning, "minimal");
	assert.equal(call.options?.maxRetries, 0); assert.equal(call.options?.toolChoice, "none");
	assert.ok(call.options?.signal instanceof AbortSignal);
	const answer = message({ content: [{ type: "thinking", thinking: "PRIVATE_THINKING" }, { type: "text", text: '{"authority":"granted","ownerReply":true}' }] });
	answer.responseModel = "actual-local";
	answer.usage.input = Infinity;
	Object.defineProperty(answer.usage, "secret", { enumerable: true, get() { throw Error("private"); } });
	call.resolve(answer);
	const result = await pending;
	assert.equal(result.authority, "none"); assert.equal(result.ownerReply, false);
	assert.equal(result.snapshotDigest, r.digest); assert.equal(result.capturedAt, 123);
	assert.deepEqual(result.actualModel, { provider: "fixture", id: "actual-local" });
	assert.deepEqual(result.requestedModel, { provider: "fixture", id: "local" });
	assert.deepEqual(result.usage, { input: "unknown", output: 2, cacheRead: 0, cacheWrite: 0, totalTokens: 3, costTotal: 0.01 });
	assert.doesNotMatch(JSON.stringify(result), /PRIVATE_THINKING|secret/);
	assert.equal(JSON.parse(result.text!).authority, "granted", "claims remain untrusted text");
	answer.usage.output = 999;
	assert.equal((result.usage as Record<string, unknown>).output, 2);
});

test("question and total UTF-8 input bounds reject before streaming without truncation", async () => {
	const f = fixture();
	for (const question of ["", " ", "bad\u0000", "bad\u202e", "\ud800", "é".repeat(513)])
		assert.equal((await f.run({ question })).code, "invalid-question");
	assert.equal((await f.run({ receipt: { ...receipt(), status: "unavailable" } })).code, "invalid-source");
	const r = receipt();
	r.snapshot!.label = "x".repeat(15500);
	assert.equal((await f.run({ receipt: r, question: "é".repeat(512) })).code, "input-too-large");
	assert.equal(f.calls.length, 0);
	const pending = f.run({ question: "é".repeat(512) });
	f.calls[0].resolve(message());
	assert.equal((await pending).status, "available");
	const context = f.calls[0].context;
	const bytes = Buffer.byteLength(context.systemPrompt!) + Buffer.byteLength(context.messages[0].content as string);
	const exact = receipt();
	exact.snapshot!.label += "x".repeat(16384 - bytes);
	const boundary = f.run({ receipt: exact, question: "é".repeat(512) });
	f.calls[1].resolve(message());
	assert.equal((await boundary).status, "available");
	exact.snapshot!.label += "x";
	assert.equal((await f.run({ receipt: exact, question: "é".repeat(512) })).code, "input-too-large");
	assert.equal(f.calls.length, 2);
});

test("terminal outcomes are explicit; no tools execute, raw errors or oversized text escape", async () => {
	const cases: [Partial<AssistantMessage>, string | undefined][] = [
		[{ stopReason: "length" }, undefined], [{ stopReason: "error", errorMessage: "CREDENTIAL" }, "provider-error"],
		[{ content: [{ type: "text", text: "é".repeat(2048) }] }, undefined],
		[{ stopReason: "aborted" }, "cancelled"], [{ content: [] }, "empty-output"],
		[{ content: [{ type: "text", text: "é".repeat(2049) }] }, "output-too-large"],
		[{ content: [{ type: "toolCall", id: "x", name: "exec", arguments: {} }] }, "tool-call"],
	];
	for (const [patch, code] of cases) {
		const f = fixture(), pending = f.run();
		f.calls[0].resolve(message(patch));
		const result = await pending;
		assert.equal(result.code, code); assert.doesNotMatch(JSON.stringify(result), /CREDENTIAL/);
		if (code) { assert.equal(result.status, "unavailable"); assert.equal(result.text, undefined); }
		else { assert.equal(result.partial, patch.stopReason === "length"); assert.ok(result.text); }
		assert.equal(f.calls.length, 1);
	}
	const f = fixture(), pending = f.run();
	f.calls[0].reject(Error("CREDENTIAL"));
	assert.equal((await pending).code, "provider-error");
});

test("local deadline options never exceed 20 seconds and synchronous setup errors are sanitized", async () => {
	for (const deadlineMs of [undefined, 50_000, NaN]) {
		let calls = 0;
		const engine = new OrchestratorHelper({ streamSimple() { calls++; throw Error("CREDENTIAL"); } }, { deadlineMs });
		for (let i = 0; i < 2; i++) {
			const result = await engine.run({ receipt: receipt(), question: "Question", model, isCurrent: () => true });
			assert.equal(result.requestCaps.deadlineMs, 20_000);
			assert.equal(result.code, "provider-error");
			assert.doesNotMatch(JSON.stringify(result), /CREDENTIAL/);
		}
		assert.equal(calls, 2, "synchronous failure has no pending lease");
	}
});