/**
 * RPC event pump: child stdout JSONL -> status.json + events.jsonl, and the
 * mandatory `extension_ui_request` answering (R-EXEC-2).
 *
 * Why the pipes are driven directly rather than through pi's exported `RpcClient`:
 *
 * - `RpcClient.start()` hardcodes `spawn("node", [cliPath, ...])` with no
 *   `detached`, so it owns the child's lifetime. R-EXEC-5 requires a detached
 *   process group and `unref()`, and R-CTRL-31 requires attaching to a child this
 *   process never spawned. Neither is expressible through `start()`.
 * - `RpcClient` has no `extension_ui_request` path at all: `handleLine` routes
 *   `response` records to pending requests and everything else to event
 *   listeners, and it never writes an `extension_ui_response`. R-EXEC-2 — the rule
 *   that keeps the child from hanging forever — cannot be implemented on top of it.
 * - `RpcClient.stop()` unconditionally SIGTERMs, which is the opposite of
 *   R-CTRL-19's bounded best-effort shutdown that deliberately leaves detached
 *   workers running.
 *
 * The protocol *types* are imported from pi rather than restated, so this pump
 * cannot drift from the wire format.
 */

import type { ChildProcess } from "node:child_process";
import { createHash } from "node:crypto";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { StringDecoder } from "node:string_decoder";
import type { RpcCommand, RpcExtensionUIRequest, RpcExtensionUIResponse, RpcResponse } from "@earendil-works/pi-coding-agent";
import { parseSteerCapability } from "../control/request.ts";
import { getProcessStartIdentity, isProcessAlive } from "../proof-of-death.ts";
import {
	CAPTURE_WINDOW_BYTES,
	EXIT_GRACE_MS,
	FINAL_STOP_GRACE_MS,
	MAX_EVENT_LINE_BYTES,
	MAX_LATE_TOOL_RESULT_IDS,
	MAX_LATE_TOOL_RESULTS,
	type RunPaths,
	type RunActiveTool,
	type RunProcessGroupAnchor,
	type RunLateToolResult,
	type RunState,
	type RunStatus,
	type RunStopFact,
	SIGKILL_GRACE_MS,
	appendEvent,
	isEndedState,
	isRunActiveTool,
	isTerminalState,
	readStatus,
	writeClosed,
	writeStatus,
} from "./status.ts";
import { hasSemanticCompletion } from "./report.ts";
import { salvageFinalAssistantMessage } from "./session-salvage.ts";
import { validateSignalTarget } from "./ownership.ts";
import { readProcessGroupIdentity, signalProcessGroup, signalProcessTree } from "./spawn.ts";
import { readDescriptor, writeDescriptor } from "./spawn.ts";
import { sanitizeInspectionField } from "./trajectory.ts";
import { appendWorkerTrace, readWorkerTraceSequence } from "./trace.ts";

/**
 * Ground truth from `modes/rpc/rpc-mode.ts`: every one of these registers in
 * `pendingExtensionRequests` and returns a Promise, and `createDialogPromise` only
 * arms a timer when `opts.timeout` is set (`:115`) — `editor` (`:255`) has no timer
 * at all. Unanswered, the child blocks forever. Verified by running one: a child
 * emitting `confirm` with no answer was still alive after 25s with no further
 * output; the same child answered `{cancelled:true}` settled in 300ms.
 */
const BLOCKING_UI_METHODS = new Set<string>(["select", "confirm", "input", "editor"]);

/**
 * Also from `rpc-mode.ts`: these return `void` and carry a fresh `crypto.randomUUID`
 * purely as a correlation courtesy. They are recorded as worker activity
 * (R-EXEC-2) and must NOT be answered — nothing is waiting, and the id is not in
 * `pendingExtensionRequests`.
 */
const FIRE_AND_FORGET_UI_METHODS = new Set<string>(["notify", "setStatus", "setWidget", "setTitle", "set_editor_text"]);
const teardownWaitCell = new Int32Array(new SharedArrayBuffer(4));
const STDIO_DRAIN_GRACE_MS = 1_000;
const MAX_LATE_TOOL_PREVIEW_BYTES = 4_096;
const MAX_EVENT_ASSISTANT_PREVIEW_BYTES = 2_048;
const MAX_EVENT_TOOL_RESULT_PREVIEW_BYTES = 2_048;

export interface PumpCallbacks {
	/** Called after every status write, so the fleet widget can refresh. */
	onStatus?: (status: RunStatus) => void;
	/** Called once when the run reaches a terminal state. */
	onTerminal?: (status: RunStatus) => void;
	/** Called when semantic evidence changes after the terminal wake was emitted. */
	onEvidence?: (status: RunStatus) => void;
}

interface WireEvent {
	type: string;
	[key: string]: unknown;
}

interface SteeringHandoff {
	pid: number;
	reqId: string;
	ts: string;
}

interface CommandLogState {
	sourcePath?: string;
	sourceOffset: number;
	finalized: boolean;
	contentBytes?: number;
}

type CommandLogOutcome = "completed" | "failed" | "timed out" | "interrupted" | "stopped";

function textOf(message: unknown): string {
	if (message === null || typeof message !== "object") return "";
	const content = (message as { content?: unknown }).content;
	if (!Array.isArray(content)) return "";
	return content
		.filter((block): block is { type: string; text: string } => {
			return block !== null && typeof block === "object" && (block as { type?: unknown }).type === "text" && typeof (block as { text?: unknown }).text === "string";
		})
		.map((block) => block.text)
		.join("");
}

function canonicalJson(value: unknown): unknown {
	if (Array.isArray(value)) return value.map(canonicalJson);
	if (value === null || typeof value !== "object") return value;
	const record = value as Record<string, unknown>;
	return Object.fromEntries(Object.keys(record).sort().map((key) => [key, canonicalJson(record[key])]));
}

function assistantEvidenceIdentity(message: unknown): string {
	// Providers can reuse response ids across turns, and timestamps are not unique at
	// millisecond resolution. Hash the complete normalized assistant message so only
	// an exact semantic replay is ignored; changed content or usage remains evidence.
	const normalized = JSON.stringify(canonicalJson(message));
	return `sha256:${createHash("sha256").update(normalized).digest("hex")}`;
}

function pathOfArgs(args: unknown): string | null {
	if (args === null || typeof args !== "object") return null;
	const record = args as Record<string, unknown>;
	for (const key of ["path", "file", "filePath", "file_path"]) {
		const value = record[key];
		if (typeof value === "string" && value.length > 0) return value;
	}
	const command = record.command;
	if (typeof command === "string") return command.slice(0, 120);
	return null;
}

function serializedToolResult(result: unknown): string {
	try {
		return result === undefined ? "" : (JSON.stringify(result) ?? "");
	} catch {
		return "";
	}
}

function toolResultText(result: unknown): string {
	if (result !== null && typeof result === "object") {
		const content = (result as { content?: unknown }).content;
		if (Array.isArray(content)) {
			const text = content
				.filter((block): block is { type: string; text: string } =>
					block !== null && typeof block === "object" && (block as { type?: unknown }).type === "text" && typeof (block as { text?: unknown }).text === "string",
				)
				.map((block) => block.text)
				.join("\n")
				.trim();
			if (text.length > 0) return text;
		}
	}
	return serializedToolResult(result);
}

function toolOutputText(result: unknown): string {
	if (result === null || typeof result !== "object") return "";
	const content = (result as { content?: unknown }).content;
	if (typeof content === "string") return content;
	if (!Array.isArray(content)) return "";
	return content
		.filter((block): block is { type: string; text: string } =>
			block !== null && typeof block === "object" && (block as { type?: unknown }).type === "text" && typeof (block as { text?: unknown }).text === "string",
		)
		.map((block) => block.text)
		.join("\n");
}

function toolFullOutputPath(result: unknown): string | undefined {
	if (result === null || typeof result !== "object") return undefined;
	const details = (result as { details?: unknown }).details;
	if (details === null || typeof details !== "object") return undefined;
	const value = (details as { fullOutputPath?: unknown }).fullOutputPath;
	return typeof value === "string" && value.length > 0 ? value : undefined;
}

function commandDuration(startedAt: string, endedAt = Date.now()): string {
	const start = Date.parse(startedAt);
	const total = Math.max(0, Math.floor((endedAt - (Number.isNaN(start) ? endedAt : start)) / 1000));
	if (total < 60) return `${total}s`;
	const minutes = Math.floor(total / 60);
	const seconds = total % 60;
	if (minutes < 60) return seconds === 0 ? `${minutes}m` : `${minutes}m${seconds}s`;
	const hours = Math.floor(minutes / 60);
	return minutes % 60 === 0 ? `${hours}h` : `${hours}h${minutes % 60}m`;
}

function replaceFileAtomic(file: string, content: string | Buffer): void {
	const temp = `${file}.tmp-${process.pid}-${Date.now()}`;
	try {
		fs.writeFileSync(temp, content, { mode: 0o600 });
		fs.renameSync(temp, file);
	} finally {
		try { fs.unlinkSync(temp); } catch {}
	}
}

