/**
 * Supervisor: owns the fleet in this process. Validates, spawns, pumps, writes
 * `result.md`, and drains the queue.
 */

import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { getProcessStartIdentity, isProcessAlive, proveDeath, type LockOwner } from "../proof-of-death.ts";
import { validateAgentName } from "../state.ts";
import { writeControlFirst } from "../control/plane.ts";
import { parseSteerCapability } from "../control/request.ts";
import { queueSteer, type SteerResult } from "../control/steer.ts";
import type { WorkerConfig } from "./config.ts";
import type { AgentProfile } from "./profiles.ts";
import { WorkerPump } from "./pump.ts";
import {
	type RegistryEntry,
	accountingEntriesForResume,
	claimAgentName,
	gitignoreRuntime,
	hasSlot,
	inheritAgentHistory,
	latestRunForName,
	liveRunForName,
	loadAllRuns,
	runsDir,
	updateAgentName,
} from "./registry.ts";
import { truncateReport } from "./report.ts";
import { buildWorkerPrompt } from "./prompt.ts";
import { acquireSessionLease, readSessionLease, releaseSessionLease, sessionLeasePath, transferSessionLease } from "./lease.ts";
import { type RecordedProcess, type RunOwner, currentOwner, ownershipOf, recordedProcessOf } from "./ownership.ts";
import { assertResumable, revivalPrompt } from "./revive.ts";
import {
	type PiResolution,
	type RunDescriptor,
	buildSpawnArgs,
	buildWorkerEnv,
	describeResolutionFailure,
	readDescriptor,
	resolvePi,
	sameRealpath,
	spawnWorker,
	writeDescriptor,
	writePromptFile,
} from "./spawn.ts";
import {
	type RunPaths,
	type RunStatus,
	type StatusClaim,
	SHUTDOWN_BUDGET_MS,
	acquireStatusClaims,
	appendEvent,
	createStopFact,
	createStatus,
	ensureRunDirs,
	isTerminalState,
	newRunId,
	patchStatus,
	patchStatusIf,
	readFileIfExists,
	readStatus,
	releaseStatusClaims,
	runPaths,
	statusClaimKey,
	writeClosed,
	writeStatus,
} from "./status.ts";

export interface DelegateRequest {
	name: string;
	agent: string;
	prompt: string;
	cwd?: string;
}

export interface SupervisorHostInfo {
	/** Orchestrator cwd; the repo root for the cwd containment check. */
	cwd: string;
	config: WorkerConfig;
	profiles: Map<string, AgentProfile>;
	/** Exact parent model and effective thinking level inherited by every worker. */
	inheritedModel?: string;
	inheritedThinking?: string;
	/** R-WORK-2: the parent's own -e paths, inherited by the worker. */
	inheritedExtensionPaths: string[];
	/** Test seam; production always uses the real detached worker launcher. */
	spawn?: typeof spawnWorker;
	sessionDir?: string;
	/**
	 * The orchestrator's own pi session id. Stamped onto every run it spawns, and the
	 * basis for R-CTRL-18 scoping: `runs/` is shared by every session in the repo, so
	 * bulk control has to be able to tell its own work from a concurrent session's.
	 * Optional only so harnesses can construct a Supervisor without a session; a
	 * Supervisor without one owns nothing beyond the pumps it is running.
	 */
	sessionId?: string;
	depth: number;
}

export interface SpawnedRun {
	runId: string;
	status: RunStatus;
	queued: boolean;
	/** Resolves only after the anchor invariant is established and the prompt is sent. */
	acceptance: Promise<{ ok: true; status: RunStatus } | { ok: false; status: RunStatus; reason: string }>;
}

type ControlResult = { runId: string; state: string; detail?: string };
export type SteeringResult = SteerResult & { runId: string; interrupt: boolean; status: RunStatus };
type AutonomousLivePhase = "starting" | "running" | "interrupting";

function pausedExitPending(entry: RegistryEntry, livePump: boolean): boolean {
	if (entry.status.state !== "paused") return false;
	if (livePump) return true;
	const sessionFile = entry.status.sessionFile;
	if (sessionFile !== null) {
		const leasePath = sessionLeasePath(sessionFile);
		if (fs.existsSync(leasePath)) {
			const holder = readSessionLease(leasePath);
			if (holder === undefined) return true;
			if (holder.runId === entry.runId) {
				const verdict = proveDeath(holder, {
					hostname: os.hostname(),
					alive: isProcessAlive(holder.pid),
					currentIdentity: getProcessStartIdentity(holder.pid),
				});
				if (!verdict.reclaimable) return true;
			}
		}
	}
	const recorded = recordedProcessOf(entry.status);
	if (recorded.pid === null) return true;
	const holder: LockOwner = {
		pid: recorded.pid,
		hostname: recorded.hostname,
		sessionId: entry.status.sessionId ?? entry.runId,
		claimedAt: entry.status.startedAt ?? entry.status.createdAt,
		...(recorded.processStartIdentity === null ? {} : { processStartIdentity: recorded.processStartIdentity }),
	};
	const verdict = proveDeath(holder, {
		hostname: os.hostname(),
		alive: isProcessAlive(holder.pid),
		currentIdentity: getProcessStartIdentity(holder.pid),
	});
	return !verdict.reclaimable;
}

function autonomousLivePhase(entry: RegistryEntry, livePump = false): AutonomousLivePhase | undefined {
	if (entry.status.state === "spawning") return "starting";
	// Paused is published before process exit. A local pump, a live/uncertain session
	// lease, or a recorded process whose death is not proven keeps it interrupting.
	if (pausedExitPending(entry, livePump)) return "interrupting";
	if (entry.status.state !== "running") return undefined;
	return fs.existsSync(entry.paths.interrupt) ? "interrupting" : "running";
}

function autonomousStopRefusal(entry: RegistryEntry, phase: AutonomousLivePhase): ControlResult {
	return {
		runId: entry.runId,
		state: "refused",
		detail: `run is ${phase}; steer a correctable course change, or interrupt it to create a resumable checkpoint before deciding whether replacement is necessary`,
	};
}

function autonomousFleetRefusal(
	owned: RegistryEntry[],
	foreign: Array<{ entry: RegistryEntry; reason: string }>,
	detail: string,
	problems: Array<{ runId: string; reason: string }> = [],
): ControlResult[] {
	return [
		...owned.map((entry) => {
			if (entry.status.state === "stopped") return { runId: entry.runId, state: "stopped", detail: "already stopped" };
			if (isTerminalState(entry.status.state)) {
				return { runId: entry.runId, state: entry.status.state, detail: "run had already settled; its result is preserved" };
			}
			return { runId: entry.runId, state: "refused", detail };
		}),
		...problems.map((problem) => ({ runId: problem.runId, state: "refused", detail: `${detail}; ${problem.reason}` })),
		...foreign.map(({ entry, reason }) => ({ runId: entry.runId, state: "skipped", detail: reason })),
	];
}

export class Supervisor {
	private host: SupervisorHostInfo;
	private pumps = new Map<string, WorkerPump>();
	private promptFiles = new Map<string, string>();
	private promptTimers = new Map<string, ReturnType<typeof setTimeout>>();
	private queue: Array<{ request: DelegateRequest; runId: string }> = [];
	private leases = new Map<string, { path: string; token: string }>();
	/** R-CTRL-18 ownership identity for every run this Supervisor spawns or adopts. */
	private owner: RunOwner;
	private onFleetChange: () => void;
	/** Phase 5: where the completion wake is armed (§11.3). */
	private onTerminal: (runId: string, status: RunStatus) => void;

