/**
 * SF Parallel Orchestrator — Core engine for parallel milestone orchestration.
 *
 * Manages worker lifecycle, budget tracking, and coordination. Workers are
 * separate processes spawned via child_process, each running in its own git
 * worktree with SF_MILESTONE_LOCK env var set. The coordinator monitors
 * workers via session status files (see session-status-io.ts).
 */
import { spawn } from "node:child_process";
import {
	appendFileSync,
	existsSync,
	mkdirSync,
	readFileSync,
	renameSync,
	unlinkSync,
	writeFileSync,
} from "node:fs";
import { join } from "node:path";
import { delay } from "./atomic-write.js";
import {
	autoWorktreeBranch,
	runWorktreePostCreateHook,
	syncSfStateToWorktree,
} from "./auto-worktree.js";
import { getErrorMessage } from "./error-utils.js";
import { readIntegrationBranch } from "./git-service.js";
import { emitJournalEvent } from "./journal.js";
import { nativeBranchExists } from "./native-git-bridge.js";
import { analyzeParallelEligibility } from "./parallel-eligibility.js";
import {
	checkIntentConflicts,
	clearAllIntents,
	declareIntent,
	releaseIntent,
} from "./parallel-intent.js";
import { sfRoot } from "./paths.js";
import { resolveParallelConfig } from "./preferences.js";
// Re-export for consumer access via parallel-orchestrator module:
export { resolveParallelConfig } from "./preferences.js";
import {
	cleanupStaleSessions,
	DEFAULT_STALE_TIMEOUT_MS,
	readAllSessionStatuses,
	readSessionStatus,
	removeSessionStatus,
	sendSignal,
	writeSessionStatus,
} from "./session-status-io.js";
import { monitorNdjsonStdout, resolveSfBin } from "./spawn-worker.js";
import { selectConflictFreeBatch } from "./uok/execution-graph.js";
import { resolveUokFlags } from "./uok/flags.js";
import { writeUnitRuntimeRecord } from "./uok/unit-runtime.js";
import { logWarning } from "./workflow-logger.js";
import { createWorktree, worktreePath } from "./worktree-manager.js";

// ─── Module State ──────────────────────────────────────────────────────────
let state = null;
function overlapKey(a, b) {
	return a < b ? `${a}::${b}` : `${b}::${a}`;
}
// ─── Persistence ──────────────────────────────────────────────────────────
const ORCHESTRATOR_STATE_FILE = "orchestrator.json";
const TMP_SUFFIX = ".tmp";
function stateFilePath(basePath) {
	return join(sfRoot(basePath), ORCHESTRATOR_STATE_FILE);
}
/**
 * Persist the current orchestrator state to .sf/orchestrator.json.
 * Uses atomic write (tmp + rename) to prevent partial reads.
 */
export function persistState(basePath) {
	if (!state) return;
	try {
		const dir = sfRoot(basePath);
		if (!existsSync(dir)) mkdirSync(dir, { recursive: true });
		const persisted = {
			active: state.active,
			workers: [...state.workers.values()].map((w) => ({
				milestoneId: w.milestoneId,
				title: w.title,
				pid: w.pid,
				worktreePath: w.worktreePath,
				startedAt: w.startedAt,
				state: w.state,
				cost: w.cost,
			})),
			totalCost: state.totalCost,
			startedAt: state.startedAt,
			configSnapshot: {
				max_workers: state.config.max_workers,
				budget_ceiling: state.config.budget_ceiling,
			},
		};
		const dest = stateFilePath(basePath);
		const tmp = dest + TMP_SUFFIX;
		writeFileSync(tmp, JSON.stringify(persisted, null, 2), "utf-8");
		renameSync(tmp, dest);
	} catch (e) {
		logWarning("parallel", `persist parallel state failed: ${e.message}`);
	}
}
/**
 * Remove the persisted state file.
 */
function removeStateFile(basePath) {
	try {
		const p = stateFilePath(basePath);
		if (existsSync(p)) unlinkSync(p);
	} catch (e) {
		logWarning("parallel", `clear parallel state file failed: ${e.message}`);
	}
}
function isPidAlive(pid) {
	if (!Number.isInteger(pid) || pid <= 0) return false;
	try {
		process.kill(pid, 0);
		return true;
	} catch (e) {
		logWarning(
			"parallel",
			`pid alive check failed for pid ${pid}: ${e.message}`,
		);
		return false;
	}
}
/**
 * Restore orchestrator state from .sf/orchestrator.json.
 * Checks PID liveness for each worker:
 * - Living PID → state "running", process stays null (no handle)
 * - Dead PID → removed from restored state
 * Returns null if no state file exists or no workers survive.
 */
