import * as fs from "node:fs";
import * as path from "node:path";
import type { AgentMessage } from "@earendil-works/pi-agent-core";
import type {
	ExtensionAPI,
	ExtensionContext,
	InputEvent,
	MessageStartEvent,
	MessageEndEvent,
	ToolExecutionEndEvent,
	ToolExecutionStartEvent,
} from "@earendil-works/pi-coding-agent";
import {
	CONTROL_POLL_MS,
	FINAL_STOP_GRACE_MS,
	STEER_ACK_TIMEOUT_MS,
	createStopFact,
	patchStatus,
	readFileIfExists,
	runPaths,
	type RunPaths,
	type RunStatus,
	writeClosed,
	writeJsonAtomic,
} from "../worker/status.ts";
import { parseControlMarker, parseSteerRequest, quarantinePath, type ControlKind, type ControlMarker } from "./request.ts";
import { formatSteering, steerAckFile, steerRequestFile, type SteerRequest } from "./steer.ts";

interface ClaimedSteer {
	request: SteerRequest;
	file: string;
	formatted: string;
}

interface PendingAck {
	reqId: string;
	state: "delivered" | "failed";
	detail?: string;
}

interface ActiveTool {
	name: string;
	startedAt: number;
}

export interface WorkerInboxDeps {
	now?: () => number;
	pollMs?: number;
	watch?: typeof fs.watch;
	setInterval?: typeof setInterval;
	clearInterval?: typeof clearInterval;
	setTimeout?: typeof setTimeout;
	clearTimeout?: typeof clearTimeout;
	/** Injection point for the durable ack write, so failure is testable. */
	writeJson?: (file: string, value: unknown) => void;
}

export interface WorkerInbox {
	start(ctx: ExtensionContext): void;
	check(ctx: ExtensionContext): void;
	handleInput(event: InputEvent): void;
	handleMessageStart(event: MessageStartEvent, ctx: ExtensionContext): void;
	handleMessageEnd(event: MessageEndEvent): { message: AgentMessage } | undefined;
	handleToolStart(event: ToolExecutionStartEvent): void;
	handleToolEnd(event: ToolExecutionEndEvent): void;
	handleSettled(ctx: ExtensionContext): void;
	close(): void;
	pending(): Map<string, string[]>;
	/** Acks whose durable write has not succeeded yet (R-CTRL-9 proof is pending). */
	unflushedAcks(): string[];
	/** The terminal control decision latched in this process, if any. */
	terminatedAs(): "paused" | "stopped" | undefined;
}

