import * as path from "node:path";
import { MessageChannel, type MessagePort, receiveMessageOnPort } from "node:worker_threads";
import { isRecord, logger, Snowflake, workerHostEntry } from "@oh-my-pi/pi-utils";
import {
	createWorkerHandle,
	createWorkerSubprocess,
	resolveWorkerSpawnCmd,
	workerEnvFromParent,
} from "../../subprocess/worker-client";
import type { ToolSession } from "../../tools";
import { ToolAbortError } from "../../tools/tool-errors";
import { ToolError } from "@oh-my-pi/pi-tui/tools/tool-errors";
import { safeSend as safeSendIpc } from "../../utils/ipc";
import { EVAL_TIMEOUT_PAUSE_OP, EVAL_TIMEOUT_RESUME_OP } from "../bridge-timeout";
import { getEnabledEvalPreludes } from "../preludes";
import {
	abortReasonError,
	attachSessionOwner,
	EvalKernelNotRunningError,
	formatKernelRestartNotice,
	type SessionOwners,
} from "../executor-base";
import { shouldDetachKernel } from "../py/spawn-options";
import { updateEvalState } from "../state";
import type { EvalShadowCellSession } from "../speculation/cell-session";
import { getActiveEvalShadowCell } from "../speculation/runtime-context";
import type { ShadowPlan } from "../speculation/types";
import type { EvalDisplayOutput, EvalToolDescriptor, EvalToolInvokeResult } from "../types";
import { type ShadowSnapshot, shadowSnapshotDigest } from "./shared/runtime";
import { projectJavaScriptShadowPlan } from "./speculation";
import { callSessionTool, type JsStatusEvent } from "./tool-bridge";
// Coding-agent binary/bundle workers route through the CLI entrypoint with a
// hidden argv mode, so compiled/npm builds only need one JavaScript entry.
import type {
	EvalPreludeSource,
	JsToolRequest,
	RunErrorPayload,
	SessionSnapshot,
	WorkerInbound,
	WorkerOutbound,
	WorkerOutboxMessage,
} from "./worker-protocol";

export { rewriteImports, wrapCode } from "./shared/rewrite-imports";

export interface VmRunState {
	signal?: AbortSignal;
	onText?: (chunk: string) => void;
	onDisplay?: (output: EvalDisplayOutput) => void;
	/** Shows the user `out://` replies of the run's bridged tool calls; absent, those calls are refused. */
	onReply?: (text: string) => void;
}

/** Isolated runtime transport used by the context manager and startup regression fixtures. */
export interface JsEvalWorkerHandle {
	mode: "process" | "worker";
	send(msg: WorkerInbound): void;
	onMessage(handler: (msg: WorkerOutbound) => void): () => void;
	onError(handler: (error: Error) => void): () => void;
	close(): Promise<boolean>;
	/**
	 * Stop the runtime at once. Resolves after every message it sent before it stopped has been
	 * passed to the `onMessage` handlers, so a killed cell's last output still reaches its run.
	 */
	terminate(): Promise<void>;
}

/** Startup dependencies overridden by tests to exercise process-to-Worker recovery. */
export interface JsEvalWorkerFactories {
	spawnProcess(): JsEvalWorkerHandle;
	spawnWorker(): JsEvalWorkerHandle;
}

interface PendingRun {
	runId: string;
	runState: VmRunState;
	toolSession: ToolSession;
	shadowCell?: EvalShadowCellSession;
	resolve(value: { value: unknown }): void;
	reject(error: Error): void;
	toolCalls: Map<string, AbortController>;
	/**
	 * Host calls currently inside a `deferExternalAbort` phase — `agent()`
	 * isolation worktree setup and merge/cherry-pick, which ignore their abort
	 * once started. Settling the run while one is live would return the cell on
	 * top of a git operation still rewriting the repo, so the abort path drains
	 * this first. Mirrors the Python bridge's shielded-signal contract.
	 */
	deferDepth: number;
	/** Resolves once {@link deferDepth} falls back to zero. */
	deferDrained?: PromiseWithResolvers<void>;
	/** Set once the turn was cancelled; blocks new bridge calls during the drain. */
	aborted: boolean;
	/**
	 * A worker `result` withheld because the cell still has bridge calls in
	 * flight. `#runOne` reports a finished run without awaiting its pending
	 * tools, so a floated or caught `agent()` would otherwise settle the run —
	 * tearing down the abort listener — while the subagent kept going with
	 * nothing left able to cancel it. Delivered once the last call drains.
	 */
	heldResult?: Extract<WorkerOutbound, { type: "result" }>;
	settled: boolean;
}

interface JsSession {
	sessionKey: string;
	sessionId: string;
	kernelId: string;
	cwd: string;
	packageRoot?: string;
	packageEnvironment?: string;
	worker: JsEvalWorkerHandle;
	state: "alive" | "dead";
	stateSessions: Set<ToolSession>;
	pending: Map<string, PendingRun>;
	pendingSnapshots: Map<string, PromiseWithResolvers<Extract<WorkerOutbound, { type: "shadow-snapshot" }>>>;
	pendingShadowRuns: Map<string, PromiseWithResolvers<Extract<WorkerOutbound, { type: "shadow-run" }>>>;
	ownerIds: Set<string>;
	hasFallbackOwner: boolean;
}

interface StartingJsSession extends SessionOwners {
	promise: Promise<JsSession>;
}

