import { fauxProvider } from "@earendil-works/pi-ai";
import { describe, expect, it, vi } from "vitest";
import {
	BACKGROUND_CONTEXT,
	createContextKey,
	getTelemetryContext,
	withAbortSignal,
	withContextValue,
	withTelemetryContext,
} from "../../src/harness/context.ts";
import { HarnessEventBus } from "../../src/harness/events.ts";
import { AbortRequested, createGate } from "../../src/harness/execution/effect-gate.ts";
import { HookRegistry } from "../../src/harness/hooks.ts";
import { InMemoryTelemetryContext } from "../../src/index.ts";

function deferred(): { promise: Promise<void>; resolve(): void } {
	let resolvePromise: (() => void) | undefined;
	const promise = new Promise<void>((resolve) => {
		resolvePromise = resolve;
	});
	return { promise, resolve: () => resolvePromise?.() };
}

describe("createGate", () => {
	it("closes starts synchronously and signals only after cancellation commits", async () => {
		const { gate, control } = createGate();
		const cancellation = deferred();
		control.beginAbort(cancellation.promise);

		let refusal: unknown;
		try {
			gate.admit(() => undefined);
		} catch (error) {
			refusal = error;
		}
		expect(refusal).toBeInstanceOf(AbortRequested);
		expect((refusal as AbortRequested).cancellation).toBe(cancellation.promise);
		expect(gate.signal.aborted).toBe(false);
		cancellation.resolve();
		await cancellation.promise;
		control.signalAbort();
		expect(gate.signal.aborted).toBe(true);
	});

	it("permanently closes and signals admitted work", () => {
		const { gate, control } = createGate();
		const error = new Error("closed");
		control.close(error);
		expect(() => gate.admit(() => undefined)).toThrow(error);
		expect(gate.signal.aborted).toBe(true);
	});
});

