import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { randomBytes } from "node:crypto";
import {
	getProcessStartIdentity,
	isProcessAlive,
	ownerToken,
	proveDeath,
	type LockOwner,
} from "../proof-of-death.ts";

/** §5.3 run directory layout, §5.4 status.json schema, R-EXEC-7 atomic writes. */

export const STATUS_SCHEMA_VERSION = 2;

/** §10.8 constants. */
export const EXIT_GRACE_MS = 5_000;
export const FINAL_STOP_GRACE_MS = 1_000;
export const SIGKILL_GRACE_MS = 2_000;
export const SHUTDOWN_BUDGET_MS = 2_000;
export const STATUS_POLL_MS = 500;
export const CONTROL_POLL_MS = 250;
export const STEER_ACK_TIMEOUT_MS = 3_000;
export const MAX_STEER_BYTES = 131_072;
export const MAX_RESULT_BYTES = 16_384;
export const MAX_EVENT_LINE_BYTES = 1_048_576;
export const MAX_EVENTS_BYTES = 52_428_800;
/** P38: rolling capture window for stdout.log, so a long child cannot exhaust memory. */
export const CAPTURE_WINDOW_BYTES = 5 * 1024 * 1024;

export type RunState =
	| "queued"
	| "spawning"
	| "running"
	| "paused"
	| "complete"
	| "stopped"
	| "timedOut"
	| "failed"
	| "orphaned"
	| "unknown";

const RUN_STATES = new Set<string>([
	"queued",
	"spawning",
	"running",
	"paused",
	"complete",
	"stopped",
	"timedOut",
	"failed",
	"orphaned",
	"unknown",
]);

/** §5.5: terminal states never transition again. `queued` is pre-spawn, not terminal. */
export const TERMINAL_STATES = new Set<RunState>(["complete", "stopped", "timedOut", "failed", "orphaned", "unknown"]);

/** Terminal worker outcomes delivered through the ordinary completion-wake path. */
export const COMPLETION_OUTCOME_STATES = new Set<RunState>(["complete", "stopped", "timedOut", "failed"]);

/**
 * States in which the run has *ended*: it has an `endedAt`, its process is going
 * away, and no live-progress write may put it back.
 *
 * `paused` is ended but deliberately **not** terminal: R-CONC-14 counts it as live
 * for concurrency, and R-TOOL-17 makes it resumable. Both facts are about
 * scheduling, not about who may rewrite the record — a pump that still holds a
 * pre-interrupt snapshot must not resurrect a `paused` run as `running`.
 */
export const ENDED_STATES = new Set<RunState>([...TERMINAL_STATES, "paused"]);

export function isTerminalState(state: RunState): boolean {
	return TERMINAL_STATES.has(state);
}

export function isCompletionOutcomeState(state: RunState): boolean {
	return COMPLETION_OUTCOME_STATES.has(state);
}

export function isEndedState(state: RunState): boolean {
	return ENDED_STATES.has(state);
}

export interface RunActiveTool {
	id: string;
	tool: string | null;
	startedAt: string;
	lastProgressAt: string;
	target: string | null;
	/** Run-directory-relative live output file for streamable tool invocations. */
	logPath?: string;
}

export function isRunActiveTool(value: unknown): value is RunActiveTool {
	if (value === null || typeof value !== "object" || Array.isArray(value)) return false;
	const record = value as Record<string, unknown>;
	return (
		typeof record.id === "string" &&
		record.id.length > 0 &&
		(record.tool === null || typeof record.tool === "string") &&
		typeof record.startedAt === "string" &&
		typeof record.lastProgressAt === "string" &&
		(record.target === null || typeof record.target === "string") &&
		(record.logPath === undefined || (typeof record.logPath === "string" && /^commands\/\d{4,}\.log$/u.test(record.logPath)))
	);
}

export interface RunActivity {
	lastEventAt: string | null;
	/**
	 * Authoritative concurrent activity for status files written by issue 9 and
	 * later. Optional so schema-1 records created by older releases remain readable.
	 */
	activeTools?: RunActiveTool[];
	currentTool: string | null;
	currentToolStartedAt: string | null;
	currentPath: string | null;
	lastAssistantPreview: string | null;
}

export interface RunCounters {
	turns: number;
	toolCalls: number;
	toolErrors: number;
	compactions: number;
}

export interface RunUsage {
	input: number;
	output: number;
	cacheWrite: number;
	costUsd: number;
}

export interface RunContextUsage {
	tokens: number;
	contextWindow: number;
	percent: number;
}

export interface RunAttention {
	reason: string;
	since: string;
	detail: string;
	/** R-SLEEP-17: durable per-run trigger dedupe across scheduler restarts. */
	fired?: string[];
}

export interface RunSteering {
	pending: number;
	delivered: number;
	failed: number;
}