const sessions = new Map<string, JsSession>();
const startingSessions = new Map<string, StartingJsSession>();
const resettingSessions = new Map<string, Promise<void>>();
// Owners of session keys whose worker died outside reset/disposal; the next cell announces the loss.
const lostSessions = new Map<string, Set<string>>();
// Worker startup (module-graph import + WorkerCore construction) is infrastructure
// cost, not user compute. Floor it independently of Bun's 5s default per-test timeout
// so a slow cold-start under load isn't aborted mid-init — terminating a still-
// initializing eval runtime triggers the same kind of terminate-race that motivates
// avoiding `vm.runInContext` (see shared/indirect-eval.ts), here surfacing as a
// SIGILL/SIGSEGV. Callers that pass a larger per-cell budget still dominate.
const WORKER_INIT_TIMEOUT_MS = 15_000;
const WORKER_CLOSE_TIMEOUT_MS = 1_000;
const JS_EVAL_PROCESS_ARG = "__omp_worker_js_eval_process";
const productionWorkerFactories: JsEvalWorkerFactories = {
	spawnProcess: spawnJsProcess,
	spawnWorker: spawnBunWorker,
};
let workerFactories = productionWorkerFactories;

/**
 * Test-only seam for exercising isolated-runtime startup transitions. Factories left out keep their
 * production implementation. Returns an idempotent restore callback so tests cannot strand a
 * process-wide factory.
 */
export function setJsEvalWorkerFactoriesForTests(factories: Partial<JsEvalWorkerFactories>): () => void {
	const previous = workerFactories;
	const installed = { ...productionWorkerFactories, ...factories };
	workerFactories = installed;
	let restored = false;
	return () => {
		if (restored) return;
		restored = true;
		if (workerFactories === installed) workerFactories = previous;
	};
}

export async function executeInVmContext(options: {
	sessionKey: string;
	sessionId: string;
	/** Logical owner identifier; scopes retained-worker cleanup via {@link disposeVmContextsByOwner}. */
	ownerId?: string;
	cwd: string;
	session: ToolSession;
	localRoots?: Record<string, string>;
	/** Selected package directory consulted only after the importing file's project. */
	packageRoot?: string;
	/** Model-visible description of the selected package environment. */
	packageEnvironment?: string;
	reset?: boolean;
	code: string;
	filename: string;
	timeoutMs?: number;
	runState: VmRunState;
}): Promise<{ value: unknown }> {
	const { sessionKey } = options;
	if (options.reset) {
		// Coalesce concurrent resets: an existing in-flight reset already
		// produces a fresh context, so a follow-up `reset: true` cell should
		// just wait for it rather than failing the user-visible call.
		const inFlight = resettingSessions.get(sessionKey);
		if (inFlight) await inFlight.catch(() => undefined);
		else {
			const resetPromise = resetVmContext(sessionKey);
			resettingSessions.set(
				sessionKey,
				resetPromise.then(() => undefined),
			);
			try {
				await resetPromise;
			} finally {
				resettingSessions.delete(sessionKey);
			}
		}
	} else {
		// Internal coordination: wait for any in-flight reset to settle and
		// then run on the freshly-rebuilt context.
		const inFlight = resettingSessions.get(sessionKey);
		if (inFlight) await inFlight.catch(() => undefined);
	}
	const session = await acquireSession(
		sessionKey,
		{
			cwd: options.cwd,
			sessionId: options.sessionId,
			localRoots: options.localRoots,
			packageRoot: options.packageRoot,
			packageEnvironment: options.packageEnvironment,
		},
		options.session,
		options.timeoutMs,
		options.ownerId,
	);
	if (!options.reset && lostSessions.delete(sessionKey)) {
		options.runState.onText?.(`${formatKernelRestartNotice("JS worker")}\n`);
	}
	const result = await runOnce(session, options);
	if (session.state === "alive" && path.isAbsolute(options.filename)) {
		updateEvalState(options.session, {
			language: "js",
			kernelId: session.kernelId,
			alive: true,
			loadedPath: options.filename,
		});
	}
	return result;
}

/** Describe or invoke tools defined in a retained JavaScript kernel. */
export async function invokeJsTool(
	request: JsToolRequest,
	options: {
		sessionKey: string;
		session: ToolSession;
		signal?: AbortSignal;
	},
): Promise<EvalToolInvokeResult | { ok: true; tools: EvalToolDescriptor[]; missing: string[] }> {
	const session = sessions.get(options.sessionKey);
	if (!session || session.state !== "alive") throw new EvalKernelNotRunningError("JavaScript");

	const runId = `tool-${crypto.randomUUID()}`;
	const { promise, resolve, reject } = Promise.withResolvers<{ value: unknown }>();
	let envelope: unknown;
	const pending: PendingRun = {
		runId,
		runState: { signal: options.signal },
		toolSession: options.session,
		resolve,
		reject,
		toolCalls: new Map(),
		deferDepth: 0,
		aborted: false,
		settled: false,
	};
	session.pending.set(runId, pending);

	const onAbort = (): void => {
		if (pending.settled) return;
		pending.aborted = true;
		pending.settled = true;
		const error = abortReasonError(options.signal?.reason, "Tool invocation aborted");
		for (const controller of pending.toolCalls.values()) controller.abort(error);
		reject(error);
	};
	if (options.signal?.aborted) onAbort();
	else options.signal?.addEventListener("abort", onAbort, { once: true });

	try {
		if (!pending.settled) safeSend(session, { type: "tool", runId, ...request });
		envelope = (await promise).value;
	} catch (error) {
		return { ok: false, error: error instanceof Error ? error.message : String(error) };
	} finally {
		options.signal?.removeEventListener("abort", onAbort);
		session.pending.delete(runId);
	}

	if (!isRecord(envelope) || envelope.ok !== true) {
		return { ok: false, error: "JavaScript tool request returned an invalid response" };
	}
	if (request.op === "call") {
		if (!("value" in envelope)) return { ok: false, error: "JavaScript tool call returned an invalid response" };
		return { ok: true, value: envelope.value };
	}

	const rawTools = Array.isArray(envelope.tools) ? envelope.tools : [];
	const tools: EvalToolDescriptor[] = [];
	for (const rawTool of rawTools) {
		if (!isRecord(rawTool)) continue;
		const { name, description, parameters } = rawTool;
		if (typeof name !== "string" || typeof description !== "string" || !isRecord(parameters)) continue;
		tools.push({ name, description, parameters, language: "js" });
	}
	const missing = Array.isArray(envelope.missing)
		? envelope.missing.filter((name): name is string => typeof name === "string")
		: [];
	return { ok: true, tools, missing };
}