export function restoreState(basePath) {
	try {
		const p = stateFilePath(basePath);
		if (!existsSync(p)) return null;
		const raw = readFileSync(p, "utf-8");
		const persisted = JSON.parse(raw);
		// Filter to only workers with living PIDs
		persisted.workers = persisted.workers.filter((w) => {
			if (
				w.state === "stopped" ||
				w.state === "error" ||
				w.state === "cancelled"
			)
				return false;
			return isPidAlive(w.pid);
		});
		if (persisted.workers.length === 0) {
			// No surviving workers — clean up and return null
			removeStateFile(basePath);
			return null;
		}
		return persisted;
	} catch (e) {
		logWarning("parallel", `readParallelState JSON parse failed: ${e.message}`);
		return null;
	}
}
function workerLogPath(basePath, milestoneId) {
	return join(sfRoot(basePath), "parallel", `${milestoneId}.stderr.log`);
}
function appendWorkerLog(basePath, milestoneId, chunk) {
	try {
		const dir = join(sfRoot(basePath), "parallel");
		if (!existsSync(dir)) mkdirSync(dir, { recursive: true });
		appendFileSync(workerLogPath(basePath, milestoneId), chunk, "utf-8");
	} catch (e) {
		logWarning(
			"parallel",
			`appendFileSync worker log failed for ${milestoneId}: ${e.message}`,
		);
	}
}
/** Read the stderr log for a worker (used by doctor checks). */
export function readWorkerStderr(basePath, milestoneId) {
	try {
		const p = workerLogPath(basePath, milestoneId);
		if (!existsSync(p)) return "";
		return readFileSync(p, "utf-8");
	} catch {
		return "";
	}
}
function restoreRuntimeState(basePath) {
	if (state?.active) {
		// Verify at least one worker is alive — if all are in terminal states,
		// the cached state is stale and we should fall through to cleanup.
		const hasLiveWorker = [...state.workers.values()].some(
			(w) =>
				w.state !== "error" && w.state !== "stopped" && w.state !== "cancelled",
		);
		if (hasLiveWorker) return true;
		// All workers dead — clear stale state so restoreState() can clean up.
		state = null;
	}
	const restored = restoreState(basePath);
	if (restored && restored.workers.length > 0) {
		const config = resolveParallelConfig(undefined);
		state = {
			active: restored.active,
			workers: new Map(),
			config: {
				...config,
				max_workers: restored.configSnapshot.max_workers,
				budget_ceiling: restored.configSnapshot.budget_ceiling,
			},
			totalCost: restored.totalCost,
			startedAt: restored.startedAt,
			failureCascadeTriggered: false,
		};
		for (const w of restored.workers) {
			const diskStatus = readSessionStatus(basePath, w.milestoneId);
			state.workers.set(w.milestoneId, {
				milestoneId: w.milestoneId,
				title: w.title,
				pid: diskStatus?.pid ?? w.pid,
				process: null,
				worktreePath: diskStatus?.worktreePath ?? w.worktreePath,
				startedAt: w.startedAt,
				state: diskStatus?.state ?? w.state,
				cost: diskStatus?.cost ?? w.cost,
			});
		}
		return true;
	}
	// Fallback: rebuild coordinator state from live session status files.
	// This covers cases where orchestrator.json is missing/corrupt but workers are
	// still running and writing heartbeats under .sf/parallel/.
	cleanupStaleSessions(basePath);
	const statuses = readAllSessionStatuses(basePath);
	if (statuses.length === 0) {
		return false;
	}
	const config = resolveParallelConfig(undefined);
	state = {
		active: true,
		workers: new Map(),
		config,
		totalCost: 0,
		startedAt: Math.min(...statuses.map((status) => status.startedAt)),
		failureCascadeTriggered: false,
	};
	for (const status of statuses) {
		state.workers.set(status.milestoneId, {
			milestoneId: status.milestoneId,
			title: status.milestoneId,
			pid: status.pid,
			process: null,
			worktreePath: status.worktreePath,
			startedAt: status.startedAt,
			state: status.state,
			cost: status.cost,
		});
		state.totalCost += status.cost;
	}
	return true;
}
async function waitForWorkerExit(worker, timeoutMs) {
	if (worker.process) {
		await new Promise((resolve) => {
			const done = () => resolve();
			const timer = setTimeout(done, timeoutMs);
			worker.process.once("exit", () => {
				clearTimeout(timer);
				resolve();
			});
		});
		return worker.process === null || !isPidAlive(worker.pid);
	}
	const startedAt = Date.now();
	while (Date.now() - startedAt < timeoutMs) {
		if (!isPidAlive(worker.pid)) return true;
		await delay(50);
	}
	return !isPidAlive(worker.pid);
}
// ─── Accessors ─────────────────────────────────────────────────────────────
/** Returns true if the orchestrator is active and has been initialized. */
export function isParallelActive() {
	return state?.active ?? false;
}
/** Returns the current orchestrator state, or null if not initialized. */
export function getOrchestratorState() {
	return state;
}
/** Returns a snapshot of all tracked workers as an array. */
export function getWorkerStatuses(basePath) {
	if (basePath) {
		refreshWorkerStatuses(basePath, { restoreIfNeeded: true });
	}
	if (!state) return [];
	return [...state.workers.values()];
}
// ─── Preparation ───────────────────────────────────────────────────────────
/**
 * Analyze eligibility and prepare for parallel start.
 * Returns the candidates report without actually starting workers.
 * Also detects orphaned sessions from prior crashes.
 */