/** Durable operator intent for an accepted terminal stop. */
export interface RunStopFact {
	reason: string;
	source: "orchestrator" | "user";
	requestedAt: string;
}

/** Exact inert process that proves the worker's detached PGID/SID still belongs to this run. */
export interface RunProcessGroupAnchor {
	pid: number;
	processStartIdentity: string;
	pgid: number;
	sid: number;
}

export function createStopFact(
	reason: string,
	source: RunStopFact["source"],
	requestedAt = new Date().toISOString(),
): RunStopFact {
	const normalized = reason.trim();
	if (normalized.length === 0) throw new Error("stop reason must be non-empty");
	return { reason: normalized, source, requestedAt };
}

/** Where the latest durable assistant verdict was observed. */
export interface RunAssistantEvidence {
	source: "rpc" | "session";
	/** True when stop intent or an ended-state verdict already existed. */
	observedDuringTerminalization: boolean;
}

/**
 * Bounded evidence from a tool that was already active when stop/terminalization
 * began and returned afterward. The complete payload remains in stdout/session.
 */
export interface RunLateToolResult {
	id: string;
	tool: string | null;
	target: string | null;
	isError: boolean;
	resultBytes: number;
	preview: string | null;
	truncated: boolean;
}

export const MAX_LATE_TOOL_RESULTS = 8;
export const MAX_LATE_TOOL_RESULT_IDS = 128;

export interface RunStatus {
	schemaVersion: number;
	runId: string;
	name: string;
	agent: string;
	state: RunState;
	createdAt: string;
	startedAt: string | null;
	endedAt: string | null;
	pid: number | null;
	processStartIdentity: string | null;
	/** Additive in schema 1; null on legacy/platforms without the POSIX anchor. */
	processGroupAnchor?: RunProcessGroupAnchor | null;
	/**
	 * R-CTRL-27 cross-host guard. Without this, `proveDeath` has no host to compare
	 * against and a run recorded on another machine — a shared NFS checkout, a
	 * container that rebuilt, a devcontainer bind mount — reads as dead here, because
	 * its pid is simply absent from *this* kernel. Declaring a live worker on another
	 * host dead is the one reclamation mistake that destroys work.
	 */
	hostname: string;
	cwd: string;
	sessionFile: string | null;
	sessionId: string | null;
	model: string | null;
	thinking: string | null;
	tools: string[];
	/** Legacy fields retained only when reading runs created before worker budgets were removed. */
	deadlineAt?: string | null;
	turnBudget?: { maxTurns: number; graceTurns: number; turnsUsed: number };
	activity: RunActivity;
	counters: RunCounters;
	usage: RunUsage;
	context: RunContextUsage | null;
	attention: RunAttention | null;
	steering: RunSteering;
	/** Durable settlement latch shared with the worker-side control inbox. */
	settled: boolean;
	/** Whether the most recent assistant message contained semantic final text. */
	finalAssistantTextPresent: boolean;
	/** Monotonic ordering for pump-owned assistant/provider evidence. */
	assistantMessageSequence: number;
	/** Additive schema-1 source/ordering fact for the latest assistant evidence. */
	assistantEvidence?: RunAssistantEvidence | null;
	/** Provider verdict attached to the most recent assistant message. */
	providerStopReason: string | null;
	/** Raw provider error retained for durable diagnosis, never routine model context. */
	providerError: string | null;
	/** The harness has begun its own post-settlement TERM/KILL teardown. */
	expectedTeardown: boolean;
	interrupted: boolean;
	stopped: boolean;
	/** Additive in schema 1: absent only on records written by older releases. */
	stop?: RunStopFact | null;
	timedOut: boolean;
	exitCode: number | null;
	processSignal: string | null;
	error: string | null;
	resultPath: string | null;
	resultConsumed: boolean;
	runIndex: number;
	/** Legacy pre-Issue-6 report heuristic; retained only when reading old statuses. */
	suspicious?: string;
	/** R-CTRL-31: adopted after an orchestrator restart, so stdin is gone. */
	detached?: boolean;
	/** R-SLEEP-22: a headless drain already handed this terminal transition back. */
	headlessDeliveredAt?: string;
	/** E25: oversized event lines skipped by the pump. */
	skippedEvents?: number;
	/** Set when `result.md` could not be written (full or read-only disk). */
	reportError?: string;
	/** Bounded post-stop/post-terminal completions for calls already in flight. */
	lateToolResults?: RunLateToolResult[];
	/** Larger bounded dedupe ring so eviction from the preview list cannot replay a call. */
	lateToolResultIds?: string[];
	/** Legacy pre-Issue-6 parser diagnostic retained for old-status compatibility. */
	reportDiagnostic?: string;
	readOnly: boolean;
	/** Set on a revived run (§10.5). */
	previousRunId?: string;
	/**
	 * The orchestrator session that spawned or adopted this run. `runs/` is shared
	 * by every pi session in the repo, and a terminal stop is unresumable, so bulk
	 * control (`stop_all`, toggle OFF, `session_shutdown`) needs to know whose work
	 * it is about to destroy. Optional because runs written before this field
	 * existed have none, and an unowned run is treated as *not* ours.
	 */
	owner?: {
		sessionId: string;
		pid: number;
		hostname: string;
		processStartIdentity?: string;
		claimedAt: string;
	};
}