/**
 * Captures a retained runtime's safe user-global snapshot without executing user
 * code. A busy runtime returns `null`; callers must execute normally.
 */
export async function snapshotVmContext(options: {
	sessionKey: string;
	cwd: string;
	sessionId: string;
	localRoots?: Record<string, string>;
	timeoutMs?: number;
}): Promise<ShadowSnapshot | null> {
	const session = sessions.get(options.sessionKey);
	if (session?.state !== "alive") return null;
	const id = `snapshot-${Snowflake.next()}`;
	const deferred = Promise.withResolvers<Extract<WorkerOutbound, { type: "shadow-snapshot" }>>();
	session.pendingSnapshots.set(id, deferred);
	try {
		session.worker.send({
			type: "shadow-snapshot",
			id,
			snapshot: {
				cwd: options.cwd,
				sessionId: options.sessionId,
				localRoots: options.localRoots,
				packageRoot: session.packageRoot,
				packageEnvironment: session.packageEnvironment,
			},
		});
		const reply = await raceWithTimeout(
			deferred.promise,
			options.timeoutMs ?? WORKER_INIT_TIMEOUT_MS,
			"JS shadow snapshot timed out",
		);
		return reply.eligible ? (reply.snapshot ?? null) : null;
	} finally {
		session.pendingSnapshots.delete(id);
	}
}

export interface JavaScriptShadowPlanningResult {
	snapshot: ShadowSnapshot;
	digest: string;
	plan: ShadowPlan;
}

/**
 * Retained-only planning seam. It never starts a worker and never executes the
 * candidate source; unavailable or busy runtimes fall back to normal eval.
 */
export async function shadowPlanIfPresent(options: {
	sessionKey: string;
	cwd: string;
	sessionId: string;
	code: string;
	localRoots?: Record<string, string>;
	timeoutMs?: number;
}): Promise<JavaScriptShadowPlanningResult | null> {
	const snapshot = await snapshotVmContext(options);
	if (!snapshot) return null;
	return {
		snapshot,
		digest: shadowSnapshotDigest(snapshot),
		plan: await projectJavaScriptShadowPlan(options.code, {
			snapshot: snapshot.values,
			initialGlobals: snapshot.initialGlobals,
		}),
	};
}

/**
 * Atomically rechecks a retained runtime's snapshot before starting a real
 * cell. `null` means the caller must discard speculative outcomes and use the
 * normal execution path.
 */
export async function runIfSnapshotMatches(options: {
	sessionKey: string;
	sessionId: string;
	cwd: string;
	session: ToolSession;
	localRoots?: Record<string, string>;
	code: string;
	filename: string;
	runState: VmRunState;
	expectedRevision: number;
	expectedDigest: string;
}): Promise<{ value: unknown } | null> {
	const session = sessions.get(options.sessionKey);
	if (session?.state !== "alive") return null;
	try {
		return await runOnce(session, options);
	} catch (error) {
		if (error instanceof ToolError && error.message === "JS shadow snapshot changed") return null;
		throw error;
	}
}

export async function resetVmContext(sessionKey: string): Promise<void> {
	lostSessions.delete(sessionKey);
	const session = sessions.get(sessionKey) ?? (await startingSessions.get(sessionKey)?.promise.catch(() => undefined));
	if (!session) return;
	sessions.delete(sessionKey);
	await killSession(session, new ToolError("JS context reset"), { force: false });
}

export async function disposeAllVmContexts(): Promise<void> {
	const pending = [...startingSessions.values()].map(starting => starting.promise);
	startingSessions.clear();
	const started = await Promise.allSettled(pending);
	const all = Array.from(sessions.values());
	for (const result of started) {
		if (result.status !== "fulfilled") continue;
		if (!all.includes(result.value)) all.push(result.value);
	}
	sessions.clear();
	// After the await: a worker that died while startups settled marked its key lost.
	lostSessions.clear();
	await Promise.all(all.map(session => killSession(session, new ToolError("JS context disposed"), { force: false })));
}

/**
 * Shut down retained JS contexts owned solely by `ownerId`; contexts with other
 * owners (e.g. an in-memory session keyed only by cwd) just drop the registration.
 */