	constructor(
		host: SupervisorHostInfo,
		onFleetChange: () => void = () => undefined,
		onTerminal: (runId: string, status: RunStatus) => void = () => undefined,
	) {
		this.host = host;
		this.onFleetChange = onFleetChange;
		this.onTerminal = onTerminal;
		this.owner = currentOwner(host.sessionId ?? `pid:${process.pid}`);
	}

	updateHost(host: SupervisorHostInfo): void {
		this.host = host;
	}

	livePumps(): WorkerPump[] {
		return [...this.pumps.values()];
	}

	/** The ownership stamp written into every run this Supervisor starts. */
	ownerIdentity(): RunOwner {
		return this.owner;
	}

	private resolveAgent(name: string): RegistryEntry {
		return latestRunForName(loadAllRuns(this.host.cwd).entries, name);
	}

	steerAgent(name: string, message: string, waitForAck = true, source: "orchestrator" | "user" = "orchestrator", interrupt = false) {
		return this.steer(this.resolveAgent(name).runId, message, waitForAck, source, interrupt);
	}

	interruptAgent(name: string, source: "orchestrator" | "user") {
		return this.interrupt(this.resolveAgent(name).runId, source);
	}

	stopAgent(name: string, reason: string) {
		return this.stop(this.resolveAgent(name).runId, reason);
	}

	resumeAgent(name: string, message: string): SpawnedRun {
		return this.resume(this.resolveAgent(name).runId, message);
	}

	/**
	 * R-CTRL-18 scoping. Split the registry into runs this orchestrator may stop in
	 * bulk and runs it must leave alone, with a reason for every exclusion so the
	 * caller can report honestly instead of silently doing less than it said.
	 */
	partitionOwned(entries: RegistryEntry[], options: { includeTerminal?: boolean } = {}): {
		owned: RegistryEntry[];
		foreign: Array<{ entry: RegistryEntry; reason: string }>;
	} {
		const owned: RegistryEntry[] = [];
		const foreign: Array<{ entry: RegistryEntry; reason: string }> = [];
		for (const entry of entries) {
			if (!options.includeTerminal && isTerminalState(entry.status.state)) continue;
			const verdict = ownershipOf(entry.status, this.owner, { livePump: this.pumps.has(entry.runId) });
			if (verdict.owned) owned.push(entry);
			else foreign.push({ entry, reason: verdict.reason });
		}
		return { owned, foreign };
	}

	/** Live pumps are exact in-process ownership evidence even if status.json is corrupt. */
	private includeLivePumps(entries: RegistryEntry[]): RegistryEntry[] {
		const byId = new Map(entries.map((entry) => [entry.runId, entry]));
		for (const [runId, pump] of this.pumps) {
			byId.set(runId, {
				runId,
				paths: pump.getPaths(),
				status: pump.getStatus(),
				live: true,
			});
		}
		return [...byId.values()];
	}

	/** R-CTRL-31: adoption transfers ownership, so the adopter may then control it. */
	claimOwnership(paths: RunPaths): void {
		try {
			patchStatus(paths, (current) => ({ ...current, owner: this.owner }));
		} catch {
			// E37: ownership is an optimisation for bulk control, never a spawn blocker.
		}
	}

	private resolveRun(prefix: string): RegistryEntry {
		const { entries } = loadAllRuns(this.host.cwd);
		const exact = entries.find((entry) => entry.runId === prefix.trim());
		if (exact !== undefined) return exact;
		const matches = entries.filter((entry) => entry.runId.startsWith(prefix.trim()));
		if (matches.length === 0) throw new Error(`No run matches '${prefix}'.`);
		if (matches.length > 1) throw new Error(`Run id prefix '${prefix}' is ambiguous: ${matches.map((entry) => entry.runId).join(", ")}.`);
		return matches[0] as RegistryEntry;
	}

	async steer(
		runId: string,
		message: string,
		waitForAck = true,
		source: "orchestrator" | "user" = "orchestrator",
		interrupt = false,
	): Promise<SteeringResult> {
		const entry = this.resolveRun(runId);
		if (isTerminalState(entry.status.state) || entry.status.state === "paused") {
			return { runId: entry.runId, interrupt, status: entry.status, reqId: "", state: "failed", detail: `run is ${entry.status.state}; no longer accepts steering` };
		}
		const result = await queueSteer(
			entry.paths,
			message,
			{
				waitForAck,
				interrupt,
				source,
				spawning: entry.status.state === "spawning" || entry.status.state === "queued",
				// R-CTRL-10: `supported: false` may only destroy this request if the record
				// belongs to the worker that is running right now.
				livePid: entry.status.pid,
			},
		);
		return { ...result, runId: entry.runId, interrupt, status: readStatus(entry.paths).status ?? entry.status };
	}

	/** R-CTRL-2: only nudge after the worker has proved its inbox is installed. */
	private readyControlProcess(entry: RegistryEntry): RecordedProcess | null {
		const recorded = recordedProcessOf(entry.status);
		if (recorded === null) return null;
		const raw = readFileIfExists(entry.paths.steerCapability);
		if (raw === undefined) return null;
		const capability = parseSteerCapability(raw);
		if (!capability.ok || capability.value.pid !== recorded.pid) return null;
		return recorded;
	}

	interrupt(runId: string, source: "orchestrator" | "user" = "orchestrator"): { runId: string; state: string; detail?: string } {
		const entry = this.resolveRun(runId);
		if (entry.status.state === "stopped") throw new Error("stopped runs cannot be resumed or interrupted; delegate a new run");
		if (entry.status.state === "paused") return { runId: entry.runId, state: "paused", detail: "already interrupted" };
		if (isTerminalState(entry.status.state)) throw new Error(`run is ${entry.status.state}; it is no longer active`);
		if (entry.status.state === "queued") throw new Error("a queued run has no session to interrupt; stop it or let it spawn first");
		const result = writeControlFirst(
			entry.paths,
			"interrupt",
			{ action: "interrupt", ts: new Date().toISOString(), source },
			this.readyControlProcess(entry),
		);
		const pump = this.pumps.get(entry.runId);
		pump?.markInterrupted();
		pump?.sendAbort();
		return { runId: entry.runId, state: "interrupting", ...(result.detail === undefined ? {} : { detail: result.detail }) };
	}

	/** Model-safe public stop entry point. Live work is pause-gated. */
	stop(runId: string, reason: string): ControlResult {
		const entry = this.resolveRun(runId);
		return this.stopEntryAutonomously(entry, reason);
	}

	/** Explicit human/lifecycle authority; never exposed through the model tool. */
	stopImmediately(runId: string, reason: string, source: "orchestrator" | "user"): ControlResult {
		return this.stopEntry(this.resolveRun(runId), reason, source);
	}