export interface RunPaths {
	dir: string;
	status: string;
	events: string;
	trace: string;
	commands: string;
	descriptor: string;
	control: string;
	steer: string;
	steerAck: string;
	interrupt: string;
	stop: string;
	closed: string;
	steerCapability: string;
	steerHandoff: string;
	stdout: string;
	stderr: string;
	result: string;
}

export interface StatusClaimOwner extends LockOwner {
	token: string;
	statusFile: string;
}

export interface StatusClaim {
	paths: RunPaths;
	lockDir: string;
	owner: StatusClaimOwner;
}

export type StatusClaimsResult =
	| { ok: true; claims: Map<string, StatusClaim> }
	| { ok: false; reason: string };

const STATUS_CLAIM_WAIT_MS = 2_000;
const statusClaimWaitCell = new Int32Array(new SharedArrayBuffer(4));
const statusClaimProcessIdentity = getProcessStartIdentity(process.pid);

export function statusClaimKey(paths: RunPaths): string {
	return path.resolve(paths.status);
}

function statusClaimPath(paths: RunPaths): string {
	return `${statusClaimKey(paths)}.claim`;
}

function statusClaimOwnerFile(lockDir: string): string {
	return path.join(lockDir, "owner.json");
}

function readStatusClaimOwner(lockDir: string): StatusClaimOwner | undefined {
	try {
		const record = JSON.parse(fs.readFileSync(statusClaimOwnerFile(lockDir), "utf8")) as Record<string, unknown>;
		if (
			typeof record.pid !== "number" ||
			typeof record.hostname !== "string" ||
			typeof record.sessionId !== "string" ||
			typeof record.claimedAt !== "string" ||
			typeof record.token !== "string" ||
			typeof record.statusFile !== "string"
		) return undefined;
		return {
			pid: record.pid,
			hostname: record.hostname,
			sessionId: record.sessionId,
			claimedAt: record.claimedAt,
			token: record.token,
			statusFile: record.statusFile,
			...(typeof record.processStartIdentity === "string" ? { processStartIdentity: record.processStartIdentity } : {}),
		};
	} catch {
		return undefined;
	}
}

function tryStatusClaim(lockDir: string, owner: StatusClaimOwner): boolean {
	const candidate = `${lockDir}.candidate-${owner.token}`;
	try {
		fs.mkdirSync(candidate, { recursive: false, mode: 0o700 });
		fs.writeFileSync(statusClaimOwnerFile(candidate), `${JSON.stringify(owner, null, 2)}\n`, { encoding: "utf8", mode: 0o600 });
		fs.renameSync(candidate, lockDir);
		return true;
	} catch {
		try {
			fs.rmSync(candidate, { recursive: true, force: true });
		} catch {}
		return false;
	}
}

function acquireOneStatusClaim(paths: RunPaths, token: string, deadline: number): StatusClaim | { reason: string } {
	const lockDir = statusClaimPath(paths);
	fs.mkdirSync(path.dirname(lockDir), { recursive: true, mode: 0o700 });
	const owner: StatusClaimOwner = {
		pid: process.pid,
		hostname: os.hostname(),
		sessionId: `status:${process.pid}`,
		claimedAt: new Date().toISOString(),
		token,
		statusFile: statusClaimKey(paths),
		...(statusClaimProcessIdentity === undefined ? {} : { processStartIdentity: statusClaimProcessIdentity }),
	};
	for (;;) {
		if (tryStatusClaim(lockDir, owner)) return { paths, lockDir, owner };
		const holder = readStatusClaimOwner(lockDir);
		if (holder === undefined) return { reason: `${lockDir} exists but owner.json is missing or unreadable` };
		const verdict = proveDeath(holder, {
			hostname: owner.hostname,
			alive: isProcessAlive(holder.pid),
			currentIdentity: getProcessStartIdentity(holder.pid),
		});
		if (verdict.reclaimable) {
			const tombstone = `${lockDir}.stale-${ownerToken(holder)}`;
			try {
				fs.renameSync(lockDir, tombstone);
			} catch {}
			if (tryStatusClaim(lockDir, owner)) {
				// Keep the per-owner tombstone. Removing it lets a late contender that
				// observed the same stale owner rename a successor out of the live path.
				return { paths, lockDir, owner };
			}
		}
		if (Date.now() >= deadline) {
			return { reason: `status mutation claim for ${path.basename(paths.dir)} is busy: ${verdict.reason}` };
		}
		Atomics.wait(statusClaimWaitCell, 0, 0, Math.min(10, Math.max(1, deadline - Date.now())));
	}
}