export async function disposeVmContextsByOwner(ownerId: string): Promise<void> {
	const toKill: JsSession[] = [];
	for (const session of Array.from(sessions.values())) {
		if (!session.ownerIds.has(ownerId)) continue;
		if (session.ownerIds.size === 1) {
			toKill.push(session);
			continue;
		}
		session.ownerIds.delete(ownerId);
	}
	const startingToKill: StartingJsSession[] = [];
	for (const [sessionKey, starting] of Array.from(startingSessions.entries())) {
		if (sessions.has(sessionKey) || !starting.ownerIds.has(ownerId)) continue;
		if (starting.ownerIds.size === 1) {
			startingSessions.delete(sessionKey);
			startingToKill.push(starting);
			continue;
		}
		starting.ownerIds.delete(ownerId);
	}
	for (const [sessionKey, ownerIds] of lostSessions) {
		if (ownerIds.delete(ownerId) && ownerIds.size === 0) lostSessions.delete(sessionKey);
	}
	for (const session of toKill) {
		if (sessions.get(session.sessionKey) === session) sessions.delete(session.sessionKey);
	}
	const started = await Promise.allSettled(startingToKill.map(starting => starting.promise));
	for (const result of started) {
		if (result.status !== "fulfilled") continue;
		const session = result.value;
		if (sessions.get(session.sessionKey) === session) sessions.delete(session.sessionKey);
		toKill.push(session);
	}
	await Promise.all(
		toKill.map(session => killSession(session, new ToolError("JS context disposed"), { force: false })),
	);
}

/**
 * Smoke probe: spawn the JS evaluator through the worker-host entry and prove
 * it answers the `init` handshake in a real isolated subprocess (not the inline
 * fallback). Catches silent process-load and init-message regressions
 * that otherwise strand every cell on the init timeout in a distribution build —
 * the failure mode that motivated `installWorkerInbox`. Wired into
 * `omp --smoke-test` so binary / source / tarball installs all exercise it.
 */
export async function smokeTestJsEvalWorker(): Promise<void> {
	const worker = spawnJsWorker();
	const session: JsSession = {
		sessionKey: "smoke",
		sessionId: "smoke",
		kernelId: `js-${Snowflake.next()}`,
		cwd: process.cwd(),
		worker,
		state: "alive",
		stateSessions: new Set(),
		pending: new Map(),
		pendingSnapshots: new Map(),
		pendingShadowRuns: new Map(),
		ownerIds: new Set(),
		hasFallbackOwner: false,
	};
	try {
		await initWorker(session, { cwd: process.cwd(), sessionId: "smoke" }, WORKER_INIT_TIMEOUT_MS);
		if (worker.mode !== "process") {
			throw new Error("JS eval worker smoke fell back from the isolated subprocess");
		}
	} finally {
		await worker.terminate().catch(() => undefined);
	}
}

function javascriptPreludeSources(session: ToolSession): EvalPreludeSource[] {
	const definitions = getEnabledEvalPreludes(session.getEvalPreludes?.() ?? []);
	return definitions.flatMap(definition =>
		definition.javascript.trim().length === 0
			? []
			: [{ name: definition.name, exports: [...definition.exports], source: definition.javascript }],
	);
}

async function runOnce(
	session: JsSession,
	options: {
		sessionId: string;
		cwd: string;
		session: ToolSession;
		localRoots?: Record<string, string>;
		packageRoot?: string;
		packageEnvironment?: string;
		code: string;
		filename: string;
		runState: VmRunState;
		expectedRevision?: number;
		expectedDigest?: string;
	},
): Promise<{ value: unknown }> {
	const runId = `r-${Snowflake.next()}`;
	const { promise, resolve, reject } = Promise.withResolvers<{ value: unknown }>();
	const pending: PendingRun = {
		runId,
		runState: options.runState,
		toolSession: options.session,
		shadowCell: getActiveEvalShadowCell(),
		resolve,
		reject,
		toolCalls: new Map(),
		deferDepth: 0,
		aborted: false,
		settled: false,
	};
	session.pending.set(runId, pending);

	const onAbort = (): void => {
		const reason = options.runState.signal?.reason;
		const abortError = abortReasonError(reason, "Execution aborted");
		// Stop delegated work at once — this is what kills spawned subagents —
		// and refuse further bridge calls so the drain below stays bounded to
		// phases that had already started.
		pending.aborted = true;
		for (const ctrl of pending.toolCalls.values()) ctrl.abort(abortError);
		// A critical host phase ignores its abort once started (isolation
		// worktree setup, merge/cherry-pick). Killing the worker now would
		// settle the cell on top of a git operation still in progress, so wait
		// for it. Hard-kill is still the only way to interrupt synchronous user
		// code, hence it stays the terminal step either way.
		const drained = pending.deferDepth > 0 ? pending.deferDrained?.promise : undefined;
		if (drained) {
			void drained.then(() => killSessionFor(session, abortError, { force: true }));
			return;
		}
		void killSessionFor(session, abortError, { force: true });
	};

	if (options.runState.signal?.aborted) {
		queueMicrotask(onAbort);
	} else {
		options.runState.signal?.addEventListener("abort", onAbort, { once: true });
	}

	try {
		if (options.packageRoot !== undefined) session.packageRoot = options.packageRoot;
		if (options.packageEnvironment !== undefined) session.packageEnvironment = options.packageEnvironment;
		const snapshot = {
			cwd: options.cwd,
			sessionId: options.sessionId,
			localRoots: options.localRoots,
			preludes: javascriptPreludeSources(options.session),
			packageRoot: session.packageRoot,
			packageEnvironment: session.packageEnvironment,
		};
		if (options.expectedRevision !== undefined && options.expectedDigest !== undefined) {
			const id = `shadow-run-${Snowflake.next()}`;
			const admission = Promise.withResolvers<Extract<WorkerOutbound, { type: "shadow-run" }>>();
			session.pendingShadowRuns.set(id, admission);
			try {
				session.worker.send({
					type: "run-if-snapshot-matches",
					id,
					runId,
					code: options.code,
					filename: options.filename,
					snapshot,
					expectedRevision: options.expectedRevision,
					expectedDigest: options.expectedDigest,
				});
				const reply = await raceWithTimeout(
					admission.promise,
					WORKER_INIT_TIMEOUT_MS,
					"JS shadow admission timed out",
				);
				if (!reply.eligible) throw new ToolError("JS shadow snapshot changed");
			} finally {
				session.pendingShadowRuns.delete(id);
			}
		} else {
			session.worker.send({ type: "run", runId, code: options.code, filename: options.filename, snapshot });
		}
		return await promise;
	} finally {
		options.runState.signal?.removeEventListener("abort", onAbort);
		session.pending.delete(runId);
	}
}