export async function prepareParallelStart(basePath, _prefs) {
	// Detect orphaned sessions before eligibility analysis
	const sessions = readAllSessionStatuses(basePath);
	const orphans = [];
	for (const session of sessions) {
		const alive = isPidAlive(session.pid);
		orphans.push({ milestoneId: session.milestoneId, pid: session.pid, alive });
		if (!alive) {
			// Clean up dead session
			removeSessionStatus(basePath, session.milestoneId);
		}
	}
	const candidates = await analyzeParallelEligibility(basePath);
	return orphans.length > 0 ? { ...candidates, orphans } : candidates;
}
// ─── Start ─────────────────────────────────────────────────────────────────
/**
 * Start parallel execution with the given eligible milestones.
 * Creates worktrees, spawns worker processes, and begins monitoring.
 */
export async function startParallel(basePath, milestoneIds, prefs) {
	// Prevent workers from spawning nested parallel sessions
	if (process.env.SF_PARALLEL_WORKER) {
		return {
			started: [],
			errors: [
				{
					mid: "all",
					error: "Cannot start parallel from within a parallel worker",
				},
			],
		};
	}
	const config = resolveParallelConfig(prefs);
	const uokFlags = resolveUokFlags(prefs);
	// Release any leftover state from a previous session before reassigning
	if (state) {
		for (const w of state.workers.values()) {
			w.cleanup?.();
			w.cleanup = undefined;
			w.process = null;
		}
		state.workers.clear();
	}
	// Try to restore from a previous crash
	const restored = restoreState(basePath);
	if (restored && restored.workers.length > 0) {
		// Adopt surviving workers instead of starting new ones
		state = {
			active: true,
			workers: new Map(),
			config,
			totalCost: restored.totalCost,
			startedAt: restored.startedAt,
			failureCascadeTriggered: false,
		};
		const adopted = [];
		for (const w of restored.workers) {
			state.workers.set(w.milestoneId, {
				milestoneId: w.milestoneId,
				title: w.title,
				pid: w.pid,
				process: null, // no handle for adopted workers
				worktreePath: w.worktreePath,
				startedAt: w.startedAt,
				state: "running",
				cost: w.cost,
				retryCount: 0, // R015 T02: retry counter for spawn-failure respawn
			});
			adopted.push(w.milestoneId);
		}
		return { started: adopted, errors: [] };
	}
	const now = Date.now();
	// Initialize orchestrator state
	state = {
		active: true,
		workers: new Map(),
		config,
		totalCost: 0,
		startedAt: now,
		failureCascadeTriggered: false,
	};
	const started = [];
	const errors = [];
	let filteredMilestoneIds = milestoneIds;
	let intentFilesByMilestone = new Map();
	if (uokFlags.executionGraph && milestoneIds.length > 1) {
		try {
			const requestedIds = new Set(milestoneIds);
			const candidates = await analyzeParallelEligibility(basePath);
			intentFilesByMilestone = candidates.fileSets ?? intentFilesByMilestone;
			const overlapPairs = new Set();
			for (const overlap of candidates.fileOverlaps) {
				if (!requestedIds.has(overlap.mid1) || !requestedIds.has(overlap.mid2))
					continue;
				overlapPairs.add(overlapKey(overlap.mid1, overlap.mid2));
			}
			filteredMilestoneIds = selectConflictFreeBatch({
				orderedIds: milestoneIds,
				maxParallel: milestoneIds.length,
				hasConflict: (candidate, existing) =>
					overlapPairs.has(overlapKey(candidate, existing)),
			});
			if (filteredMilestoneIds.length < milestoneIds.length) {
				const skipped = milestoneIds.filter(
					(mid) => !filteredMilestoneIds.includes(mid),
				);
				logWarning(
					"parallel",
					`uok execution graph filtered ${skipped.length} conflicting milestone(s): ${skipped.join(", ")}`,
				);
			}
		} catch (e) {
			logWarning(
				"parallel",
				`uok execution graph overlap analysis failed; using legacy milestone selection: ${e.message}`,
			);
			filteredMilestoneIds = milestoneIds;
		}
	}
	// Cap to max_workers
	const toStart = filteredMilestoneIds.slice(0, config.max_workers);
	for (const mid of toStart) {
		const intendedFiles = intentFilesByMilestone.get(mid) ?? [];
		const conflicts = checkIntentConflicts(basePath, mid, intendedFiles);
		if (conflicts.length > 0) {
			errors.push({
				mid,
				error: `File intent conflict with ${conflicts.map((c) => c.milestoneId).join(", ")}`,
			});
			continue;
		}
		const intent = declareIntent(basePath, mid, intendedFiles);
		if (!intent.ok) {
			errors.push({ mid, error: intent.error ?? "Could not declare intent" });
			continue;
		}
		// Check budget ceiling before each spawn
		if (isBudgetExceeded()) {
			releaseIntent(basePath, mid);
			errors.push({
				mid,
				error: `Budget ceiling ($${config.budget_ceiling}) reached — skipping`,
			});
			continue;
		}
		try {
			// Create the worktree (without chdir — coordinator stays in project root)
			let wtPath;
			try {
				wtPath = createMilestoneWorktree(basePath, mid);
			} catch (e) {
				logWarning(
					"parallel",
					`createMilestoneWorktree fallback for ${mid}: ${e.message}`,
				);
				wtPath = worktreePath(basePath, mid);
			}
			const worker = {
				milestoneId: mid,
				title: mid,
				pid: 0, // placeholder — real PID set by spawnWorker()
				process: null,
				worktreePath: wtPath,
				startedAt: now,
				state: "running",
				cost: 0,
				retryCount: 0, // R015 T02: retry counter for spawn-failure respawn
			};
			state.workers.set(mid, worker);
			// Spawn BEFORE writing session status so the file gets the real worker PID.
			const spawned = spawnWorker(basePath, mid);
			if (!spawned) {
				worker.state = "error";
			}
			// Write session status with real PID (or 0 if spawn failed)
			writeSessionStatus(basePath, {
				milestoneId: mid,
				pid: worker.pid,
				state: worker.state,
				currentUnit: null,
				completedUnits: 0,
				cost: 0,
				lastHeartbeat: now,
				startedAt: now,
				worktreePath: wtPath,
			});
			started.push(mid);
		} catch (err) {
			releaseIntent(basePath, mid);
			const message = getErrorMessage(err);
			errors.push({ mid, error: message });
		}
	}
	// If nothing started successfully, deactivate
	if (started.length === 0) {
		state.active = false;
	}
	// Persist state for crash recovery
	persistState(basePath);
	return { started, errors };
}
// ─── Worktree Creation ────────────────────────────────────────────────────
/**
 * Create a git worktree for a milestone without changing the coordinator's cwd.
 * Uses milestone/<MID> branch naming (same as auto-worktree.ts).
 */