	private stopEntryAutonomously(entry: RegistryEntry, reason: string): ControlResult {
		const acquired = acquireStatusClaims([entry.paths]);
		if (!acquired.ok) return { runId: entry.runId, state: "refused", detail: `${acquired.reason}; no stop was attempted` };
		const claim = acquired.claims.get(statusClaimKey(entry.paths));
		if (claim === undefined) {
			releaseStatusClaims(acquired.claims);
			return { runId: entry.runId, state: "refused", detail: "status mutation claim was not returned; no stop was attempted" };
		}
		let stopped: RunStatus | undefined;
		try {
			const fresh = readStatus(entry.paths);
			if (fresh.status === undefined) {
				return {
					runId: entry.runId,
					state: "refused",
					detail: `current run status is unreadable; no stop was attempted${fresh.problem === undefined ? "" : ` (${fresh.problem})`}`,
				};
			}
			const currentEntry: RegistryEntry = { ...entry, status: fresh.status };
			const phase = autonomousLivePhase(currentEntry, this.pumps.has(entry.runId));
			if (phase !== undefined) return autonomousStopRefusal(currentEntry, phase);
			if (currentEntry.status.state === "stopped") return { runId: entry.runId, state: "stopped", detail: "already stopped" };
			if (isTerminalState(currentEntry.status.state)) {
				return { runId: entry.runId, state: currentEntry.status.state, detail: "run had already settled; its result is preserved" };
			}
			if (currentEntry.status.state !== "queued" && currentEntry.status.state !== "paused") {
				return { runId: entry.runId, state: "refused", detail: `run is ${currentEntry.status.state}; no stop was attempted` };
			}
			stopped = this.commitAutonomousStop(currentEntry, reason, claim);
		} finally {
			releaseStatusClaims(acquired.claims);
		}
		if (stopped === undefined) throw new Error(`claimed autonomous stop did not produce ${entry.runId}`);
		this.emitTerminal(entry.runId, stopped);
		this.onFleetChange();
		return { runId: entry.runId, state: "stopped" };
	}

	/** Commit only after the caller has decided eligibility while holding this claim. */
	private commitAutonomousStop(entry: RegistryEntry, reason: string, claim: StatusClaim): RunStatus {
		const stop = createStopFact(reason, "orchestrator");
		const mutation = patchStatusIf(entry.paths, (latest) => {
			if (latest.state !== "queued" && latest.state !== "paused") {
				throw new Error(`claimed autonomous stop invariant failed: ${entry.runId} changed to ${latest.state}`);
			}
			return {
				...latest,
				state: "stopped",
				endedAt: new Date().toISOString(),
				stopped: true,
				stop,
			};
		}, claim);
		if (!mutation.changed || mutation.status === undefined) throw new Error(`claimed autonomous stop did not mutate ${entry.runId}`);

		this.dequeue(entry.runId);
		appendEvent(entry.paths, { ts: stop.requestedAt, kind: "stop_requested", source: stop.source, detail: stop.reason });
		writeClosed(entry.paths, mutation.status);
		return mutation.status;
	}

	/**
	 * Stop one already-resolved run.
	 *
	 * Takes an entry rather than a run id so bulk callers read the registry once:
	 * resolving per run made `stopAll` re-parse every `status.json` in the repo for
	 * each run it stopped, which is what pushed `session_shutdown` past its
	 * R-CTRL-19 budget.
	 */
	private stopEntry(
		entry: RegistryEntry,
		reason: string,
		source: "orchestrator" | "user",
		statusWaitMs?: number,
	): { runId: string; state: string; detail?: string } {
		const acquired = acquireStatusClaims([entry.paths], statusWaitMs === undefined ? {} : { waitMs: statusWaitMs });
		if (!acquired.ok) throw new Error(acquired.reason);
		const claim = acquired.claims.get(statusClaimKey(entry.paths));
		if (claim === undefined) {
			releaseStatusClaims(acquired.claims);
			throw new Error("status mutation claim was not returned");
		}

		let currentEntry: RegistryEntry | undefined;
		let immediateResult: ControlResult | undefined;
		let stopped: RunStatus | undefined;
		let liveStop: RunStatus["stop"];
		try {
			const fresh = readStatus(entry.paths);
			if (fresh.status === undefined) {
				throw new Error(`current run status is unreadable${fresh.problem === undefined ? "" : ` (${fresh.problem})`}`);
			}
			currentEntry = { ...entry, status: fresh.status };
			if (currentEntry.status.state === "stopped") {
				immediateResult = { runId: entry.runId, state: "stopped", detail: "already stopped" };
			} else if (isTerminalState(currentEntry.status.state)) {
				immediateResult = {
					runId: entry.runId,
					state: currentEntry.status.state,
					detail: "run had already settled; its result is preserved",
				};
			} else {
				const pausedStillLive = currentEntry.status.state === "paused" && pausedExitPending(currentEntry, this.pumps.has(entry.runId));
				if (currentEntry.status.state === "queued" || (currentEntry.status.state === "paused" && !pausedStillLive)) {
					const stop = createStopFact(reason, source);
					const mutation = patchStatusIf(entry.paths, (latest) => {
						if (latest.state !== currentEntry?.status.state) {
							throw new Error(`claimed immediate stop invariant failed: ${entry.runId} changed to ${latest.state}`);
						}
						return {
							...latest,
							state: "stopped",
							endedAt: new Date().toISOString(),
							stopped: true,
							stop,
						};
					}, claim);
					if (!mutation.changed || mutation.status === undefined) throw new Error(`claimed immediate stop did not mutate ${entry.runId}`);
					stopped = mutation.status;
					this.dequeue(entry.runId);
					appendEvent(entry.paths, { ts: stop.requestedAt, kind: "stop_requested", source: stop.source, detail: stop.reason });
					writeClosed(entry.paths, mutation.status);
					try {
						fs.rmSync(entry.paths.stop, { force: true });
					} catch {
						// The durable stopped status and closed tombstone have superseded the marker.
					}
					immediateResult = { runId: entry.runId, state: "stopped" };
				} else {
					// A live stop is accepted here, while the exact fresh status is still
					// claimed. The marker, nudge, abort, and escalation are delivery side
					// effects; none of them may be the only durable copy of operator intent.
					const proposed = createStopFact(reason, source);
					const mutation = patchStatusIf(entry.paths, (latest) => {
						if (latest.state !== currentEntry?.status.state) {
							throw new Error(`claimed live stop invariant failed: ${entry.runId} changed to ${latest.state}`);
						}
						return { ...latest, stopped: true, stop: latest.stop ?? proposed };
					}, claim);
					if (!mutation.changed || mutation.status === undefined) throw new Error(`claimed live stop did not persist ${entry.runId}`);
					currentEntry = { ...currentEntry, status: mutation.status };
					liveStop = mutation.status.stop ?? proposed;
				}
			}
		} finally {
			releaseStatusClaims(acquired.claims);
		}

		if (stopped !== undefined) {
			this.emitTerminal(entry.runId, stopped);
			this.onFleetChange();
		}
		if (immediateResult !== undefined) return immediateResult;
		if (currentEntry === undefined) throw new Error(`claimed immediate stop did not resolve ${entry.runId}`);

		// Live stop side effects happen only after releasing the status claim: pump
		// status writes use the same claim, and callbacks must never run under it.
		const stop = liveStop;
		if (stop === undefined || stop === null) throw new Error(`claimed live stop did not retain intent for ${entry.runId}`);
		const result = writeControlFirst(
			currentEntry.paths,
			"stop",
			{ action: "stop", ts: stop.requestedAt, source: stop.source, reason: stop.reason },
			this.readyControlProcess(currentEntry),
		);
		const pump = this.pumps.get(currentEntry.runId);
		if (pump === undefined) appendEvent(currentEntry.paths, { ts: stop.requestedAt, kind: "stop_requested", source: stop.source, detail: stop.reason });
		pump?.markStopped(stop, statusWaitMs);
		pump?.sendAbort();
		pump?.requestStopEscalation();
		return { runId: entry.runId, state: "stopping", ...(result.detail === undefined ? {} : { detail: result.detail }) };
	}