async function acquireSession(
	sessionKey: string,
	snapshot: SessionSnapshot,
	toolSession: ToolSession,
	timeoutMs?: number,
	ownerId?: string,
): Promise<JsSession> {
	const existing = sessions.get(sessionKey);
	if (existing && existing.state === "alive") {
		existing.sessionId = snapshot.sessionId;
		existing.cwd = snapshot.cwd;
		existing.packageRoot = snapshot.packageRoot;
		existing.packageEnvironment = snapshot.packageEnvironment;
		attachSessionOwner(existing, snapshot.sessionId, ownerId);
		markJsSessionAlive(existing, toolSession, snapshot.packageEnvironment);
		return existing;
	}
	const starting = startingSessions.get(sessionKey);
	if (starting) {
		attachSessionOwner(starting, snapshot.sessionId, ownerId);
		const session = await starting.promise;
		markJsSessionAlive(session, toolSession, snapshot.packageEnvironment);
		return session;
	}
	// oxlint-disable-next-line prefer-const -- captured by the startup closure before assignment
	let startingSession!: StartingJsSession;

	const startup = (async (): Promise<JsSession> => {
		// Attach the message listener before sending init. Both Bun Worker messages
		// and subprocess IPC can arrive immediately after the evaluator loads.
		const worker = spawnJsWorker();
		const session: JsSession = {
			sessionKey,
			sessionId: snapshot.sessionId,
			kernelId: `js-${Snowflake.next()}`,
			cwd: snapshot.cwd,
			packageRoot: snapshot.packageRoot,
			packageEnvironment: snapshot.packageEnvironment,
			worker,
			state: "alive",
			stateSessions: new Set(),
			pending: new Map(),
			pendingSnapshots: new Map(),
			pendingShadowRuns: new Map(),
			ownerIds: new Set(),
			hasFallbackOwner: false,
		};
		// Init headroom is the fixed infrastructure floor; the caller's per-cell timeout
		// dominates when larger so users can grant more by raising `timeout` on a cell.
		const readyTimeoutMs = Math.max(WORKER_INIT_TIMEOUT_MS, timeoutMs ?? 0);
		while (true) {
			try {
				await initWorker(session, snapshot, readyTimeoutMs);
				break;
			} catch (error) {
				// Runtime crash/load failures surface asynchronously via the runtime's
				// error callback, after the synchronous spawn try/catch has returned.
				// Recover once from the subprocess into a Bun Worker. A second isolated
				// runtime failure must surface; running the cell on this host thread
				// would make synchronous user code impossible to cancel.
				const failed = session.worker;
				await failed.terminate().catch(() => undefined);
				if (failed.mode === "worker") {
					throw new Error(
						`Failed to initialize isolated JS eval worker: ${error instanceof Error ? error.message : String(error)}`,
						{ cause: error },
					);
				}
				logger.warn("JS eval subprocess init failed; retrying with a Bun Worker", {
					error: error instanceof Error ? error.message : String(error),
				});
				session.worker = workerFactories.spawnWorker();
				session.state = "alive";
			}
		}
		session.ownerIds = new Set(startingSession.ownerIds);
		session.hasFallbackOwner = startingSession.hasFallbackOwner;
		// Publish only while this startup still owns the key: owner disposal or
		// a concurrent dispose-all may have already reaped the starting record,
		// and publishing here would resurrect a context that was just torn down.
		if (startingSessions.get(sessionKey) === startingSession) {
			sessions.set(sessionKey, session);
		}
		return session;
	})();
	startingSession = {
		ownerIds: new Set(),
		hasFallbackOwner: false,
		promise: startup,
	};
	attachSessionOwner(startingSession, snapshot.sessionId, ownerId);
	startingSessions.set(sessionKey, startingSession);
	try {
		const session = await startup;
		markJsSessionAlive(session, toolSession, snapshot.packageEnvironment);
		return session;
	} finally {
		if (startingSessions.get(sessionKey) === startingSession) startingSessions.delete(sessionKey);
	}
}