function replaceFileFromPrefix(file: string, source: string, length: number): number {
	const temp = `${file}.tmp-${process.pid}-${Date.now()}`;
	let copied = 0;
	try {
		const sourceFd = fs.openSync(source, "r");
		const targetFd = fs.openSync(temp, "w", 0o600);
		try {
			const buffer = Buffer.allocUnsafe(64 * 1024);
			while (copied < length) {
				const read = fs.readSync(sourceFd, buffer, 0, Math.min(buffer.length, length - copied), copied);
				if (read <= 0) break;
				fs.writeSync(targetFd, buffer, 0, read);
				copied += read;
			}
		} finally {
			fs.closeSync(sourceFd);
			fs.closeSync(targetFd);
		}
		fs.renameSync(temp, file);
		return copied;
	} finally {
		try { fs.unlinkSync(temp); } catch {}
	}
}

function utf8Prefix(text: string, maxBytes: number): string {
	if (Buffer.byteLength(text, "utf8") <= maxBytes) return text;
	let bytes = 0;
	let result = "";
	for (const point of text) {
		const size = Buffer.byteLength(point, "utf8");
		if (bytes + size > maxBytes) break;
		bytes += size;
		result += point;
	}
	return result;
}

function monotonicTimestamp(previous: string, candidate: string): string {
	const previousAt = Date.parse(previous);
	const candidateAt = Date.parse(candidate);
	if (Number.isNaN(previousAt)) return candidate;
	if (Number.isNaN(candidateAt) || candidateAt < previousAt) return previous;
	return candidate;
}

/**
 * LF-only JSONL framing, matching pi's own `modes/rpc/jsonl.ts` (which is internal
 * and not exported from the package). Node readline is deliberately not used: it
 * splits on U+2028/U+2029, which are legal inside a JSON string, so it does not
 * implement strict JSONL.
 *
 * Chunks are decoded through a `StringDecoder` rather than `chunk.toString("utf8")`:
 * a stream boundary can fall inside a multi-byte character, and decoding each chunk
 * independently turns that character into U+FFFD in whatever event carries it
 * (worker text ends up in `result.md`, so the corruption is durable). The decoder
 * holds the incomplete tail until its remaining bytes arrive.
 */
export function createLineSplitter(onLine: (line: string) => void): (chunk: Buffer | string) => void {
	const decoder = new StringDecoder("utf8");
	let buffer = "";
	return (chunk) => {
		buffer += typeof chunk === "string" ? chunk : decoder.write(chunk);
		while (true) {
			const index = buffer.indexOf("\n");
			if (index === -1) {
				// E25: a child emitting one enormous line must not grow the buffer without
				// bound. The oversized prefix is dropped and the pump resynchronizes at the
				// next newline. The cap is a byte budget, so it is measured in bytes —
				// `buffer.length` counts UTF-16 code units and would admit 2-4x the limit.
				if (Buffer.byteLength(buffer, "utf8") > MAX_EVENT_LINE_BYTES) buffer = "";
				return;
			}
			const line = buffer.slice(0, index);
			buffer = buffer.slice(index + 1);
			onLine(line.endsWith("\r") ? line.slice(0, -1) : line);
		}
	};
}

/**
 * Drives one worker run. Owns `status.json` and `events.jsonl` for that run and
 * nothing else; there is exactly one pump per live run in this process.
 */
export class WorkerPump {
	readonly paths: RunPaths;
	private status: RunStatus;
	private child: ChildProcess | undefined;
	private callbacks: PumpCallbacks;
	private settled = false;
	private lastAssistantText: string | null = null;
	private lastAssistantStopReason: string | null = null;
	private lastAssistantError: string | null = null;
	private sawAnyEvent = false;
	private stdoutBytes = 0;
	private stderrBytes = 0;
	private skipped = 0;
	private exitTimer: ReturnType<typeof setTimeout> | undefined;
	private killTimer: ReturnType<typeof setTimeout> | undefined;
	private controlExitTimer: ReturnType<typeof setTimeout> | undefined;
	private finalStopTimer: ReturnType<typeof setTimeout> | undefined;
	private drainTimer: ReturnType<typeof setTimeout> | undefined;
	/** Once identity proof fails, this pump never attempts another process signal. */
	private signalsRejected = false;
	private closed = false;
	/** Terminal callbacks are externally consequential (they arm durable wakes). */
	private terminalEmitted = false;
	/** Child `exit` may precede the final stdout data; semantic classification waits for `close`. */
	private processExitObserved = false;
	/** `close` or the bounded drain fallback has made the semantic verdict once. */
	private drainClassified = false;
	/** Exact assistant text already handed to onReport; prevents duplicate rewrites. */
	private reportedAssistantText: string | null = null;
	/** Stable RPC message identities already accounted into text/usage evidence. */
	private assistantEvidenceSeen = new Set<string>();
	private onReport: (status: RunStatus, finalText: string | null) => void;
	/**
	 * Pi can execute several tool calls concurrently in one assistant turn. Keep
	 * the complete live set in memory while projecting one invocation through the
	 * backward-compatible status.activity fields.
	 */
	private activeTools = new Map<string, RunActiveTool>();
	private anonymousToolSequence = 0;
	private commandLogSequence = 0;
	private commandLogs = new Map<string, CommandLogState>();
	/** Monotonic cursor for the concise main-agent-facing trace. */
	private traceSequence: number;

	constructor(options: {
		paths: RunPaths;
		status: RunStatus;
		callbacks?: PumpCallbacks;
		onReport: (status: RunStatus, finalText: string | null) => void;
	}) {
		this.paths = options.paths;
		this.status = options.status;
		this.callbacks = options.callbacks ?? {};
		this.onReport = options.onReport;
		this.traceSequence = readWorkerTraceSequence(options.paths);
		for (const invocation of options.status.activity.activeTools ?? []) {
			if (!isRunActiveTool(invocation)) continue;
			this.activeTools.set(invocation.id, { ...invocation });
			if (invocation.logPath !== undefined) {
				const match = /^commands\/(\d+)\.log$/u.exec(invocation.logPath);
				if (match !== null) this.commandLogSequence = Math.max(this.commandLogSequence, Number(match[1]));
				this.commandLogs.set(invocation.id, { sourceOffset: 0, finalized: false });
			}
			const anonymous = /^anonymous-(\d+)$/.exec(invocation.id);
			if (anonymous !== null) {
				const sequence = Number(anonymous[1]);
				if (Number.isSafeInteger(sequence)) this.anonymousToolSequence = Math.max(this.anonymousToolSequence, sequence);
			}
		}
		if (options.status.activity.activeTools !== undefined) this.projectActiveTool();
		this.seedCommandLogSequence();
	}

	getStatus(): RunStatus {
		return this.status;
	}

	getPaths(): RunPaths {
		return this.paths;
	}

	private seedCommandLogSequence(): void {
		try {
			for (const name of fs.readdirSync(this.paths.commands)) {
				const match = /^(\d+)\.log$/u.exec(name);
				if (match !== null) this.commandLogSequence = Math.max(this.commandLogSequence, Number(match[1]));
			}
		} catch {}
	}

	private allocateCommandLog(id: string): string {
		let name: string;
		do name = `${String(++this.commandLogSequence).padStart(4, "0")}.log`;
		while (fs.existsSync(path.join(this.paths.commands, name)));
		const relative = `commands/${name}`;
		try {
			fs.mkdirSync(this.paths.commands, { recursive: true, mode: 0o700 });
			fs.writeFileSync(path.join(this.paths.dir, relative), "", { encoding: "utf8", mode: 0o600 });
			this.commandLogs.set(id, { sourceOffset: 0, finalized: false });
		} catch {
			return "";
		}
		return relative;
	}

	private commandLogFile(active: RunActiveTool): string | undefined {
		return active.logPath === undefined ? undefined : path.join(this.paths.dir, active.logPath);
	}

	private mirrorCommandSource(active: RunActiveTool, sourcePath: string): boolean {
		const file = this.commandLogFile(active);
		if (file === undefined) return false;
		const state = this.commandLogs.get(active.id) ?? { sourceOffset: 0, finalized: false };
		try {
			const stat = fs.statSync(sourcePath);
			if (!stat.isFile()) return false;
			if (state.sourcePath !== sourcePath) {
				state.sourcePath = sourcePath;
				state.sourceOffset = replaceFileFromPrefix(file, sourcePath, stat.size);
				state.contentBytes = state.sourceOffset;
				this.commandLogs.set(active.id, state);
				return true;
			}
			if (stat.size < state.sourceOffset) {
				state.sourceOffset = replaceFileFromPrefix(file, sourcePath, stat.size);
				state.contentBytes = state.sourceOffset;
				this.commandLogs.set(active.id, state);
				return true;
			}
			if (stat.size > state.sourceOffset) {
				const descriptor = fs.openSync(sourcePath, "r");
				try {
					const buffer = Buffer.allocUnsafe(64 * 1024);
					while (state.sourceOffset < stat.size) {
						const length = Math.min(buffer.length, stat.size - state.sourceOffset);
						const read = fs.readSync(descriptor, buffer, 0, length, state.sourceOffset);
						if (read <= 0) break;
						fs.appendFileSync(file, buffer.subarray(0, read), { mode: 0o600 });
						state.sourceOffset += read;
						state.contentBytes = state.sourceOffset;
					}
				} finally {
					fs.closeSync(descriptor);
				}
			}
			this.commandLogs.set(active.id, state);
			return true;
		} catch {
			return false;
		}
	}