/**
 * Claim every status in lexical order. All supported status writers use the same
 * per-run filesystem claim, so holding a set is a real isolation boundary rather
 * than another read-before-rename check.
 */
export function acquireStatusClaims(pathsList: RunPaths[], options: { waitMs?: number } = {}): StatusClaimsResult {
	const unique = [...new Map(pathsList.map((paths) => [statusClaimKey(paths), paths])).values()]
		.sort((left, right) => statusClaimKey(left).localeCompare(statusClaimKey(right)));
	const deadline = Date.now() + (options.waitMs ?? STATUS_CLAIM_WAIT_MS);
	const token = randomBytes(12).toString("hex");
	const claims = new Map<string, StatusClaim>();
	for (const paths of unique) {
		const claim = acquireOneStatusClaim(paths, token, deadline);
		if ("reason" in claim) {
			releaseStatusClaims(claims);
			return { ok: false, reason: claim.reason };
		}
		claims.set(statusClaimKey(paths), claim);
	}
	return { ok: true, claims };
}

function releaseStatusClaim(claim: StatusClaim): void {
	const moved = `${claim.lockDir}.released-${claim.owner.token}`;
	try {
		fs.rmSync(moved, { recursive: true, force: true });
	} catch {}
	try {
		fs.renameSync(claim.lockDir, moved);
	} catch {
		return;
	}
	const owner = readStatusClaimOwner(moved);
	if (owner === undefined || owner.token !== claim.owner.token || owner.statusFile !== claim.owner.statusFile) {
		try {
			fs.renameSync(moved, claim.lockDir);
		} catch {}
		return;
	}
	try {
		fs.rmSync(moved, { recursive: true, force: true });
	} catch {}
}

export function releaseStatusClaims(claims: Map<string, StatusClaim>): void {
	for (const claim of [...claims.values()].reverse()) releaseStatusClaim(claim);
}

function requireStatusClaim(paths: RunPaths, claim: StatusClaim): void {
	if (statusClaimKey(claim.paths) !== statusClaimKey(paths) || claim.owner.statusFile !== statusClaimKey(paths)) {
		throw new Error(`status claim does not cover ${paths.status}`);
	}
	const owner = readStatusClaimOwner(claim.lockDir);
	if (owner === undefined || owner.token !== claim.owner.token || owner.statusFile !== statusClaimKey(paths)) {
		throw new Error(`status claim for ${paths.status} is no longer held`);
	}
}

function withStatusClaim<T>(
	paths: RunPaths,
	claim: StatusClaim | undefined,
	body: (held: StatusClaim) => T,
	options: { waitMs?: number } = {},
): T {
	if (claim !== undefined) {
		requireStatusClaim(paths, claim);
		return body(claim);
	}
	const acquired = acquireStatusClaims([paths], options);
	if (!acquired.ok) throw new Error(acquired.reason);
	const held = acquired.claims.get(statusClaimKey(paths));
	if (held === undefined) {
		releaseStatusClaims(acquired.claims);
		throw new Error(`status claim for ${paths.status} was not returned`);
	}
	try {
		return body(held);
	} finally {
		releaseStatusClaims(acquired.claims);
	}
}

export function runPaths(runsDir: string, runId: string): RunPaths {
	const dir = path.join(runsDir, runId);
	const control = path.join(dir, "control");
	return {
		dir,
		status: path.join(dir, "status.json"),
		events: path.join(dir, "events.jsonl"),
		trace: path.join(dir, "trace.log"),
		commands: path.join(dir, "commands"),
		descriptor: path.join(dir, "descriptor.json"),
		control,
		steer: path.join(control, "steer"),
		steerAck: path.join(control, "steer-ack"),
		interrupt: path.join(control, "interrupt.json"),
		stop: path.join(control, "stop.json"),
		closed: path.join(control, "closed.json"),
		steerCapability: path.join(control, "steer-capability.json"),
		steerHandoff: path.join(control, "steer-handoff.json"),
		stdout: path.join(dir, "stdout.log"),
		stderr: path.join(dir, "stderr.log"),
		result: path.join(dir, "result.md"),
	};
}

/**
 * Sortable run id. Millisecond timestamp in fixed-width base36 followed by
 * randomness, so lexicographic order is chronological order for the lifetime of
 * the scheme and two runs created in the same millisecond still differ.
 */
export function newRunId(now = Date.now(), random = randomBytes(5)): string {
	return `r_${now.toString(36).padStart(9, "0").toUpperCase()}${random.toString("hex").toUpperCase()}`;
}

const RUN_ID_PATTERN = /^r_[0-9A-Z]{9,40}$/;