async function initWorker(session: JsSession, snapshot: SessionSnapshot, timeoutMs: number): Promise<void> {
	const worker = session.worker;
	const { promise: readyPromise, resolve: resolveReady, reject: rejectReady } = Promise.withResolvers<void>();
	let resolved = false;
	const unsubscribeMessage = worker.onMessage(msg => {
		if (!resolved && msg.type === "ready") {
			resolved = true;
			resolveReady();
			return;
		}
		if (!resolved && msg.type === "init-failed") {
			resolved = true;
			rejectReady(errorFromPayload(msg.error));
			return;
		}
		handleSessionMessage(session, msg);
	});
	const unsubscribeError = worker.onError(error => {
		if (!resolved) {
			resolved = true;
			rejectReady(error);
			return;
		}
		// Worker died after a successful handshake: tear the session down so the
		// in-flight run (and the next acquire) fail fast instead of hanging on a
		// worker that will never reply.
		void killSessionFor(session, error, { force: true });
	});
	try {
		// Attach listeners and send init before awaiting ready. The worker now
		// emits ready only in response to init, so this ordering is race-free.
		worker.send({ type: "init", snapshot });
		await raceWithTimeout(readyPromise, timeoutMs, "Timed out initializing JS eval worker");
	} catch (error) {
		// Handshake failed (timeout, init-failed, or worker error): drop both listeners
		// so the abandoned worker can't keep routing messages into a session the caller
		// is about to discard or retry on the isolated Worker fallback.
		unsubscribeMessage();
		unsubscribeError();
		throw error;
	}
}

function handleSessionMessage(session: JsSession, msg: WorkerOutbound): void {
	switch (msg.type) {
		case "text": {
			const pending = session.pending.get(msg.runId);
			pending?.runState.onText?.(msg.chunk);
			return;
		}
		case "display": {
			const pending = session.pending.get(msg.runId);
			pending?.runState.onDisplay?.(msg.output);
			return;
		}
		case "tool-call":
			void handleToolCall(session, msg);
			return;
		case "result":
			settlePending(session, msg);
			return;
		case "shadow-snapshot": {
			const pending = session.pendingSnapshots.get(msg.id);
			if (pending) pending.resolve(msg);
			return;
		}
		case "shadow-run": {
			const pending = session.pendingShadowRuns.get(msg.id);
			if (pending) pending.resolve(msg);
			return;
		}
		case "log":
			logWorkerMessage(msg);
			return;
		case "ready":
		case "init-failed":
		case "closed":
			return;
	}
}

/**
 * Maintain {@link PendingRun.deferDepth} from the bridge's pause/resume status
 * events so an abort can wait out a critical `agent()` phase instead of
 * settling the cell over a half-applied merge.
 */
function trackDeferPhase(pending: PendingRun, event: JsStatusEvent): void {
	if (event.deferExternalAbort !== true) return;
	if (event.op === EVAL_TIMEOUT_PAUSE_OP) {
		pending.deferDepth++;
		pending.deferDrained ??= Promise.withResolvers<void>();
		return;
	}
	if (event.op !== EVAL_TIMEOUT_RESUME_OP || pending.deferDepth === 0) return;
	pending.deferDepth--;
	if (pending.deferDepth > 0) return;
	pending.deferDrained?.resolve();
	pending.deferDrained = undefined;
}

async function handleToolCall(session: JsSession, msg: Extract<WorkerOutbound, { type: "tool-call" }>): Promise<void> {
	const pending = session.pending.get(msg.runId);
	// A dead session is only handing over what its runtime sent before it stopped; never start host
	// work for a runtime that is gone.
	if (!pending || session.state === "dead") {
		safeSend(session, {
			type: "tool-reply",
			id: msg.id,
			reply: { ok: false, error: { message: "Run no longer active" } },
		});
		return;
	}
	if (pending.aborted) {
		safeSend(session, {
			type: "tool-reply",
			id: msg.id,
			reply: { ok: false, error: { message: "Run was interrupted" } },
		});
		return;
	}
	const ctrl = new AbortController();
	pending.toolCalls.set(msg.id, ctrl);
	try {
		const value = await callSessionTool(msg.name, msg.args, {
			session: pending.toolSession,
			signal: ctrl.signal,
			identity: msg.identity,
			shadowCell: pending.shadowCell,
			onReply: pending.runState.onReply,
			emitStatus: (event: JsStatusEvent) => {
				trackDeferPhase(pending, event);
				pending.runState.onDisplay?.({ type: "status", event });
			},
		});
		safeSend(session, { type: "tool-reply", id: msg.id, reply: { ok: true, value } });
	} catch (error) {
		safeSend(session, { type: "tool-reply", id: msg.id, reply: { ok: false, error: toErrorPayload(error) } });
	} finally {
		pending.toolCalls.delete(msg.id);
		// Last call of a run whose worker result was withheld: settle it now.
		const held = pending.heldResult;
		if (held && !pending.settled && !pending.aborted && pending.toolCalls.size === 0) {
			finishPending(pending, held);
		}
	}
}

/** Deliver a worker `result` to the waiting {@link runOnce}. */
function finishPending(pending: PendingRun, msg: Extract<WorkerOutbound, { type: "result" }>): void {
	pending.settled = true;
	pending.heldResult = undefined;
	if (msg.ok) {
		pending.resolve({ value: msg.value });
		return;
	}
	pending.reject(errorFromPayload(msg.error));
}

function settlePending(session: JsSession, msg: Extract<WorkerOutbound, { type: "result" }>): void {
	const pending = session.pending.get(msg.runId);
	if (!pending || pending.settled) return;
	// Once the turn is cancelled the scheduled kill is the sole settler, so a
	// late worker result can't cut the abort drain short.
	if (pending.aborted) return;
	// A cell owns every bridge call it starts. The worker finishes a run without
	// awaiting its outstanding tool calls, so `agent(...)` that is floated or
	// caught would settle the run here — `runOnce` then drops the abort listener
	// and the pending entry, leaving the subagent running with nothing able to
	// cancel it. Hold the result until the last call drains.
	if (pending.toolCalls.size > 0) {
		pending.heldResult = msg;
		return;
	}
	finishPending(pending, msg);
}