export function createWorkerInbox(pi: ExtensionAPI, runDir: string, deps: WorkerInboxDeps = {}): WorkerInbox {
	const paths = runPaths(path.dirname(runDir), path.basename(runDir));
	const pending = new Map<string, string[]>();
	const pendingRequests = new Map<string, ClaimedSteer>();
	const deliveryQueue: ClaimedSteer[] = [];
	const observedNativeInputs = new Set<string>();
	const ackTimers = new Map<string, ReturnType<typeof setTimeout>>();
	const unflushed = new Map<string, PendingAck>();
	const shutdownTimers = new Set<ReturnType<typeof setTimeout>>();
	const watchers: fs.FSWatcher[] = [];
	let poll: ReturnType<typeof setInterval> | undefined;
	let checking = false;
	let closed = false;
	/**
	 * R-CTRL-7: once stop precedence has been applied in this process, pending
	 * steering is irrelevant *immediately*. Latched in memory as well as in
	 * `closed.json` so an `input` event arriving in the same tick cannot be
	 * acknowledged as delivered.
	 */
	let terminated: "paused" | "stopped" | undefined;
	let liveCtx: ExtensionContext | undefined;
	let interruptShutdownCtx: ExtensionContext | undefined;
	let interruptFallbackTimer: ReturnType<typeof setTimeout> | undefined;
	let activeDelivery: ClaimedSteer | undefined;
	/** Accepted by Pi's native queue, but not yet inserted at a model boundary. */
	let acceptedNonInterrupting: ClaimedSteer | undefined;
	let abortRequestedForSteering = false;
	let dispatchTimer: ReturnType<typeof setTimeout> | undefined;
	const activeTools = new Map<string, ActiveTool>();
	const steeringInterruptedTools = new Map<string, ActiveTool>();
	const now = deps.now ?? Date.now;
	const setTimer = deps.setTimeout ?? setTimeout;
	const clearTimer = deps.clearTimeout ?? clearTimeout;
	const writeJson = deps.writeJson ?? writeJsonAtomic;

	function isClosedOnDisk(): boolean {
		try {
			return readFileIfExists(paths.closed) !== undefined;
		} catch {
			return false;
		}
	}

	function bumpSteeringCounters(state: "delivered" | "failed"): void {
		try {
			patchStatus(paths, (current) => ({
				...current,
				steering: {
					...current.steering,
					pending: Math.max(0, current.steering.pending - 1),
					delivered: current.steering.delivered + (state === "delivered" ? 1 : 0),
					failed: current.steering.failed + (state === "failed" ? 1 : 0),
				},
			}));
		} catch {
			// Ack proof is authoritative; counters are only fleet observability.
		}
	}

	/**
	 * The ack file *is* the proof (R-CTRL-9), so nothing is discarded until it is on
	 * disk. Clearing the timer and the claimed request before the write meant a
	 * failed write lost the request, the proof, and any way to retry at once.
	 */
	function writeAck(reqId: string, state: "delivered" | "failed", detail?: string): boolean {
		try {
			writeJson(steerAckFile(paths, reqId), {
				reqId,
				state,
				ts: new Date().toISOString(),
				...(detail === undefined ? {} : { detail }),
			});
		} catch {
			// Retained for retry on the next inbox check. The instruction itself was
			// already accepted by the runtime for `delivered`, so it must not be resent.
			unflushed.set(reqId, { reqId, state, ...(detail === undefined ? {} : { detail }) });
			return false;
		}
		unflushed.delete(reqId);
		const timer = ackTimers.get(reqId);
		if (timer !== undefined) clearTimer(timer);
		ackTimers.delete(reqId);
		const claimed = pendingRequests.get(reqId);
		if (claimed !== undefined) {
			removePending(claimed.formatted, reqId);
			pendingRequests.delete(reqId);
			const queuedIndex = deliveryQueue.findIndex((entry) => entry.request.reqId === reqId);
			if (queuedIndex >= 0) deliveryQueue.splice(queuedIndex, 1);
			if (activeDelivery?.request.reqId === reqId) activeDelivery = undefined;
		}
		bumpSteeringCounters(state);
		return true;
	}

	/** Retry every ack whose durable write has not landed yet. */
	function flushAcks(): void {
		if (unflushed.size === 0) return;
		for (const entry of [...unflushed.values()]) {
			writeAck(entry.reqId, entry.state, entry.detail);
		}
	}

	function removePending(formatted: string, reqId: string): void {
		const ids = pending.get(formatted);
		if (ids === undefined) return;
		const index = ids.indexOf(reqId);
		if (index >= 0) ids.splice(index, 1);
		if (ids.length === 0) pending.delete(formatted);
	}

	function requeue(claimed: ClaimedSteer): void {
		removePending(claimed.formatted, claimed.request.reqId);
		pendingRequests.delete(claimed.request.reqId);
		const queuedIndex = deliveryQueue.findIndex((entry) => entry.request.reqId === claimed.request.reqId);
		if (queuedIndex >= 0) deliveryQueue.splice(queuedIndex, 1);
		if (activeDelivery?.request.reqId === claimed.request.reqId) activeDelivery = undefined;
		// R-CTRL-7 / R-CTRL-13: after stop or closure a re-enqueued request is a
		// poison file that a later revive would consume. Fail it instead.
		if (terminated !== undefined || closed || isClosedOnDisk()) {
			writeAck(claimed.request.reqId, "failed", `run is ${terminated ?? "closed"}; steering was not delivered`);
			return;
		}
		if (fs.existsSync(claimed.file)) return;
		try {
			writeJsonAtomic(claimed.file, claimed.request);
		} catch {
			writeAck(claimed.request.reqId, "failed", "steering could not be re-enqueued after acknowledgment timeout");
		}
	}

	function armAckTimeout(claimed: ClaimedSteer): void {
		const timer = setTimer(() => {
			ackTimers.delete(claimed.request.reqId);
			requeue(claimed);
		}, STEER_ACK_TIMEOUT_MS);
		timer.unref?.();
		ackTimers.set(claimed.request.reqId, timer);
	}

	/**
	 * R-CTRL-7 / R-CTRL-13. Every claimed steer is failed and every timer cancelled,
	 * so no callback can recreate a request after the run has closed.
	 */
	function cancelPendingSteers(detail: string): void {
		for (const timer of ackTimers.values()) clearTimer(timer);
		ackTimers.clear();
		const claimedIds = [...pendingRequests.keys()];
		pending.clear();
		pendingRequests.clear();
		deliveryQueue.length = 0;
		activeDelivery = undefined;
		acceptedNonInterrupting = undefined;
		observedNativeInputs.clear();
		for (const reqId of claimedIds) writeAck(reqId, "failed", detail);
	}

	function claimFile(file: string): boolean {
		try {
			fs.unlinkSync(file);
			return true;
		} catch {
			// ENOENT means a concurrent check already took it; any other failure means we
			// could not claim it, and an unclaimed request is never delivered.
			return false;
		}
	}

	/** Move an invalid record out of the scanned directory, preserving the bytes. */
	function quarantine(file: string, reason: string): void {
		const target = quarantinePath(paths, file);
		try {
			fs.mkdirSync(path.dirname(target), { recursive: true, mode: 0o700 });
			fs.renameSync(file, target);
			try {
				fs.writeFileSync(`${target}.reason.txt`, `${reason}\n`, { encoding: "utf8", mode: 0o600 });
			} catch {}
			return;
		} catch {
			// Fall back to consuming it: an unmovable malformed record must still not be
			// re-examined on every poll.
			claimFile(file);
		}
	}

	function steerFiles(): string[] {
		try {
			return fs.readdirSync(paths.steer).filter((name) => name.endsWith(".json")).sort().map((name) => path.join(paths.steer, name));
		} catch {
			return [];
		}
	}

	function readRequest(file: string): { ok: true; value: SteerRequest } | { ok: false; reason: string } {
		let raw: string;
		try {
			raw = fs.readFileSync(file, "utf8");
		} catch (error) {
			return { ok: false, reason: `unreadable: ${(error as Error).message}` };
		}
		return parseSteerRequest(raw, file);
	}

	function claimSteers(): ClaimedSteer[] {
		const result: ClaimedSteer[] = [];
		for (const file of steerFiles()) {
			const parsed = readRequest(file);
			if (!parsed.ok) {
				quarantine(file, parsed.reason);
				continue;
			}
			if (pendingRequests.has(parsed.value.reqId)) continue;
			result.push({ request: parsed.value, file, formatted: formatSteering(parsed.value.message) });
			// One correction at a time. Its source file stays durable while the stale
			// turn is interrupted; it is removed only after exact input acceptance.
			break;
		}
		return result;
	}

	/**
	 * A marker is validated before it is allowed to abort the session. Consuming by
	 * existence alone meant any malformed byte sequence terminally stopped the run.
	 */
	function consumeMarker(kind: ControlKind): ControlMarker | undefined {
		const file = kind === "stop" ? paths.stop : paths.interrupt;
		let raw: string | undefined;
		try {
			raw = readFileIfExists(file);
		} catch {
			return undefined;
		}
		if (raw === undefined) return undefined;
		const parsed = parseControlMarker(raw, kind);
		if (!parsed.ok) {
			quarantine(file, parsed.reason);
			try {
				appendControlEvent({ kind: "control_rejected", request: kind, detail: parsed.reason });
			} catch {}
			return undefined;
		}
		return claimFile(file) ? parsed.value : undefined;
	}

	function appendControlEvent(fields: Record<string, unknown>): void {
		try {
			fs.appendFileSync(paths.events, `${JSON.stringify({ ts: new Date().toISOString(), ...fields })}\n`, { encoding: "utf8", mode: 0o600 });
		} catch {
			// Event logging is observability; it never blocks control.
		}
	}

	function scheduleShutdown(ctx: ExtensionContext, delayMs: number, beforeShutdown?: () => void): ReturnType<typeof setTimeout> {
		const timer = setTimer(() => {
			shutdownTimers.delete(timer);
			beforeShutdown?.();
			try {
				ctx.shutdown();
			} catch {}
		}, delayMs);
		timer.unref?.();
		shutdownTimers.add(timer);
		return timer;
	}

	function terminate(ctx: ExtensionContext, state: "paused" | "stopped", marker?: ControlMarker): void {
		// Latched first: everything after this point must already know that pending
		// steering is irrelevant, including an `input` event in the same tick.
		terminated = state;
		let settlementLatched = false;
		let terminalStatus: RunStatus | undefined;
		try {
			const updated = patchStatus(paths, (current) => {
				settlementLatched = current.settled === true;
				if (settlementLatched) return current;
				return {
					...current,
					state,
					endedAt: new Date().toISOString(),
					interrupted: current.interrupted || state === "paused",
					stopped: current.stopped || state === "stopped",
					...(state === "stopped" && marker?.reason !== undefined
						? { stop: current.stop ?? createStopFact(marker.reason, marker.source, marker.ts) }
						: {}),
				};
			});
			terminalStatus = updated;
			settlementLatched = settlementLatched || updated?.settled === true;
		} catch {}
		cancelPendingSteers(`run is ${state}; steering was not delivered`);
		if (state === "stopped") {
			// R-CTRL-7: a stop makes every pending steer irrelevant, including files
			// that were never claimed. Left behind, they are consumed by a later revive.
			for (const file of steerFiles()) claimFile(file);
		}
		// The supervisor has already persisted the report at settlement. Abort the
		// process, but leave its latched outcome for the exit path to classify.
		if (!settlementLatched) writeClosed(paths, terminalStatus ?? state);
		try {
			ctx.abort();
		} catch {}
		if (state === "paused" && !settlementLatched) {
			// New pi sessions are not materialized on disk until an assistant entry
			// exists. Let the aborted turn settle before shutdown so resume has a file.
			interruptShutdownCtx = ctx;
			interruptFallbackTimer = scheduleShutdown(ctx, Math.max(1, Math.floor(FINAL_STOP_GRACE_MS / 2)), () => {
				interruptShutdownCtx = undefined;
				interruptFallbackTimer = undefined;
			});
			return;
		}
		scheduleShutdown(ctx, 0);
	}

	function contextIsIdle(ctx: ExtensionContext): boolean {
		try {
			return typeof ctx.isIdle === "function" ? ctx.isIdle() : true;
		} catch {
			return false;
		}
	}

	function interruptCurrentTurn(ctx: ExtensionContext): void {
		if (abortRequestedForSteering) return;
		abortRequestedForSteering = true;
		for (const [toolCallId, tool] of activeTools) steeringInterruptedTools.set(toolCallId, tool);
		try {
			ctx.abort();
		} catch (error) {
			abortRequestedForSteering = false;
			const item = deliveryQueue[0];
			if (item !== undefined) writeAck(item.request.reqId, "failed", `could not interrupt the current worker turn: ${(error as Error).message}`);
		}
	}

	function dispatchNext(ctx: ExtensionContext): void {
		if (closed || terminated !== undefined || activeDelivery !== undefined || acceptedNonInterrupting !== undefined || deliveryQueue.length === 0) return;
		const item = deliveryQueue[0];
		if (item === undefined) return;
		if (!contextIsIdle(ctx)) {
			if (item.request.interrupt) {
				interruptCurrentTurn(ctx);
				return;
			}
			activeDelivery = item;
			let nativeSend: ReturnType<ExtensionAPI["sendUserMessage"]>;
			try {
				nativeSend = pi.sendUserMessage(item.formatted, { deliverAs: "steer" });
			} catch (error) {
				writeAck(item.request.reqId, "failed", (error as Error).message);
				return;
			}
			void Promise.resolve(nativeSend).then(() => {
				if (closed || terminated !== undefined || activeDelivery !== item || !pendingRequests.has(item.request.reqId)) return;
				if (!observedNativeInputs.delete(item.request.reqId)) {
					writeAck(item.request.reqId, "failed", "Pi accepted the steering call without an exactly correlated input event");
					return;
				}
				try {
					fs.rmSync(item.file, { force: true });
				} catch {}
				// Pi has accepted the exact text into its native queue, but a later abort
				// can still discard that queue until its user message begins.
				acceptedNonInterrupting = item;
				writeAck(item.request.reqId, "delivered");
			}).catch((error: unknown) => {
				if (activeDelivery === item && pendingRequests.has(item.request.reqId)) {
					observedNativeInputs.delete(item.request.reqId);
					writeAck(item.request.reqId, "failed", (error as Error).message);
				}
			});
			return;
		}
		activeDelivery = item;
		if (abortRequestedForSteering) {
			try {
				// The parent sees the stale RPC settlement only after extension handlers
				// return. This one-shot marker tells it that a plain continuation was
				// launched inside that settle hook, so it must not close stdin yet.
				writeJsonAtomic(paths.steerHandoff, {
					pid: process.pid,
					reqId: item.request.reqId,
					ts: new Date().toISOString(),
				});
			} catch (error) {
				writeAck(item.request.reqId, "failed", `could not arm the steering continuation: ${(error as Error).message}`);
				return;
			}
		}
		const send = () => {
			try {
				pi.sendUserMessage(item.formatted);
			} catch (error) {
				try {
					fs.rmSync(paths.steerHandoff, { force: true });
				} catch {}
				writeAck(item.request.reqId, "failed", (error as Error).message);
			}
		};
		if (!abortRequestedForSteering) {
			send();
			return;
		}
		// Pi marks itself idle before invoking agent_settled extensions, but starting
		// another prompt synchronously would re-enter the previous run's finally block.
		// The durable marker protects the intervening RPC settlement; the next tick
		// starts an ordinary user turn without a streaming delivery mode.
		dispatchTimer = setTimer(() => {
			dispatchTimer = undefined;
			if (closed || terminated !== undefined || activeDelivery !== item) return;
			send();
		}, 0);
		dispatchTimer.unref?.();
	}

	function handleSettled(ctx: ExtensionContext): void {
		if (interruptShutdownCtx !== undefined) {
			interruptShutdownCtx = undefined;
			if (interruptFallbackTimer !== undefined) {
				clearTimer(interruptFallbackTimer);
				shutdownTimers.delete(interruptFallbackTimer);
				interruptFallbackTimer = undefined;
			}
			scheduleShutdown(ctx, 0);
			return;
		}
		checkInternal(ctx);
	}

	function checkInternal(ctx: ExtensionContext): void {
		if (closed || checking) return;
		if (terminated !== undefined) return;
		if (isClosedOnDisk()) return;
		checking = true;
		liveCtx = ctx;
		try {
			flushAcks();
			// R-CTRL-7 precedence: stop, then interrupt, then steering.
			const stop = consumeMarker("stop");
			if (stop !== undefined) {
				consumeMarker("interrupt");
				terminate(ctx, "stopped", stop);
				return;
			}
			if (consumeMarker("interrupt")) {
				terminate(ctx, "paused");
				return;
			}

			const claimed = activeDelivery === undefined && acceptedNonInterrupting === undefined && deliveryQueue.length === 0 ? claimSteers() : [];
			for (const item of claimed) {
				if (item === undefined) continue;
				const ids = pending.get(item.formatted) ?? [];
				ids.push(item.request.reqId);
				pending.set(item.formatted, ids);
				pendingRequests.set(item.request.reqId, item);
				deliveryQueue.push(item);
				armAckTimeout(item);
			}
			dispatchNext(ctx);
		} catch {
			// A malformed control payload or transient filesystem error never escapes an event hook.
		} finally {
			checking = false;
		}
	}

	function check(ctx: ExtensionContext): void {
		checkInternal(ctx);
	}

	function requestAcceptsInput(request: SteerRequest, streamingBehavior: InputEvent["streamingBehavior"]): boolean {
		if (streamingBehavior === "followUp") return false;
		return request.interrupt ? streamingBehavior === undefined : streamingBehavior === undefined || streamingBehavior === "steer";
	}

	function claimMatchingFile(formatted: string, streamingBehavior: InputEvent["streamingBehavior"]): ClaimedSteer | undefined {
		for (const file of steerFiles()) {
			const parsed = readRequest(file);
			if (!parsed.ok) {
				quarantine(file, parsed.reason);
				continue;
			}
			// A coincidental later request must never leapfrog the oldest valid durable
			// request merely because its text or delivery mode matches this input event.
			if (formatSteering(parsed.value.message) !== formatted || !requestAcceptsInput(parsed.value, streamingBehavior)) return undefined;
			if (!claimFile(file)) continue;
			return { request: parsed.value, file, formatted };
		}
		return undefined;
	}

	/**
	 * R-CTRL-9. The *only* proof that pi accepted a steer is an `input` event whose
	 * text matches exactly. Non-interrupting steering is accepted through Pi's native
	 * `steer` queue; interrupting steering must still arrive as ordinary input only.
	 */
	function handleInput(event: InputEvent): void {
		if (event.source !== "extension" && event.source !== "rpc") return;
		if (event.streamingBehavior === "followUp") return;
		if (terminated !== undefined || closed || isClosedOnDisk()) return;
		let reqId: string | undefined;
		let claimed: ClaimedSteer | undefined;
		const ids = pending.get(event.text);
		if (ids !== undefined) {
			const oldest = ids[0];
			const oldestClaimed = oldest === undefined ? undefined : pendingRequests.get(oldest);
			if (oldestClaimed !== undefined && requestAcceptsInput(oldestClaimed.request, event.streamingBehavior)) {
				reqId = oldest;
				claimed = oldestClaimed;
			}
		}
		if (reqId === undefined && ids === undefined && event.streamingBehavior === undefined) {
			claimed = claimMatchingFile(event.text, event.streamingBehavior);
			reqId = claimed?.request.reqId;
		}
		if (reqId === undefined) return;
		claimed ??= pendingRequests.get(reqId);
		if (claimed?.request.interrupt === false && event.streamingBehavior === "steer") {
			observedNativeInputs.add(reqId);
			return;
		}
		if (claimed !== undefined) {
			try {
				fs.rmSync(claimed.file, { force: true });
			} catch {}
		}
		writeAck(reqId, "delivered");
		abortRequestedForSteering = false;
		if (liveCtx !== undefined) {
			const ctx = liveCtx;
			const timer = setTimer(() => checkInternal(ctx), 0);
			timer.unref?.();
		}
	}

	function messageText(message: AgentMessage): string {
		if (!("content" in message) || !Array.isArray(message.content)) return "";
		return message.content
			.filter((part): part is { type: "text"; text: string } => part?.type === "text" && typeof part.text === "string")
			.map((part) => part.text)
			.join("\n");
	}

	function handleMessageStart(event: MessageStartEvent, ctx: ExtensionContext): void {
		const accepted = acceptedNonInterrupting;
		if (accepted === undefined || event.message.role !== "user" || messageText(event.message) !== accepted.formatted) return;
		acceptedNonInterrupting = undefined;
		observedNativeInputs.clear();
		checkInternal(ctx);
	}

	function handleToolStart(event: ToolExecutionStartEvent): void {
		activeTools.set(event.toolCallId, { name: event.toolName, startedAt: now() });
		if (abortRequestedForSteering) steeringInterruptedTools.set(event.toolCallId, activeTools.get(event.toolCallId) as ActiveTool);
	}

	function handleToolEnd(event: ToolExecutionEndEvent): void {
		activeTools.delete(event.toolCallId);
	}

	function interruptedCommandMessage(message: AgentMessage, tool: ActiveTool): AgentMessage {
		const elapsedSeconds = Math.max(0, now() - tool.startedAt) / 1000;
		const footer = `Command interrupted by steering after ${elapsedSeconds.toFixed(1)} seconds.`;
		const content = "content" in message && Array.isArray(message.content) ? [...message.content] : [];
		let replaced = false;
		for (let index = content.length - 1; index >= 0; index--) {
			const part = content[index];
			if (part?.type !== "text") continue;
			const body = part.text.replace(/(?:\n\n)?Command aborted\s*$/u, "").trimEnd();
			content[index] = { ...part, text: `${body}${body.length > 0 ? "\n\n" : ""}${footer}` };
			replaced = true;
			break;
		}
		if (!replaced) content.push({ type: "text", text: footer });
		return { ...message, content } as AgentMessage;
	}

	function handleMessageEnd(event: MessageEndEvent): { message: AgentMessage } | undefined {
		if (event.message.role !== "toolResult") return undefined;
		const tool = steeringInterruptedTools.get(event.message.toolCallId);
		if (tool === undefined) return undefined;
		steeringInterruptedTools.delete(event.message.toolCallId);
		if (tool.name !== "bash") return undefined;
		return { message: interruptedCommandMessage(event.message, tool) };
	}

	function start(ctx: ExtensionContext): void {
		if (closed || poll !== undefined) return;
		liveCtx = ctx;
		// R-CTRL-4: claim anything already present before relying on a watcher.
		check(ctx);
		try {
			writeJsonAtomic(paths.steerCapability, { pid: process.pid, readyAt: new Date().toISOString(), supported: typeof pi.sendUserMessage === "function" });
		} catch {}
		const watch = deps.watch ?? fs.watch;
		for (const dir of [paths.control, paths.steer]) {
			try {
				const watcher = watch(dir, () => {
					if (liveCtx !== undefined) check(liveCtx);
				});
				watcher.on("error", () => undefined);
				watchers.push(watcher);
			} catch {
				// Polling below is the guarantee.
			}
		}
		const startInterval = deps.setInterval ?? setInterval;
		poll = startInterval(() => {
			if (liveCtx !== undefined) check(liveCtx);
		}, deps.pollMs ?? CONTROL_POLL_MS);
		poll.unref?.();
	}

	function close(): void {
		closed = true;
		if (poll !== undefined) (deps.clearInterval ?? clearInterval)(poll);
		poll = undefined;
		for (const watcher of watchers) watcher.close();
		watchers.length = 0;
		// Last chance to make the proof durable before this process goes away.
		flushAcks();
		for (const timer of ackTimers.values()) clearTimer(timer);
		for (const timer of shutdownTimers) clearTimer(timer);
		if (dispatchTimer !== undefined) clearTimer(dispatchTimer);
		ackTimers.clear();
		shutdownTimers.clear();
		pending.clear();
		pendingRequests.clear();
		deliveryQueue.length = 0;
		activeDelivery = undefined;
		acceptedNonInterrupting = undefined;
		abortRequestedForSteering = false;
		activeTools.clear();
		steeringInterruptedTools.clear();
		dispatchTimer = undefined;
		interruptShutdownCtx = undefined;
		interruptFallbackTimer = undefined;
		liveCtx = undefined;
	}

	return {
		start,
		check,
		handleInput,
		handleMessageStart,
		handleMessageEnd,
		handleToolStart,
		handleToolEnd,
		handleSettled,
		close,
		pending: () => pending,
		unflushedAcks: () => [...unflushed.keys()],
		terminatedAs: () => terminated,
	};
}

