agame/addons/godot_ai/utils/port_resolver.gd

@tool
class_name McpPortResolver
extends RefCounted

## Pure-static port discovery / OS-specific scrapers. No instance state,
## no editor dependencies. plugin.gd has thin instance shims that wrap
## these and increment the cold-start trace counters.

## Canonical pid-file path. plugin.gd::SERVER_PID_FILE re-exports this so
## external readers and tests can use either name.
const SERVER_PID_FILE := "user://godot_ai_server.pid"
## Where a plugin-spawned server reports a failure that happens before it
## publishes its capability record (`--startup-report`). Removed before every
## spawn, so a report that exists afterwards belongs to that launch.
const SERVER_STARTUP_REPORT := "user://godot_ai_server_startup.json"
const WindowsPortReservation := preload("res://addons/godot_ai/utils/windows_port_reservation.gd")
const LinuxProc := preload("res://addons/godot_ai/utils/linux_proc.gd")
const SNAPSHOT_DIAGNOSTIC_STAGES := ["single", "pair", "first", "final"]
const SNAPSHOT_DIAGNOSTIC_CATEGORIES := [
	"invalid_pid", "shell_exit", "empty_output", "outer_size", "outer_json", "outer_shape",
	"snapshot_size", "snapshot_json", "snapshot_shape", "collector_null", "row_shape",
	"row_pid", "row_chain", "row_identity", "lineage_cycle",
]
static var _process_spawn_mutex := Mutex.new()

enum PortOccupancy { UNKNOWN, FREE, OCCUPIED }


## Serialize plugin-owned subprocess creation across the brief process-global
## environment override used for server capabilities. Callers must keep this
## window to environment mutation + one spawn and release it immediately.
static func lock_process_spawn() -> void:
	_process_spawn_mutex.lock()


static func unlock_process_spawn() -> void:
	_process_spawn_mutex.unlock()


## Linux can prove ownership through procfs inside minimal desktop sandboxes.
## Query each launch so installing the missing tool makes Retry work.
static func listener_tools_problem() -> String:
	if OS.get_name() != "Linux" or LinuxProc.available():
		return ""
	var output: Array = []
	var available := OS.execute("/bin/sh", ["-c",
		"command -v lsof >/dev/null 2>&1 || command -v ss >/dev/null 2>&1"
	], output, true)
	if available == 0:
		return ""
	return "Cannot verify Linux listener ownership: /proc is unavailable and neither lsof nor ss is available on the editor's PATH. Make /proc accessible inside the editor environment, or install lsof or iproute2 there, then retry starting the server."


static func can_bind_local_port(port: int) -> bool:
	var server := TCPServer.new()
	var err := server.listen(port, "127.0.0.1")
	if err == OK:
		server.stop()
		return true
	return false


## Windows permits a loopback bind beside a wildcard listener. Use the OS
## listener table, and treat failed queries as unavailable rather than free.
static func is_port_in_use(port: int) -> bool:
	if OS.get_name() == "Windows":
		return windows_port_occupancy(port) != PortOccupancy.FREE
	return is_port_in_use_via_scrape(port)


## `trace` mirrors `find_all_pids_on_port`'s hook: one call per OS
## invocation with the counter name of the scraper that actually ran, so
## a wrapping caller's startup trace sees a genuine PowerShell fallback
## as `powershell`, not as a silent extra second under `netstat`.
static func is_port_in_use_via_scrape(port: int, trace: Callable = Callable()) -> bool:
	if OS.get_name() == "Linux":
		var linux := LinuxProc.listener_snapshot()
		if linux.known:
			return linux.listeners.has(port)
		## A failed observation cannot prove that a port is free.
		if not can_bind_local_port(port):
			return true
	var output: Array = []
	if OS.get_name() == "Windows":
		return windows_port_occupancy(port, windows_listener_snapshot(trace)) != PortOccupancy.FREE
	_trace(trace, "lsof")
	var exit_code := OS.execute("lsof", ["-ti:%d" % port, "-sTCP:LISTEN"], output, true)
	if exit_code == 0 and output.size() > 0 and not output[0].strip_edges().is_empty():
		return true
	if OS.get_name() != "Linux":
		return false
	_trace(trace, "ss")
	output.clear()
	return (
		OS.execute("ss", ["-H", "-ltnp"], output, true) == 0
		and not parse_linux_ss_pids(str(output[0]) if not output.is_empty() else "", port).is_empty()
	)