	/**
	 * R-CTRL-18. Stop every active run **this orchestrator owns**.
	 *
	 * Scoping is the point. A terminal stop is unresumable, `runs/` is shared by every
	 * pi session in the repo, and this is reached from `session_shutdown` — so an
	 * unscoped version destroyed a concurrent session's work, `paused` runs included,
	 * every time an unrelated tab was closed. Runs that are skipped are returned with
	 * a reason instead of being silently ignored.
	 */
	stopAllImmediately(
		reason: string,
		source: "orchestrator" | "user",
		options: { entries?: RegistryEntry[]; deadline?: number; includePumps?: boolean } = {},
	): Array<{ runId: string; state: string; detail?: string }> {
		const base = options.entries ?? loadAllRuns(this.host.cwd).entries;
		const entries = options.includePumps === false ? base : this.includeLivePumps(base);
		const { owned, foreign } = this.partitionOwned(entries);
		const results: Array<{ runId: string; state: string; detail?: string }> = [];
		for (const entry of owned) {
			if (options.deadline !== undefined && Date.now() >= options.deadline) {
				results.push({ runId: entry.runId, state: "skipped", detail: "shutdown budget was exhausted before this run was reached" });
				continue;
			}
			try {
				const statusWaitMs = options.deadline === undefined ? undefined : Math.max(0, options.deadline - Date.now());
				results.push(this.stopEntry(entry, reason, source, statusWaitMs));
			} catch (error) {
				results.push({ runId: entry.runId, state: "failed", detail: (error as Error).message });
			}
		}
		for (const skipped of foreign) {
			results.push({ runId: skipped.entry.runId, state: "skipped", detail: skipped.reason });
		}
		return results;
	}

	/** Model-safe public bulk entry point. All owned statuses are claimed before mutation. */
	stopAll(reason: string): ControlResult[] {
		// Issue 4: explicit human/lifecycle callers use stopAllImmediately instead.
		const loaded = loadAllRuns(this.host.cwd);
		const initial = this.partitionOwned(loaded.entries, { includeTerminal: true });
		if (loaded.problems.length > 0) {
			return autonomousFleetRefusal(
				initial.owned,
				initial.foreign,
				"the durable fleet status is incomplete; no owned run was stopped",
				loaded.problems,
			);
		}
		const claimable = loaded.entries.filter((entry) => !isTerminalState(entry.status.state));
		const acquired = acquireStatusClaims(claimable.map((entry) => entry.paths));
		if (!acquired.ok) {
			return autonomousFleetRefusal(initial.owned, initial.foreign, `${acquired.reason}; no owned run was stopped`);
		}
		const committed: Array<{ runId: string; status: RunStatus }> = [];
		let completedResults: ControlResult[] | undefined;
		try {
			const claimableKeys = new Set(claimable.map((entry) => statusClaimKey(entry.paths)));
			const freshEntries: RegistryEntry[] = [];
			for (const entry of loaded.entries) {
				const fresh = readStatus(entry.paths);
				if (fresh.status === undefined) {
					return autonomousFleetRefusal(
						initial.owned,
						initial.foreign,
						"the durable fleet status became incomplete while its mutation claims were acquired; no owned run was stopped",
						[{ runId: entry.runId, reason: fresh.problem ?? "status.json disappeared" }],
					);
				}
				freshEntries.push({ ...entry, status: fresh.status });
			}
			const refreshed = this.partitionOwned(freshEntries, { includeTerminal: true });
			const freshOwned = refreshed.owned;
			const blockers = freshOwned.flatMap((entry) => {
				const phase = autonomousLivePhase(entry, this.pumps.has(entry.runId));
				return phase === undefined ? [] : [{ entry, phase }];
			});
			if (blockers.length > 0) {
				const blockerText = blockers.map(({ entry, phase }) => `${entry.runId} (${phase})`).join(", ");
				return autonomousFleetRefusal(
					freshOwned,
					refreshed.foreign,
					`live owned worker${blockers.length === 1 ? "" : "s"}: ${blockerText}; no owned run was stopped — inspect or steer live work, or interrupt it before deciding to replace it`,
				);
			}
			const eligible = freshOwned.filter((entry) => !isTerminalState(entry.status.state));
			if (eligible.some((entry) => !claimableKeys.has(statusClaimKey(entry.paths)) || !acquired.claims.has(statusClaimKey(entry.paths)))) {
				return autonomousFleetRefusal(freshOwned, refreshed.foreign, "an owned status mutation claim was lost; no owned run was stopped");
			}
			for (const entry of eligible) {
				const claim = acquired.claims.get(statusClaimKey(entry.paths)) as StatusClaim;
				committed.push({ runId: entry.runId, status: this.commitAutonomousStop(entry, reason, claim) });
			}
			const committedByRun = new Map(committed.map((item) => [item.runId, item.status]));
			completedResults = [
				...freshOwned.map((entry): ControlResult => {
					if (committedByRun.has(entry.runId)) return { runId: entry.runId, state: "stopped" };
					if (entry.status.state === "stopped") return { runId: entry.runId, state: "stopped", detail: "already stopped" };
					return { runId: entry.runId, state: entry.status.state, detail: "run had already settled; its result is preserved" };
				}),
				...refreshed.foreign.map(({ entry, reason: skippedReason }) => ({ runId: entry.runId, state: "skipped", detail: skippedReason })),
			];
		} finally {
			releaseStatusClaims(acquired.claims);
		}
		for (const item of committed) this.emitTerminal(item.runId, item.status);
		if (committed.length > 0) this.onFleetChange();
		return completedResults ?? [];
	}

	resume(runId: string, message: string): SpawnedRun {
		const resolved = this.resolveRun(runId);
		const acquired = acquireStatusClaims([resolved.paths]);
		if (!acquired.ok) throw new Error(`cannot resume while status is being mutated: ${acquired.reason}`);
		try {
			const fresh = readStatus(resolved.paths);
			if (fresh.status === undefined) throw new Error(`cannot resume because status is unreadable: ${fresh.problem ?? "status.json disappeared"}`);
			const previous = { ...resolved, status: fresh.status };
			assertResumable(previous.status.state);
			if (pausedExitPending(previous, this.pumps.has(previous.runId))) {
				throw new Error("worker is still finishing interruption; retry resume after its process exits");
			}
			return this.resumeClaimed(previous, message);
		} finally {
			releaseStatusClaims(acquired.claims);
		}
	}