function createMilestoneWorktree(basePath, milestoneId) {
	const branch = autoWorktreeBranch(milestoneId);
	const branchExists = nativeBranchExists(basePath, branch);
	let info;
	if (branchExists) {
		info = createWorktree(basePath, milestoneId, {
			branch,
			reuseExistingBranch: true,
		});
	} else {
		const integrationBranch =
			readIntegrationBranch(basePath, milestoneId) ?? undefined;
		info = createWorktree(basePath, milestoneId, {
			branch,
			startPoint: integrationBranch,
		});
	}
	// Run post-create hook if configured
	runWorktreePostCreateHook(basePath, info.path);
	// Copy .sf/ planning artifacts (milestones, CONTEXT, ROADMAP, etc.) from the
	// project root into the worktree. Without this, workers for newly-planned
	// milestones can't find their roadmap and exit immediately (#2184 Bug 4).
	syncSfStateToWorktree(basePath, info.path);
	return info.path;
}
// ─── Worker Spawning ───────────────────────────────────────────────────
/**
 * Spawn a worker process for a milestone.
 * The worker runs `sf headless --json autonomous` in the milestone's worktree
 * with SF_MILESTONE_LOCK set to isolate state derivation.
 *
 * IMPORTANT: We use `headless --json autonomous` instead of `--print "/autonomous"`.
 * --print mode calls session.prompt() which returns immediately after the
 * extension command handler fires, because autonomous mode's ctx.newSession()
 * resets the session and unblocks the outer prompt() await. This causes
 * process.exit(0) to fire before any LLM work happens. See #2792.
 *
 * The headless subcommand uses an RPC client that keeps the process alive
 * until autonomous mode emits a terminal notification or the idle timer fires.
 * It outputs NDJSON events to stdout (with --json), which our
 * processWorkerLine() parser already understands.
 */