export function registerWorkerControl(pi: ExtensionAPI, runDir: string): void {
	const inbox = createWorkerInbox(pi, runDir);
	let signalHandler: (() => void) | undefined;
	pi.on("session_start", async (_event, ctx) => {
		inbox.start(ctx);
		signalHandler = () => inbox.check(ctx);
		try {
			process.on(process.platform === "win32" ? "SIGBREAK" : "SIGUSR2", signalHandler);
		} catch {
			// Signals are only a latency optimization.
		}
	});
	pi.on("message_start", async (event, ctx) => {
		inbox.handleMessageStart(event, ctx);
		inbox.check(ctx);
	});
	pi.on("message_end", async (event, ctx) => {
		const replacement = inbox.handleMessageEnd(event);
		inbox.check(ctx);
		return replacement;
	});
	pi.on("tool_execution_start", async (event, ctx) => {
		inbox.handleToolStart(event);
		inbox.check(ctx);
	});
	pi.on("tool_execution_end", async (event, ctx) => {
		inbox.handleToolEnd(event);
		inbox.check(ctx);
	});
	pi.on("turn_end", async (_event, ctx) => inbox.check(ctx));
	pi.on("agent_settled", async (_event, ctx) => {
		inbox.handleSettled(ctx);
	});
	pi.on("input", async (event) => inbox.handleInput(event));
	pi.on("session_shutdown", async () => {
		if (signalHandler !== undefined) {
			try {
				process.off(process.platform === "win32" ? "SIGBREAK" : "SIGUSR2", signalHandler);
			} catch {}
		}
		inbox.close();
	});
}

export { steerRequestFile };