	private resumeClaimed(previous: RegistryEntry, message: string): SpawnedRun {
		const inheritedModel = this.host.inheritedModel;
		if (inheritedModel === undefined) throw new Error("cannot resume because the parent session has no resolved provider/model");
		assertResumable(previous.status.state);
		if (message.trim().length === 0) throw new Error("resume requires a follow-up message");
		const rawDescriptor = readFileIfExists(previous.paths.descriptor);
		if (rawDescriptor === undefined) throw new Error("relaunch contract unavailable; delegate a new run");
		// R-CTRL-21: the descriptor is checked against the run's own independently
		// written facts, not just for well-formedness, so a forged one cannot widen
		// capabilities or change the worker identity.
		const parsed = readDescriptor(rawDescriptor, {
			runId: previous.runId,
			name: previous.status.name,
			agent: previous.status.agent,
			cwd: previous.status.cwd,
			model: previous.status.model,
			...(previous.status.sessionId === null ? {} : { sessionId: previous.status.sessionId }),
		});
		if (!parsed.ok) throw new Error(`relaunch contract unavailable; delegate a new run (${parsed.reason})`);
		const descriptor = parsed.descriptor;
		if (
			previous.status.sessionFile !== null &&
			descriptor.sessionFile !== null &&
			path.resolve(previous.status.sessionFile) !== path.resolve(descriptor.sessionFile)
		) {
			throw new Error("descriptor session file identity does not match status.json");
		}
		const sessionFile = previous.status.sessionFile ?? descriptor.sessionFile;
		if (sessionFile === null || !fs.existsSync(sessionFile)) {
			throw new Error(`session file is missing${sessionFile === null ? "" : `: ${sessionFile}`}; delegate a new run`);
		}
		if (descriptor.sessionId === null || previous.status.sessionId !== descriptor.sessionId) {
			throw new Error("descriptor session identity does not match status.json");
		}
		// R-CTRL-24: `--session-id` is resolved *inside a directory*, so continuing the
		// same session requires the same directory. The orchestrator's current
		// `--session-dir` is not evidence of anything — it can have changed since the
		// original spawn — so the contract is the recorded one, and it has to actually
		// contain the verified session file.
		const sessionDir = descriptor.sessionDir ?? path.dirname(sessionFile);
		if (path.resolve(path.dirname(sessionFile)) !== path.resolve(sessionDir)) {
			throw new Error(
				`recorded session directory ${sessionDir} does not contain the session file ${sessionFile}; ` +
					"resuming there would continue a different session",
			);
		}
		const { entries } = loadAllRuns(this.host.cwd);
		const accountingEntries = accountingEntriesForResume(entries, previous.runId);
		const conflicting = liveRunForName(accountingEntries, previous.status.name);
		if (conflicting !== undefined) {
			throw new Error(`task '${previous.status.name}' already has a live successor ${conflicting.runId} (${conflicting.status.state})`);
		}
		if (!hasSlot(accountingEntries, false, this.host.config)) {
			throw new Error("no worker concurrency slot is available for this resume");
		}
		const nextRunId = newRunId();
		// Everything that can throw is done *before* the lease is acquired, and
		// everything after it is inside a failure path that releases exactly the token
		// we took. A throw between the two used to leave a lease owned by this very
		// process, which `proveDeath` correctly refuses to reclaim — permanently.
		const prompt = revivalPrompt(previous.status.name, previous.status.state, message);
		const lease = acquireSessionLease(sessionFile, nextRunId, descriptor.sessionId);
		if (!lease.ok) {
			throw new Error(
				`session lease is held${lease.holder === undefined ? "" : ` by run ${lease.holder.runId}, pid ${lease.holder.pid}, host ${lease.holder.hostname}`}: ${lease.reason}`,
			);
		}
		let leaseHandedOver = false;
		try {
			const paths = runPaths(runsDir(this.host.cwd), nextRunId);
			ensureRunDirs(paths);
			inheritAgentHistory(previous.paths, paths);
			const status = createStatus({
				runId: nextRunId,
				name: previous.status.name,
				agent: descriptor.agent,
				state: "spawning",
				cwd: descriptor.cwd,
				model: inheritedModel,
				thinking: this.host.inheritedThinking ?? null,
				// Legacy status field only: an empty list means the harness did not impose
				// a tool allowlist. Pi discovers the worker's normal tools itself.
				tools: [],
				readOnly: false,
				sessionId: descriptor.sessionId,
				runIndex: previous.status.runIndex + 1,
				previousRunId: previous.runId,
				owner: this.owner,
			});
			status.sessionFile = sessionFile;
			writeStatus(paths, status);
			const nextDescriptor: RunDescriptor = {
				...descriptor,
				schemaVersion: 5,
				runId: nextRunId,
				sourceRunId: nextRunId,
				prompt,
				sessionFile,
				sessionDir,
				createdAt: status.createdAt,
				model: inheritedModel,
				thinking: this.host.inheritedThinking ?? null,
			};
			writeDescriptor(paths, nextDescriptor);
			updateAgentName(this.host.cwd, status.name, nextRunId);
			this.leases.set(nextRunId, { path: lease.path, token: lease.owner.token });
			leaseHandedOver = true;
			return this.reviveNow(previous, nextRunId, paths, status, nextDescriptor, prompt, sessionDir, lease);
		} finally {
			// The lease is only ever owned by a run that made it into `this.leases`;
			// anything else must give it straight back.
			if (!leaseHandedOver) releaseSessionLease(lease.path, lease.owner.token);
		}
	}

	private reviveNow(
		previous: RegistryEntry,
		nextRunId: string,
		paths: RunPaths,
		status: RunStatus,
		descriptor: RunDescriptor,
		prompt: string,
		sessionDir: string,
		lease: { path: string; owner: { token: string } },
	): SpawnedRun {
		const pump = new WorkerPump({
			paths,
			status,
			callbacks: {
				onStatus: () => this.onFleetChange(),
				onTerminal: (finalStatus) => this.onRunTerminal(nextRunId, finalStatus),
				onEvidence: (evidenceStatus) => this.emitTerminal(nextRunId, evidenceStatus),
			},
			onReport: (finalStatus, finalText) => this.writeResult(paths, finalStatus, finalText),
		});
		this.pumps.set(nextRunId, pump);
		try {
			const resolution = resolvePi({ configPiBinary: this.host.config.piBinary });
			const promptFile = writePromptFile(descriptor.appendSystemPrompt);
			this.promptFiles.set(nextRunId, promptFile);
			const args = buildSpawnArgs({
				sessionId: descriptor.sessionId,
				sessionDir,
				...(status.model === null ? {} : { model: status.model }),
				...(status.thinking === null ? {} : { thinking: status.thinking }),
				promptFile,
				approve: sameRealpath(descriptor.cwd, this.host.cwd) ? undefined : false,
				extensionPaths: this.host.inheritedExtensionPaths,
			});
			const env = buildWorkerEnv(process.env, {
				runId: nextRunId,
				name: previous.status.name,
				runDir: paths.dir,
				depth: this.host.depth + 1,
				maxDepth: this.host.config.maxDepth,
			});
			pump.event({ kind: "revive", previousRunId: previous.runId });
			const spawned = (this.host.spawn ?? spawnWorker)({ resolution, args, cwd: descriptor.cwd, env });
			if (spawned.pid <= 0) throw new Error("the child reported no pid");
			pump.setSpawned(spawned.pid, spawned.processStartIdentity);
			if (!transferSessionLease(lease.path, lease.owner.token, spawned.pid, spawned.processStartIdentity)) {
				pump.terminateNow();
				throw new Error("session lease ownership could not be transferred to the revived worker");
			}
			const acceptance = this.acceptSpawnedWorker(nextRunId, pump, spawned, prompt);
			this.onFleetChange();
			return { runId: nextRunId, status: pump.getStatus(), queued: false, acceptance };
		} catch (error) {
			// `finalize` reaches `onRunTerminal`, which releases the lease this run holds.
			const reason = `worker revive failed: ${(error as Error).message}`;
			pump.finalize("failed", reason);
			this.cleanupPromptFile(nextRunId);
			this.onFleetChange();
			const failed = pump.getStatus();
			return {
				runId: nextRunId,
				status: failed,
				queued: false,
				acceptance: Promise.resolve({ ok: false as const, status: failed, reason }),
			};
		}
	}