export function spawnWorker(basePath, milestoneId) {
	if (!state) return false;
	const worker = state.workers.get(milestoneId);
	if (!worker) return false;
	if (worker.process) return true; // already spawned
	// Resolve the SF CLI binary path
	const binPath = resolveSfBin();
	if (!binPath) return false;
	let child;
	try {
		const workerEnv = {
			...process.env,
			SF_MILESTONE_LOCK: milestoneId,
			// Pass the real project root so workers don't need to re-derive it.
			// Without this, process.cwd() resolves symlinks and the worktree
			// path heuristic can match the user-level ~/.sf instead of the
			// project .sf, causing writes to ~ and corrupting user config.
			SF_PROJECT_ROOT: basePath,
			// Prevent workers from spawning their own parallel sessions
			SF_PARALLEL_WORKER: "1",
		};
		// Apply worker model override if configured, so workers use a cheaper
		// model (e.g. Haiku) rather than inheriting the coordinator's model.
		if (state.config.worker_model) {
			workerEnv.SF_WORKER_MODEL = state.config.worker_model;
		}
		const shellWrapper = state.config.shell_wrapper ?? [];
		const [spawnCmd, spawnArgs] =
			shellWrapper.length > 0
				? [
						shellWrapper[0],
						[
							...shellWrapper.slice(1),
							process.execPath,
							binPath,
							"headless",
							"--json",
							"autonomous",
						],
					]
				: [process.execPath, [binPath, "headless", "--json", "autonomous"]];
		child = spawn(spawnCmd, spawnArgs, {
			cwd: worker.worktreePath,
			env: workerEnv,
			stdio: ["ignore", "pipe", "pipe"],
			detached: false,
		});
	} catch (e) {
		logWarning(
			"parallel",
			`spawnSync worker failed for ${milestoneId}: ${e.message}`,
		);
		return false;
	}
	// Handle spawn errors (e.g., ENOENT when binary doesn't exist)
	child.on("error", () => {
		if (!state) return;
		const w = state.workers.get(milestoneId);
		if (w) {
			w.process = null;
			// Don't change state — spawn failure is non-fatal, coordinator can retry
		}
	});
	worker.process = child;
	worker.pid = child.pid ?? 0;
	if (!child.pid) {
		// Spawn returned but no PID — process failed to start
		worker.process = null;
		return false;
	}
	// ── NDJSON stdout monitoring ────────────────────────────────────────
	// Workers run via `headless --json`, which forwards all RPC events
	// as NDJSON to stdout. We parse message_end events to extract
	// cost/token usage, keeping the coordinator's cost tracking in sync
	// with actual API spend.
	if (child.stdout) {
		monitorNdjsonStdout(child.stdout, (line) =>
			processWorkerLine(basePath, milestoneId, line),
		);
	}
	if (child.stderr) {
		child.stderr.on("data", (data) => {
			appendWorkerLog(basePath, milestoneId, data.toString());
		});
	}
	// Update session status with real PID
	writeSessionStatus(basePath, {
		milestoneId,
		pid: worker.pid,
		state: "running",
		currentUnit: null,
		completedUnits: 0,
		cost: worker.cost,
		lastHeartbeat: Date.now(),
		startedAt: worker.startedAt,
		worktreePath: worker.worktreePath,
	});
	// Store cleanup function to remove all listeners from the child process.
	// This prevents listener accumulation when workers are respawned, since
	// handler closures capture milestoneId and other data that would otherwise
	// be retained indefinitely.
	worker.cleanup = () => {
		child.stdout?.removeAllListeners();
		child.stderr?.removeAllListeners();
		child.removeAllListeners();
	};
	// Handle worker exit
	child.on("exit", (code) => {
		if (!state) return;
		const w = state.workers.get(milestoneId);
		if (!w) return;
		// Remove all stream listeners to release closure references
		w.cleanup?.();
		w.cleanup = undefined;
		w.process = null;
		if (w.state === "stopped" || w.state === "cancelled") return; // terminal, already handled
		if (code === 0) {
			w.state = "stopped";
		} else {
			w.state = "error";
			appendWorkerLog(
				basePath,
				milestoneId,
				`\n[orchestrator] worker exited with code ${code ?? "null"}\n`,
			);
			// ── Sibling-stop cascade (opt-in) ──────────────────────────────
			// If stop_on_failure is enabled and no cascade has fired yet,
			// gracefully stop all siblings that are still running.
			if (state.config.stop_on_failure && !state.failureCascadeTriggered) {
				state.failureCascadeTriggered = true;
				const triggeringWorkerKind = milestoneId;
				for (const [siblingId, sibling] of state.workers) {
					if (siblingId === milestoneId) continue;
					if (sibling.state !== "running" && sibling.state !== "paused")
						continue;
					// Graceful stop: file-based IPC signal + SIGTERM
					sendSignal(basePath, siblingId, "stop");
					if (sibling.pid > 0) {
						try {
							if (sibling.process) {
								sibling.process.kill("SIGTERM");
							} else if (sibling.pid !== process.pid) {
								process.kill(sibling.pid, "SIGTERM");
							}
						} catch (e) {
							logWarning(
								"parallel",
								`stop-on-failure SIGTERM failed for pid ${sibling.pid}: ${e.message}`,
							);
						}
					}
					sibling.state = "cancelled";
					sibling.process = null;
					releaseIntent(basePath, siblingId);
					// Update session status so dashboard reflects the cancellation
					writeSessionStatus(basePath, {
						milestoneId: siblingId,
						pid: sibling.pid,
						state: "cancelled",
						currentUnit: null,
						completedUnits: 0,
						cost: sibling.cost,
						lastHeartbeat: Date.now(),
						startedAt: sibling.startedAt,
						worktreePath: sibling.worktreePath,
					});
					// Emit structured journal event
					emitJournalEvent(basePath, {
						ts: new Date().toISOString(),
						flowId: siblingId,
						seq: 0,
						eventType: "parallel-cancelled-by-sibling",
						data: {
							workerId: siblingId,
							triggeringWorkerId: milestoneId,
							triggeringWorkerKind,
						},
					});
				}
			}
		}
		// Update session status and persist orchestrator state for crash recovery
		writeSessionStatus(basePath, {
			milestoneId,
			pid: w.pid,
			state: w.state,
			currentUnit: null,
			completedUnits: 0,
			cost: w.cost,
			lastHeartbeat: Date.now(),
			startedAt: w.startedAt,
			worktreePath: w.worktreePath,
		});
		releaseIntent(basePath, milestoneId);
		persistState(basePath);
	});
	return true;
}
// ─── NDJSON Processing ──────────────────────────────────────────────────────
/**
 * Process a single NDJSON line from a worker's stdout.
 * Extracts cost and token usage from message_end events and updates
 * the worker's tracking state + session status file.
 */