/** R-SEC-7: a run id becomes a path segment, so it is validated before use. */
export function validateRunId(raw: string): string {
	const runId = raw.trim();
	if (!RUN_ID_PATTERN.test(runId) || runId.includes("..") || runId.includes("/") || runId.includes("\\")) {
		throw new Error(`Invalid run id '${raw}'. Expected r_ followed by uppercase base36 characters.`);
	}
	return runId;
}

const LEGACY_REGISTRY_RUN_ID_PATTERN = /^r_[A-Z0-9_]{1,40}$/;

/**
 * Registry directories may contain bounded legacy test/early-release IDs, but never
 * control characters, prose, path separators, or unbounded text. New IDs still use
 * validateRunId's strict sortable format.
 */
export function validateRegistryRunId(raw: string): string {
	try {
		return validateRunId(raw);
	} catch (error) {
		if (LEGACY_REGISTRY_RUN_ID_PATTERN.test(raw)) return raw;
		throw error;
	}
}

/** R-EXEC-7: temp-then-rename at mode 0600. */
export function writeJsonAtomic(file: string, value: unknown): void {
	fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });
	const tmp = `${file}.tmp-${randomBytes(8).toString("hex")}`;
	try {
		const fd = fs.openSync(tmp, "wx", 0o600);
		try {
			fs.writeFileSync(fd, `${JSON.stringify(value, null, 2)}\n`, "utf8");
		} finally {
			fs.closeSync(fd);
		}
		fs.renameSync(tmp, file);
	} catch (error) {
		try {
			fs.unlinkSync(tmp);
		} catch {
			// A leftover temp file is inert.
		}
		throw error;
	}
}

function latchSettlement(current: RunStatus, incoming: RunStatus): RunStatus {
	const next = { ...incoming };
	next.processGroupAnchor = current.processGroupAnchor ?? incoming.processGroupAnchor ?? null;
	// Stop intent is an operator fact, not pump telemetry. The first durable intent
	// wins and stale whole-status snapshots may never erase or replace it.
	next.stop = current.stop ?? (isTerminalState(current.state) && current.state !== "stopped" ? null : incoming.stop ?? null);
	if (current.settled === true) next.settled = true;
	// These fields were added without a schema-version bump and are pump-owned. A
	// monotonic sequence lets the latest assistant message replace true with false
	// while preventing a stale whole-status snapshot from erasing newer evidence.
	const currentAssistantSequence = current.assistantMessageSequence ?? 0;
	const incomingAssistantSequence = incoming.assistantMessageSequence;
	if (incomingAssistantSequence === undefined || currentAssistantSequence >= incomingAssistantSequence) {
		next.finalAssistantTextPresent = current.finalAssistantTextPresent;
		next.assistantMessageSequence = currentAssistantSequence;
		next.assistantEvidence = current.assistantEvidence ?? incoming.assistantEvidence ?? null;
		next.providerStopReason = current.providerStopReason;
		next.providerError = current.providerError;
	}
	const lateToolResults = new Map<string, RunLateToolResult>();
	for (const result of current.lateToolResults ?? []) lateToolResults.set(result.id, result);
	for (const result of incoming.lateToolResults ?? []) {
		if (!lateToolResults.has(result.id)) lateToolResults.set(result.id, result);
	}
	next.lateToolResults = [...lateToolResults.values()].slice(-MAX_LATE_TOOL_RESULTS);
	const lateToolResultIds = new Set<string>();
	for (const id of current.lateToolResultIds ?? []) lateToolResultIds.add(id);
	for (const result of current.lateToolResults ?? []) lateToolResultIds.add(result.id);
	for (const id of incoming.lateToolResultIds ?? []) lateToolResultIds.add(id);
	for (const result of incoming.lateToolResults ?? []) lateToolResultIds.add(result.id);
	next.lateToolResultIds = [...lateToolResultIds].slice(-MAX_LATE_TOOL_RESULT_IDS);
	next.expectedTeardown = current.expectedTeardown === true || incoming.expectedTeardown === true;
	if (isTerminalState(current.state)) {
		// Terminal verdict and operator/process facts are immutable, but terminal is
		// not synonymous with evidence-complete. A final assistant message can race a
		// stop, so a later snapshot may enrich report/counter/usage fields without
		// turning the stopped run into complete or changing when/how it ended.
		next.state = current.state;
		next.endedAt = current.endedAt;
		next.stopped = current.stopped;
		next.interrupted = current.interrupted;
		next.timedOut = current.timedOut;
		// A stop can become terminal before the child exits. Exit facts therefore
		// latch from unknown to known exactly once, then become immutable.
		next.exitCode = current.exitCode ?? incoming.exitCode;
		next.processSignal = current.processSignal ?? incoming.processSignal;
		next.error = current.error;
		next.resultPath = current.resultPath ?? incoming.resultPath;
		next.settled = current.settled || incoming.settled;
		next.activity = {
			...incoming.activity,
			lastEventAt: current.activity.lastEventAt,
			activeTools: current.activity.activeTools,
			currentTool: current.activity.currentTool,
			currentToolStartedAt: current.activity.currentToolStartedAt,
			currentPath: current.activity.currentPath,
			lastAssistantPreview: incoming.activity.lastAssistantPreview ?? current.activity.lastAssistantPreview,
		};
		next.counters = {
			turns: Math.max(current.counters.turns, incoming.counters.turns),
			toolCalls: Math.max(current.counters.toolCalls, incoming.counters.toolCalls),
			toolErrors: Math.max(current.counters.toolErrors, incoming.counters.toolErrors),
			compactions: Math.max(current.counters.compactions, incoming.counters.compactions),
		};
		next.usage = {
			input: Math.max(current.usage.input, incoming.usage.input),
			output: Math.max(current.usage.output, incoming.usage.output),
			cacheWrite: Math.max(current.usage.cacheWrite, incoming.usage.cacheWrite),
			costUsd: Math.max(current.usage.costUsd, incoming.usage.costUsd),
		};
		if (current.context !== null && (incoming.context === null || current.context.tokens > incoming.context.tokens)) {
			next.context = current.context;
		}
		if (current.reportError !== undefined) next.reportError = current.reportError;
		if (current.reportDiagnostic !== undefined) next.reportDiagnostic = current.reportDiagnostic;
		if (current.suspicious !== undefined) next.suspicious = current.suspicious;
		if (current.skippedEvents !== undefined) {
			next.skippedEvents = Math.max(current.skippedEvents, incoming.skippedEvents ?? 0);
		}
	} else if (isEndedState(current.state) && !isEndedState(incoming.state)) {
		// The run has ended (`paused` via the worker's own interrupt handling) and the
		// incoming record still describes it as live. That is a stale snapshot, not a
		// transition: keeping `incoming` would erase the interrupt and leave the
		// impossible pair `state: "running", interrupted: true` on disk, which
		// `assertResumable` then refuses to resume. A genuine later transition out of
		// `paused` — a stop, or a settled report — is itself an ended state and takes
		// the branch above instead.
		next.state = current.state;
		next.endedAt = current.endedAt;
		next.stopped = current.stopped;
		next.interrupted = current.interrupted;
		next.activity = {
			...current.activity,
			activeTools: current.activity.activeTools?.map((invocation) => ({ ...invocation })),
		};
	}
	return next;
}