	private persistCommandUpdate(active: RunActiveTool, result: unknown): void {
		const file = this.commandLogFile(active);
		if (file === undefined) return;
		const sourcePath = toolFullOutputPath(result);
		if (sourcePath !== undefined && this.mirrorCommandSource(active, sourcePath)) return;
		const state = this.commandLogs.get(active.id) ?? { sourceOffset: 0, finalized: false };
		if (state.sourcePath !== undefined) {
			this.mirrorCommandSource(active, state.sourcePath);
			return;
		}
		try {
			const text = toolOutputText(result);
			replaceFileAtomic(file, text);
			state.contentBytes = Buffer.byteLength(text, "utf8");
			this.commandLogs.set(active.id, state);
		} catch {}
	}

	private finalizeCommandLog(active: RunActiveTool, result: unknown, outcome: CommandLogOutcome): void {
		const file = this.commandLogFile(active);
		if (file === undefined) return;
		const state = this.commandLogs.get(active.id) ?? { sourceOffset: 0, finalized: false };
		if (state.finalized) return;
		const sourcePath = toolFullOutputPath(result) ?? state.sourcePath;
		const mirrored = sourcePath !== undefined && this.mirrorCommandSource(active, sourcePath);
		if (!mirrored && state.sourcePath === undefined) {
			try {
				const text = toolOutputText(result);
				replaceFileAtomic(file, text);
				state.contentBytes = Buffer.byteLength(text, "utf8");
			} catch {}
		}
		const footer = `--- command ${outcome} after ${commandDuration(active.startedAt)} ---`;
		try {
			const size = state.contentBytes ?? fs.statSync(file).size;
			fs.appendFileSync(file, `${size > 0 ? "\n\n" : ""}${footer}\n`, { encoding: "utf8", mode: 0o600 });
		} catch {}
		state.finalized = true;
		this.commandLogs.set(active.id, state);
	}

	private finalizeActiveCommandLogs(outcome: CommandLogOutcome): void {
		for (const active of this.activeTools.values()) this.finalizeCommandLog(active, undefined, outcome);
	}

	/** Attach to a live child this process spawned. */
	attach(child: ChildProcess): void {
		this.child = child;
		// Create the capture files up front. E23's diagnosis attaches the stderr tail,
		// and the child that fails hardest is the one that dies before writing a byte —
		// so the file has to exist before any data arrives, or the most actionable
		// failure is the one with no evidence attached.
		for (const file of [this.paths.stdout, this.paths.stderr, this.paths.trace]) {
			try {
				fs.appendFileSync(file, "", { mode: 0o600 });
			} catch {
				// E37: capture is diagnostics, never a reason to fail a run.
			}
		}
		const onStdout = createLineSplitter((line) => this.handleLine(line));
		child.stdout?.on("data", (chunk: Buffer) => {
			this.captureStdout(chunk);
			onStdout(chunk);
		});
		child.stderr?.on("data", (chunk: Buffer) => this.captureStderr(chunk));
		child.stdout?.on("error", () => undefined);
		child.stderr?.on("error", () => undefined);
		// A detached child whose stdin closes shuts pi down (rpc-mode's stdin "end"
		// handler), so an EPIPE here must not become an unhandled error.
		child.stdin?.on("error", () => undefined);
		child.once("error", (error) => this.onSpawnError(error));
		child.once("exit", (code, signal) => this.onProcessExit(code, signal));
		// Node guarantees `close` only after the process has exited and stdio has
		// closed. Final assistant/settlement records can still arrive between these
		// two events, so lifecycle truth is decided here, after the stream is drained.
		child.once("close", (code, signal) => this.onExit(code, signal));
	}

	/** R-EXEC-1: exactly one line at start, with the task text verbatim. */
	sendPrompt(message: string): boolean {
		return this.send({ type: "prompt", id: "initial", message });
	}

	/**
	 * pi 0.83's RPC `steer` command bypasses extension input hooks. A prompt with
	 * streamingBehavior=steer takes the same queue path but emits `input`, which is
	 * the only place the worker can correlate exact text and write an honest ack.
	 */
	sendSteering(message: string): boolean {
		return this.send({ type: "prompt", id: `steer-${Date.now()}`, message, streamingBehavior: "steer" });
	}

	sendAbort(): boolean {
		return this.send({ type: "abort", id: `control-abort-${Date.now()}` });
	}

	requestSessionInfo(): boolean {
		return this.send({ type: "get_state", id: "agi-session-info" });
	}

	send(command: RpcCommand): boolean {
		const stdin = this.child?.stdin;
		if (stdin === undefined || stdin === null || stdin.destroyed || !stdin.writable) return false;
		try {
			stdin.write(`${JSON.stringify(command)}\n`);
			return true;
		} catch {
			return false;
		}
	}

	/**
	 * Stop observing the child entirely, without signalling it.
	 *
	 * Clearing a reference is not enough: the `exit`/`data` listeners live on the
	 * ChildProcess object, so a pump that has merely been forgotten still writes
	 * `status.json` when the child dies. That is the difference between "this process
	 * stopped supervising" and "this process crashed", and only the latter leaves the
	 * non-terminal status that §10.7 reconciliation reads.
	 */
	detach(): void {
		const child = this.child;
		this.child = undefined;
		this.clearTimers();
		// Detach is "stop supervising", not "kill": toggling AGI mode off or adopting a
		// run must not terminate it. Callers that do mean to terminate escalate
		// explicitly and inside a bounded budget (R-CTRL-19).
		this.clearTerminalSweep();
		if (child === undefined) return;
		child.removeAllListeners("exit");
		child.removeAllListeners("close");
		child.removeAllListeners("error");
		child.stdout?.removeAllListeners("data");
		child.stderr?.removeAllListeners("data");
	}

	/** R-CTRL-31: adoption marks the run as controllable only through the filesystem. */
	markDetached(): void {
		this.status.detached = true;
		this.writeStatus();
	}

	private captureStdout(chunk: Buffer): void {
		// P38 (mjakl's rolling window). Unbounded capture of a long-running child is a
		// disk-exhaustion vector, so the raw log stops growing at the window size and
		// records that it did.
		if (this.stdoutBytes >= CAPTURE_WINDOW_BYTES) return;
		this.stdoutBytes += chunk.length;
		try {
			fs.appendFileSync(this.paths.stdout, chunk, { mode: 0o600 });
			if (this.stdoutBytes >= CAPTURE_WINDOW_BYTES) {
				fs.appendFileSync(this.paths.stdout, `\n[capture stopped at ${CAPTURE_WINDOW_BYTES} bytes]\n`);
			}
		} catch {
			// E37: a failed log write must never crash the orchestrator.
		}
	}

	private captureStderr(chunk: Buffer): void {
		if (this.stderrBytes >= CAPTURE_WINDOW_BYTES) return;
		this.stderrBytes += chunk.length;
		try {
			fs.appendFileSync(this.paths.stderr, chunk, { mode: 0o600 });
		} catch {
			// Same reasoning as captureStdout.
		}
	}

	private handleLine(line: string): void {
		if (line.trim().length === 0) return;
		if (Buffer.byteLength(line) > MAX_EVENT_LINE_BYTES) {
			// E25: skipped with a counter rather than parsed.
			this.skipped += 1;
			this.status.skippedEvents = this.skipped;
			return;
		}
		let parsed: unknown;
		try {
			parsed = JSON.parse(line);
		} catch {
			// E24: non-JSON lines are already in stdout.log and are ignored for state.
			// A malformed line must never crash the poller.
			return;
		}
		if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return;
		const event = parsed as WireEvent;
		if (typeof event.type !== "string") return;
		this.sawAnyEvent = true;
		if (event.type === "extension_ui_request") {
			// Belt and braces for F1: every field below is untrusted child input, and a
			// throw here would propagate out of the stdout `data` handler and crash the
			// orchestrator. A malformed event is dropped, never fatal.
			try {
				this.handleUiRequest(event as unknown as RpcExtensionUIRequest);
			} catch (error) {
				this.event({ kind: "worker_ui", method: "(malformed)", detail: `unparseable ui request: ${(error as Error).message}` });
			}
			return;
		}
		if (event.type === "response") {
			this.handleResponse(event as unknown as RpcResponse);
			return;
		}
		this.applyEvent(event);
	}

	/**
	 * E37. Everything on the child's stdout is untrusted: it is produced by another
	 * process, may come from a different pi version, and reaches this method inside
	 * the orchestrator's stdout handler where a throw is unrecoverable. The previous
	 * version dereferenced `response.data.sessionFile` directly, so a response with
	 * `data: null` — well-formed JSON, wrong shape — threw a TypeError out of the
	 * handler. Nothing is read before the shape is checked, and the whole body sits
	 * inside a catch.
	 */
	private handleResponse(response: RpcResponse): void {
		try {
			const record = response as unknown as Record<string, unknown>;
			if (record.command !== "get_state" || record.success !== true) return;
			const data = record.data;
			if (data === null || typeof data !== "object" || Array.isArray(data)) {
				this.event({ kind: "worker_response", command: "get_state", detail: "response data was not an object; ignored" });
				return;
			}
			const fields = data as Record<string, unknown>;
			const sessionFile = typeof fields.sessionFile === "string" && fields.sessionFile.length > 0 ? fields.sessionFile : undefined;
			const sessionId = typeof fields.sessionId === "string" && fields.sessionId.length > 0 ? fields.sessionId : undefined;
			if (sessionFile === undefined && sessionId === undefined) {
				this.event({ kind: "worker_response", command: "get_state", detail: "response carried no usable session identity; ignored" });
				return;
			}
			this.setSessionInfo(sessionFile ?? null, sessionId ?? null);
			if (sessionFile === undefined) return;
			const raw = fs.readFileSync(this.paths.descriptor, "utf8");
			const parsed = readDescriptor(raw, { runId: this.status.runId, name: this.status.name });
			if (!parsed.ok) return;
			// R-CTRL-20/24: the descriptor must record where this session actually lives,
			// so a resume resolves the same file even if the orchestrator's session
			// directory has since changed.
			writeDescriptor(this.paths, {
				...parsed.descriptor,
				sessionFile,
				sessionDir: path.dirname(sessionFile),
				...(sessionId === undefined ? {} : { sessionId }),
			});
		} catch {
			// Resume will honestly refuse if the relaunch contract cannot be completed,
			// and a malformed response must never reach the orchestrator's event loop.
		}
	}