function processWorkerLine(basePath, milestoneId, line) {
	if (!line.trim() || !state) return;
	let event;
	try {
		event = JSON.parse(line);
	} catch {
		return; // Non-NDJSON lines (progress text, tool output) are expected — silent drop
	}
	const type = String(event.type ?? "");
	// message_end carries usage data with cost
	if (type === "message_end" && event.message) {
		const msg = event.message;
		const usage = msg.usage;
		if (usage) {
			const cost = usage.cost?.total;
			if (typeof cost === "number") {
				const worker = state.workers.get(milestoneId);
				if (worker) {
					worker.cost += cost;
					// Update aggregate
					state.totalCost = 0;
					for (const w of state.workers.values()) {
						state.totalCost += w.cost;
					}
				}
			}
		}
		// Update session status file so dashboard sees live cost
		const worker = state.workers.get(milestoneId);
		if (worker) {
			writeSessionStatus(basePath, {
				milestoneId,
				pid: worker.pid,
				state: worker.state,
				currentUnit: null,
				completedUnits: 0,
				cost: worker.cost,
				lastHeartbeat: Date.now(),
				startedAt: worker.startedAt,
				worktreePath: worker.worktreePath,
			});
		}
	}
	// tool_execution_start can track current unit
	if (type === "extension_ui_request" && event.method === "notify") {
		// SF autonomous mode sends notifications about current unit
		const worker = state.workers.get(milestoneId);
		if (worker) {
			writeSessionStatus(basePath, {
				milestoneId,
				pid: worker.pid,
				state: worker.state,
				currentUnit: null,
				completedUnits: 0,
				cost: worker.cost,
				lastHeartbeat: Date.now(),
				startedAt: worker.startedAt,
				worktreePath: worker.worktreePath,
			});
		}
	}
}
// ─── Stop ──────────────────────────────────────────────────────────────────
/**
 * Stop all workers or a specific milestone's worker.
 * Sends stop signals and updates tracking state.
 */