describe("HookRegistry", () => {
	it("aggregates before_run messages in registration order with each handler seeing prior output", async () => {
		const errors: Error[] = [];
		const hooks = new HookRegistry((error) => {
			errors.push(error);
		});
		const second = vi.fn((event: { prompt: Array<{ role: string }> }) => {
			expect(event.prompt).toHaveLength(2);
			return { messages: [{ role: "user" as const, content: "second", timestamp: 3 }] };
		});
		hooks.on("before_run", () => ({ messages: [{ role: "user", content: "first", timestamp: 2 }] }));
		hooks.on("before_run", second);

		const result = await hooks.runWithGate(
			"before_run",
			{
				lane: "main",
				runId: "run",
				prompt: [{ role: "user", content: "prompt", timestamp: 1 }],
				resources: {},
			},
			createGate().gate,
			BACKGROUND_CONTEXT,
		);

		expect(result).toMatchObject({
			messages: [
				{ role: "user", content: "first" },
				{ role: "user", content: "second" },
			],
		});
		expect(second).toHaveBeenCalledOnce();
		expect(errors).toEqual([]);
	});

	it("checks the effect gate immediately before the complete before_run pipeline", async () => {
		const hooks = new HookRegistry(() => {});
		const calls: string[] = [];
		const release = deferred();
		hooks.on("before_run", async () => {
			calls.push("first:start");
			await release.promise;
			calls.push("first:end");
			return undefined;
		});
		hooks.on("before_run", () => {
			calls.push("second");
			return undefined;
		});
		const event = { lane: "main", runId: "run", prompt: [], resources: {} };
		const closed = new Error("closed");
		const closedGate = createGate();
		closedGate.control.close(closed);
		expect(() => hooks.runWithGate("before_run", event, closedGate.gate, BACKGROUND_CONTEXT)).toThrow(closed);
		expect(calls).toEqual([]);

		const running = hooks.runWithGate("before_run", event, createGate().gate, BACKGROUND_CONTEXT);
		expect(calls).toEqual(["first:start"]);
		hooks.on("before_run", () => {
			calls.push("late");
			return undefined;
		});
		hooks.close(closed);
		release.resolve();
		await running;
		expect(calls).toEqual(["first:start", "first:end", "second"]);
	});

	it("rejects pre-aborted invocations and combines admitted hook cancellation with the effect gate", async () => {
		const event = { operation: "run" as const, lane: "main", runId: "run" };
		const preAbortedHooks = new HookRegistry(() => {});
		const handler = vi.fn();
		preAbortedHooks.on("before_drive", handler);
		const preAbortedGate = createGate();
		const invocationController = new AbortController();
		const abortReason = new Error("invocation aborted");
		invocationController.abort(abortReason);

		expect(() =>
			preAbortedHooks.runWithGate(
				"before_drive",
				event,
				preAbortedGate.gate,
				withAbortSignal(invocationController.signal, BACKGROUND_CONTEXT),
			),
		).toThrow(abortReason);
		expect(handler).not.toHaveBeenCalled();
		expect(() => preAbortedGate.gate.admit(() => undefined)).not.toThrow();

		const admittedHooks = new HookRegistry(() => {});
		const admittedGate = createGate();
		let admittedSignal: AbortSignal | undefined;
		admittedHooks.on("before_drive", (_event, context) => {
			admittedSignal = context.abortSignal;
		});
		await admittedHooks.runWithGate("before_drive", event, admittedGate.gate, BACKGROUND_CONTEXT);
		expect(admittedSignal?.aborted).toBe(false);

		const gateReason = new Error("gate closed");
		admittedGate.control.close(gateReason);
		expect(admittedSignal?.aborted).toBe(true);
		expect(admittedSignal?.reason).toBe(gateReason);
	});

	it("passes tool handlers a child context of the active hook span", async () => {
		const telemetry = new InMemoryTelemetryContext();
		const valueKey = createContextKey<string>("hook.test.value");
		const hooks = new HookRegistry(() => {});
		let receivedValue: string | undefined;
		hooks.on("before_tool", async (_event, context) => {
			receivedValue = context.value(valueKey);
			await getTelemetryContext(context).startSpan({ name: "handler.child" }, () => undefined);
			return undefined;
		});

		await telemetry.startSpan({ name: "invocation" }, (invocationSpan) =>
			hooks.runToolWithGate(
				"before_tool",
				{
					lane: "main",
					runId: "run",
					toolCallId: "call",
					toolName: "tool",
					args: {},
				},
				createGate().gate,
				withContextValue(valueKey, "preserved", withTelemetryContext(invocationSpan, BACKGROUND_CONTEXT)),
			),
		);

		expect(receivedValue).toBe("preserved");
		const spans = telemetry.getSpans();
		const invocationSpan = spans.find((span) => span.name === "invocation");
		const hookSpan = spans.find((span) => span.name === "pi.harness.hook");
		const handlerSpan = spans.find((span) => span.name === "handler.child");
		expect(hookSpan?.parentId).toBe(invocationSpan?.id);
		expect(handlerSpan?.parentId).toBe(hookSpan?.id);
	});

	it("treats registration ids as optional metadata and fails before_drive closed", async () => {
		const errors: Error[] = [];
		const hooks = new HookRegistry((error) => {
			errors.push(error);
		});
		const failure = new Error("prerequisite failed");
		const later = vi.fn();
		hooks.on(
			"before_drive",
			() => {
				throw failure;
			},
			{ id: "duplicate" },
		);
		hooks.on("before_drive", later, { id: "duplicate" });

		await expect(
			hooks.runWithGate(
				"before_drive",
				{ lane: "main", runId: "run", operation: "run" },
				createGate().gate,
				BACKGROUND_CONTEXT,
			),
		).rejects.toBe(failure);
		expect(later).not.toHaveBeenCalled();
		expect(errors).toEqual([failure]);
	});

	it("chains transform_context output and isolates handler failures", async () => {
		const errors: Error[] = [];
		const hooks = new HookRegistry((error) => {
			errors.push(error);
		});
		const messages = [{ role: "user" as const, content: "transformed", timestamp: 2 }];
		hooks.on("transform_context", () => ({ messages, systemPrompt: "first" }));
		hooks.on("transform_context", (event) => {
			expect(event.messages).toBe(messages);
			expect(event.systemPrompt).toBe("first");
			throw new Error("ignored transform failure");
		});
		hooks.on("transform_context", (event) => {
			expect(event.messages).toBe(messages);
			expect(event.systemPrompt).toBe("first");
			return { systemPrompt: "final" };
		});

		const result = await hooks.runWithGate(
			"transform_context",
			{
				lane: "main",
				runId: "run",
				messages: [{ role: "user", content: "original", timestamp: 1 }],
				systemPrompt: "base",
			},
			createGate().gate,
			BACKGROUND_CONTEXT,
		);

		expect(result).toEqual({ messages, systemPrompt: "final" });
		expect(errors).toHaveLength(1);
		expect(errors[0]?.message).toBe("ignored transform failure");
	});

	it("preserves clear-all before_request patches across later handlers", async () => {
		const hooks = new HookRegistry(() => {});
		hooks.on("before_request", () => ({ streamOptions: { headers: undefined, metadata: undefined } }));
		hooks.on("before_request", (event) => {
			expect(event.streamOptions.headers).toBeUndefined();
			expect(event.streamOptions.metadata).toBeUndefined();
			return { streamOptions: { headers: { c: "3" }, metadata: { y: 2 } } };
		});
		const result = await hooks.runWithGate(
			"before_request",
			{
				lane: "main",
				runId: "run",
				model: fauxProvider().getModel(),
				step: "assistant",
				attempt: 1,
				streamOptions: { headers: { a: "1", b: "2" }, metadata: { x: 1 } },
			},
			createGate().gate,
			BACKGROUND_CONTEXT,
		);
		expect(result).toEqual({
			streamOptions: {
				headers: { a: undefined, b: undefined, c: "3" },
				metadata: { x: undefined, y: 2 },
			},
		});
	});

	it("preserves earlier after_tool fields when a later patch returns undefined", async () => {
		const hooks = new HookRegistry(() => {});
		hooks.on("after_tool", () => ({ content: [{ type: "text", text: "patched" }] }));
		hooks.on("after_tool", () => ({ content: undefined, isError: false }));
		const result = await hooks.runWithGate(
			"after_tool",
			{
				lane: "main",
				runId: "run",
				toolCallId: "call",
				toolName: "tool",
				args: {},
				content: [{ type: "text", text: "raw" }],
				isError: true,
			},
			createGate().gate,
			BACKGROUND_CONTEXT,
		);
		expect(result).toEqual({ content: [{ type: "text", text: "patched" }], isError: false });
	});

	it("accepts explicit false structural declines and rejects true conflicts", async () => {
		const errors: Error[] = [];
		const hooks = new HookRegistry((error) => {
			errors.push(error);
		});
		const ignored = { summary: "ignored", readFiles: [], modifiedFiles: [] };
		const selected = { summary: "selected", readFiles: [], modifiedFiles: [] };
		hooks.on("before_navigation", () => ({ decline: true, summary: ignored }));
		hooks.on("before_navigation", () => ({ decline: false, summary: selected }));

		const result = await hooks.runWithGate(
			"before_navigation",
			{
				lane: "main",
				runId: "run",
				targetId: "target",
				preparation: {
					messages: [],
					fileOps: { read: new Set(), written: new Set(), edited: new Set() },
					totalTokens: 0,
				},
			},
			createGate().gate,
			BACKGROUND_CONTEXT,
		);
		expect(result).toEqual({ decline: false, summary: selected });
		expect(errors).toHaveLength(1);
		expect(errors[0].message).toMatch(/cannot return both decline and summary/);
	});

	it("admits the complete before_drive pipeline as one gated effect", async () => {
		const hooks = new HookRegistry(() => {});
		const calls: string[] = [];
		const release = deferred();
		hooks.on("before_drive", async () => {
			calls.push("first:start");
			await release.promise;
			calls.push("first:end");
		});
		hooks.on("before_drive", () => {
			calls.push("second");
		});
		const event = { operation: "run" as const, lane: "main", runId: "run" };

		const abortFirstGate = createGate();
		const cancellation = Promise.resolve();
		abortFirstGate.control.beginAbort(cancellation);
		expect(() => hooks.runWithGate("before_drive", event, abortFirstGate.gate, BACKGROUND_CONTEXT)).toThrow(
			AbortRequested,
		);
		expect(calls).toEqual([]);

		const startFirstGate = createGate();
		const running = hooks.runWithGate("before_drive", event, startFirstGate.gate, BACKGROUND_CONTEXT);
		expect(calls).toEqual(["first:start"]);
		startFirstGate.control.beginAbort(cancellation);
		startFirstGate.control.signalAbort();
		release.resolve();
		await running;
		expect(calls).toEqual(["first:start", "first:end", "second"]);
	});
});