	/**
	 * P35 / R-EXEC-2 (mandatory). Every blocking request is answered, with
	 * cancellation as the default — the orchestrator is headless and has no user to
	 * ask. Fire-and-forget methods are recorded as activity and deliberately not
	 * answered: their ids are not registered in the child, so a response would be
	 * silently dropped and would suggest a correlation that does not exist.
	 */
	private handleUiRequest(request: RpcExtensionUIRequest): void {
		const method = request.method;
		if (FIRE_AND_FORGET_UI_METHODS.has(method)) {
			this.event({ kind: "worker_ui", method, detail: describeFireAndForget(request) });
			this.touch();
			return;
		}
		if (!BLOCKING_UI_METHODS.has(method)) {
			// An unknown method in a newer pi. Answering cancellation is the safe
			// direction: a spurious response is dropped, an unanswered blocking request
			// hangs the child forever.
			this.event({ kind: "worker_ui", method, detail: "unknown method; answered cancelled" });
		} else {
			this.event({ kind: "worker_ui", method, detail: `${method} dialog answered cancelled (E26)` });
		}
		if (typeof request.id !== "string" || request.id.length === 0) {
			// No id means nothing in the child's `pendingExtensionRequests` can be
			// resolved, so there is no correlation to answer. Record it and stop, rather
			// than writing a response whose id is `undefined`.
			this.event({ kind: "worker_ui_unanswerable", method, detail: "ui request had no usable id; nothing to answer" });
			return;
		}
		const response: RpcExtensionUIResponse = { type: "extension_ui_response", id: request.id, cancelled: true };
		const stdin = this.child?.stdin;
		if (stdin !== undefined && stdin !== null && !stdin.destroyed && stdin.writable) {
			try {
				stdin.write(`${JSON.stringify(response)}\n`);
			} catch {
				// R-CTRL-31: an adopted run has no stdin. Nothing can be answered, and the
				// child will sit on the dialog until its deadline. Recorded, not hidden.
				this.event({ kind: "worker_ui_unanswerable", method, detail: "stdin is not writable; the child may block until its deadline" });
			}
		} else {
			this.event({ kind: "worker_ui_unanswerable", method, detail: "stdin is not available (adopted run); the child may block until its deadline" });
		}
		this.touch();
	}

	private applyEvent(event: WireEvent): void {
		if (isEndedState(this.status.state) || this.status.stopped) {
			// Lifecycle is final, evidence is not. Only final assistant/settlement records
			// and completions for already-active tools are accepted after stop/ended
			// checkpoint; ordinary turns, new tools and activity remain frozen so a dead
			// run cannot render as live or consume lifecycle budget.
			if (event.type === "message_end") {
				const message = event.message as { role?: unknown; usage?: unknown } | undefined;
				if (message?.role === "assistant") {
					const changed = this.captureAssistantEvidence(message, true);
					this.reportSafely();
					this.writeStatus();
					if (changed) this.emitEvidenceUpdate();
				}
			} else if (event.type === "agent_settled") {
				this.settled = true;
				this.status.settled = true;
				this.reportSafely();
				this.writeStatus();
			} else if (event.type === "tool_execution_end") {
				if (this.captureLateToolResult(event)) this.emitEvidenceUpdate();
			}
			return;
		}
		const now = new Date().toISOString();
		this.status.activity.lastEventAt = now;
		if (this.status.state === "spawning") {
			this.status.state = "running";
			this.status.startedAt = this.status.startedAt ?? now;
		}
		switch (event.type) {
		case "turn_start":
				this.status.counters.turns += 1;
				break;
			case "tool_execution_start": {
				this.status.counters.toolCalls += 1;
				const tool = typeof event.toolName === "string" ? event.toolName : null;
				const target = pathOfArgs(event.args);
				let id: string;
				if (typeof event.toolCallId === "string" && event.toolCallId.length > 0) {
					id = event.toolCallId;
				} else {
					do id = `anonymous-${++this.anonymousToolSequence}`;
					while (this.activeTools.has(id));
				}
				// A repeated start id is malformed but harmless: it still counts/logs as
				// the wire event it is, while its one invocation is refreshed in place.
				this.activeTools.delete(id);
				const allocatedLogPath = tool === "bash" ? this.allocateCommandLog(id) : "";
				const logPath = allocatedLogPath.length > 0 ? allocatedLogPath : undefined;
				this.activeTools.set(id, { id, tool, startedAt: now, lastProgressAt: now, target, ...(logPath === undefined ? {} : { logPath }) });
				this.projectActiveTool();
				this.event({ kind: "tool_start", toolCallId: id, tool, target, ...(logPath === undefined ? {} : { logPath }) });
				break;
			}
			case "tool_execution_update": {
				const id = typeof event.toolCallId === "string" && event.toolCallId.length > 0 ? event.toolCallId : null;
				const active = id === null ? undefined : this.activeTools.get(id);
				if (active !== undefined) {
					active.lastProgressAt = monotonicTimestamp(active.lastProgressAt, now);
					this.persistCommandUpdate(active, event.partialResult);
				}
				this.projectActiveTool();
				// Progress metadata remains durable for stall detection, but the explicit
				// inspection renderer filters these lifecycle pulses from model context.
				this.event({ kind: "tool_progress", toolCallId: id, tool: active?.tool ?? (typeof event.toolName === "string" ? event.toolName : null), target: active?.target ?? null, matched: active !== undefined });
				break;
			}
			case "tool_execution_end": {
				if (event.isError === true) this.status.counters.toolErrors += 1;
				const id = typeof event.toolCallId === "string" && event.toolCallId.length > 0 ? event.toolCallId : null;
				let ended: RunActiveTool | undefined;
				if (id !== null) {
					ended = this.activeTools.get(id);
					this.activeTools.delete(id);
				} else {
					// Compatibility for older/malformed streams without a call id: remove
					// at most one matching anonymous invocation, never arbitrary activity.
					const tool = typeof event.toolName === "string" ? event.toolName : null;
					const matching = [...this.activeTools.values()].reverse().find((invocation) =>
						invocation.id.startsWith("anonymous-") && invocation.tool === tool,
					);
					if (matching !== undefined) {
						ended = matching;
						this.activeTools.delete(matching.id);
					}
				}
				const endText = toolOutputText(event.result);
				const commandOutcome: CommandLogOutcome = this.status.interrupted
					? "interrupted"
					: this.status.stopped
						? "stopped"
						: /Command timed out after/u.test(endText)
							? "timed out"
							: event.isError === true ? "failed" : "completed";
				if (ended !== undefined) this.finalizeCommandLog(ended, event.result, commandOutcome);
				this.projectActiveTool();
				const serialized = serializedToolResult(event.result);
				const resultText = toolResultText(event.result);
				const preview = ended?.logPath === undefined ? sanitizeInspectionField(resultText, MAX_EVENT_TOOL_RESULT_PREVIEW_BYTES) : null;
				const resultIsError = event.isError === true || (event.result !== null && typeof event.result === "object" && (event.result as Record<string, unknown>).isError === true);
				this.event({
					kind: "tool_end",
					toolCallId: id ?? ended?.id ?? null,
					tool: typeof event.toolName === "string" ? event.toolName : (ended?.tool ?? null),
					target: ended?.target ?? null,
					...(ended?.logPath === undefined ? {} : { logPath: ended.logPath }),
					isError: resultIsError,
					resultBytes: Buffer.byteLength(serialized, "utf8"),
					preview,
					truncated: ended?.logPath === undefined && Buffer.byteLength(resultText, "utf8") > MAX_EVENT_TOOL_RESULT_PREVIEW_BYTES,
					matched: ended !== undefined,
				});
				break;
			}
			case "message_end": {
				const message = event.message as { role?: unknown; usage?: unknown } | undefined;
				if (message?.role === "assistant") {
					this.captureAssistantEvidence(message, true);
				}
				break;
			}
			case "compaction_end": {
				// R-WORK-12: the count is what the orchestrator sees; a worker that has
				// compacted repeatedly is a signal the task was too large.
				this.status.counters.compactions += 1;
				this.event({ kind: "compaction", reason: typeof event.reason === "string" ? event.reason : null });
				break;
			}
			case "auto_retry_start":
				this.event({ kind: "retry", detail: typeof event.errorMessage === "string" ? event.errorMessage.slice(0, 200) : null });
				break;
			case "agent_settled":
				if (this.consumeSteeringHandoff()) {
					// Pi emits the stale settlement after extension settle handlers. The
					// worker already started an ordinary user correction in that handler,
					// so this is an internal handoff rather than the run's terminal idle.
					this.event({ kind: "steering_handoff" });
					break;
				}
				// F2 / R-EXEC-3: the only true idle signal. `agent_end` was observed firing
				// four times before settlement in a real child under provider retries.
				this.settled = true;
				this.status.settled = true;
				this.onSettled();
				break;
			default:
				break;
		}
		this.writeStatus();
	}