## One snapshot per selection operation avoids a subprocess for every candidate.
## Unknown observations never prove a port free. This is not ownership evidence.
static func windows_listener_snapshot(trace: Callable = Callable()) -> Dictionary:
	var output: Array = []
	_trace(trace, "netstat")
	var code := OS.execute("netstat", ["-ano"], output, true)
	var snapshot := windows_snapshot_from_netstat(code, output)
	if snapshot.known:
		return snapshot
	_trace(trace, "powershell")
	output.clear()
	var script := "$ErrorActionPreference='Stop'; try { $listeners = @(Get-NetTCPConnection -ErrorAction Stop | Where-Object { $_.State -eq 'Listen' } | Select-Object LocalPort,OwningProcess); ConvertTo-Json -Compress -Depth 3 -InputObject @{ listeners = $listeners }; exit 0 } catch { exit 1 }"
	code = execute_windows_powershell(script, output)
	return windows_snapshot_from_powershell(code, output)


static func windows_port_occupancy(port: int, snapshot: Dictionary = {}) -> PortOccupancy:
	if snapshot.is_empty():
		snapshot = windows_listener_snapshot()
	if not snapshot.known:
		return PortOccupancy.UNKNOWN
	return PortOccupancy.OCCUPIED if snapshot.listeners.has(port) else PortOccupancy.FREE


static func windows_snapshot_from_netstat(exit_code: int, output: Array) -> Dictionary:
	var unknown := {"known": false, "listeners": {}}
	if exit_code != 0 or output.is_empty():
		return unknown
	var listeners := {}
	var saw_tcp := false
	for line in str(output[0]).split("\n"):
		var fields := split_on_whitespace(line.strip_edges())
		if fields.is_empty() or fields[0].to_upper() != "TCP":
			continue
		if fields.size() < 5 or not fields[1].contains(":") or not fields[2].contains(":"):
			return unknown
		var port_text := fields[1].get_slice(":", fields[1].get_slice_count(":") - 1)
		var remote_port_text := fields[2].get_slice(":", fields[2].get_slice_count(":") - 1)
		var pid_text := fields[fields.size() - 1]
		if not port_text.is_valid_int() or not remote_port_text.is_valid_int() or not pid_text.is_valid_int():
			return unknown
		var port := int(port_text)
		var pid := int(pid_text)
		if port < 1 or port > 65535 or pid < 0 or pid > 4294967295 or int(remote_port_text) < 0 or int(remote_port_text) > 65535:
			return unknown
		saw_tcp = true
		## Windows localizes the state column, but listeners keep a wildcard foreign port.
		if fields[2].ends_with(":0"):
			_record_windows_listener(listeners, port, pid)
	return {"known": true, "listeners": listeners} if saw_tcp else unknown


static func windows_snapshot_from_powershell(exit_code: int, output: Array) -> Dictionary:
	var unknown := {"known": false, "listeners": {}}
	if exit_code != 0 or output.is_empty():
		return unknown
	var parsed: Variant = JSON.parse_string(str(output[0]))
	if not (parsed is Dictionary) or not (parsed.get("listeners") is Array):
		return unknown
	var listeners := {}
	for row in parsed.listeners:
		if not (row is Dictionary):
			return unknown
		var port: Variant = row.get("LocalPort")
		var pid: Variant = row.get("OwningProcess")
		if not (port is int or port is float) or not is_finite(float(port)) or float(port) != floor(float(port)) or port < 1 or port > 65535:
			return unknown
		if not (pid is int or pid is float) or not is_finite(float(pid)) or float(pid) != floor(float(pid)) or pid < 0 or pid > 4294967295:
			return unknown
		_record_windows_listener(listeners, int(port), int(pid))
	return {"known": true, "listeners": listeners}


static func _record_windows_listener(listeners: Dictionary, port: int, pid: int) -> void:
	if not listeners.has(port):
		listeners[port] = []
	## PID 0 still occupies the port, but cannot identify a process we may stop.
	if pid > 0 and not listeners[port].has(pid):
		listeners[port].append(pid)


## Return the PID currently listening on the given TCP port, or 0 if
## no positive listener PID could be observed. Thin wrapper around `find_all_pids_on_port`
## — the per-OS scraping logic lives in one place.
static func find_pid_on_port(port: int, trace: Callable = Callable()) -> int:
	var pids := find_all_pids_on_port(port, trace)
	return pids[0] if not pids.is_empty() else 0