function mergeStatusForWrite(current: RunStatus | undefined, incoming: RunStatus): RunStatus {
	if (current === undefined) return incoming;
	const next = latchSettlement(current, incoming);

	if (current.detached === true) next.detached = true;
	if (current.resultConsumed === true) next.resultConsumed = true;
	if (current.attention !== null) next.attention = current.attention;
	if (current.headlessDeliveredAt !== undefined) next.headlessDeliveredAt = current.headlessDeliveredAt;
	if (current.stopped) next.stopped = true;
	if (current.interrupted) next.interrupted = true;
	// Ownership is set once at spawn and re-stamped only by adoption, which writes
	// through `patchStatus`. A pump snapshot that predates it must not drop it.
	if (next.owner === undefined && current.owner !== undefined) next.owner = current.owner;
	return next;
}

/**
 * status.json has multiple writers. Re-read immediately before every whole-file
 * replacement so scheduler/tool-owned fields and a latched settlement survive a
 * stale pump or supervisor snapshot.
 */
export function writeStatus(paths: RunPaths, status: RunStatus, claim?: StatusClaim, options: { waitMs?: number } = {}): void {
	withStatusClaim(paths, claim, () => {
		let current: RunStatus | undefined;
		try {
			current = readStatus(paths).status;
		} catch {
			current = undefined;
		}
		writeJsonAtomic(paths.status, mergeStatusForWrite(current, status));
	}, options);
}

/** Apply one status mutation while holding the run's filesystem claim. */
export function patchStatus(
	paths: RunPaths,
	patch: (current: RunStatus) => RunStatus,
	claim?: StatusClaim,
	options: { waitMs?: number } = {},
): RunStatus | undefined {
	return withStatusClaim(paths, claim, () => {
		const current = readStatus(paths).status;
		if (current === undefined) return undefined;
		const next = patch(current);
		const latched = latchSettlement(current, next);
		writeJsonAtomic(paths.status, latched);
		return latched;
	}, options);
}