	private consumeSteeringHandoff(): boolean {
		let parsed: unknown;
		try {
			parsed = JSON.parse(fs.readFileSync(this.paths.steerHandoff, "utf8"));
		} catch {
			return false;
		}
		try {
			fs.rmSync(this.paths.steerHandoff, { force: true });
		} catch {}
		if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return false;
		const marker = parsed as Partial<SteeringHandoff>;
		let runtimePid: number | undefined;
		try {
			const capability = parseSteerCapability(fs.readFileSync(this.paths.steerCapability, "utf8"));
			if (capability.ok && capability.value.supported) runtimePid = capability.value.pid;
		} catch {}
		const validPid = marker.pid === runtimePid || marker.pid === this.status.pid;
		const valid = validPid && typeof marker.reqId === "string" && marker.reqId.length > 0 &&
			typeof marker.ts === "string" && !Number.isNaN(Date.parse(marker.ts));
		if (!valid) {
			this.event({
				kind: "steering_handoff_rejected",
				markerPid: marker.pid ?? null,
				runtimePid: runtimePid ?? null,
				statusPid: this.status.pid,
				reqIdType: typeof marker.reqId,
				reqIdLength: typeof marker.reqId === "string" ? marker.reqId.length : null,
				tsType: typeof marker.ts,
				tsParsed: typeof marker.ts === "string" ? Date.parse(marker.ts) : null,
			});
		}
		return valid;
	}

	private captureAssistantEvidence(
		message: { usage?: unknown; stopReason?: unknown; errorMessage?: unknown },
		includeUsage: boolean,
		source: "rpc" | "session" = "rpc",
	): boolean {
		const identity = assistantEvidenceIdentity(message);
		if (this.assistantEvidenceSeen.has(identity)) return false;
		this.assistantEvidenceSeen.add(identity);
		// Final-message semantics are positional, not set-like: every assistant
		// message replaces the candidate, while an exact wire replay is ignored before
		// it can alter source/order or manufacture post-terminal evidence.
		this.lastAssistantStopReason = typeof message.stopReason === "string" ? message.stopReason : null;
		this.lastAssistantError = typeof message.errorMessage === "string" && message.errorMessage.length > 0 ? message.errorMessage : null;
		const text = textOf(message);
		this.lastAssistantText = text.trim().length > 0 ? text : null;
		this.status.assistantMessageSequence = (this.status.assistantMessageSequence ?? 0) + 1;
		this.status.finalAssistantTextPresent = this.lastAssistantText !== null;
		this.status.assistantEvidence = {
			source,
			observedDuringTerminalization: this.status.stopped || isEndedState(this.status.state),
		};
		this.status.providerStopReason = this.lastAssistantStopReason;
		this.status.providerError = this.lastAssistantError;
		this.status.activity.lastAssistantPreview =
			this.lastAssistantText === null ? null : this.lastAssistantText.replace(/\s+/g, " ").slice(0, 160);
		const preview = sanitizeInspectionField(this.lastAssistantText, MAX_EVENT_ASSISTANT_PREVIEW_BYTES);
		if (preview !== null) {
			this.event({
				kind: "assistant",
				preview,
				truncated: Buffer.byteLength(this.lastAssistantText ?? "", "utf8") > MAX_EVENT_ASSISTANT_PREVIEW_BYTES,
				source,
			});
		}

		if (includeUsage) this.applyUsage(message.usage);
		return true;
	}

	private captureLateToolResult(event: WireEvent): boolean {
		const id = typeof event.toolCallId === "string" && event.toolCallId.length > 0 ? event.toolCallId : null;
		const seen = [...new Set([
			...(this.status.lateToolResultIds ?? []),
			...(this.status.lateToolResults ?? []).map((result) => result.id),
		])];
		if (id === null || seen.includes(id)) return false;
		const active = this.activeTools.get(id);
		if (active === undefined) return false;
		this.finalizeCommandLog(active, event.result, this.status.interrupted ? "interrupted" : this.status.stopped ? "stopped" : event.isError === true ? "failed" : "completed");
		const serialized = serializedToolResult(event.result);
		const resultBytes = Buffer.byteLength(serialized, "utf8");
		const preview = serialized.length === 0 ? null : utf8Prefix(serialized, MAX_LATE_TOOL_PREVIEW_BYTES);
		const result: RunLateToolResult = {
			id,
			tool: active.tool ?? (typeof event.toolName === "string" ? event.toolName : null),
			target: active.target,
			isError: event.isError === true || (event.result !== null && typeof event.result === "object" && (event.result as Record<string, unknown>).isError === true),
			resultBytes,
			preview,
			truncated: resultBytes > MAX_LATE_TOOL_PREVIEW_BYTES,
		};
		this.status.lateToolResultIds = [...seen, id].slice(-MAX_LATE_TOOL_RESULT_IDS);
		this.status.lateToolResults = [...(this.status.lateToolResults ?? []), result].slice(-MAX_LATE_TOOL_RESULTS);
		this.event({ kind: "late_tool_end", ...result, ...(active.logPath === undefined ? {} : { logPath: active.logPath }) });
		this.writeStatus();
		return true;
	}

	/** Bounded, local-only fallback for a final message missed by the RPC event path. */
	private salvageSessionEvidence(): void {
		let expectedDirectory: string | undefined;
		try {
			const raw = fs.readFileSync(this.paths.descriptor, "utf8");
			const parsed = readDescriptor(raw, {
				runId: this.status.runId,
				name: this.status.name,
				agent: this.status.agent,
				cwd: this.status.cwd,
				model: this.status.model,
				sessionId: this.status.sessionId ?? undefined,
			});
			if (!parsed.ok || parsed.descriptor.sessionFile === null || this.status.sessionFile === null) return;
			if (path.resolve(parsed.descriptor.sessionFile) !== path.resolve(this.status.sessionFile)) return;
			expectedDirectory = parsed.descriptor.sessionDir ?? path.dirname(parsed.descriptor.sessionFile);
		} catch (error) {
			// Early failures may have status session identity before descriptor creation.
			// Any other descriptor read failure is an identity ambiguity, so refuse salvage.
			if ((error as NodeJS.ErrnoException).code !== "ENOENT") return;
		}
		const message = salvageFinalAssistantMessage(this.status.sessionFile, { expectedDirectory });
		if (message === null) return;
		if (
			message.text === this.lastAssistantText &&
			message.stopReason === this.lastAssistantStopReason &&
			message.errorMessage === this.lastAssistantError
		) return;
		this.lastAssistantText = message.text;
		this.lastAssistantStopReason = message.stopReason;
		this.lastAssistantError = message.errorMessage;
		this.status.assistantMessageSequence = (this.status.assistantMessageSequence ?? 0) + 1;
		this.status.finalAssistantTextPresent = message.text !== null && message.text.trim().length > 0;
		this.status.assistantEvidence = {
			source: "session",
			observedDuringTerminalization: this.status.stopped || isEndedState(this.status.state),
		};
		this.status.providerStopReason = message.stopReason;
		this.status.providerError = message.errorMessage;
		this.status.activity.lastAssistantPreview =
			this.status.finalAssistantTextPresent && message.text !== null ? message.text.replace(/\s+/g, " ").slice(0, 160) : null;
		this.event({ kind: "report_salvaged", source: "session" });
	}

	private projectActiveTool(): void {
		const invocations = [...this.activeTools.values()];
		this.status.activity.activeTools = invocations.map((invocation) => ({ ...invocation }));
		const active = invocations.reduce<RunActiveTool | undefined>((latest, invocation) => {
			if (latest === undefined) return invocation;
			return invocation.lastProgressAt >= latest.lastProgressAt ? invocation : latest;
		}, undefined);
		this.status.activity.currentTool = active === undefined ? null : (active.tool ?? "unknown");
		this.status.activity.currentToolStartedAt = active?.startedAt ?? null;
		this.status.activity.currentPath = active?.target ?? null;
	}

	private applyUsage(usage: unknown): void {
		if (usage === null || typeof usage !== "object") return;
		const record = usage as Record<string, unknown>;
		const num = (value: unknown): number => (typeof value === "number" && Number.isFinite(value) ? value : 0);
		this.status.usage.input += num(record.input);
		this.status.usage.output += num(record.output);
		this.status.usage.cacheWrite += num(record.cacheWrite);
		const cost = record.cost;
		if (cost !== null && typeof cost === "object") {
			this.status.usage.costUsd += num((cost as Record<string, unknown>).total);
		}
		const total = num(record.totalTokens);
		if (total > 0 && this.status.context !== null) {
			this.status.context = {
				tokens: total,
				contextWindow: this.status.context.contextWindow,
				percent: this.status.context.contextWindow > 0 ? Math.round((total / this.status.context.contextWindow) * 100) : 0,
			};
		}
	}

	setContextWindow(contextWindow: number): void {
		if (contextWindow > 0) this.status.context = { tokens: 0, contextWindow, percent: 0 };
	}

	private touch(): void {
		this.status.activity.lastEventAt = new Date().toISOString();
		this.writeStatus();
	}