async function killSessionFor(session: JsSession, error: Error, options: { force: boolean }): Promise<void> {
	if (sessions.get(session.sessionKey) === session) {
		sessions.delete(session.sessionKey);
		lostSessions.set(session.sessionKey, new Set(session.ownerIds));
	}
	await killSession(session, error, options);
}

async function killSession(session: JsSession, error: Error, options: { force: boolean }): Promise<void> {
	if (session.state === "dead") return;
	session.state = "dead";
	for (const toolSession of session.stateSessions) {
		updateEvalState(toolSession, { language: "js", kernelId: session.kernelId, alive: false });
	}
	// This kill is the sole settler of every run from here on. Mark them all before a single tool
	// controller is aborted: an aborted bridge call's `finally` runs during the drain below, and an
	// unmarked run would let it release a worker success the runtime never got to stand behind —
	// reporting a cell whose host work was cut short as having completed. `aborted` also keeps a
	// late `result` frame out of `settlePending`. Neither blocks text or display frames, so the
	// output the runtime sent before it stopped still reaches the run.
	for (const pending of session.pending.values()) pending.aborted = true;
	// Stop delegated host work at once; the runtime itself goes next.
	for (const pending of session.pending.values()) {
		for (const ctrl of pending.toolCalls.values()) ctrl.abort(error);
	}
	// Settle the runs only once the runtime is gone and everything it sent before has been handled:
	// a cell's last output (often the only clue why it was killed) belongs to the cancelled run, and
	// the caller reads that output as soon as the run settles.
	if (options.force || !(await session.worker.close().catch(() => false))) {
		await session.worker.terminate().catch(() => undefined);
	}
	for (const pending of session.pending.values()) {
		if (pending.settled) continue;
		pending.settled = true;
		pending.reject(error);
	}
	session.pending.clear();
	for (const pending of session.pendingSnapshots.values()) pending.reject(error);
	session.pendingSnapshots.clear();
	for (const pending of session.pendingShadowRuns.values()) pending.reject(error);
	session.pendingShadowRuns.clear();
}

function markJsSessionAlive(session: JsSession, toolSession: ToolSession, environment: string | undefined): void {
	if (session.state !== "alive") return;
	session.stateSessions.add(toolSession);
	updateEvalState(toolSession, {
		language: "js",
		kernelId: session.kernelId,
		alive: true,
		environment,
		interpreter: `Bun ${Bun.version}`,
	});
}

function safeSend(session: JsSession, msg: WorkerInbound): void {
	if (session.state !== "alive") return;
	try {
		session.worker.send(msg);
	} catch (err) {
		logger.debug("js worker send failed", { error: err instanceof Error ? err.message : String(err) });
	}
}

function errorFromPayload(payload: RunErrorPayload): Error {
	if (payload.isAbort) {
		const err = new ToolAbortError(payload.message || "Execution aborted");
		if (payload.stack) err.stack = payload.stack;
		return err;
	}
	const ctor = payload.isToolError ? ToolError : Error;
	const error = new ctor(payload.message);
	if (payload.name) error.name = payload.name;
	if (payload.stack) error.stack = payload.stack;
	return error;
}

function toErrorPayload(error: unknown): RunErrorPayload {
	if (error instanceof Error) {
		return {
			name: error.name,
			message: error.message,
			stack: error.stack,
			isAbort: error.name === "AbortError" || error.name === "ToolAbortError",
			isToolError: error instanceof ToolError || error.name === "ToolError",
		};
	}
	return { message: String(error) };
}

function logWorkerMessage(msg: Extract<WorkerOutbound, { type: "log" }>): void {
	if (msg.level === "debug") logger.debug(msg.msg, msg.meta);
	else if (msg.level === "warn") logger.warn(msg.msg, msg.meta);
	else logger.error(msg.msg, msg.meta);
}

async function raceWithTimeout<T>(promise: Promise<T>, timeoutMs: number, reason: string): Promise<T> {
	const timeoutSignal = AbortSignal.timeout(timeoutMs);
	const { promise: timeoutPromise, reject } = Promise.withResolvers<never>();
	const onAbort = (): void => reject(new ToolError(reason));
	timeoutSignal.addEventListener("abort", onAbort, { once: true });
	try {
		return await Promise.race([promise, timeoutPromise]);
	} finally {
		timeoutSignal.removeEventListener("abort", onAbort);
	}
}

function spawnJsWorker(): JsEvalWorkerHandle {
	try {
		return workerFactories.spawnProcess();
	} catch (error) {
		// A worker thread remains isolated and can interrupt synchronous user code
		// via terminate(), so it is the only safe recovery from subprocess spawn.
		logger.warn("JS eval subprocess spawn failed; falling back to a Bun Worker", {
			error: error instanceof Error ? error.message : String(error),
		});
	}
	return workerFactories.spawnWorker();
}