	refreshControlState(): void {
		for (const pump of this.pumps.values()) pump.syncExternalStatus();
	}

	/**
	 * R-TOOL-12: validate everything before any process starts, and reject the
	 * *whole* call on any failure. A partial spawn leaves the fleet in a state the
	 * orchestrator did not intend.
	 */
	validate(request: DelegateRequest, entries: RegistryEntry[]): { profile: AgentProfile; cwd: string; readOnly: boolean } {
		validateAgentName(request.name);
		const profile = this.host.profiles.get(request.agent);
		if (profile === undefined) {
			const available = [...this.host.profiles.values()].filter((candidate) => !candidate.disabled).map((candidate) => candidate.name);
			throw new Error(`Unknown agent profile '${request.agent}'. Available: ${available.join(", ")}.`);
		}
		if (profile.disabled) throw new Error(`Agent profile '${request.agent}' is disabled.`);

		const cwd = path.resolve(request.cwd ?? this.host.cwd);
		let stat: fs.Stats;
		try {
			stat = fs.statSync(cwd);
		} catch {
			throw new Error(`cwd '${cwd}' does not exist.`);
		}
		if (!stat.isDirectory()) throw new Error(`cwd '${cwd}' is not a directory.`);
		if (!this.host.config.allowExternalCwd) {
			const root = path.resolve(this.host.cwd);
			const rel = path.relative(root, cwd);
			// A cwd outside the repo is refused by default: a worker with edit/write in an
			// unrelated directory is a much larger blast radius than the orchestrator
			// intended.
			if (rel.startsWith("..") || path.isAbsolute(rel)) {
				throw new Error(
					`cwd '${cwd}' is outside the project root '${root}'. Set "allowExternalCwd": true in .pi/agi/config.json to permit this.`,
				);
			}
		}

		const existing = entries.find((entry) => entry.status.name === request.name);
		if (existing !== undefined) {
			throw new Error(
				`Agent '${request.name}' already exists. Resume it to continue the saved conversation, ` +
					"or choose a different name for a fresh agent.",
			);
		}

		// R-CONC-8 / D12: workers cannot delegate at all (the role guard never
		// registers the AGI tools), so this is an observability check, not the
		// enforcement mechanism.
		if (this.host.depth >= this.host.config.maxDepth) {
			throw new Error(`Delegation depth ${this.host.depth} has reached the configured cap of ${this.host.config.maxDepth}.`);
		}

		if (this.host.inheritedModel === undefined) {
			throw new Error("cannot delegate because the parent session has no resolved provider/model");
		}
		return { profile, cwd, readOnly: false };
	}

	/** Create and spawn one run. A live worker rejects another delegation. */
	start(request: DelegateRequest): SpawnedRun {
		const { entries } = loadAllRuns(this.host.cwd);
		const validated = this.validate(request, entries);
		if (!hasSlot(entries, validated.readOnly, this.host.config)) {
			throw new Error("another worker is still active; wait for it to finish, review its result, then decide whether to delegate the next task");
		}
		const dir = runsDir(this.host.cwd);
		// R-EXEC-8: `.runtime/` must be gitignored, and the run directory is about to be
		// created inside the worker's own cwd. Done here rather than only on activation
		// so the invariant holds for every path that creates a run.
		gitignoreRuntime(this.host.cwd);
		const runId = newRunId();
		const paths = runPaths(dir, runId);
		claimAgentName(this.host.cwd, request.name, runId);
		try {
			ensureRunDirs(paths);
		} catch (error) {
			try { fs.unlinkSync(path.join(this.host.cwd, ".pi", "agi", ".runtime", "agents", request.name)); } catch {}
			throw error;
		}

		const profile = validated.profile;
		// The worker model is never selected by a delegation call or profile. Use the
		// exact provider/model captured from the parent session.
		const model = this.host.inheritedModel;
		const thinking = this.host.inheritedThinking;
		const status = createStatus({
			runId,
			name: request.name,
			agent: profile.name,
			state: "queued",
			cwd: validated.cwd,
			model: model ?? null,
			thinking: thinking ?? null,
			// Legacy status field; actual capabilities come from normal Pi discovery.
			tools: [],
			readOnly: validated.readOnly,
			// A stable per-task session id makes a resume continue that session rather
			// than fork a divergent copy (R-CTRL-24). --no-session would make resume
			// impossible, so a worker always has one.
			sessionId: `agi.${runId}`,
			// R-CTRL-18 scoping: record who is responsible for this run, so a concurrent
			// session in the same repo cannot terminally stop it in bulk.
			owner: this.owner,
		});
		writeStatus(paths, status);

		const spawned = this.spawnNow(request, runId, paths, profile, {
			model,
			thinking,
			cwd: validated.cwd,
			readOnly: validated.readOnly,
			status,
		});
		return { runId, status: spawned.status, queued: false, acceptance: spawned.acceptance };
	}

	private spawnNow(
		request: DelegateRequest,
		runId: string,
		paths: RunPaths,
		profile: AgentProfile,
		resolved: {
			model: string | undefined;
			thinking: string | undefined;
			cwd: string;
			readOnly: boolean;
			status: RunStatus;
		},
	): { status: RunStatus; acceptance: SpawnedRun["acceptance"] } {
		const appendPrompt = buildWorkerPrompt({
			name: request.name,
			agent: profile.name,
			profileBody: profile.body,
		});

		const status: RunStatus = { ...resolved.status, state: "spawning" };
		writeStatus(paths, status);

		// R-CTRL-20: the descriptor is the relaunch contract and holds the appended
		// prompt *inline*, because the temp file below is deleted as soon as the child
		// has read it.
		const descriptor: RunDescriptor = {
			schemaVersion: 5,
			runId,
			name: request.name,
			agent: profile.name,
			model: resolved.model ?? null,
			thinking: resolved.thinking ?? null,
			appendSystemPrompt: appendPrompt,
			prompt: request.prompt,
			cwd: resolved.cwd,
			sessionFile: null,
			sessionId: status.sessionId,
			// Recorded so a resume resolves `--session-id` in the same place even if the
			// orchestrator's own `--session-dir` has changed since (R-CTRL-24). Filled in
			// exactly once more, from the worker's own `get_state`, when the real session
			// file path is known.
			sessionDir: this.host.sessionDir ?? null,
			sourceRunId: runId,
			createdAt: status.createdAt,
		};
		writeDescriptor(paths, descriptor);

		const pump = new WorkerPump({
			paths,
			status,
			callbacks: {
				onStatus: () => this.onFleetChange(),
				onTerminal: (finalStatus) => this.onRunTerminal(runId, finalStatus),
				onEvidence: (evidenceStatus) => this.emitTerminal(runId, evidenceStatus),
			},
			onReport: (finalStatus, finalText) => this.writeResult(paths, finalStatus, finalText),
		});
		this.pumps.set(runId, pump);

		let resolution: PiResolution;
		try {
			resolution = resolvePi({ configPiBinary: this.host.config.piBinary });
		} catch (error) {
			const reason = (error as Error).message;
			pump.finalize("failed", reason);
			const failed = pump.getStatus();
			return { status: failed, acceptance: Promise.resolve({ ok: false as const, status: failed, reason }) };
		}

		const promptFile = writePromptFile(appendPrompt);
		this.promptFiles.set(runId, promptFile);

		const args = buildSpawnArgs({
			sessionId: status.sessionId,
			...(this.host.sessionDir === undefined ? {} : { sessionDir: this.host.sessionDir }),
			...(resolved.model === undefined ? {} : { model: resolved.model }),
			...(resolved.thinking === undefined ? {} : { thinking: resolved.thinking }),
			promptFile,
			// R-WORK-2: project trust transfers only on realpath equality, and
			// --no-approve is always preserved.
			approve: sameRealpath(resolved.cwd, this.host.cwd) ? undefined : false,
			extensionPaths: this.host.inheritedExtensionPaths,
		});

		const env = buildWorkerEnv(process.env, {
			runId,
			name: request.name,
			runDir: paths.dir,
			depth: this.host.depth + 1,
			maxDepth: this.host.config.maxDepth,
		});

		pump.event({ kind: "spawn", command: resolution.command, args, via: resolution.via, cwd: resolved.cwd });

		try {
			const spawned = (this.host.spawn ?? spawnWorker)({ resolution, args, cwd: resolved.cwd, env });
			if (spawned.pid <= 0) throw new Error("the child reported no pid");
			pump.setSpawned(spawned.pid, spawned.processStartIdentity);
			const acceptance = this.acceptSpawnedWorker(runId, pump, spawned, request.prompt);
			this.onFleetChange();
			return { status: pump.getStatus(), acceptance };
		} catch (error) {
			// E21/E22: the failure names the resolution path that was used, so "cannot
			// find pi" is actionable rather than mysterious.
			const detail =
				resolution.via === "PATH"
					? `${(error as Error).message}\n\n${describeResolutionFailure(resolution)}`
					: `${(error as Error).message} (pi resolved via ${resolution.via}: ${resolution.command})`;
			const reason = `worker spawn failed: ${detail}`;
			pump.finalize("failed", reason);
			this.cleanupPromptFile(runId);
			this.onFleetChange();
			const failed = pump.getStatus();
			return { status: failed, acceptance: Promise.resolve({ ok: false as const, status: failed, reason }) };
		}
	}