	/**
	 * `onReport` writes `result.md` and can therefore fail on a full or read-only
	 * disk. A throw must not abort the caller: in `finalize()` it would leave the run
	 * non-terminal forever (no state write, no inbox close, no completion callback)
	 * and propagate into the stdout handler. The report failure is recorded on the
	 * run and the lifecycle continues.
	 */
	private reportSafely(): void {
		if (this.lastAssistantText === null || this.lastAssistantText === this.reportedAssistantText) return;
		try {
			this.onReport(this.status, this.lastAssistantText);
			if (this.reportIsDurable(this.lastAssistantText)) this.reportedAssistantText = this.lastAssistantText;
		} catch (error) {
			this.event({ kind: "report_error", detail: (error as Error).message });
			this.status.reportError = (error as Error).message;
		}
	}

	private reportIsDurable(text: string): boolean {
		const resultPath = this.status.resultPath;
		if (resultPath === null) return false;
		try {
			const stat = fs.statSync(resultPath);
			if (!stat.isFile()) return false;
			const stored = fs.readFileSync(resultPath, "utf8");
			return stored === text || (!text.endsWith("\n") && stored === `${text}\n`);
		} catch {
			return false;
		}
	}

	/**
	 * R-EXEC-3: after settlement, close stdin and wait EXIT_GRACE_MS for a natural
	 * exit, then escalate SIGTERM -> SIGKILL_GRACE_MS -> SIGKILL against the
	 * process group. E27: the state stays whatever settlement decided; a worker that
	 * did its work and then failed to exit is still complete.
	 *
	 * Settlement latches the outcome before the process is torn down.
	 */
	private onSettled(): void {
		this.reportSafely();
		try {
			this.child?.stdin?.end();
		} catch {
			// Already gone.
		}
		if (this.exitTimer !== undefined) return;
		this.exitTimer = setTimeout(() => {
			if (this.child?.exitCode === null && this.child?.signalCode === null) {
				this.beginSettlementTeardown();
			}
		}, EXIT_GRACE_MS);
		this.exitTimer.unref?.();
	}

	private beginSettlementTeardown(): void {
		if (!this.beginTerminationEscalation()) return;
		// process.kill is synchronous and the exit callback is queued, so this latch
		// is set before onExit can classify the harness-issued signal.
		this.status.expectedTeardown = true;
		this.event({ kind: "settlement_teardown", detail: "worker remained alive after settlement grace" });
		this.writeStatus();
	}

	/** E22: a spawn failure is the run's own failure, with the errno preserved. */
	private onSpawnError(error: Error): void {
		const code = (error as NodeJS.ErrnoException).code ?? "unknown";
		this.event({ kind: "spawn_error", detail: `${code}: ${error.message}` });
		this.finalize("failed", `worker could not be spawned (${code}): ${error.message}`);
	}

	private onProcessExit(code: number | null, signal: NodeJS.Signals | null): void {
		if (this.processExitObserved) return;
		this.processExitObserved = true;
		this.clearTimers();
		// The pi process is gone, but a descendant it spawned may still hold the process
		// group open. That is the R-CTRL-17 case, so the grace is armed here rather than
		// waiting for stdio close — `clearTimers` deliberately leaves it alone.
		this.armTerminalSweep();
		try {
			const disk = readStatus(this.paths).status;
			if (disk !== undefined) {
				this.status.stopped = this.status.stopped || disk.stopped;
				this.status.interrupted = this.status.interrupted || disk.interrupted;
				if (!this.settled && (disk.state === "stopped" || disk.state === "paused")) {
					this.status.state = disk.state;
					this.status.endedAt = disk.endedAt;
					if (disk.state === "paused") {
						this.status.activity = {
							...disk.activity,
							activeTools: disk.activity.activeTools?.map((invocation) => ({ ...invocation })),
						};
					}
				}
			}
		} catch {
			// The in-memory lifecycle remains usable when status.json is unreadable.
		}
		this.status.exitCode = code;
		this.status.processSignal = signal;
		this.event({ kind: "exit", code, signal });
		this.writeStatus();
		// A descendant can keep inherited stdio open indefinitely. Prefer the real
		// `close` drain barrier, but never leave the run non-terminal forever when the
		// direct child is already gone.
		this.drainTimer = setTimeout(() => this.onExit(code, signal), STDIO_DRAIN_GRACE_MS);
		this.drainTimer.unref?.();
	}

	/** Classify only after stdout/stderr have drained and the child emits `close`. */
	private onExit(code: number | null, signal: NodeJS.Signals | null): void {
		if (!this.processExitObserved) this.onProcessExit(code, signal);
		if (this.drainClassified) return;
		this.drainClassified = true;
		if (this.drainTimer !== undefined) clearTimeout(this.drainTimer);
		this.drainTimer = undefined;
		if (this.processExitObserved) {
			// `close` carries the same exit facts, but keep the final values truthful if
			// a platform reports them only here.
			if (this.status.exitCode === null && code !== null) this.status.exitCode = code;
			if (this.status.processSignal === null && signal !== null) this.status.processSignal = signal;
		}
		const finalCode = this.status.exitCode ?? code;
		const finalSignal = (this.status.processSignal ?? signal) as NodeJS.Signals | null;
		// Session persistence happens after Pi emits message_end. If stop wins that
		// narrow window, stdout may never deliver the final event even though the
		// session now contains it. Exit is the last bounded opportunity to recover it.
		this.salvageSessionEvidence();
		this.reportSafely();

		if (isTerminalState(this.status.state)) {
			// R-EXEC-10 (sticky) and E27: a state already decided is not rewritten by the
			// exit. A stop always produces a child error, and letting that error rewrite
			// the state would erase the operator's intent.
			this.finalizeActiveCommandLogs(this.status.state === "stopped" ? "stopped" : this.status.state === "timedOut" ? "timed out" : this.status.state === "complete" ? "completed" : "failed");
			this.writeStatus();
			writeClosed(this.paths, readStatus(this.paths).status ?? this.status);
			this.emitTerminal();
			return;
		}

		// R-EXEC-11 state precedence at settle: stopped > timedOut > failed > complete.
		if (this.status.stopped) {
			this.finalize("stopped", null);
			return;
		}
		if (this.status.timedOut) {
			this.finalize("timedOut", this.status.error ?? "legacy worker timeout");
			return;
		}
		if (this.status.interrupted) {
			this.finalize("paused", null);
			return;
		}
		if (finalSignal !== null && !this.status.expectedTeardown) {
			// R-EXEC-12: a signal with no flag set is a failure, never a success. E30
			// (OOM kill) lands here with exitCode 137.
			const signum = SIGNUM[finalSignal] ?? 0;
			this.status.exitCode = 128 + signum;
			this.finalize("failed", `worker terminated unexpectedly by ${finalSignal}`);
			return;
		}
		if (!this.sawAnyEvent) {
			const stderrTail = tail(this.paths.stderr, 2000);
			this.finalize(
				"failed",
				stderrTail.length > 0
					? `worker failed before RPC activity:\n${stderrTail}`
					: "worker exited before producing model, tool, or output activity",
			);
			return;
		}
		if (this.lastAssistantStopReason === "error" || this.lastAssistantStopReason === "aborted") {
			const detail = this.lastAssistantError === null ? "" : `: ${this.lastAssistantError}`;
			this.finalize("failed", `worker model ended with ${this.lastAssistantStopReason}${detail}`);
			return;
		}
		if (!hasSemanticCompletion(this.settled, this.lastAssistantText)) {
			// E28 / R-EXEC-11: exit 0 with no final assistant text is `failed`. pi
			// resolves some provider failures rather than rejecting, so the exit code
			// alone lies — observed directly: a child with a broken provider emitted four
			// `agent_end`s, settled, and exited 0 with empty content.
			this.finalize("failed", "worker completed without producing a final report");
			return;
		}
		if (finalCode !== null && finalCode !== 0) {
			if (this.status.expectedTeardown && finalSignal !== null) {
				this.finalize("complete", null);
				return;
			}
			this.finalize("failed", `worker exited with code ${finalCode}`);
			return;
		}
		this.finalize("complete", null);
	}

	/** Apply a terminal (or paused) state once, then close the control inbox. */
	finalize(state: RunState, error: string | null): void {
		if (this.closed) return;
		// R-EXEC-10: `stopped` is sticky. Nothing overwrites it.
		if (this.status.state === "stopped" && state !== "stopped") return;
		this.status.state = state;
		this.status.endedAt = new Date().toISOString();
		if (state === "complete") this.finalizeActiveCommandLogs("completed");
		else if (state === "paused") this.finalizeActiveCommandLogs("interrupted");
		else if (state === "stopped") this.finalizeActiveCommandLogs("stopped");
		else if (state === "timedOut") this.finalizeActiveCommandLogs("timed out");
		else this.finalizeActiveCommandLogs("failed");
		if (error !== null) this.status.error = error;
		if (state === "complete" || state === "timedOut" || state === "failed" || state === "paused" || state === "stopped") {
			this.salvageSessionEvidence();
			this.reportSafely();
		}
		this.closed = true;
		this.clearTimers();
		// R-CTRL-17: the run is terminal, so whatever is left of its process tree gets
		// FINAL_STOP_GRACE_MS and is then swept.
		this.armTerminalSweep();
		this.writeStatus();
		// R-CTRL-13: a terminal run stops accepting control, so a later steer fails
		// immediately instead of waiting out the ack timeout. The tombstone records
		// `this.status.state`, not the `state` parameter, so the tombstone always records
		// the final in-memory lifecycle verdict and the two files cannot disagree.
		writeClosed(this.paths, this.status);
		this.emitTerminal();
	}