function spawnBunWorker(): JsEvalWorkerHandle {
	try {
		const hostEntry = workerHostEntry();
		const worker = hostEntry
			? new Worker(hostEntry, { type: "module", argv: ["__omp_worker_js_eval"] })
			: new Worker(new URL("./worker-entry.ts", import.meta.url).href, { type: "module" });
		const channel = new MessageChannel();
		const outbox: WorkerOutboxMessage = { type: "outbox", port: channel.port2 };
		worker.postMessage(outbox, [channel.port2]);
		return wrapBunWorker(worker, channel.port1);
	} catch (error) {
		throw new Error(
			`Failed to start isolated JS eval worker: ${error instanceof Error ? error.message : String(error)}`,
			{ cause: error },
		);
	}
}

function spawnJsProcess(): JsEvalWorkerHandle {
	const spawned = createWorkerSubprocess<WorkerOutbound>({
		spawnCommand: resolveWorkerSpawnCmd(JS_EVAL_PROCESS_ARG),
		env: workerEnvFromParent(),
		exitLabel: "JS eval worker",
		detached: shouldDetachKernel(process.platform),
		reportCleanExit: true,
		unref: false,
	});
	const base = createWorkerHandle<WorkerInbound, WorkerOutbound>(spawned, message =>
		safeSendIpc(spawned.proc, message, "js-eval"),
	);
	return {
		mode: "process",
		send: message => base.send(message),
		onMessage: handler => base.onMessage(handler),
		onError: handler => base.onError(handler),
		async close() {
			const { promise, resolve } = Promise.withResolvers<boolean>();
			let settled = false;
			let unsubscribe = (): void => {};
			const finish = (value: boolean): void => {
				if (settled) return;
				settled = true;
				if (timeout) clearTimeout(timeout);
				unsubscribe();
				resolve(value);
			};
			unsubscribe = base.onMessage(message => {
				if (message.type !== "closed") return;
				void base.terminate().finally(() => finish(true));
			});
			const timeout = setTimeout(() => finish(false), WORKER_CLOSE_TIMEOUT_MS);
			base.send({ type: "close" });
			return await promise;
		},
		async terminate() {
			await base.terminate();
			// Bun hands the child's IPC messages to `ipc` until the channel closes, a SIGKILLed
			// child's included; wait for that so its last output reaches the handlers. Bounded: a
			// descendant that inherited the channel could keep it open.
			const bound = Promise.withResolvers<void>();
			const timer = setTimeout(bound.resolve, WORKER_CLOSE_TIMEOUT_MS);
			try {
				await Promise.race([spawned.disconnected, bound.promise]);
			} finally {
				clearTimeout(timer);
			}
		},
	};
}

/**
 * `outbox` is the port the worker sends every message on (see {@link WorkerOutboxMessage}). Unlike
 * the Worker's own channel it keeps what the worker posted after `terminate()`, so `terminate()`
 * can hand that over before the session settles its runs.
 */
function wrapBunWorker(worker: Worker, outbox: MessagePort): JsEvalWorkerHandle {
	const handlers = new Set<(msg: WorkerOutbound) => void>();
	const deliver = (data: unknown): void => {
		for (const handler of handlers) handler(data as WorkerOutbound);
	};
	outbox.on("message", deliver);
	// The Worker itself decides process lifetime; the port never adds to it.
	outbox.unref();
	let outboxOpen = true;
	return {
		mode: "worker",
		send(msg) {
			worker.postMessage(msg);
		},
		onMessage(handler) {
			handlers.add(handler);
			return () => handlers.delete(handler);
		},
		onError(handler) {
			const onError = (event: ErrorEvent): void => handler(errorFromWorkerEvent(event));
			const onMessageError = (event: MessageEvent): void =>
				handler(new ToolError(`JS eval worker message error: ${String(event.data)}`));
			const onClose = (): void => handler(new Error("JS eval worker exited"));
			worker.addEventListener("error", onError);
			worker.addEventListener("messageerror", onMessageError);
			worker.addEventListener("close", onClose);
			return () => {
				worker.removeEventListener("error", onError);
				worker.removeEventListener("messageerror", onMessageError);
				worker.removeEventListener("close", onClose);
			};
		},
		async close() {
			const { promise: closed, resolve } = Promise.withResolvers<boolean>();
			let settled = false;
			let sawClosedAck = false;
			let sawWorkerExit = false;
			let unsubscribe = (): void => {};
			const finish = (value: boolean): void => {
				if (settled) return;
				settled = true;
				if (timeout) clearTimeout(timeout);
				unsubscribe();
				worker.removeEventListener("close", onClose);
				resolve(value);
			};
			const finishIfClosed = (): void => {
				if (sawClosedAck && sawWorkerExit) finish(true);
			};
			const onClose = (): void => {
				sawWorkerExit = true;
				finishIfClosed();
			};
			unsubscribe = this.onMessage(msg => {
				if (msg.type !== "closed") return;
				sawClosedAck = true;
				finishIfClosed();
			});
			worker.addEventListener("close", onClose);
			const timeout = setTimeout(() => finish(false), WORKER_CLOSE_TIMEOUT_MS);
			worker.postMessage({ type: "close" } satisfies WorkerInbound);
			return await closed;
		},
		async terminate() {
			worker.terminate();
			if (!outboxOpen) return;
			outboxOpen = false;
			for (let received = receiveMessageOnPort(outbox); received; received = receiveMessageOnPort(outbox)) {
				deliver(received.message);
			}
			outbox.close();
		},
	};
}

function errorFromWorkerEvent(event: ErrorEvent): Error {
	if (event.error instanceof Error) return event.error;
	if (event.message) return new Error(event.message);
	return new Error("Unknown JS eval worker error");
}