export async function stopParallel(basePath, milestoneId) {
	if (!state) return;
	const targets = milestoneId ? [milestoneId] : [...state.workers.keys()];
	for (const mid of targets) {
		const worker = state.workers.get(mid);
		if (!worker) continue;
		// Send stop signal via file-based IPC (worker checks on next dispatch)
		sendSignal(basePath, mid, "stop");
		// Send SIGTERM to the process for immediate response.
		// Use process handle when available, fall back to PID-based kill
		// (handles are null after coordinator restart / deserialization).
		if (worker.pid > 0) {
			try {
				if (worker.process) {
					worker.process.kill("SIGTERM");
				} else if (worker.pid !== process.pid) {
					process.kill(worker.pid, "SIGTERM");
				}
			} catch (e) {
				logWarning(
					"parallel",
					`process.kill SIGTERM failed for pid ${worker.pid}: ${e.message}`,
				);
			}
		}
		// Wait for the headless process to cascade SIGTERM to its RPC child.
		// The headless signal handler calls client.stop() which sends SIGTERM
		// to the RPC child and waits up to 1000ms. The previous 750ms window
		// was insufficient — the parent got SIGKILL before the child died,
		// leaving orphaned RPC processes holding auto.lock. See #2798.
		const exitedAfterTerm = await waitForWorkerExit(worker, 3000);
		if (!exitedAfterTerm && worker.pid > 0) {
			try {
				if (worker.process) {
					worker.process.kill("SIGKILL");
				} else if (worker.pid !== process.pid) {
					process.kill(worker.pid, "SIGKILL");
				}
			} catch (e) {
				logWarning(
					"parallel",
					`process.kill SIGKILL failed for pid ${worker.pid}: ${e.message}`,
				);
			}
			await waitForWorkerExit(worker, 250);
		}
		// Remove stream listeners before releasing the process handle
		worker.cleanup?.();
		worker.cleanup = undefined;
		// Update in-memory state
		worker.state = "stopped";
		worker.process = null;
		// Clean up session status file
		removeSessionStatus(basePath, mid);
	}
	// If stopping all workers, deactivate the orchestrator
	if (!milestoneId) {
		state.active = false;
	}
	// Clear all intent claims on shutdown
	clearAllIntents(basePath);
	// Persist final state and clean up state file
	removeStateFile(basePath);
}
export async function shutdownParallel(basePath) {
	if (!state) return;
	await stopParallel(basePath);
	resetOrchestrator();
}
// ─── Pause / Resume ────────────────────────────────────────────────────────
/** Pause a specific worker or all workers. */
export function pauseWorker(basePath, milestoneId) {
	if (!state) return;
	const targets = milestoneId ? [milestoneId] : [...state.workers.keys()];
	for (const mid of targets) {
		const worker = state.workers.get(mid);
		if (!worker || worker.state !== "running") continue;
		sendSignal(basePath, mid, "pause");
		worker.state = "paused";
	}
}
/** Resume a specific worker or all workers. */
export function resumeWorker(basePath, milestoneId) {
	if (!state) return;
	const targets = milestoneId ? [milestoneId] : [...state.workers.keys()];
	for (const mid of targets) {
		const worker = state.workers.get(mid);
		if (!worker || worker.state !== "paused") continue;
		sendSignal(basePath, mid, "resume");
		worker.state = "running";
	}
}
// ─── Status Refresh ────────────────────────────────────────────────────────
/**
 * Poll worker statuses from disk and update orchestrator state.
 * Call this periodically from the dashboard refresh cycle.
 */