	/**
	 * A stop that arrives *after* the worker already settled with a report does not
	 * erase that result (F7). R-EXEC-10's sticky-`stopped` rule exists so the child
	 * error a stop provokes cannot overwrite the operator's intent; applying it to a
	 * run whose work was already finished inverts it, discarding a completed result.
	 * The stop is still recorded, so the operator sees it was requested and raced.
	 */
	markStopped(stop: RunStopFact, statusWaitMs?: number): void {
		if (this.settled) {
			this.event({ kind: "stop_requested", detail: "worker had already settled; the completed result is kept" });
			this.writeStatus(statusWaitMs);
			return;
		}
		this.status.stopped = true;
		this.status.stop ??= stop;
		this.event({ kind: "stop_requested", source: stop.source, detail: stop.reason, requestedAt: stop.requestedAt });
		this.writeStatus(statusWaitMs);
	}

	markInterrupted(): void {
		if (this.settled) {
			this.event({ kind: "interrupt_requested", detail: "worker had already settled; the completed result is kept" });
			return;
		}
		this.status.interrupted = true;
		this.event({ kind: "interrupt_requested" });
		this.writeStatus();
	}

	/**
	 * Stop escalation after a durable stop request: grace, SIGTERM, then SIGKILL.
	 *
	 * The escalation targets the process *group* we created at spawn, and only after
	 * `validateSignalTarget` proves the recorded pid is still the worker. A pid read
	 * from a status record can have been reused, and SIGTERM/SIGKILL to the wrong
	 * group is unrecoverable.
	 */
	requestStopEscalation(): void {
		if (this.controlExitTimer !== undefined || this.closed) return;
		this.controlExitTimer = setTimeout(() => {
			this.controlExitTimer = undefined;
			if (this.child !== undefined && (this.child.exitCode !== null || this.child.signalCode !== null)) return;
			this.beginTerminationEscalation();
		}, EXIT_GRACE_MS);
		this.controlExitTimer.unref?.();
	}

	/** Bounded teardown: one identity-validated SIGTERM to the run's own group. */
	terminateNow(): boolean {
		return this.signalOwnTree("SIGTERM");
	}

	/** Immediate bounded TERM→KILL escalation, revalidating identity at both stages. */
	terminateWithEscalation(): boolean {
		// Headless drain may otherwise have no referenced handles after it returns.
		// Keep this one timer alive so the promised second-stage KILL cannot vanish
		// merely because the orchestrator reached the end of its event loop.
		return this.beginTerminationEscalation(true);
	}

	/**
	 * Lifecycle halt cannot leave escalation to timers that detach/shutdown clears.
	 * Give SIGTERM a small slice of the remaining budget, then synchronously sweep
	 * the exact spawned process group with SIGKILL before releasing this pump.
	 */
	terminateWithin(deadline: number): void {
		if (!this.signalOwnTree("SIGTERM")) return;
		const graceUntil = Math.min(deadline, Date.now() + 100);
		while (Date.now() < graceUntil) {
			Atomics.wait(teardownWaitCell, 0, 0, Math.min(10, Math.max(1, graceUntil - Date.now())));
		}
		this.sweepProcessGroup();
	}

	/** SIGTERM/SIGKILL the tree, refusing to signal a pid that is no longer ours. */
	private signalOwnTree(signal: NodeJS.Signals): boolean {
		if (this.signalsRejected) return false;
		const verdict = validateSignalTarget({
			pid: this.status.pid,
			processStartIdentity: this.status.processStartIdentity,
			hostname: this.status.hostname,
		});
		if (verdict.ok) return signalProcessTree(verdict.pid, signal);
		const anchor = this.validateProcessGroupAnchor();
		if (anchor.ok) {
			if (signalProcessGroup(anchor.pgid, signal)) {
				this.event({ kind: "process_group_anchor_used", signal, pid: anchor.pid, pgid: anchor.pgid, sid: anchor.sid });
				return true;
			}
			const detail = `identity-proven process group ${anchor.pgid} disappeared before ${signal}`;
			if (signal === "SIGTERM") {
				this.event({ kind: "signal_deferred", signal, detail });
				return false;
			}
			this.rejectSignals(signal, detail);
			return false;
		}
		const leaderPid = this.status.pid;
		const leaderDead =
			this.status.hostname === os.hostname() &&
			leaderPid !== null &&
			Number.isInteger(leaderPid) &&
			leaderPid > 1 &&
			!isProcessAlive(leaderPid);
		// A dead leader plus an anchor that has not become durable yet is a temporary
		// absence of proof, not authorization to fall back to a positive PID and not a
		// reason to cancel the later terminal sweep. A subsequent exact anchor proof may
		// safely authorize the group-only TERM/KILL path.
		if (signal === "SIGTERM" && leaderDead) {
			this.event({ kind: "signal_deferred", signal, detail: `${verdict.reason}; ${anchor.reason}` });
			return false;
		}
		this.rejectSignals(signal, verdict.reason);
		return false;
	}

	private rejectSignals(signal: NodeJS.Signals, detail: string): void {
		this.signalsRejected = true;
		this.clearSignalEscalationTimers();
		this.event({ kind: "signal_skipped", signal, detail });
	}

	private validateProcessGroupAnchor():
		| { ok: true; pid: number; pgid: number; sid: number }
		| { ok: false; reason: string } {
		const leaderPid = this.status.pid;
		if (this.status.hostname !== os.hostname()) return { ok: false, reason: `anchor host mismatch: run was started on '${this.status.hostname}'` };
		if (leaderPid === null || !Number.isInteger(leaderPid) || leaderPid <= 1) return { ok: false, reason: "no valid process-group leader pid is recorded" };
		if (isProcessAlive(leaderPid)) return { ok: false, reason: `leader pid ${leaderPid} is still alive but failed its identity proof` };
		const anchor = this.status.processGroupAnchor;
		if (anchor === undefined || anchor === null) return { ok: false, reason: "no process-group anchor is recorded" };
		if (!Number.isInteger(anchor.pid) || anchor.pid <= 1 || anchor.pid === process.pid) return { ok: false, reason: `recorded anchor pid ${String(anchor.pid)} is not signal-safe` };
		if (anchor.pgid !== leaderPid || anchor.sid !== leaderPid) {
			return { ok: false, reason: `recorded anchor group/session ${anchor.pgid}/${anchor.sid} does not match leader ${leaderPid}` };
		}
		if (!isProcessAlive(anchor.pid)) return { ok: false, reason: `anchor pid ${anchor.pid} is gone` };
		const identity = getProcessStartIdentity(anchor.pid);
		if (identity === undefined || identity !== anchor.processStartIdentity) {
			return { ok: false, reason: `anchor pid ${anchor.pid} failed process-start identity proof` };
		}
		const group = readProcessGroupIdentity(anchor.pid);
		if (group === undefined || group.pgid !== anchor.pgid || group.sid !== anchor.sid) {
			return { ok: false, reason: `anchor pid ${anchor.pid} moved outside recorded group/session ${anchor.pgid}/${anchor.sid}` };
		}
		return { ok: true, pid: anchor.pid, pgid: anchor.pgid, sid: anchor.sid };
	}

	/** Start TERM→KILL only when the TERM target is freshly identity-proven. */
	private beginTerminationEscalation(keepProcessAlive = false): boolean {
		if (!this.signalOwnTree("SIGTERM")) return false;
		if (this.killTimer !== undefined) clearTimeout(this.killTimer);
		this.killTimer = setTimeout(() => {
			this.killTimer = undefined;
			this.sweepProcessGroup();
		}, SIGKILL_GRACE_MS);
		if (!keepProcessAlive) this.killTimer.unref?.();
		return true;
	}

	/**
	 * R-CTRL-17's identity-safe force kill. A live leader uses its exact PID/start
	 * identity. After that leader exits, the exact inert anchor must still prove the
	 * recorded PID/start identity, PGID, and SID before a group-only signal is sent.
	 */
	private sweepProcessGroup(): void {
		const pid = this.status.pid;
		if (pid === null || pid <= 1) return;
		if (!this.signalOwnTree("SIGKILL")) return;
		this.event({ kind: "process_group_swept", pid, detail: "identity-validated worker process group received SIGKILL" });
	}

	/** Arm the R-CTRL-17 grace once the run has ended, whatever ended it. */
	private armTerminalSweep(): void {
		if (this.finalStopTimer !== undefined || this.signalsRejected) return;
		this.finalStopTimer = setTimeout(() => {
			this.finalStopTimer = undefined;
			this.sweepProcessGroup();
		}, FINAL_STOP_GRACE_MS);
		this.finalStopTimer.unref?.();
	}

	timerState(): { controlExit: boolean; finalStop: boolean; referenced: boolean } {
		const timers = [this.controlExitTimer, this.finalStopTimer].filter((timer): timer is ReturnType<typeof setTimeout> => timer !== undefined);
		return {
			controlExit: this.controlExitTimer !== undefined,
			finalStop: this.finalStopTimer !== undefined,
			referenced: timers.some((timer) => timer.hasRef?.() === true),
		};
	}