	private acceptSpawnedWorker(
		runId: string,
		pump: WorkerPump,
		spawned: ReturnType<typeof spawnWorker>,
		prompt: string,
	): SpawnedRun["acceptance"] {
		// Attach immediately so a stop or launcher failure before readiness is still
		// observed. Prompt/session traffic is withheld until the anchor child has
		// installed its signal traps and Node has verified its identity and group.
		pump.attach(spawned.child);
		const isSupervised = (): boolean => this.pumps.get(runId) === pump;
		const reject = (reason: string): { ok: false; status: RunStatus; reason: string } => {
			if (isSupervised()) pump.event({ kind: "worker_acceptance_rejected", detail: reason });
			this.cleanupPromptFile(runId);
			return { ok: false, status: pump.getStatus(), reason };
		};
		const accept = (): { ok: true; status: RunStatus } | { ok: false; status: RunStatus; reason: string } => {
			const current = pump.getStatus();
			if (current.stopped || current.interrupted || isTerminalState(current.state) || current.state === "paused") {
				return reject("worker stopped before launcher readiness was accepted");
			}
			// R-EXEC-1: exactly one line at start, the task text verbatim in a JSON
			// string. Withholding it until this point prevents pre-anchor work.
			if (!pump.sendPrompt(prompt)) {
				pump.finalize("failed", "worker launcher became unavailable before the initial prompt could be sent");
				return reject("worker launcher became unavailable before prompt acceptance");
			}
			pump.requestSessionInfo();
			const timer = setTimeout(() => this.cleanupPromptFile(runId), 60_000);
			timer.unref?.();
			this.promptTimers.set(runId, timer);
			return { ok: true, status: pump.getStatus() };
		};

		if (spawned.anchorReady === null) return Promise.resolve(accept());
		return spawned.anchorReady.then(async (result) => {
			if (!isSupervised()) {
				return { ok: false as const, status: pump.getStatus(), reason: "worker supervision ended before launcher readiness was accepted" };
			}
			if (!result.ok) {
				pump.event({ kind: "process_group_anchor_unavailable", detail: result.reason });
				const beforeExit = pump.getStatus();
				if (
					!beforeExit.stopped &&
					!beforeExit.interrupted &&
					beforeExit.processSignal === null &&
					!isTerminalState(beforeExit.state) &&
					beforeExit.state !== "paused" &&
					spawned.child.exitCode === null &&
					spawned.child.signalCode === null
				) {
					// A group signal can close the private anchor pipe before Node emits the
					// child exit event. Wait boundedly for that already-owned child so its
					// signal listener can publish R-EXEC-12 facts before readiness failure is
					// allowed to classify the run.
					await new Promise<void>((resolve) => {
						let timer: ReturnType<typeof setTimeout> | undefined;
						const finish = (): void => {
							if (timer !== undefined) clearTimeout(timer);
							spawned.child.off("exit", finish);
							resolve();
						};
						spawned.child.once("exit", finish);
						timer = setTimeout(finish, 1_000);
					});
				}
				const current = pump.getStatus();
				if (
					isSupervised() &&
					!current.stopped &&
					!current.interrupted &&
					current.processSignal === null &&
					!isTerminalState(current.state) &&
					current.state !== "paused"
				) {
					pump.finalize("failed", `worker launcher anchor was not ready: ${result.reason}`);
				}
				return reject(result.reason);
			}
			pump.setProcessGroupAnchor(result.anchor);
			if (pump.getStatus().stopped) pump.terminateWithEscalation();
			return accept();
		});
	}

	private cleanupPromptFile(runId: string): void {
		const timer = this.promptTimers.get(runId);
		if (timer !== undefined) clearTimeout(timer);
		this.promptTimers.delete(runId);
		const file = this.promptFiles.get(runId);
		if (file === undefined) return;
		this.promptFiles.delete(runId);
		try {
			fs.unlinkSync(file);
		} catch {
			// A leftover temp file is inert.
		}
	}

	/**
	 * R-WORK-8/10. Write any non-empty final assistant text to `result.md` in full.
	 * Called from the pump at settle and at any finalize, so a timed-out or stopped
	 * worker's partial report is still saved (E31). Lifecycle state is decided only
	 * by the pump; report prose never reclassifies it.
	 */
	private writeResult(paths: RunPaths, status: RunStatus, finalText: string | null): void {
		if (finalText === null || finalText.trim().length === 0) return;
		try {
			fs.writeFileSync(paths.result, finalText.endsWith("\n") ? finalText : `${finalText}\n`, { encoding: "utf8", mode: 0o600 });
			status.resultPath = paths.result;
		} catch {
			// E37: losing result.md is bad but must not crash the orchestrator.
		}
	}

	/** R-CONC-4: drain on settle, tolerating late validation failures. */
	private onRunTerminal(runId: string, status: RunStatus): void {
		this.pumps.delete(runId);
		const lease = this.leases.get(runId);
		if (lease !== undefined) {
			releaseSessionLease(lease.path, lease.token);
			this.leases.delete(runId);
		}
		this.cleanupPromptFile(runId);
		this.drain();
		this.onFleetChange();
		if (isTerminalState(status.state)) this.emitTerminal(runId, status);
	}