describe("HarnessEventBus", () => {
	it("buffers between snapshot and start, then delivers each event once in order", async () => {
		const bus = new HarnessEventBus();
		const watcher = bus.watch({ tipId: null }, () => true, BACKGROUND_CONTEXT);
		await bus.emit({ type: "run_start", runId: "one", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		const seen: string[] = [];
		watcher.start((event) => {
			seen.push(`${event.type}:${"runId" in event ? event.runId : ""}`);
		});
		await bus.emit({ type: "run_start", runId: "two", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await new Promise((resolve) => setTimeout(resolve, 0));
		expect(watcher.snapshot).toEqual({ tipId: null });
		expect(seen).toEqual(["run_start:one", "run_start:two"]);
		watcher.unsubscribe();
		await bus.emit({ type: "run_start", runId: "three", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		expect(seen).toHaveLength(2);
	});

	it("drops pre-snapshot delivery and holds later events when resnapshotting inside a listener", async () => {
		const bus = new HarnessEventBus();
		const listenerStarted = deferred();
		const releaseListener = deferred();
		const resnapshotDone = deferred();
		const watcher = bus.watch(
			{ version: "old" },
			() => true,
			BACKGROUND_CONTEXT,
			async (_context, markBoundary) => {
				markBoundary();
				return { version: "fresh" };
			},
		);
		const queueEvents: string[] = [];
		watcher.start(async (event, context) => {
			if (event.type === "run_start" && event.runId === "blocking") {
				listenerStarted.resolve();
				await releaseListener.promise;
			} else if (event.type === "navigation_end") {
				await watcher.resnapshot(context);
				resnapshotDone.resolve();
			} else if (event.type === "queue_update") {
				queueEvents.push(event.queues[0]?.entryId ?? "empty");
			}
		});
		await bus.emit({ type: "run_start", runId: "blocking", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await listenerStarted.promise;
		const queued = bus.emitBatch(
			[
				{
					type: "navigation_end",
					runId: "navigation",
					status: "completed",
					fromTipId: null,
					tipId: null,
					endedAt: 2,
					lane: "main",
				},
				{
					type: "queue_update",
					queues: [
						{
							entryId: "stale",
							kind: "nextRun",
							type: "message",
							message: { role: "user", content: "stale", timestamp: 2 },
						},
					],
					lane: "main",
				},
			],
			BACKGROUND_CONTEXT,
		);
		releaseListener.resolve();
		await Promise.all([queued, resnapshotDone.promise]);

		expect(watcher.snapshot).toEqual({ version: "fresh" });
		expect(queueEvents).toEqual([]);
		await bus.emit(
			{
				type: "queue_update",
				queues: [
					{
						entryId: "later",
						kind: "nextRun",
						type: "message",
						message: { role: "user", content: "later", timestamp: 3 },
					},
				],
				lane: "main",
			},
			BACKGROUND_CONTEXT,
		);
		await expect.poll(() => queueEvents).toEqual(["later"]);
	});

	it("isolates each listener from payload mutation", async () => {
		const bus = new HarnessEventBus();
		let observed: string[] = [];
		bus.on("config_update", (event) => {
			if (event.property === "activeTools") event.value.push("mutated");
		});
		bus.on("config_update", (event) => {
			if (event.property === "activeTools") observed = event.value;
		});
		await bus.emit(
			{
				type: "config_update",
				property: "activeTools",
				value: ["read"],
				previous: [],
				lane: "main",
			},
			BACKGROUND_CONTEXT,
		);
		expect(observed).toEqual(["read"]);
	});

	it("serializes concurrent publications in process order", async () => {
		const bus = new HarnessEventBus();
		const started = deferred();
		const release = deferred();
		const seen: string[] = [];
		bus.on("run_start", async (event) => {
			seen.push(`${event.runId}:start`);
			if (event.runId === "one") {
				started.resolve();
				await release.promise;
			}
			seen.push(`${event.runId}:end`);
		});

		const one = bus.emit({ type: "run_start", runId: "one", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await started.promise;
		const two = bus.emit({ type: "run_start", runId: "two", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await new Promise((resolve) => setTimeout(resolve, 0));
		expect(seen).toEqual(["one:start"]);
		release.resolve();
		await Promise.all([one, two]);
		expect(seen).toEqual(["one:start", "one:end", "two:start", "two:end"]);
	});

	it("keeps concurrent batches contiguous with their emitting contexts", async () => {
		const bus = new HarnessEventBus();
		const sourceKey = createContextKey<string>("event.batch.source");
		const firstContext = withContextValue(sourceKey, "first", BACKGROUND_CONTEXT);
		const secondContext = withContextValue(sourceKey, "second", BACKGROUND_CONTEXT);
		const seen: Array<{ runId: string; context: typeof firstContext }> = [];
		bus.on("run_start", async (event, context) => {
			seen.push({ runId: event.runId, context });
			await Promise.resolve();
		});

		const first = bus.emitBatch(
			[
				{ type: "run_start", runId: "a1", startedAt: 1, lane: "main" },
				{ type: "run_start", runId: "a2", startedAt: 1, lane: "main" },
			],
			firstContext,
		);
		const second = bus.emitBatch(
			[
				{ type: "run_start", runId: "b1", startedAt: 1, lane: "worker" },
				{ type: "run_start", runId: "b2", startedAt: 1, lane: "worker" },
			],
			secondContext,
		);

		await Promise.all([first, second]);
		expect(seen.map(({ runId }) => runId)).toEqual(["a1", "a2", "b1", "b2"]);
		expect(seen.slice(0, 2).every(({ context }) => context === firstContext)).toBe(true);
		expect(seen.slice(2).every(({ context }) => context === secondContext)).toBe(true);
	});

	it("binds listeners and watchers when a batch is emitted", async () => {
		const bus = new HarnessEventBus();
		const started = deferred();
		const release = deferred();
		bus.on("run_start", async (event) => {
			if (event.runId !== "blocking") return;
			started.resolve();
			await release.promise;
		});
		const blocking = bus.emit(
			{ type: "run_start", runId: "blocking", startedAt: 1, lane: "main" },
			BACKGROUND_CONTEXT,
		);
		await started.promise;

		const queued = bus.emit({ type: "run_start", runId: "queued", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		const lateListenerEvents: string[] = [];
		bus.on("run_start", (event) => {
			lateListenerEvents.push(event.runId);
		});
		const lateWatcherEvents: string[] = [];
		const lateWatcher = bus.watch({}, (event) => event.type === "run_start", BACKGROUND_CONTEXT);
		lateWatcher.start((event) => {
			if (event.type === "run_start") lateWatcherEvents.push(event.runId);
		});

		release.resolve();
		await Promise.all([blocking, queued]);
		expect(lateListenerEvents).toEqual([]);
		expect(lateWatcherEvents).toEqual([]);

		await bus.emit({ type: "run_start", runId: "later", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		expect(lateListenerEvents).toEqual(["later"]);
		expect(lateWatcherEvents).toEqual(["later"]);
	});

	it("resolves an empty batch without delivery", async () => {
		const bus = new HarnessEventBus();
		const listener = vi.fn();
		bus.on("run_start", listener);

		await bus.emitBatch([], BACKGROUND_CONTEXT);

		expect(listener).not.toHaveBeenCalled();
	});

	it("drains already emitted contiguous batches during close and ignores later publication", async () => {
		const bus = new HarnessEventBus();
		const started = deferred();
		const release = deferred();
		const seen: string[] = [];
		bus.on("run_start", async (event) => {
			seen.push(event.runId);
			if (event.runId === "blocking") {
				started.resolve();
				await release.promise;
			}
		});
		const blocking = bus.emit(
			{ type: "run_start", runId: "blocking", startedAt: 1, lane: "main" },
			BACKGROUND_CONTEXT,
		);
		await started.promise;
		const batch = bus.emitBatch(
			[
				{ type: "run_start", runId: "one", startedAt: 1, lane: "main" },
				{ type: "run_start", runId: "two", startedAt: 1, lane: "main" },
			],
			BACKGROUND_CONTEXT,
		);

		bus.close(new Error("closed"));
		await bus.emit({ type: "run_start", runId: "late", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		release.resolve();
		await Promise.all([blocking, batch]);

		expect(seen).toEqual(["blocking", "one", "two"]);
	});

	it("continues watcher delivery after listener failure and reports it", async () => {
		const bus = new HarnessEventBus();
		const failures: string[] = [];
		bus.on("handler_error", (event) => {
			failures.push(event.error);
		});
		const watcher = bus.watch({}, (event) => event.type === "run_start", BACKGROUND_CONTEXT);
		const seen: string[] = [];
		watcher.start((event) => {
			if (event.type !== "run_start") return;
			if (event.runId === "one") throw new Error("watcher failed");
			seen.push(event.runId);
		});

		await bus.emit({ type: "run_start", runId: "one", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await bus.emit({ type: "run_start", runId: "two", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		await new Promise((resolve) => setTimeout(resolve, 0));
		expect(seen).toEqual(["two"]);
		expect(failures).toEqual(["watcher failed"]);
	});

	it("isolates listener failures and emits handler_error", async () => {
		const bus = new HarnessEventBus();
		const failures: string[] = [];
		bus.on("run_start", () => {
			throw new Error("listener failed");
		});
		bus.on("handler_error", (event) => {
			failures.push(event.error);
		});
		await bus.emit({ type: "run_start", runId: "run", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);
		expect(failures).toEqual(["listener failed"]);
	});

	it("does not recurse when a handler_error listener fails", async () => {
		const bus = new HarnessEventBus();
		let handlerErrors = 0;
		bus.on("run_start", () => {
			throw new Error("listener failed");
		});
		bus.on("handler_error", () => {
			handlerErrors += 1;
			throw new Error("error listener failed");
		});

		await bus.emit({ type: "run_start", runId: "run", startedAt: 1, lane: "main" }, BACKGROUND_CONTEXT);

		expect(handlerErrors).toBe(1);
	});
});