	/**
	 * Adopt an externally-written ended state (the worker's own interrupt or stop
	 * handling) and give its process tree the R-CTRL-17 grace.
	 */
	syncExternalStatus(): void {
		let disk: RunStatus | undefined;
		try {
			disk = readStatus(this.paths).status;
		} catch {
			return;
		}
		if (disk === undefined) return;
		this.status.stopped = this.status.stopped || disk.stopped;
		if (disk.stop !== undefined && disk.stop !== null) this.status.stop = disk.stop;
		if (disk.processGroupAnchor !== undefined && disk.processGroupAnchor !== null) this.status.processGroupAnchor ??= disk.processGroupAnchor;
		this.status.interrupted = this.status.interrupted || disk.interrupted;
		if (this.settled) return;
		if (disk.state !== "stopped" && disk.state !== "paused") return;
		this.status.state = disk.state;
		this.status.endedAt = disk.endedAt;
		if (disk.state === "paused") {
			this.status.activity = {
				...disk.activity,
				activeTools: disk.activity.activeTools?.map((invocation) => ({ ...invocation })),
			};
		}
		this.armTerminalSweep();
	}

	setSessionInfo(sessionFile: string | null, sessionId: string | null): void {
		if (sessionFile !== null) this.status.sessionFile = sessionFile;
		if (sessionId !== null) this.status.sessionId = sessionId;
		this.writeStatus();
	}

	setSpawned(pid: number, identity: string | undefined): void {
		this.status.pid = pid;
		this.status.processStartIdentity = identity ?? null;
		this.status.startedAt = new Date().toISOString();
		// `spawnWorker` passes `detached: true` off Windows, so the child leads its own
		// process group and the group id equals this pid. That is what makes the
		// R-CTRL-17 sweep addressable after the leader itself has exited.
		this.writeStatus();
	}

	setProcessGroupAnchor(anchor: RunProcessGroupAnchor): void {
		this.status.processGroupAnchor ??= anchor;
		this.event({ kind: "process_group_anchor", pid: anchor.pid, pgid: anchor.pgid, sid: anchor.sid });
		this.writeStatus();
	}

	private emitTerminal(): void {
		if (this.terminalEmitted) return;
		if (!isTerminalState(this.status.state) && this.status.state !== "paused") return;
		this.terminalEmitted = true;
		this.callbacks.onTerminal?.(this.status);
	}

	private emitEvidenceUpdate(): void {
		if (!this.terminalEmitted) return;
		this.callbacks.onEvidence?.(this.status);
	}

	private clearTimers(): void {
		for (const timer of [this.exitTimer, this.killTimer, this.controlExitTimer, this.drainTimer]) {
			if (timer !== undefined) clearTimeout(timer);
		}
		this.exitTimer = undefined;
		this.killTimer = undefined;
		this.controlExitTimer = undefined;
		this.drainTimer = undefined;
	}

	/** Cancel every timer whose only remaining action is a process signal. */
	private clearSignalEscalationTimers(): void {
		for (const timer of [this.exitTimer, this.killTimer, this.controlExitTimer, this.finalStopTimer]) {
			if (timer !== undefined) clearTimeout(timer);
		}
		this.exitTimer = undefined;
		this.killTimer = undefined;
		this.controlExitTimer = undefined;
		this.finalStopTimer = undefined;
	}

	/**
	 * The R-CTRL-17 grace is deliberately *not* cleared by `clearTimers`: the whole
	 * point is to outlive the direct child so a surviving descendant is still swept.
	 * Only detach and dispose cancel it, because neither is an instruction to kill.
	 */
	private clearTerminalSweep(): void {
		if (this.finalStopTimer !== undefined) clearTimeout(this.finalStopTimer);
		this.finalStopTimer = undefined;
	}

	event(fields: Record<string, unknown>): void {
		const event = { ts: new Date().toISOString(), kind: String(fields.kind ?? "event"), ...fields };
		appendEvent(this.paths, event);
		const nextSequence = this.traceSequence + 1;
		if (appendWorkerTrace(this.paths, nextSequence, event)) this.traceSequence = nextSequence;
	}

	/**
	 * `status.json` has more than one legitimate writer: this pump, the reconciler
	 * (which sets `detached` when it adopts a run) and `agi_worker` (which sets
	 * `resultConsumed`, and `attention`). R-EXEC-7's temp-then-rename makes each write
	 * atomic, but atomicity is not isolation — serialising our private copy over the
	 * file reverts whatever another writer changed since we last wrote, a plain lost
	 * update. So re-read and adopt the externally-owned fields first.
	 *
	 * Operator intent (`stopped`, `interrupted`) is merged the same way, latching on:
	 * once anyone has requested a stop, no pump write may clear it.
	 */
	private writeStatus(statusWaitMs?: number): void {
		try {
			const current = readStatus(this.paths).status;
			if (current !== undefined) {
				if (current.detached === true) this.status.detached = true;
				if (current.resultConsumed === true) this.status.resultConsumed = true;
				if (current.stopped === true) this.status.stopped = true;
				if (current.stop !== undefined && current.stop !== null) this.status.stop = current.stop;
				if (current.processGroupAnchor !== undefined && current.processGroupAnchor !== null) this.status.processGroupAnchor ??= current.processGroupAnchor;
				if (current.interrupted === true) this.status.interrupted = true;
				if (current.settled === true) this.status.settled = true;
				if (this.status.attention === null || this.status.attention === undefined) {
					this.status.attention = current.attention;
				}
			}
		} catch {
			// An unreadable or half-written file is not a reason to skip our own write.
		}
		try {
			writeStatus(this.paths, this.status, undefined, statusWaitMs === undefined ? {} : { waitMs: statusWaitMs });
		} catch (error) {
			// E37: a status write that fails (ENOSPC) marks the run failed in memory and
			// is surfaced, but must never throw into the orchestrator's event loop.
			this.status.error = `status write failed: ${(error as Error).message}`;
		}
		this.callbacks.onStatus?.(this.status);
	}
}

/**
 * Describes a fire-and-forget request for the event log.
 *
 * Every field is treated as untrusted. The payload comes from arbitrary extension
 * code in the child (`ui.notify(undefined)` is enough), and this runs inside the
 * orchestrator's stdout handler, so an unguarded `.slice()` on a missing field
 * throws where nothing catches it and takes the orchestrator down with it. A
 * worker must never be able to crash its supervisor.
 */
function describeFireAndForget(request: RpcExtensionUIRequest): string {
	const text = (value: unknown, limit = 200): string => (typeof value === "string" ? value.slice(0, limit) : "(none)");
	const record = request as unknown as Record<string, unknown>;
	switch (request.method) {
		case "notify":
			return `${text(record.notifyType, 20) === "(none)" ? "info" : text(record.notifyType, 20)}: ${text(record.message)}`;
		case "setStatus":
			return `${text(record.statusKey, 80)}=${record.statusText === undefined || record.statusText === null ? "(cleared)" : text(record.statusText)}`;
		case "setWidget":
			return `${text(record.widgetKey, 80)}: ${Array.isArray(record.widgetLines) ? `${record.widgetLines.length} line(s)` : "(cleared)"}`;
		case "setTitle":
			return text(record.title);
		case "set_editor_text":
			return `${typeof record.text === "string" ? record.text.length : 0} char(s)`;
		default:
			return "";
	}
}

const SIGNUM: Record<string, number> = {
	SIGHUP: 1,
	SIGINT: 2,
	SIGQUIT: 3,
	SIGILL: 4,
	SIGABRT: 6,
	SIGFPE: 8,
	SIGKILL: 9,
	SIGSEGV: 11,
	SIGPIPE: 13,
	SIGALRM: 14,
	SIGTERM: 15,
	SIGUSR1: 10,
	SIGUSR2: 12,
};

function tail(file: string, bytes: number): string {
	try {
		const stat = fs.statSync(file);
		const start = Math.max(0, stat.size - bytes);
		const fd = fs.openSync(file, "r");
		try {
			const buffer = Buffer.alloc(Math.min(bytes, stat.size));
			fs.readSync(fd, buffer, 0, buffer.length, start);
			return buffer.toString("utf8").trim();
		} finally {
			fs.closeSync(fd);
		}
	} catch {
		return "";
	}
}

/**
 * R-CTRL-31 adoption. A worker that survived the orchestrator restart is still
 * doing useful work, so it is adopted rather than killed. Its stdout pipe died
 * with the old parent, which means the pump cannot see events any more: the run is
 * polled from `status.json` and marked `detached`, and control is filesystem-only.
 */
export function pollAdoptedRun(paths: RunPaths, status: RunStatus): RunStatus {
	const pid = status.pid;
	if (pid === null || pid <= 0) return status;
	const identity = getProcessStartIdentity(pid);
	let alive = false;
	try {
		process.kill(pid, 0);
		alive = true;
	} catch (error) {
		alive = (error as NodeJS.ErrnoException).code === "EPERM";
	}
	// P12: pid liveness alone is not proof, because pids are reused. The recorded
	// start identity is what distinguishes "still running" from "a different process
	// now holds this pid".
	if (alive && (status.processStartIdentity === null || identity === undefined || identity === status.processStartIdentity)) {
		return status;
	}
	const next: RunStatus = { ...status, state: "orphaned", endedAt: new Date().toISOString() };
	appendEvent(paths, { ts: next.endedAt ?? new Date().toISOString(), kind: "orphaned", detail: "process is no longer alive" });
	return next;
}