## Returns every PID bound LISTEN on `port`. Used by the kill paths so
## both the uvicorn reloader parent AND its worker child are caught when
## both bind the same port.
##
## `trace` is an optional Callable that fires once per OS invocation with
## a counter name (`"netstat"` / `"powershell"` / `"lsof"` / `"ss"`) so the plugin
## can keep its cold-start trace accurate. The Windows path may fall
## through netstat → PowerShell, and a wrapping caller can't see which
## scraper actually ran without the hook.
static func find_all_pids_on_port(port: int, trace: Callable = Callable(), snapshot: Dictionary = {}) -> Array[int]:
	if OS.get_name() == "Windows":
		if snapshot.is_empty():
			snapshot = windows_listener_snapshot(trace)
		var pids: Array[int] = []
		if snapshot.known:
			pids.assign(snapshot.listeners.get(port, []))
		return pids
	if OS.get_name() == "Linux":
		var linux := LinuxProc.listener_snapshot()
		if linux.known:
			var linux_pids := LinuxProc.listener_pids(port, linux)
			if not linux_pids.is_empty() or not linux.listeners.has(port):
				return linux_pids
	var output: Array = []
	_trace(trace, "lsof")
	var exit_code := OS.execute("lsof", ["-ti:%d" % port, "-sTCP:LISTEN"], output, true)
	if exit_code == 0 and not output.is_empty():
		var lsof_pids := parse_lsof_pids(str(output[0]))
		if not lsof_pids.is_empty() or OS.get_name() != "Linux":
			return lsof_pids
	elif OS.get_name() != "Linux":
		var empty: Array[int] = []
		return empty
	_trace(trace, "ss")
	output.clear()
	if OS.execute("ss", ["-H", "-ltnp"], output, true) != 0 or output.is_empty():
		var empty: Array[int] = []
		return empty
	return parse_linux_ss_pids(str(output[0]), port)


static func _trace(trace: Callable, counter: String) -> void:
	if trace.is_valid():
		trace.call(counter)