/** Re-read and mutate only when the callback accepts the claimed durable state. */
export function patchStatusIf(
	paths: RunPaths,
	patch: (current: RunStatus) => RunStatus | undefined,
	claim?: StatusClaim,
	options: { waitMs?: number } = {},
): { status: RunStatus | undefined; changed: boolean } {
	return withStatusClaim(paths, claim, () => {
		const current = readStatus(paths).status;
		if (current === undefined) return { status: undefined, changed: false };
		const next = patch(current);
		if (next === undefined) return { status: current, changed: false };
		const latched = latchSettlement(current, next);
		writeJsonAtomic(paths.status, latched);
		return { status: latched, changed: true };
	}, options);
}

export interface ReadStatusResult {
	status?: RunStatus;
	/** Present when the file exists but cannot be used (R-EXEC-9). */
	problem?: string;
}

/**
 * R-EXEC-9: an unknown `schemaVersion` yields `unknown`, never a guess. A
 * missing file is not a problem, it is an absent run.
 */
export function readStatus(paths: RunPaths): ReadStatusResult {
	let raw: string;
	try {
		raw = fs.readFileSync(paths.status, "utf8");
	} catch (error) {
		if ((error as NodeJS.ErrnoException).code === "ENOENT") return {};
		return { problem: `status.json unreadable: ${(error as Error).message}` };
	}
	let parsed: unknown;
	try {
		parsed = JSON.parse(raw);
	} catch (error) {
		return { problem: `status.json is not valid JSON: ${(error as Error).message}` };
	}
	if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
		return { problem: "status.json is not a JSON object" };
	}
	const record = parsed as Record<string, unknown>;
	if (record.schemaVersion !== STATUS_SCHEMA_VERSION) {
		return { problem: `status.json has unsupported schemaVersion ${String(record.schemaVersion)}` };
	}
	const state = typeof record.state === "string" && RUN_STATES.has(record.state) ? (record.state as RunState) : undefined;
	if (state === undefined) {
		return { problem: `status.json has unknown state '${String(record.state)}'` };
	}
	if (typeof record.runId !== "string" || typeof record.name !== "string") {
		return { problem: "status.json is missing runId or name" };
	}
	return { status: { ...(record as unknown as RunStatus), state } };
}

export function createStatus(fields: {
	runId: string;
	name: string;
	agent: string;
	state: RunState;
	cwd: string;
	model: string | null;
	thinking: string | null;
	tools: string[];
	/** Legacy fixture/import compatibility; new workers omit both fields. */
	deadlineAt?: string | null;
	turnBudget?: { maxTurns: number; graceTurns: number; turnsUsed: number };
	readOnly: boolean;
	sessionId: string | null;
	runIndex?: number;
	previousRunId?: string;
	owner?: RunStatus["owner"];
	now?: Date;
}): RunStatus {
	const now = (fields.now ?? new Date()).toISOString();
	return {
		schemaVersion: STATUS_SCHEMA_VERSION,
		runId: fields.runId,
		name: fields.name,
		agent: fields.agent,
		state: fields.state,
		createdAt: now,
		startedAt: null,
		endedAt: null,
		pid: null,
		processStartIdentity: null,
		processGroupAnchor: null,
		hostname: os.hostname(),
		cwd: fields.cwd,
		sessionFile: null,
		sessionId: fields.sessionId,
		model: fields.model,
		thinking: fields.thinking,
		tools: fields.tools,
		...(fields.deadlineAt === undefined ? {} : { deadlineAt: fields.deadlineAt }),
		...(fields.turnBudget === undefined ? {} : { turnBudget: fields.turnBudget }),
		activity: {
			lastEventAt: null,
			activeTools: [],
			currentTool: null,
			currentToolStartedAt: null,
			currentPath: null,
			lastAssistantPreview: null,
		},
		counters: { turns: 0, toolCalls: 0, toolErrors: 0, compactions: 0 },
		usage: { input: 0, output: 0, cacheWrite: 0, costUsd: 0 },
		context: null,
		attention: null,
		steering: { pending: 0, delivered: 0, failed: 0 },
		settled: false,
		finalAssistantTextPresent: false,
		assistantMessageSequence: 0,
		assistantEvidence: null,
		providerStopReason: null,
		providerError: null,
		expectedTeardown: false,
		interrupted: false,
		stopped: false,
		stop: null,
		timedOut: false,
		exitCode: null,
		processSignal: null,
		error: null,
		resultPath: null,
		resultConsumed: false,
		lateToolResults: [],
		lateToolResultIds: [],
		runIndex: fields.runIndex ?? 1,
		readOnly: fields.readOnly,
		...(fields.previousRunId === undefined ? {} : { previousRunId: fields.previousRunId }),
		...(fields.owner === undefined ? {} : { owner: fields.owner }),
	};
}

export interface RunEvent {
	ts: string;
	kind: string;
	[key: string]: unknown;
}

/**
 * R-EXEC-7: append-only, one JSON object per line. E25: the file rotates rather
 * than growing without bound, because unbounded capture of a long-running child
 * is a disk-exhaustion vector.
 */