export function refreshWorkerStatuses(basePath, options = {}) {
	if (!state && options.restoreIfNeeded) {
		restoreRuntimeState(basePath);
	}
	if (!state) return;
	// Clean up stale sessions first
	const staleIds = cleanupStaleSessions(basePath);
	for (const mid of staleIds) {
		const worker = state.workers.get(mid);
		if (worker) {
			worker.cleanup?.();
			worker.cleanup = undefined;
			worker.state = "error";
			worker.process = null;
		}
	}
	// Read all live session statuses from disk
	const statuses = readAllSessionStatuses(basePath);
	const statusMap = new Map();
	for (const s of statuses) {
		statusMap.set(s.milestoneId, s);
	}
	// Update in-memory worker state from disk data
	for (const [mid, worker] of state.workers) {
		const diskStatus = statusMap.get(mid);
		if (!diskStatus) {
			if (!isPidAlive(worker.pid)) {
				worker.cleanup?.();
				worker.cleanup = undefined;
				worker.state = "error";
				worker.process = null;
			}
			continue;
		}
		worker.state = diskStatus.state;
		worker.cost = diskStatus.cost;
		worker.pid = diskStatus.pid;
	}
	// Recalculate aggregate cost
	state.totalCost = 0;
	for (const worker of state.workers.values()) {
		state.totalCost += worker.cost;
	}
	// Kill workers that have exceeded their timeout
	const workerTimeoutMs = (state.config.worker_timeout_minutes ?? 120) * 60_000;
	for (const [, worker] of state.workers) {
		if (
			worker.state === "running" &&
			Date.now() - worker.startedAt > workerTimeoutMs
		) {
			if (worker.process) worker.process.kill("SIGTERM");
			worker.cleanup?.();
			worker.cleanup = undefined;
			worker.state = "error";
			worker.process = null;
		}
	}

	// ── Spawn-failure watchdog (R015) ─────────────────────────────────────
	// Detect silent worker failure: spawned worker alive > grace+timeout, no heartbeat,
	// zero progress. Transitions to 'failed' state, kills the process, captures stderr,
	// emits a journal event, and triggers T02's retry/respawn path.
	const spawnFailureTimeoutMs =
		state.config.spawn_failure_timeout_ms ?? DEFAULT_STALE_TIMEOUT_MS;
	const spawnFailureGracePeriodMs =
		state.config.spawn_failure_grace_period_ms ?? 10_000;
	for (const [mid, worker] of state.workers) {
		if (worker.state !== "running") continue;
		// Grace period: don't fire before worker has had a fair chance to start (avoids
		// false positives on slow LLM cold-starts, especially with Gemini).
		if (Date.now() - worker.startedAt < spawnFailureGracePeriodMs) continue;
		// Check if heartbeat is stale AND no progress was ever made.
		const diskStatus = readSessionStatus(basePath, mid);
		if (!diskStatus) continue;
		const heartbeatAge = Date.now() - (diskStatus.lastHeartbeat ?? 0);
		const hasProgress = (diskStatus.progressCount ?? 0) > 0;
		if (heartbeatAge <= spawnFailureTimeoutMs || hasProgress) continue;
		// Silent failure confirmed — transition to 'failed'.
		logWarning(
			"parallel",
			`spawn watchdog: worker ${mid} silent (${heartbeatAge}ms heartbeat age, progress=${diskStatus.progressCount}). Transitioning to failed.`,
		);
		worker.state = "failed";
		// Kill the subprocess (SIGKILL — the worker is already dead from the loop's perspective)
		if (worker.process) {
			try {
				worker.process.kill("SIGKILL");
			} catch {
				// Process may have already exited; non-fatal
			}
		}
		worker.cleanup?.();
		worker.cleanup = undefined;
		worker.process = null;
		// Write 'failed' session status so the IPC protocol reflects the terminal state.
		writeSessionStatus(basePath, {
			...diskStatus,
			state: "failed",
			lastHeartbeat: Date.now(),
		});
		// Emit structured journal event for observability.
		const stderrExcerpt = readWorkerStderr(basePath, mid);
		emitJournalEvent(basePath, {
			ts: new Date().toISOString(),
			flowId: mid,
			seq: 0,
			eventType: "worker-spawn-failure",
			data: {
				milestoneId: mid,
				pid: worker.pid,
				elapsedMsSinceSpawn: Date.now() - worker.startedAt,
				heartbeatAgeMs: heartbeatAge,
				stderrExcerpt:
					stderrExcerpt.length > 1000
						? stderrExcerpt.slice(-1000)
						: stderrExcerpt,
			},
		});

		// ── T02: Retry/respawn path (R015) ─────────────────────────────────
		// After watchdog fires and worker is marked 'failed', check if we have
		// retry budget remaining. If so, respawn the worker; otherwise leave it
		// in terminal 'failed' state (doctor signal already fired).
		const maxRetries = state.config.max_retries ?? 3;
		const retryCount = worker.retryCount ?? 0;
		if (retryCount < maxRetries) {
			// Increment recoveryAttempts in the unit runtime record for observability.
			writeUnitRuntimeRecord(
				basePath,
				"execute-milestone",
				milestoneId,
				worker.startedAt,
				{
					recoveryAttempts: retryCount + 1,
					lineageEvent: {
						unitType: "execute-milestone",
						unitId: milestoneId,
						status: "failed",
						workerSessionId: `spawn-${worker.pid}`,
						note: `spawn-failure retry ${retryCount + 1}/${maxRetries}`,
					},
				},
			);
			worker.retryCount = retryCount + 1;
			worker.startedAt = Date.now();
			worker.state = "running";
			// Respawn the worker — will write session status with new PID.
			spawnWorker(basePath, milestoneId);
			// Update session status with new PID and 'running' state after respawn.
			const refreshedStatus = readSessionStatus(basePath, milestoneId);
			if (refreshedStatus) {
				writeSessionStatus(basePath, {
					...refreshedStatus,
					pid: worker.pid,
					state: "running",
					lastHeartbeat: Date.now(),
				});
			}
		} else {
			// Exhausted retry budget — write failed status to disk for consistency.
			logWarning(
				"parallel",
				`worker ${mid} exhausted retry budget (${retryCount}/${maxRetries}). Left in terminal failed state.`,
			);
			writeSessionStatus(basePath, {
				...diskStatus,
				state: "failed",
				lastHeartbeat: Date.now(),
			});
		}
	}

	// If all workers are in a terminal state (error/stopped/cancelled/failed), the
	// orchestration is finished — deactivate and clean up so zombie workers don't persist.
	const allDead =
		state.workers.size > 0 &&
		[...state.workers.values()].every(
			(w) =>
				w.state === "error" ||
				w.state === "stopped" ||
				w.state === "cancelled" ||
				w.state === "failed",
		);
	if (allDead) {
		state.active = false;
		removeStateFile(basePath);
		state = null;
		return;
	}
	// Persist updated state for crash recovery
	persistState(basePath);
}
// ─── Budget ────────────────────────────────────────────────────────────────
/** Get aggregate cost across all workers. */
export function getAggregateCost() {
	if (!state) return 0;
	return state.totalCost;
}
/** Check if budget ceiling has been reached. */
export function isBudgetExceeded() {
	if (!state) return false;
	if (state.config.budget_ceiling == null) return false;
	return state.totalCost >= state.config.budget_ceiling;
}
// ─── Reset ─────────────────────────────────────────────────────────────────
/** Reset orchestrator state. Called on clean shutdown. */
export function resetOrchestrator() {
	if (state) {
		// Explicitly release all WorkerInfo references and run any pending
		// cleanup callbacks so child process stream closures are freed.
		for (const w of state.workers.values()) {
			w.cleanup?.();
			w.cleanup = undefined;
			w.process = null;
		}
		state.workers.clear();
	}
	state = null;
}