static func execute_windows_powershell(script: String, output: Array) -> int:
	var args := ["-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script]
	for exe in windows_powershell_candidates():
		output.clear()
		var exit_code := OS.execute(exe, args, output, true)
		if exit_code == 0:
			return exit_code
	return -1


static func windows_powershell_candidates() -> Array[String]:
	var candidates: Array[String] = []
	## Prefer an already-installed PowerShell 7 at its fixed machine path.
	## Missing or failing installations retain every existing fallback below.
	for variable in ["ProgramW6432", "ProgramFiles"]:
		var program_files := OS.get_environment(variable).replace("\\", "/").trim_suffix("/")
		if not program_files.is_empty():
			var pwsh := program_files.path_join("PowerShell/7/pwsh.exe")
			if FileAccess.file_exists(pwsh) and not candidates.has(pwsh):
				candidates.append(pwsh)
	var system_root := OS.get_environment("SystemRoot")
	if system_root.is_empty():
		system_root = "C:/Windows"
	system_root = system_root.replace("\\", "/").trim_suffix("/")
	candidates.append(system_root + "/System32/WindowsPowerShell/v1.0/powershell.exe")
	candidates.append("powershell.exe")
	candidates.append("pwsh.exe")
	return candidates


## Pure parser for `lsof -ti` output — newline-separated decimal PIDs.
## Empty lines, non-positive values, non-numeric tokens, and duplicate
## IPv4/IPv6 rows for the same listener are dropped.
static func parse_lsof_pids(raw: String) -> Array[int]:
	return parse_pid_lines(raw)


## Parse Linux `ss -H -ltnp` without trusting process names or localized
## state text. The local endpoint is the fourth whitespace-delimited field;
## every `pid=` value on an exact-port row is retained for later brand,
## lineage, and fingerprint checks.
static func parse_linux_ss_pids(raw: String, port: int) -> Array[int]:
	var result: Array[int] = []
	if port < 1 or port > 65535:
		return result
	var expression := RegEx.new()
	if expression.compile("(?:^|,)pid=([0-9]+)(?:,|\\))") != OK:
		return result
	for line in raw.split("\n", false):
		var fields := line.split(" ", false)
		if fields.size() < 4 or not str(fields[3]).ends_with(":%d" % port):
			continue
		for matched in expression.search_all(line):
			var pid := int(matched.get_string(1))
			if pid > 0 and not result.has(pid):
				result.append(pid)
	return result


static func parse_pid_lines(raw: String) -> Array[int]:
	var pids: Array[int] = []
	for line in raw.strip_edges().split("\n", false):
		var stripped := line.strip_edges()
		if stripped.is_valid_int():
			var pid := int(stripped)
			if pid > 0 and not pids.has(pid):
				pids.append(pid)
	return pids


## Parse a Windows `netstat -ano` dump and return PIDs of rows whose
## local address ends with `:port` and foreign port is zero. Substring
## matching the whole dump is wrong: a remote address containing
## `:port` would false-positive against an unrelated ESTABLISHED row.
static func parse_windows_netstat_pid(stdout: String, port: int) -> int:
	var pids := parse_windows_netstat_pids(stdout, port)
	return pids[0] if not pids.is_empty() else 0


static func parse_windows_netstat_pids(stdout: String, port: int) -> Array[int]:
	var snapshot := windows_snapshot_from_netstat(0, [stdout])
	var pids: Array[int] = []
	if snapshot.known:
		pids.assign(snapshot.listeners.get(port, []))
	return pids


static func windows_netstat_dump_parseable(stdout: String) -> bool:
	return bool(windows_snapshot_from_netstat(0, [stdout]).known)


## `String.split(" ", false)` only splits on single spaces; netstat
## columns are separated by runs of spaces / tabs. Collapse manually.
static func split_on_whitespace(s: String) -> PackedStringArray:
	var out: PackedStringArray = []
	var cur := ""
	for i in s.length():
		var c := s.substr(i, 1)
		if c == " " or c == "\t":
			if not cur.is_empty():
				out.append(cur)
				cur = ""
		else:
			cur += c
	if not cur.is_empty():
		out.append(cur)
	return out


static func read_pid_file(path := SERVER_PID_FILE) -> int:
	if not FileAccess.file_exists(path):
		return 0
	var f := FileAccess.open(path, FileAccess.READ)
	if f == null:
		return 0
	var content := f.get_as_text().strip_edges()
	f.close()
	if content.is_empty() or not content.is_valid_int():
		return 0
	var pid := int(content)
	return pid if pid > 0 else 0


static func clear_pid_file() -> void:
	if FileAccess.file_exists(SERVER_PID_FILE):
		DirAccess.remove_absolute(ProjectSettings.globalize_path(SERVER_PID_FILE))


## `kill -0` returns 0 for both running and zombie processes; Godot
## never `waitpid`s on `OS.create_process` children, so a fast-failing
## uvx launcher lingers as a zombie forever and `kill -0` would block
## the spawn-failure branch in check_server_health from firing. Use
## `ps -o stat=` instead. State codes: R/S/D/I/T (live), Z (zombie). #172.
static func pid_alive(pid: int, snapshot: Variant = null) -> bool:
	if pid <= 0:
		return false
	if snapshot != null:
		return not _process_snapshot_row(snapshot, pid).is_empty()
	if OS.get_name() == "Linux":
		return LinuxProc.is_alive(LinuxProc.process_stat(pid))
	if OS.get_name() == "Windows":
		var output: Array = []
		var exit_code := OS.execute("tasklist", ["/FI", "PID eq %d" % pid, "/NH", "/FO", "CSV"], output, true)
		if exit_code != 0 or output.is_empty():
			return false
		for line in output:
			if str(line).find("\"%d\"" % pid) >= 0:
				return true
		return false
	var output: Array = []
	var exit_code := OS.execute("ps", ["-p", str(pid), "-o", "stat="], output, true)
	if exit_code != 0 or output.is_empty():
		return false
	var stat := str(output[0]).strip_edges()
	return not stat.is_empty() and not stat.begins_with("Z")


## Read-only process identity helpers.  Lifecycle authority is granted only
## after capturing a fingerprint and every later stop re-reads the fingerprint
## before acting, so PID reuse fails closed.
## One Windows query captures the target and at most fifteen ancestors. The
## dictionary belongs to one proof boundary, never a cache: callers must take
## a fresh snapshot when closing the capture window or authorizing a kill.
## Linux captures the same bounded ancestry using procfs; macOS keeps ps.
## The process enumeration stays inside PowerShell; only the bounded target
## ancestry crosses back into Godot, without logging command-line metadata.
static func capture_process_snapshot(pid: int, diagnostics: Variant = null) -> Variant:
	if OS.get_name() == "Linux":
		return LinuxProc.process_snapshot(pid)
	if OS.get_name() != "Windows":
		return null
	if pid <= 1:
		return _snapshot_failure(diagnostics, "invalid_pid", "single")
	var script := _windows_process_snapshot_script(pid)
	var output: Array = []
	var began := Time.get_ticks_msec()
	var code := execute_windows_powershell(script, output)
	if code != 0 or output.is_empty():
		return _snapshot_failure(
			diagnostics, "shell_exit" if code != 0 else "empty_output",
			"single", -1, Time.get_ticks_msec() - began
		)
	return parse_process_snapshot(str(output[0]), pid, diagnostics)


static func _windows_process_snapshot_script(pid: int) -> String:
	return (
		"$rows = @(); $current = %d; $seen = @{}; $capture_failed = $false; "
		+ "$processes = @{}; try { Get-CimInstance Win32_Process -ErrorAction Stop | "
		+ "ForEach-Object { $processes[[int]$_.ProcessId] = $_ } } catch {}; "
		+ "for ($depth = 0; $depth -lt 16 -and $current -gt 1; $depth++) { "
		+ "if ($seen.ContainsKey($current)) { break }; $seen[$current] = $true; "
		+ "$p = $processes[$current]; "
		+ "if ($null -eq $p) { "
		+ "try { $identity = [string](Get-Process -Id $current -ErrorAction Stop).StartTime.ToFileTimeUtc(); "
		+ "$rows += @{pid=$current;parent_pid=0;identity=$identity;commandline=''} } catch { if ($_.CategoryInfo.Category -ne 'ObjectNotFound') { $capture_failed = $true } }; break }; "
		+ "$rows += @{pid=[int]$p.ProcessId;parent_pid=[int]$p.ParentProcessId; "
		+ "identity=([string]$p.CreationDate + '|' + [string]$p.CommandLine);commandline=[string]$p.CommandLine}; "
		+ "$current = [int]$p.ParentProcessId }; "
		+ "if ($capture_failed) { 'null' } else { ConvertTo-Json -InputObject @($rows) -Compress -Depth 3 }"
	) % pid


## Independent CIM maps in separate scriptblock scopes are
## captured within one shell invocation. No snapshot crosses a grant call.
static func _capture_windows_process_snapshot_pair(pid: int, diagnostics: Variant = null) -> Array:
	var query := _windows_process_snapshot_script(pid)
	var script := (
		"$first = & { %s }; $final = & { %s }; "
		+ "ConvertTo-Json -InputObject @([string]$first,[string]$final) -Compress"
	) % [query, query]
	var output: Array = []
	var began := Time.get_ticks_msec()
	var code := execute_windows_powershell(script, output)
	if code != 0 or output.is_empty():
		_snapshot_failure(
			diagnostics, "shell_exit" if code != 0 else "empty_output",
			"pair", -1, Time.get_ticks_msec() - began
		)
		return [{"capture_error": true}, {"capture_error": true}]
	return _parse_process_snapshot_pair(str(output[0]), pid, diagnostics)


static func _parse_process_snapshot_pair(
	raw: String, pid: int, diagnostics: Variant = null
) -> Array:
	if raw.length() > 4 * 1024 * 1024 + 16:
		_snapshot_failure(diagnostics, "outer_size", "pair")
		return [{"capture_error": true}, {"capture_error": true}]
	var json := JSON.new()
	if json.parse(raw) != OK:
		_snapshot_failure(diagnostics, "outer_json", "pair")
		return [{"capture_error": true}, {"capture_error": true}]
	var pair: Variant = json.data
	if not (pair is Array) or pair.size() != 2 or not (pair[0] is String and pair[1] is String):
		_snapshot_failure(diagnostics, "outer_shape", "pair")
		return [{"capture_error": true}, {"capture_error": true}]
	return [
		parse_process_snapshot(pair[0], pid, diagnostics, "first"),
		parse_process_snapshot(pair[1], pid, diagnostics, "final"),
	]


static func parse_process_snapshot(
	raw: String, expected_pid: int, diagnostics: Variant = null, stage := "single"
) -> Dictionary:
	if raw.length() > 1024 * 1024:
		return _snapshot_failure(diagnostics, "snapshot_size", stage, -1)
	var json := JSON.new()
	if json.parse(raw) != OK:
		return _snapshot_failure(diagnostics, "snapshot_json", stage, -1)
	var rows: Variant = json.data
	if not (rows is Array) or rows.size() > 16:
		return _snapshot_failure(diagnostics, "collector_null" if rows == null else "snapshot_shape", stage)
	var snapshot := {}
	var next_pid := expected_pid
	for row in rows:
		if not (row is Dictionary):
			return _snapshot_failure(diagnostics, "row_shape", stage, snapshot.size())
		var value: Variant = row.get("pid")
		if (
			not (value is int or value is float) or not is_finite(float(value))
			or float(value) <= 1.0 or float(value) > 4294967295.0
			or float(value) != float(int(value))
		):
			return _snapshot_failure(diagnostics, "row_pid", stage, snapshot.size())
		var pid := int(value)
		if pid <= 1 or pid != next_pid or snapshot.has(pid):
			return _snapshot_failure(diagnostics, "row_chain", stage, snapshot.size())
		snapshot[pid] = row
		if _process_snapshot_row(snapshot, pid).is_empty():
			return _snapshot_failure(diagnostics, "row_identity", stage, snapshot.size() - 1)
		next_pid = int(row.parent_pid)
		if snapshot.has(next_pid):
			return _snapshot_failure(diagnostics, "lineage_cycle", stage, snapshot.size() - 1)
	return snapshot


## Diagnostics never carry proof rows, command lines, or shell output. The
## caller owns this sink: at most eight distinct failures, with bounded counts.
## Snapshots and authority remain unchanged.
static func _snapshot_failure(
	diagnostics: Variant, category: String, stage: String, depth := -1, elapsed_ms := -1
) -> Dictionary:
	stage = stage if stage in SNAPSHOT_DIAGNOSTIC_STAGES else "unknown"
	category = category if category in SNAPSHOT_DIAGNOSTIC_CATEGORIES else "unknown"
	if diagnostics is Array:
		for item in diagnostics:
			if item.category == category and item.stage == stage and item.depth == depth:
				item.count = mini(int(item.count) + 1, 10000)
				return {"capture_error": true}
		if diagnostics.size() < 8:
			diagnostics.append({
				"category": category, "stage": stage, "depth": clampi(depth, -1, 16),
				"elapsed_ms": clampi(elapsed_ms, -1, 600000), "count": 1,
			})
	return {"capture_error": true}



## An unavailable capture is not evidence that its PID has exited.
static func capture_failed(snapshot: Variant) -> bool:
	return snapshot is Dictionary and bool(snapshot.get("capture_error", false))


static func _process_snapshot_row(snapshot: Variant, pid: int) -> Dictionary:
	if not (snapshot is Dictionary) or capture_failed(snapshot) or pid <= 1:
		return {}
	var row: Variant = snapshot.get(pid)
	if not (row is Dictionary) or row.size() != 4:
		return {}
	var parent: Variant = row.get("parent_pid")
	if (
		row.get("pid") != pid
		or not (parent is int or parent is float)
		or not is_finite(float(parent)) or float(parent) < 0.0 or float(parent) > 4294967295.0
		or float(parent) != float(int(parent))
		or not (row.get("identity") is String) or str(row.identity).strip_edges().is_empty()
		or not (row.get("commandline") is String)
	):
		return {}
	var identity := str(row.identity)
	var commandline := str(row.commandline)
	var separator := identity.find("|")
	if separator >= 0:
		if separator == 0 or identity.substr(separator + 1) != commandline:
			return {}
	elif not identity.is_valid_int() or not commandline.is_empty() or int(parent) != 0:
		return {}
	return row


static func process_commandline(pid: int, snapshot: Variant = null) -> String:
	if pid <= 1:
		return ""
	if snapshot != null:
		return str(_process_snapshot_row(snapshot, pid).get("commandline", ""))
	if OS.get_name() == "Windows":
		var output: Array = []
		var script := (
			"Get-CimInstance Win32_Process -Filter 'ProcessId = %d' | "
			+ "Select-Object -ExpandProperty CommandLine"
		) % pid
		if execute_windows_powershell(script, output) != 0 or output.is_empty():
			return ""
		return str(output[0]).strip_edges()
	if OS.get_name() == "Linux":
		return LinuxProc.commandline(pid)
	var proc_path := "/proc/%d/cmdline" % pid
	if FileAccess.file_exists(proc_path):
		var file := FileAccess.open(proc_path, FileAccess.READ)
		if file != null:
			var bytes := PackedByteArray()
			while bytes.size() < (1 << 20):
				var chunk := file.get_buffer(4096)
				if chunk.is_empty():
					break
				bytes.append_array(chunk)
				if file.eof_reached():
					break
			file.close()
			for index in range(bytes.size()):
				if bytes[index] == 0:
					bytes[index] = 0x20
			return bytes.get_string_from_utf8().strip_edges()
	var output: Array = []
	if OS.execute("ps", ["-ww", "-p", str(pid), "-o", "args="], output, true) != 0:
		return ""
	return str(output[0]).strip_edges() if not output.is_empty() else ""


static func process_parent(pid: int, snapshot: Variant = null) -> int:
	if pid <= 1:
		return 0
	if snapshot != null:
		return int(_process_snapshot_row(snapshot, pid).get("parent_pid", 0))
	if OS.get_name() == "Linux":
		return int(LinuxProc.process_stat(pid).get("parent_pid", 0))
	var output: Array = []
	if OS.get_name() == "Windows":
		var script := (
			"Get-CimInstance Win32_Process -Filter 'ProcessId = %d' | "
			+ "Select-Object -ExpandProperty ParentProcessId"
		) % pid
		if execute_windows_powershell(script, output) != 0 or output.is_empty():
			return 0
	else:
		if OS.execute("ps", ["-o", "ppid=", "-p", str(pid)], output, true) != 0 or output.is_empty():
			return 0
	var raw := str(output[0]).strip_edges()
	return int(raw) if raw.is_valid_int() else 0


static func process_descends_from(pid: int, ancestor_pid: int, snapshot: Variant = null) -> bool:
	if pid <= 1 or ancestor_pid <= 1:
		return false
	if snapshot != null and _process_snapshot_row(snapshot, pid).is_empty():
		return false
	var current := pid
	for _depth in range(16):
		if snapshot != null and _process_snapshot_row(snapshot, current).is_empty():
			return false
		if current == ancestor_pid:
			return true
		var parent := process_parent(current, snapshot)
		if parent <= 1 or parent == current:
			return false
		current = parent
	return false


static func commandline_is_godot_ai_server(commandline: String) -> bool:
	if commandline.is_empty():
		return false
	var lower := commandline.to_lower()
	var expression := RegEx.new()
	var brand_search := lower
	## Paths we choose ourselves must not satisfy the brand: strip the pid-file
	## and startup-report values (they live under user:// and carry our name)
	## before searching for it.
	if expression.compile("--(pid-file|startup-report)(?:=|\\s+)\\S+") == OK:
		brand_search = expression.sub(lower, "--$1 ", true)
	var branded := brand_search.contains("godot-ai") or brand_search.contains("godot_ai")
	return branded and (lower.contains("--pid-file") or lower.contains("--transport"))


static func pid_cmdline_is_godot_ai(pid: int, snapshot: Variant = null) -> bool:
	var current := pid
	for _depth in range(5):
		if current <= 1:
			return false
		if commandline_is_godot_ai_server(process_commandline(current, snapshot)):
			return true
		current = process_parent(current, snapshot)
	return false


static func process_fingerprint(pid: int, snapshot: Variant = null) -> String:
	if snapshot == null and OS.get_name() == "Linux":
		snapshot = capture_process_snapshot(pid)
	if not pid_alive(pid, snapshot):
		return ""
	var output: Array = []
	var identity := ""
	if snapshot != null:
		identity = str(_process_snapshot_row(snapshot, pid).get("identity", "")).strip_edges()
	elif OS.get_name() == "Windows":
		var script := (
			"Get-CimInstance Win32_Process -Filter 'ProcessId = %d' | "
			+ "ForEach-Object { [string]$_.CreationDate + '|' + [string]$_.CommandLine }"
		) % pid
		if execute_windows_powershell(script, output) == 0 and not output.is_empty():
			identity = str(output[0]).strip_edges()
		if identity.is_empty():
			## Some restricted Windows hosts expose process start time through
			## Get-Process while CIM returns no Win32_Process projection. PID +
			## creation time is the reuse-resistant identity; command-line brand
			## is checked independently before destructive authority is minted.
			output.clear()
			script = (
				"(Get-Process -Id %d -ErrorAction Stop).StartTime.ToFileTimeUtc()"
			) % pid
			if execute_windows_powershell(script, output) == 0 and not output.is_empty():
				identity = str(output[0]).strip_edges()
	else:
		if OS.execute(
			"ps", ["-ww", "-p", str(pid), "-o", "lstart=", "-o", "args="], output, true
		) == 0 and not output.is_empty():
			identity = str(output[0]).strip_edges()
	if identity.is_empty():
		return ""
	var context := HashingContext.new()
	context.start(HashingContext.HASH_SHA256)
	context.update((str(pid) + "|" + identity).to_utf8_buffer())
	return context.finish().hex_encode()


## Capture the process identity that a later destructive call must present.
## A PID or command-line brand alone is never kill authority: either can refer
## to a different process by the time a worker reaches the effect boundary.
## `diagnostics`, when given, receives one word naming the check that
## refused the grant. It costs no extra probe and never changes the result.
static func capture_process_kill_grant(
	pid: int, require_brand := false, diagnostics: Array = [], snapshot_diagnostics: Variant = null
) -> Dictionary:
	if pid <= 1 or pid == OS.get_process_id():
		diagnostics.append("invalid_pid")
		return {}
	var pair: Array = _capture_windows_process_snapshot_pair(pid, snapshot_diagnostics) if OS.get_name() == "Windows" else []
	var first: Variant = pair[0] if not pair.is_empty() else capture_process_snapshot(pid)
	if capture_failed(first):
		diagnostics.append("identity_unavailable")
		return {}
	if not pid_alive(pid, first):
		diagnostics.append("not_alive")
		return {}
	if require_brand and not pid_cmdline_is_godot_ai(pid, first):
		diagnostics.append("unbranded")
		return {}
	var fingerprint := process_fingerprint(pid, first)
	if fingerprint.is_empty():
		diagnostics.append("fingerprint_unavailable")
		return {}
	## Close the capture window: both identity and optional lineage/brand must
	## still describe the same process in the independently collected final snapshot.
	var final: Variant = pair[1] if not pair.is_empty() else capture_process_snapshot(pid)
	if capture_failed(final):
		diagnostics.append("identity_unavailable")
		return {}
	if process_fingerprint(pid, final) != fingerprint:
		diagnostics.append("fingerprint_changed")
		return {}
	if require_brand and not pid_cmdline_is_godot_ai(pid, final):
		diagnostics.append("brand_changed")
		return {}
	return {"pid": pid, "fingerprint": fingerprint}


## Kill only exact process grants and re-read identity immediately before the
## OS call. On Windows, taskkill /T already handles real descendants; do not
## append command-line lookalikes such as spoofed multiprocessing children.
static func kill_exact_processes(
	grants: Array[Dictionary], require_brand := false, require_tree_proof := false
) -> Array[int]:
	var killed: Array[int] = []
	var seen: Array[int] = []
	for grant in grants:
		if grant.keys().size() != 2 or not grant.has("pid") or not grant.has("fingerprint"):
			continue
		var pid := int(grant.get("pid", 0))
		var fingerprint := str(grant.get("fingerprint", ""))
		if pid <= 1 or pid == OS.get_process_id() or seen.has(pid) or fingerprint.is_empty():
			continue
		var snapshot: Variant = capture_process_snapshot(pid)
		if (
			process_fingerprint(pid, snapshot) != fingerprint
			or (require_brand and not pid_cmdline_is_godot_ai(pid, snapshot))
		):
			continue
		seen.append(pid)
		if OS.get_name() == "Windows":
			var output: Array = []
			var code := OS.execute("taskkill", ["/PID", str(pid), "/T", "/F"], output, true)
			## A vanished direct PID is enough for ordinary exact-process callers.
			## Timeout safety needs stronger evidence: only taskkill /T success
			## proves that Windows accepted termination of the whole child tree.
			if code == 0 or (not require_tree_proof and not pid_alive(pid)):
				killed.append(pid)
		elif OS.kill(pid) == OK or not pid_alive(pid):
			killed.append(pid)
	return killed


## Poll until the given port is no longer bound, or the timeout elapses.
## Used after `OS.kill` so we don't race the port-in-use check on rebind.
static func wait_for_port_free(port: int, timeout_s: float) -> PortOccupancy:
	var deadline := Time.get_ticks_msec() + int(timeout_s * 1000.0)
	while true:
		var occupancy := windows_port_occupancy(port) if OS.get_name() == "Windows" else (PortOccupancy.OCCUPIED if is_port_in_use(port) else PortOccupancy.FREE)
		if occupancy != PortOccupancy.OCCUPIED:
			return occupancy
		if Time.get_ticks_msec() >= deadline:
			push_warning("MCP | port %d still in use after %.1fs — proceeding anyway" % [port, timeout_s])
			return occupancy
		OS.delay_msec(100)
	return PortOccupancy.UNKNOWN


## Choose a non-Windows-reserved WS port. Returns `configured` when free;
## otherwise the first non-excluded port within `span` of it. Optional
## `log_buffer` is a duck-typed sink (`log(String)`) that gets the
## remap notice so users see why the port shifted.
static func resolve_ws_port(configured: int, max_port: int, log_buffer = null) -> int:
	var resolved := WindowsPortReservation.suggest_non_excluded_port(
		configured,
		2048,
		max_port
	)
	if resolved != configured:
		var message := "WebSocket port %d is reserved by Windows; using %d" % [configured, resolved]
		print("MCP | %s" % message)
		if log_buffer != null:
			log_buffer.log(message)
	return resolved


static func resolve_ws_port_from_output(
	configured_port: int,
	netsh_output: String,
	max_port: int,
	span: int = 2048
) -> int:
	return WindowsPortReservation.suggest_non_excluded_port_from_output(
		netsh_output,
		configured_port,
		span,
		max_port
	)