	private emitTerminal(runId: string, status: RunStatus): void {
		// Last, and guarded: the scheduler arms a wake here, and a throw from that path
		// would otherwise abort the queue drain that already ran above — leaving a
		// finished slot unfilled because a notification failed.
		try {
			this.onTerminal(runId, status);
		} catch {
			// A failed wake costs one notification; the next tick recovers it (E67).
		}
	}

	drain(): void {
		this.refreshControlState();
		if (this.queue.length === 0) return;
		const dir = runsDir(this.host.cwd);
		let progressed = true;
		while (progressed && this.queue.length > 0) {
			progressed = false;
			const { entries } = loadAllRuns(this.host.cwd);
			for (let i = 0; i < this.queue.length; i++) {
				const queued = this.queue[i];
				if (queued === undefined) continue;
				const paths = runPaths(dir, queued.runId);
				let validated: { profile: AgentProfile; cwd: string; readOnly: boolean };
				try {
					// R-CONC-5: everything validated at enqueue is re-validated at spawn. A
					// queued run's cwd may have vanished and its profile may have been deleted;
					// validating only at enqueue is a TOCTOU bug.
					validated = this.validate(queued.request, entries.filter((entry) => entry.runId !== queued.runId));
				} catch (error) {
					// R-CONC-4: a queued run whose validation now fails is marked failed with a
					// reason and the drain continues.
					const existing = entries.find((entry) => entry.runId === queued.runId)?.status;
					if (existing !== undefined) {
						const failed: RunStatus = {
							...existing,
							state: "failed",
							endedAt: new Date().toISOString(),
							error: `queued run failed re-validation at spawn: ${(error as Error).message}`,
						};
						writeStatus(paths, failed);
						this.emitTerminal(queued.runId, failed);
					}
					this.queue.splice(i, 1);
					i -= 1;
					progressed = true;
					continue;
				}
				if (!hasSlot(entries, validated.readOnly, this.host.config)) continue;
				const existing = entries.find((entry) => entry.runId === queued.runId)?.status;
				if (existing === undefined) {
					this.queue.splice(i, 1);
					i -= 1;
					progressed = true;
					continue;
				}
				this.queue.splice(i, 1);
				i -= 1;
				progressed = true;
				const profile = validated.profile;
				this.spawnNow(queued.request, queued.runId, paths, profile, {
					model: this.host.inheritedModel,
					thinking: this.host.inheritedThinking,
					cwd: validated.cwd,
					readOnly: validated.readOnly,
					status: existing,
				});
				break;
			}
		}
	}

	/** R-CONC-3: a stop against a queued run removes it without ever spawning. */
	dequeue(runId: string): boolean {
		const index = this.queue.findIndex((queued) => queued.runId === runId);
		if (index < 0) return false;
		this.queue.splice(index, 1);
		return true;
	}

	queuedRunIds(): string[] {
		return this.queue.map((queued) => queued.runId);
	}

	/**
	 * R-SLEEP-22 deadline path. Stop only the active runs named by the headless
	 * drain, never every process in the user's session. Queued runs are failed without
	 * spawning; live pumps latch `stopped` before SIGTERM so the child error cannot
	 * rewrite the operator's intent (R-EXEC-10).
	 */
	stopForHeadlessDrain(entries: RegistryEntry[], reason: string): void {
		const ids = new Set(entries.map((entry) => entry.runId));
		for (const runId of [...this.queuedRunIds()]) {
			if (!ids.has(runId) || !this.dequeue(runId)) continue;
			const entry = entries.find((candidate) => candidate.runId === runId);
			if (entry === undefined) continue;
			try {
				writeStatus(entry.paths, {
					...entry.status,
					state: "failed",
					endedAt: new Date().toISOString(),
					error: reason,
				});
			} catch {
				// The process never started; there is nothing else to stop.
			}
		}
		for (const [runId, pump] of this.pumps) {
			if (!ids.has(runId)) continue;
			const status = pump.getStatus();
			if (isTerminalState(status.state)) continue;
			pump.markStopped(createStopFact(reason, "orchestrator"));
			pump.event({ kind: "headless_drain_timeout", detail: reason });
			if (status.pid !== null) pump.terminateWithEscalation();
		}
		this.onFleetChange();
	}

	/**
	 * R-CTRL-19. Shutdown is bounded: write stop intent, perform identity-validated
	 * TERM/KILL escalation within the budget, and return. An unprovable target is
	 * deliberately left for reconciliation rather than risking an unrelated group.
	 */
	shutdown(): void {
		this.terminateOwned("orchestrator session is shutting down", "orchestrator");
		this.pumps.clear();
		this.queue = [];
	}

	/**
	 * R-CTRL-18 + R-CTRL-19: stop every owned run, then escalate to the process group,
	 * all inside `SHUTDOWN_BUDGET_MS`.
	 *
	 * Shared by `session_shutdown` and toggle OFF because both tell the user the runs
	 * are being stopped terminally. Toggle OFF previously armed the escalation through
	 * `stop_all` and then destroyed it by detaching in the same call, so a worker that
	 * never consumed the durable request simply kept running.
	 *
	 * The budget is real: the handler is synchronous and pi's exit blocks on it, so the
	 * registry is read once, the deadline is checked between runs, and remaining work
	 * is stopped rather than allowed to hang the exit.
	 */
	terminateOwned(
		reason: string,
		source: "orchestrator" | "user",
		budgetMs = SHUTDOWN_BUDGET_MS,
	): Array<{ runId: string; state: string; detail?: string }> {
		const deadline = Date.now() + budgetMs;
		const entries = this.includeLivePumps(loadAllRuns(this.host.cwd).entries);
		const pumpIds = new Set(this.pumps.keys());
		const pumpEntries = entries.filter((entry) => pumpIds.has(entry.runId));
		const results: Array<{ runId: string; state: string; detail?: string }> = [];
		for (const entry of pumpEntries) {
			try {
				results.push(this.stopEntry(entry, reason, source, 0));
			} catch (error) {
				results.push({ runId: entry.runId, state: "failed", detail: (error as Error).message });
			}
		}
		for (const [runId, pump] of this.pumps) {
			pump.event({ kind: "shutdown", detail: reason });
			pump.terminateWithin(deadline);
			pump.detach();
			this.cleanupPromptFile(runId);
		}
		const remaining = entries.filter((entry) => !pumpIds.has(entry.runId));
		results.push(...this.stopAllImmediately(reason, source, { entries: remaining, deadline, includePumps: false }));
		return results;
	}

	/**
	 * Drop every in-process handle **without signalling anything**, which is what an
	 * orchestrator crash looks like from the run directory's point of view.
	 *
	 * Distinct from `shutdown()` on purpose: `shutdown()` is the R-CTRL-19 clean path
	 * that asks workers to stop, while a crash asks nothing and leaves detached
	 * children running — which is exactly the case §10.7 reconciliation exists to
	 * handle, and the only way to exercise adoption and orphaning honestly.
	 */
	detachAll(): void {
		for (const [runId, pump] of this.pumps) {
			pump.detach();
			this.cleanupPromptFile(runId);
		}
		this.pumps.clear();
		this.queue = [];
	}

	/** R-WORK-10 result text for the orchestrator, with the full copy left on disk. */
	resultText(paths: RunPaths, raw: string): string {
		return truncateReport(raw, this.host.config.maxResultBytes, paths.result);
	}
}