export function appendEvent(paths: RunPaths, event: RunEvent): void {
	const line = `${JSON.stringify(event)}\n`;
	try {
		fs.mkdirSync(paths.dir, { recursive: true, mode: 0o700 });
		let size = 0;
		try {
			size = fs.statSync(paths.events).size;
		} catch {
			size = 0;
		}
		if (size + line.length > MAX_EVENTS_BYTES) {
			try {
				fs.renameSync(paths.events, `${paths.events}.1`);
			} catch {
				// Rotation is best effort; on failure the append below still runs.
			}
		}
		fs.appendFileSync(paths.events, line, { encoding: "utf8", mode: 0o600 });
	} catch {
		// E37: a failed event write must never crash the orchestrator.
	}
}

/**
 * R-EXEC-7: readers tolerate a truncated final line, which happens when a write
 * is in flight. The partial line is discarded rather than reported as corrupt.
 */
export function readEvents(paths: RunPaths, tailLines?: number): RunEvent[] {
	let raw: string;
	try {
		raw = fs.readFileSync(paths.events, "utf8");
	} catch {
		return [];
	}
	const lines = raw.split("\n");
	// R-EXEC-7: the last element is either the empty string after a final "\n" or a
	// record still being written. Either way it is discarded.
	lines.pop();
	const selected = tailLines === undefined ? lines : lines.slice(-tailLines);
	const events: RunEvent[] = [];
	for (const line of selected) {
		if (line.trim().length === 0) continue;
		try {
			const parsed = JSON.parse(line) as unknown;
			if (parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)) {
				events.push(parsed as RunEvent);
			}
		} catch {
			// A single unparseable line never fails the read (E24).
		}
	}
	return events;
}

export function readFileIfExists(file: string): string | undefined {
	try {
		return fs.readFileSync(file, "utf8");
	} catch (error) {
		if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined;
		throw error;
	}
}

export interface ClosedRunRecord {
	state: RunState;
	closedAt: string;
	stop?: RunStopFact | null;
	endedAt?: string | null;
	resultPath?: string | null;
	exitCode?: number | null;
	processSignal?: string | null;
}

/**
 * R-CTRL-13: terminal runs stop accepting control. New tombstones mirror the
 * durable terminal facts needed after a stop marker is consumed. The state-only
 * form remains for compatibility with callers that have no readable status.
 */
export function writeClosed(paths: RunPaths, statusOrState: RunStatus | RunState): void {
	try {
		const status = typeof statusOrState === "string" ? undefined : statusOrState;
		const incoming: ClosedRunRecord = {
			state: typeof statusOrState === "string" ? statusOrState : statusOrState.state,
			closedAt: new Date().toISOString(),
			...(status === undefined
				? {}
				: {
						stop: status.stop ?? null,
						endedAt: status.endedAt,
						resultPath: status.resultPath,
						exitCode: status.exitCode,
						processSignal: status.processSignal,
					}),
		};
		let current: Partial<ClosedRunRecord> | undefined;
		try {
			const parsed = JSON.parse(fs.readFileSync(paths.closed, "utf8")) as unknown;
			if (parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)) current = parsed as Partial<ClosedRunRecord>;
		} catch {}
		const currentState = typeof current?.state === "string" && RUN_STATES.has(current.state) ? current.state : undefined;
		const terminalVerdictLatched = currentState !== undefined && isTerminalState(currentState);
		const merged: ClosedRunRecord = terminalVerdictLatched
			? {
					...incoming,
					state: currentState,
					closedAt: current?.closedAt ?? incoming.closedAt,
					stop: current?.stop ?? (currentState === "stopped" ? incoming.stop ?? null : null),
					endedAt: current?.endedAt ?? incoming.endedAt ?? null,
					resultPath: current?.resultPath ?? incoming.resultPath ?? null,
					exitCode: current?.exitCode ?? incoming.exitCode ?? null,
					processSignal: current?.processSignal ?? incoming.processSignal ?? null,
				}
			: incoming;
		writeJsonAtomic(paths.closed, merged);
	} catch {
		// Best effort: a missing tombstone only costs a steer the ack timeout.
	}
}

export function ensureRunDirs(paths: RunPaths): void {
	fs.mkdirSync(paths.commands, { recursive: true, mode: 0o700 });
	fs.mkdirSync(paths.steer, { recursive: true, mode: 0o700 });
	fs.mkdirSync(paths.steerAck, { recursive: true, mode: 0o700 });
}

export function listRunIds(runsDir: string): string[] {
	try {
		return fs
			.readdirSync(runsDir, { withFileTypes: true })
			.filter((entry) => entry.isDirectory() && entry.name.startsWith("r_"))
			.map((entry) => entry.name)
			.sort();
	} catch (error) {
		if ((error as NodeJS.ErrnoException).code === "ENOENT") return [];
		throw error;
	}
}