godot ai 更新

addons/godot_ai/utils/client_mutation_lock.gd

@@ -102,6 +102,14 @@ static func _root_path() -> String:
 	## Unlike user://, the OS config directory is shared by every Godot
 	## project owned by this account. User-scope client config is shared too,
 	## so a project-specific lock would leave the real cross-editor race open.
+	##
+	## A Flatpak editor that shares the home directory writes the host's client
+	## configs, the same files an editor outside the sandbox writes, while its
+	## own OS config directory is the per-app ~/.var/app/<id>/config. It takes
+	## the lock where those other editors look for it.
+	var host_config_home := McpPathTemplate.flatpak_host_config_home()
+	if not host_config_home.is_empty():
+		return host_config_home.path_join(ROOT_DIR_NAME)
 	return OS.get_config_dir().path_join(ROOT_DIR_NAME)
 
 

addons/godot_ai/utils/error_codes.gd

@@ -26,6 +26,11 @@ const EDITOR_NOT_READY := "EDITOR_NOT_READY"
 const UNKNOWN_COMMAND := "UNKNOWN_COMMAND"
 const INTERNAL_ERROR := "INTERNAL_ERROR"
 const DEFERRED_TIMEOUT := "DEFERRED_TIMEOUT"
+## #1120: filesystem mutation discovery failed before any disk effect
+## (deadline/cancellation, entry/byte limits, or unreadable owner metadata).
+## Keep this separate from INVALID_PARAMS so callers can distinguish an
+## environmental discovery failure from a bad filesystem_manage request.
+const FILESYSTEM_DISCOVERY_FAILED := "FILESYSTEM_DISCOVERY_FAILED"
 ## Python-originated transport/attach bridge codes. GDScript has no emit path,
 ## but the public registry intentionally mirrors protocol/errors.py.
 const TRANSPORT_OUTCOME_UNKNOWN := "TRANSPORT_OUTCOME_UNKNOWN"

addons/godot_ai/utils/port_resolver.gd

@@ -21,6 +21,12 @@ const SNAPSHOT_DIAGNOSTIC_CATEGORIES := [
 	"snapshot_size", "snapshot_json", "snapshot_shape", "collector_null", "row_shape",
 	"row_pid", "row_chain", "row_identity", "lineage_cycle",
 ]
+## Public lifecycle evidence uses a smaller, stable vocabulary than the
+## internal collector diagnostics. Keep the detailed category internally so a
+## future fix can distinguish malformed JSON from an inaccessible ancestor.
+const SNAPSHOT_FAILURE_CATEGORIES := [
+	"process_query", "json_shape", "ancestor_capture", "identity_mismatch",
+]
 static var _process_spawn_mutex := Mutex.new()
 
 enum PortOccupancy { UNKNOWN, FREE, OCCUPIED }
@@ -531,6 +537,39 @@ static func _snapshot_failure(
 	return {"capture_error": true}
 
 
+## Map one internal collector refusal to the bounded category retained in the
+## editor log. `row_identity` at depth zero is malformed target metadata;
+## the same refusal on an ancestor is an ancestor-capture failure. A refusal
+## this table does not know stays "unknown" instead of borrowing a category.
+static func snapshot_failure_category(detail: String, depth := -1) -> String:
+	match detail:
+		"shell_exit", "empty_output", "invalid_pid", "collector_null":
+			return "process_query"
+		"lineage_cycle":
+			return "ancestor_capture"
+		"row_identity":
+			return "ancestor_capture" if depth > 0 else "json_shape"
+		"outer_size", "outer_json", "outer_shape", "snapshot_size", "snapshot_json", \
+		"snapshot_shape", "row_shape", "row_pid", "row_chain":
+			return "json_shape"
+	return "unknown"
+
+
+## Return only the creation-time portion of a captured identity. Windows rows
+## append the command line after `|`; Linux rows use `linux:<start>|<command>`.
+## The command line is intentionally never copied into diagnostics.
+static func process_creation_identity(pid: int, snapshot: Variant = null) -> String:
+	if pid <= 1 or not (snapshot is Dictionary) or capture_failed(snapshot):
+		return ""
+	var identity := str(_process_snapshot_row(snapshot, pid).get("identity", "")).strip_edges()
+	if identity.is_empty():
+		return ""
+	var separator := identity.find("|")
+	if separator > 0:
+		identity = identity.substr(0, separator).strip_edges()
+	return identity.substr(0, 128)
+
+
 
 ## An unavailable capture is not evidence that its PID has exited.
 static func capture_failed(snapshot: Variant) -> bool:

addons/godot_ai/utils/server_lifecycle.gd

@@ -51,6 +51,7 @@ const STALE_PRE_V4_HINT := "stale_pre_v4_server"
 const DEFAULT_PROVE_TIMEOUT_MS := 180_000
 const LAUNCH_FINGERPRINT_TIMEOUT_MS := 15_000
 const REPLACEMENT_TTL_MS := 15_000
+const IDENTITY_CAPTURE_STAGES := ["launch", "first_server", "final_server"]
 ## An authenticated endpoint that drops (a late keepalive, a server restart)
 ## is re-probed automatically with this backoff, each attempt re-proving the
 ## server exactly as a start does; after the last one the block stays until
@@ -126,6 +127,8 @@ func _begin_start_episode(existing_id := 0, probe := true) -> void:
 		"blocked_target": {},
 		"launch": {},
 		"after_stop": "",
+		"snapshot_diagnostic": {},
+		"proof_stage_attempts": {"launch": 0, "first_server": 0, "final_server": 0},
 	}
 	_publish()
 	if not probe:
@@ -461,6 +464,7 @@ func _complete_launch(result: Dictionary) -> void:
 
 
 func _complete_prove(result: Dictionary) -> void:
+	_remember_snapshot_evidence(result)
 	if bool(result.get("pending", false)):
 		_episode["proof_pending_reason"] = str(result.get("reason", "unknown"))
 		if Time.get_ticks_msec() >= int(_episode.get("prove_deadline_msec", 0)):
@@ -468,6 +472,7 @@ func _complete_prove(result: Dictionary) -> void:
 				"proof_timeout",
 				"The managed server proof timed out at %s."
 				% str(_episode.get("proof_pending_reason", "unknown"))
+				+ _snapshot_diagnostic_sentence()
 				+ startup_report_summary(str(_plan.get("startup_report", ""))),
 			)
 			return
@@ -478,6 +483,7 @@ func _complete_prove(result: Dictionary) -> void:
 		var pending_reason := str(_episode.get("proof_pending_reason", ""))
 		if not pending_reason.is_empty():
 			message += " Last pending proof: %s." % pending_reason
+		message += _snapshot_diagnostic_sentence()
 		_block(str(result.get("reason", "proof_failed")), message)
 		return
 	var launch: Dictionary = _episode.get("launch", {})
@@ -485,7 +491,11 @@ func _complete_prove(result: Dictionary) -> void:
 	var fingerprint := str(result.get("fingerprint", ""))
 	var exact_grant = _owned_process_grant(pid, fingerprint)
 	if not exact_grant.is_valid():
-		_block("process_proof_failed", "The server process identity changed before proof completed.")
+		_block(
+			"process_proof_failed",
+			"The server process identity changed before proof completed."
+			+ _snapshot_diagnostic_sentence(),
+		)
 		return
 	_process_grant = exact_grant
 	launch["pid"] = pid
@@ -861,8 +871,9 @@ func _effect_launch(payload: Dictionary) -> Dictionary:
 		"ok": not fingerprint.is_empty(),
 		"reason": "launch_unproven" if fingerprint.is_empty() else "",
 		"message": (
-			_launch_unproven_message(pid, attempts, Time.get_ticks_msec() - capture_started)
-			+ _snapshot_diagnostic_summary(snapshot_diagnostics)
+			_launch_unproven_message(
+				pid, attempts, Time.get_ticks_msec() - capture_started, snapshot_diagnostics
+			)
 			if fingerprint.is_empty()
 			else ""
 		),
@@ -896,12 +907,13 @@ static func _snapshot_diagnostic_summary(diagnostics: Array) -> String:
 ## (#988, #1012 both surfaced as this bare sentence). Diagnostic only: the
 ## values are read once more after the capture budget and never grant
 ## authority.
-func _launch_unproven_message(pid: int, attempts: Array, elapsed_ms: int) -> String:
-	var snapshot: Variant = _capture_process_snapshot(pid)
+func _launch_unproven_message(
+	pid: int, attempts: Array, elapsed_ms: int, snapshot_diagnostics: Array = []
+) -> String:
+	var capture_diagnostics: Array = []
+	var snapshot: Variant = _capture_process_snapshot(pid, capture_diagnostics)
 	var alive := PortResolver.pid_alive(pid, snapshot)
-	var commandline := PortResolver.process_commandline(pid, snapshot) if alive else ""
-	if commandline.length() > 200:
-		commandline = commandline.substr(0, 199) + "…"
+	var creation_identity := PortResolver.process_creation_identity(pid, snapshot)
 	## Summarise the per-attempt refusals as "reason×count" in first-seen order.
 	var counts := {}
 	var order: Array[String] = []
@@ -914,43 +926,218 @@ func _launch_unproven_message(pid: int, attempts: Array, elapsed_ms: int) -> Str
 	var summary: Array[String] = []
 	for key in order:
 		summary.append("%s×%d" % [key, int(counts[key])])
-	return (
+	var message := (
 		"The launched process identity could not be captured in %d attempts over %.1f s "
-		+ "(pid %d, now alive=%s, refusals: %s%s)."
+		+ "(pid %d, creation_identity=%s, now alive=%s, refusals: %s)."
 	) % [
 		attempts.size(),
 		elapsed_ms / 1000.0,
 		pid,
+		"unavailable" if creation_identity.is_empty() else creation_identity,
 		"unknown" if PortResolver.capture_failed(snapshot) else ("yes" if alive else "no"),
 		", ".join(summary) if not summary.is_empty() else "none recorded",
-		"" if commandline.is_empty() else "; command: " + commandline,
+	]
+	return message + _snapshot_diagnostic_summary(snapshot_diagnostics + capture_diagnostics)
+
+
+func _capture_process_snapshot(pid: int, diagnostics: Variant = null) -> Variant:
+	return PortResolver.capture_process_snapshot(pid, diagnostics)
+
+
+func _remember_snapshot_evidence(result: Dictionary) -> void:
+	var diagnostic: Variant = result.get("snapshot_diagnostic")
+	if diagnostic is Dictionary:
+		_episode["snapshot_diagnostic"] = diagnostic.duplicate(true)
+	var attempts: Variant = result.get("proof_stage_attempts")
+	if attempts is Dictionary:
+		_episode["proof_stage_attempts"] = attempts.duplicate(true)
+
+
+func _snapshot_diagnostic_sentence() -> String:
+	var diagnostic: Variant = _episode.get("snapshot_diagnostic", {})
+	if not (diagnostic is Dictionary) or diagnostic.is_empty():
+		return ""
+	var stage := str(diagnostic.get("stage", ""))
+	if stage not in IDENTITY_CAPTURE_STAGES:
+		stage = "unknown"
+	var category := str(diagnostic.get("category", ""))
+	if category not in PortResolver.SNAPSHOT_FAILURE_CATEGORIES:
+		category = "unknown"
+	var identity := str(diagnostic.get("creation_identity", "")).strip_edges()
+	if identity.is_empty():
+		identity = "unavailable"
+	identity = identity.get_slice("|", 0).replace("\n", " ").replace("\r", " ").substr(0, 128)
+	## The collector's own refusal, when one was recorded: the fixed category
+	## alone cannot tell a shell that exited from one that printed nothing.
+	var detail := str(diagnostic.get("detail", ""))
+	if not detail.is_empty():
+		if detail not in PortResolver.SNAPSHOT_DIAGNOSTIC_CATEGORIES:
+			detail = "unknown"
+		category += " detail=%s" % detail
+		var depth := int(diagnostic.get("depth", -1))
+		if depth >= 0:
+			category += " depth=%d" % clampi(depth, 0, 16)
+	return (
+		" Identity capture: stage=%s category=%s pid=%d creation_identity=%s attempt=%d elapsed_ms=%d."
+	) % [
+		stage,
+		category,
+		int(diagnostic.get("pid", -1)),
+		identity,
+		clampi(int(diagnostic.get("attempt", 1)), 1, 100000),
+		clampi(int(diagnostic.get("elapsed_ms", 0)), 0, 600000),
 	]
 
 
-func _capture_process_snapshot(pid: int) -> Variant:
-	return PortResolver.capture_process_snapshot(pid)
+func _capture_identity(
+	stage: String, pid: int, stage_attempts: Dictionary
+) -> Dictionary:
+	var attempt := int(stage_attempts.get(stage, 0)) + 1
+	stage_attempts[stage] = attempt
+	var diagnostics: Array = []
+	var began := Time.get_ticks_msec()
+	var snapshot: Variant = _capture_process_snapshot(pid, diagnostics)
+	return {
+		"snapshot": snapshot,
+		"diagnostic": _snapshot_evidence(
+			stage, pid, snapshot, diagnostics, attempt, Time.get_ticks_msec() - began
+		),
+	}
+
+
+func _record_snapshot_capture(
+	stage: String, pid: int, snapshot: Variant, stage_attempts: Dictionary, elapsed_ms: int = 0
+) -> Dictionary:
+	var attempt := int(stage_attempts.get(stage, 0)) + 1
+	stage_attempts[stage] = attempt
+	return {
+		"snapshot": snapshot,
+		"diagnostic": _snapshot_evidence(stage, pid, snapshot, [], attempt, elapsed_ms),
+	}
+
+
+static func _capture_elapsed_ms(capture: Dictionary) -> int:
+	var diagnostic: Variant = capture.get("diagnostic", {})
+	return int(diagnostic.get("elapsed_ms", 0)) if diagnostic is Dictionary else 0
+
+
+static func _retarget_snapshot_capture(
+	capture: Dictionary, stage: String, pid: int
+) -> Dictionary:
+	var snapshot: Variant = capture.get("snapshot")
+	var old_diagnostic: Variant = capture.get("diagnostic", {})
+	var attempt := int(old_diagnostic.get("attempt", 1)) if old_diagnostic is Dictionary else 1
+	var elapsed_ms := int(old_diagnostic.get("elapsed_ms", 0)) if old_diagnostic is Dictionary else 0
+	return {
+		"snapshot": snapshot,
+		"diagnostic": _snapshot_evidence(stage, pid, snapshot, [], attempt, elapsed_ms),
+	}
+
+
+static func _snapshot_evidence(
+	stage: String,
+	pid: int,
+	snapshot: Variant,
+	diagnostics: Array,
+	attempt: int,
+	elapsed_ms: int,
+	category_override := "",
+) -> Dictionary:
+	var detail := ""
+	var depth := -1
+	if not diagnostics.is_empty():
+		var last: Variant = diagnostics[diagnostics.size() - 1]
+		if last is Dictionary:
+			detail = str(last.get("category", ""))
+			depth = int(last.get("depth", -1))
+	var category := str(category_override)
+	if category.is_empty() and not detail.is_empty():
+		category = PortResolver.snapshot_failure_category(detail, depth)
+	elif category.is_empty() and PortResolver.capture_failed(snapshot):
+		category = "process_query"
+	var creation_identity := PortResolver.process_creation_identity(pid, snapshot)
+	return {
+		"stage": stage if stage in IDENTITY_CAPTURE_STAGES else "unknown",
+		"category": (
+			category
+			if category.is_empty() or category in PortResolver.SNAPSHOT_FAILURE_CATEGORIES
+			else "unknown"
+		),
+		"detail": (
+			detail
+			if detail.is_empty() or detail in PortResolver.SNAPSHOT_DIAGNOSTIC_CATEGORIES
+			else "unknown"
+		),
+		"depth": clampi(depth, -1, 16),
+		"pid": pid,
+		"creation_identity": "unavailable" if creation_identity.is_empty() else creation_identity,
+		"attempt": clampi(attempt, 1, 100000),
+		"elapsed_ms": clampi(elapsed_ms, 0, 600000),
+	}
+
+
+func _proof_pending(
+	reason: String, stage_attempts: Dictionary, diagnostic: Dictionary = {}
+) -> Dictionary:
+	var result := {
+		"pending": true,
+		"reason": reason,
+		"proof_stage_attempts": stage_attempts.duplicate(true),
+	}
+	if not diagnostic.is_empty():
+		result["snapshot_diagnostic"] = diagnostic.duplicate(true)
+	return result
+
+
+func _capture_failure(
+	stage_attempts: Dictionary, capture: Dictionary, category_override := ""
+) -> Dictionary:
+	var diagnostic: Dictionary = capture.get("diagnostic", {}).duplicate(true)
+	if not category_override.is_empty():
+		diagnostic["category"] = category_override
+	return _proof_pending("identity_unavailable", stage_attempts, diagnostic)
+
+
+func _identity_mismatch(
+	reason: String, stage_attempts: Dictionary, capture: Dictionary
+) -> Dictionary:
+	var diagnostic: Dictionary = capture.get("diagnostic", {}).duplicate(true)
+	diagnostic["category"] = "identity_mismatch"
+	return _proof_pending(reason, stage_attempts, diagnostic)
 
 
 func _effect_prove(payload: Dictionary) -> Dictionary:
 	var launch_grant = payload.get("grant")
 	var launch_pid := int(launch_grant.process_id()) if launch_grant != null else -1
+	var stage_attempts: Variant = payload.get("proof_stage_attempts", {})
+	if not (stage_attempts is Dictionary):
+		stage_attempts = {}
+	else:
+		stage_attempts = stage_attempts.duplicate(true)
+	for stage in IDENTITY_CAPTURE_STAGES:
+		stage_attempts[stage] = maxi(0, int(stage_attempts.get(stage, 0)))
 	## The pid file is an untrusted hint. Reuse its ancestor snapshot only
 	## when it includes the launcher; the original PID read and all final
 	## capability, listener, identity, and lineage checks still follow.
 	var hinted_pid := -1
 	var first_snapshot: Variant = null
 	var launch_snapshot: Variant = null
+	var launch_capture: Dictionary = {}
+	var first_capture: Dictionary = {}
 	if OS.get_name() == "Windows" and launch_pid > 1:
 		hinted_pid = PortResolver.read_pid_file(str(payload.get("pid_file", "")))
 		if hinted_pid > 1:
-			first_snapshot = _capture_process_snapshot(hinted_pid)
+			first_capture = _capture_identity("launch", hinted_pid, stage_attempts)
+			first_snapshot = first_capture.get("snapshot")
 			if PortResolver.process_descends_from(hinted_pid, launch_pid, first_snapshot):
+				launch_capture = _retarget_snapshot_capture(first_capture, "launch", launch_pid)
 				launch_snapshot = first_snapshot
 	if launch_snapshot == null:
-		launch_snapshot = _capture_process_snapshot(launch_pid)
+		launch_capture = _capture_identity("launch", launch_pid, stage_attempts)
+		launch_snapshot = launch_capture.get("snapshot")
 		first_snapshot = null
 	if PortResolver.capture_failed(launch_snapshot):
-		return {"pending": true, "reason": "identity_unavailable"}
+		return _capture_failure(stage_attempts, launch_capture)
 	var launch_alive := PortResolver.pid_alive(launch_pid, launch_snapshot)
 	var launch_fingerprint := (
 		PortResolver.process_fingerprint(launch_pid, launch_snapshot)
@@ -963,72 +1150,103 @@ func _effect_prove(payload: Dictionary) -> Dictionary:
 		launch_fingerprint,
 	)
 	if launch_disposition != "owned":
-		return {
-			"ok": false,
-			"reason": "launch_%s" % launch_disposition,
-			"message": (
-				"The launched server process exited or changed identity before publishing capabilities."
-				+ startup_report_summary(str(payload.get("startup_report", "")))
-			),
-		}
+		if launch_disposition == "gone":
+			return {
+				"ok": false,
+				"reason": "launch_gone",
+				"message": (
+					"The launched server process exited or changed identity before publishing capabilities."
+					+ startup_report_summary(str(payload.get("startup_report", "")))
+				),
+				"proof_stage_attempts": stage_attempts.duplicate(true),
+			}
+		var mismatch := _identity_mismatch("launch_%s" % launch_disposition, stage_attempts, launch_capture)
+		mismatch.erase("pending")
+		mismatch["ok"] = false
+		mismatch["message"] = (
+			"The launched server process exited or changed identity before publishing capabilities."
+			+ startup_report_summary(str(payload.get("startup_report", "")))
+		)
+		return mismatch
 	var port := int(payload.http_port)
 	var capability := _read_capability(port)
 	if capability.is_empty() or str(capability.get("instance_nonce", "")) == str(payload.get("baseline_instance_id", "")):
-		return {"pending": true, "reason": "capability_record"}
+		return _proof_pending("capability_record", stage_attempts)
 	if (
 		str(capability.get("http", "")) != str(payload.get("http_capability", ""))
 		or str(capability.get("websocket", "")) != str(payload.get("ws_capability", ""))
 	):
-		return {"pending": true, "reason": "capability_pair"}
+		return _proof_pending("capability_pair", stage_attempts)
 	var listeners := PortResolver.windows_listener_snapshot() if OS.get_name() == "Windows" else {}
 	var live := _probe_with_capability(port, capability, int(payload.timeout_ms), listeners)
 	if not _authenticated_status_matches_record(live, capability):
-		return {"pending": true, "reason": "authenticated_status"}
+		return _proof_pending("authenticated_status", stage_attempts)
 	var version := str(live.get("version", ""))
 	var ws_port := int(live.get("ws_port", 0))
 	if not _server_status_compatibility(
 		version, str(payload.expected_version), ws_port, int(payload.expected_ws_port)
 	).get("compatible", false):
-		return {"ok": false, "reason": "incompatible", "message": "The launched server does not match this plugin."}
+		return {
+			"ok": false,
+			"reason": "incompatible",
+			"message": "The launched server does not match this plugin.",
+			"proof_stage_attempts": stage_attempts.duplicate(true),
+		}
 	var pid := PortResolver.read_pid_file(str(payload.get("pid_file", "")))
 	if pid <= 1:
-		return {"pending": true, "reason": "pid_file"}
+		return _proof_pending("pid_file", stage_attempts)
 	if not PortResolver.find_all_pids_on_port(port, Callable(), listeners).has(pid):
-		return {"pending": true, "reason": "listener_pid"}
+		return _proof_pending("listener_pid", stage_attempts)
 	if first_snapshot == null or pid != hinted_pid:
-		first_snapshot = (
-			launch_snapshot if pid == launch_pid else _capture_process_snapshot(pid)
+		if pid == launch_pid:
+			first_snapshot = launch_snapshot
+			first_capture = _record_snapshot_capture(
+				"first_server", pid, first_snapshot, stage_attempts,
+				_capture_elapsed_ms(launch_capture)
+			)
+		else:
+			first_capture = _capture_identity("first_server", pid, stage_attempts)
+			first_snapshot = first_capture.get("snapshot")
+	else:
+		first_capture = _record_snapshot_capture(
+			"first_server", pid, first_snapshot, stage_attempts,
+			_capture_elapsed_ms(first_capture)
 		)
 	if PortResolver.capture_failed(first_snapshot):
-		return {"pending": true, "reason": "identity_unavailable"}
+		return _capture_failure(stage_attempts, first_capture)
 	if not PortResolver.pid_cmdline_is_godot_ai(pid, first_snapshot):
-		return {"pending": true, "reason": "process_brand"}
+		return _identity_mismatch("process_brand", stage_attempts, first_capture)
 	if not PortResolver.process_descends_from(pid, launch_pid, first_snapshot):
-		return {"pending": true, "reason": "launch_lineage"}
+		return _identity_mismatch("launch_lineage", stage_attempts, first_capture)
 	var fingerprint := PortResolver.process_fingerprint(pid, first_snapshot)
 	if fingerprint.is_empty():
-		return {"pending": true, "reason": "process_fingerprint"}
+		return _identity_mismatch("process_fingerprint", stage_attempts, first_capture)
 	## Close every capture window before minting process authority. The same
 	## launch-lineage PID must still own the listener, the private capability
 	## record must still name the authenticated instance, and the process
 	## fingerprint must remain unchanged after the final probe.
 	var final_capability := _read_capability(port)
 	var final_live := _probe_with_capability(port, final_capability, int(payload.timeout_ms), listeners)
-	var final_snapshot: Variant = _capture_process_snapshot(pid)
+	var final_capture := _capture_identity("final_server", pid, stage_attempts)
+	var final_snapshot: Variant = final_capture.get("snapshot")
 	if PortResolver.capture_failed(final_snapshot):
-		return {"pending": true, "reason": "identity_unavailable"}
+		return _capture_failure(stage_attempts, final_capture)
+	var final_fingerprint := PortResolver.process_fingerprint(pid, final_snapshot)
 	var final_listeners := PortResolver.windows_listener_snapshot() if OS.get_name() == "Windows" else {}
 	if (
 		PortResolver.read_pid_file(str(payload.get("pid_file", ""))) != pid
 		or not PortResolver.find_all_pids_on_port(port, Callable(), final_listeners).has(pid)
 		or not PortResolver.process_descends_from(pid, launch_pid, final_snapshot)
 		or not PortResolver.pid_cmdline_is_godot_ai(pid, final_snapshot)
-		or PortResolver.process_fingerprint(pid, final_snapshot) != fingerprint
-		or final_capability != capability
+		or final_fingerprint != fingerprint
+	):
+		return _identity_mismatch("final_capture_window", stage_attempts, final_capture)
+	if (
+		final_capability != capability
 		or not _authenticated_status_matches_record(final_live, final_capability)
 		or str(final_live.get("instance_id", "")) != str(live.get("instance_id", ""))
 	):
-		return {"pending": true, "reason": "final_capture_window"}
+		return _proof_pending("final_capture_window", stage_attempts)
 	var final_version := str(final_live.get("version", ""))
 	var final_ws_port := int(final_live.get("ws_port", 0))
 	if not _server_status_compatibility(
@@ -1041,6 +1259,7 @@ func _effect_prove(payload: Dictionary) -> Dictionary:
 			"ok": false,
 			"reason": "incompatible",
 			"message": "The launched server changed compatibility before authority was granted.",
+			"proof_stage_attempts": stage_attempts.duplicate(true),
 		}
 	return {
 		"ok": true,
@@ -1048,6 +1267,7 @@ func _effect_prove(payload: Dictionary) -> Dictionary:
 		"fingerprint": fingerprint,
 		"version": final_version,
 		"transport": _transport_from(port, final_ws_port, final_live, final_capability),
+		"proof_stage_attempts": stage_attempts.duplicate(true),
 	}
 
 
@@ -1289,6 +1509,9 @@ func _prove_payload() -> Dictionary:
 		"ws_capability": str(launch.get("ws_capability", "")),
 		"baseline_instance_id": str(launch.get("baseline_instance_id", "")),
 		"grant": _process_grant,
+		"proof_stage_attempts": _episode.get(
+			"proof_stage_attempts", {"launch": 0, "first_server": 0, "final_server": 0}
+		),
 	}
 
 
@@ -1742,7 +1965,8 @@ static func _dormant_episode(episode_id: int) -> Dictionary:
 		"id": episode_id, "state": DORMANT, "phase": "", "reason": "",
 		"message": "Server lifecycle is dormant.", "ready_kind": "", "effect": "",
 		"expected_version": "", "actual_version": "", "blocked_target": {},
-		"launch": {}, "after_stop": "",
+		"launch": {}, "after_stop": "", "snapshot_diagnostic": {},
+		"proof_stage_attempts": {"launch": 0, "first_server": 0, "final_server": 0},
 	}
 
 

addons/godot_ai/utils/transport_capability.gd

@@ -13,6 +13,19 @@ const _GROUP_OTHER_PERMISSION_MASK := 0x3f  ## 0077
 const _GROUP_OTHER_WRITE_MASK := 0x12  ## 0022
 const _MAX_LINK_HOPS := 8
 const _SYSTEM_TEMP_ROOTS: Array[String] = ["/tmp", "/private/tmp", "/var/tmp"]
+const _FLATPAK_INFO_PATH := "/.flatpak-info"
+## Every read-write spelling of the two grants that expose the home directory.
+## Flatpak 1.16 writes plain read-write as the bare name; `:create` is
+## read-write too. `:ro` is absent on purpose.
+const _FLATPAK_HOME_GRANTS: Array[String] = [
+	"host", "host:rw", "host:create", "home", "home:rw", "home:create",
+]
+## Root directories a `host` grant leaves out of the sandbox, where Flatpak's
+## own runtime lives (`dont_mount_in_root` in flatpak-context.c).
+const _FLATPAK_HOST_EXCLUDED: Array[String] = [
+	"lib", "lib32", "lib64", "bin", "sbin", "usr", "boot", "root",
+	"tmp", "etc", "app", "run", "proc", "sys", "dev", "var",
+]
 const _KEYS: Array[String] = [
 	"version", "http", "websocket", "instance_nonce",
 ]
@@ -322,10 +335,371 @@ static func path_for_http_port(http_port: int) -> String:
 				"Library/Application Support/godot-ai/capabilities"
 			)
 		else:
-			directory = OS.get_environment("XDG_CONFIG_HOME").strip_edges()
-			if directory.is_empty():
-				directory = OS.get_environment("HOME").path_join(".config")
-			directory = directory.path_join("godot-ai/capabilities")
+			directory = linux_config_home(_flatpak_info()).path_join("godot-ai/capabilities")
 	if directory.is_empty() or not directory.is_absolute_path():
 		return ""
 	return directory.simplify_path().path_join("http-%d.json" % http_port)
+
+
+## The config directory a Linux process outside this one would also name.
+## Flatpak points `XDG_CONFIG_HOME` at the app's own `~/.var/app/<id>/config`,
+## which a client outside that sandbox never reads. When the sandbox shares
+## the host's config directory, that is `HOST_XDG_CONFIG_HOME` if the host set
+## one, else the `~/.config` default. Python's `capability_directory` applies
+## the same rule, so the server publishes where this reads.
+static func linux_config_home(flatpak_info: String) -> String:
+	var config := OS.get_environment(linux_config_home_variable(flatpak_info)).strip_edges()
+	if config.is_empty():
+		config = OS.get_environment("HOME").path_join(".config")
+	return config
+
+
+## The variable that names that directory; unset or empty means `~/.config`.
+static func linux_config_home_variable(flatpak_info: String) -> String:
+	return (
+		"HOST_XDG_CONFIG_HOME" if flatpak_shares_config_home(flatpak_info) else "XDG_CONFIG_HOME"
+	)
+
+
+## Whether `/.flatpak-info` text describes a sandbox that can write the host's
+## config directory at its own path: one that shares the home, or holds a
+## read-write grant for the whole of `xdg-config`. Flatpak mounts that grant at
+## the host's path only and leaves `XDG_CONFIG_HOME` on the per-app directory.
+## An `xdg-config/<dir>` grant does not count: Flatpak mounts it inside the
+## per-app directory as well, so `XDG_CONFIG_HOME` already reaches it.
+static func flatpak_shares_config_home(flatpak_info: String) -> bool:
+	return (
+		flatpak_shares_home(flatpak_info)
+		or bool(_flatpak_filesystems(flatpak_info).get("xdg-config", false))
+	)
+
+
+## Whether `/.flatpak-info` text describes a sandbox that can write the host's
+## home directory. A read-write `host` or `home` grant exposes the real home at
+## its own path; a `:ro` grant, a narrower one, or none leaves the app's private
+## directories as the only ones a process outside the sandbox can also see.
+static func flatpak_shares_home(flatpak_info: String) -> bool:
+	var group := ""
+	for line in flatpak_info.split("\n"):
+		if line.begins_with("["):
+			group = line.strip_edges()
+		elif group == "[Context]" and line.begins_with("filesystems="):
+			for grant in line.trim_prefix("filesystems=").strip_edges().split(";"):
+				if grant in _FLATPAK_HOME_GRANTS:
+					return true
+			return false
+	return false
+
+
+## The grant in `/.flatpak-info` text that decides whether the sandbox shares
+## `path` with the host, so that a process outside it finds the same file at
+## the same location. Returned as {root, location, writable}: the directory or
+## file the grant covers, the grant as Flatpak spells it (`home`, `~/dir`,
+## `xdg-config/dir`), and whether it is read-write. {} when no grant covers
+## the path.
+##
+## `home` and `host` share the home directory, except `~/.var/app`: there
+## Flatpak hides every app's directory but the sandbox's own, whatever the
+## grant, and shares that one with no grant at all. `host` also shares the
+## root directories its runtime does not occupy. `~/dir` and `/dir` share that
+## location, and `xdg-config[/dir]` the host's config directory,
+## `host_config_home`. The narrowest grant decides, as the narrowest mount
+## does. The other `xdg-` names share nothing here: where they lie depends on
+## host settings the sandbox cannot read, so their paths fail closed.
+##
+## A grant says what Flatpak was asked for, not what it mounted. It skips a
+## location that does not exist when the app starts, and `/.flatpak-info`
+## leaves out denials (`--nofilesystem`). `flatpak_blocking_mount` checks the
+## answer against the mounts.
+static func flatpak_grant_over(
+	flatpak_info: String, path: String, home: String, host_config_home: String
+) -> Dictionary:
+	var target := path.simplify_path()
+	if not target.is_absolute_path():
+		return {}
+	var grants := _flatpak_filesystems(flatpak_info)
+	var found := {}
+	for location in grants:
+		var root := _flatpak_grant_path(str(location), home, host_config_home)
+		if root.is_empty() or not path_is_within(root, target):
+			continue
+		var covered := str(found.get("root", ""))
+		if root.length() > covered.length():
+			found = {"root": root, "location": str(location), "writable": bool(grants[location])}
+		elif root == covered and found["writable"] and not grants[location]:
+			## Two spellings of one location: Flatpak mounts whichever it
+			## reaches last, so only agreement counts as writable.
+			found = {"root": root, "location": str(location), "writable": false}
+	var home_dir := home.simplify_path()
+	var top := target.get_slice("/", 1)
+	var wide := {}
+	if not home.is_empty() and path_is_within(home_dir, target):
+		var apps_dir := home_dir.path_join(".var/app")
+		var app_id := flatpak_application_id(flatpak_info)
+		if path_is_within(apps_dir, target):
+			if not app_id.is_empty() and path_is_within(apps_dir.path_join(app_id), target):
+				wide = {
+					"root": apps_dir.path_join(app_id),
+					"location": apps_dir.path_join(app_id),
+					"writable": true,
+				}
+		elif grants.has("home") or grants.has("host"):
+			## Flatpak applies the wider of the two modes to the home directory,
+			## so a read-write `home` is always enough for a path inside it.
+			wide = {
+				"root": home_dir,
+				"location": "home",
+				"writable": bool(grants.get("home", false)) or bool(grants.get("host", false)),
+			}
+	elif grants.has("host") and not top.is_empty() and top not in _FLATPAK_HOST_EXCLUDED:
+		wide = {"root": "/" + top, "location": "host", "writable": bool(grants["host"])}
+	if str(wide.get("root", "")).length() > str(found.get("root", "")).length():
+		return wide
+	return found
+
+
+## The sandboxed app's ID from `/.flatpak-info` text, or "" outside Flatpak.
+static func flatpak_application_id(flatpak_info: String) -> String:
+	var group := ""
+	for line in flatpak_info.split("\n"):
+		if line.begins_with("["):
+			group = line.strip_edges()
+		elif group == "[Application]" and line.begins_with("name="):
+			return line.trim_prefix("name=").strip_edges()
+	return ""
+
+
+## Whether `directory` is `path` or one of its ancestors.
+static func path_is_within(directory: String, path: String) -> bool:
+	return path == directory or path.begins_with(directory.trim_suffix("/") + "/")
+
+
+## Where a grant's location lies in the sandbox, or "" for one this cannot
+## place (`host-os`, `xdg-documents`, ...).
+static func _flatpak_grant_path(location: String, home: String, host_config_home: String) -> String:
+	var placed := ""
+	if location == "~" or location.begins_with("~/"):
+		placed = home.path_join(location.substr(2))
+	elif location.begins_with("/"):
+		placed = _home_spelling(location, home)
+	elif not host_config_home.is_empty() and (
+		location == "xdg-config" or location.begins_with("xdg-config/")
+	):
+		placed = host_config_home + location.trim_prefix("xdg-config")
+	return placed.simplify_path() if placed.is_absolute_path() else ""
+
+
+## `/home` and `/var/home` are one directory on ostree systems, and Flatpak
+## mounts a grant at whichever is real, keeping the link. A grant spelled the
+## way the home directory is not therefore reads as the one that is.
+static func _home_spelling(path: String, home: String) -> String:
+	for spelling in [["/var/home/", "/home/"], ["/home/", "/var/home/"]]:
+		if home.begins_with(spelling[0]) and path.begins_with(spelling[1]):
+			return str(spelling[0]) + path.trim_prefix(spelling[1])
+	return path
+
+
+## The `[Context] filesystems=` grants of `/.flatpak-info` text as
+## {location: writable}, a location being the grant without its mode: `host`,
+## `~/dir`, `/dir`, `xdg-config/dir`. Plain and `:create` grants are writable.
+## `:ro`, a denial (`!`) and a mode this does not know are not.
+static func _flatpak_filesystems(flatpak_info: String) -> Dictionary:
+	var grants := {}
+	var group := ""
+	for line in flatpak_info.split("\n"):
+		if line.begins_with("["):
+			group = line.strip_edges()
+		elif group == "[Context]" and line.begins_with("filesystems="):
+			for entry in _key_file_list(line.trim_prefix("filesystems=").strip_edges()):
+				var grant := _flatpak_location_and_mode(entry.trim_prefix("!"))
+				## `!host:reset` withdraws inherited grants. It names no location.
+				if grant[0].is_empty() or grant[1] == "reset":
+					continue
+				grants[grant[0]] = not entry.begins_with("!") and grant[1] in ["", "rw", "create"]
+			break
+	return grants
+
+
+## A key file list value split on its unescaped `;`, with the key file's own
+## escapes resolved. Flatpak's `\:` and `\\` reach the entry as `\:` and `\\`.
+static func _key_file_list(value: String) -> PackedStringArray:
+	var entries := PackedStringArray()
+	var entry := ""
+	var index := 0
+	while index < value.length():
+		var character := value[index]
+		if character == "\\" and index + 1 < value.length():
+			index += 1
+			match value[index]:
+				"s":
+					entry += " "
+				"n":
+					entry += "\n"
+				"t":
+					entry += "\t"
+				"r":
+					entry += "\r"
+				_:
+					entry += value[index]
+		elif character == ";":
+			entries.append(entry)
+			entry = ""
+		else:
+			entry += character
+		index += 1
+	if not entry.is_empty():
+		entries.append(entry)
+	return entries
+
+
+## One grant as [location, mode]. Flatpak escapes a `:` or `\` that belongs to
+## the location, and the first bare `:` starts the mode (`ro`, `create`).
+static func _flatpak_location_and_mode(entry: String) -> PackedStringArray:
+	var location := ""
+	var index := 0
+	while index < entry.length():
+		var character := entry[index]
+		if character == "\\" and index + 1 < entry.length():
+			index += 1
+			location += entry[index]
+		elif character == ":":
+			return PackedStringArray([location, entry.substr(index + 1)])
+		else:
+			location += character
+		index += 1
+	return PackedStringArray([location, ""])
+
+
+## `/proc/self/mountinfo` text as one {mount_point, root, fstype, read_only}
+## per mount, in the kernel's order. A line that does not parse is left out.
+static func parse_mountinfo(mountinfo: String) -> Array[Dictionary]:
+	var mounts: Array[Dictionary] = []
+	for line in mountinfo.split("\n", false):
+		var fields := line.split(" ", false)
+		## Optional fields end at a lone "-"; the filesystem type follows it.
+		var separator := fields.find("-", 6)
+		if separator < 0 or separator + 1 >= fields.size():
+			continue
+		mounts.append({
+			"mount_point": _unescape_mount_path(fields[4]),
+			"root": _unescape_mount_path(fields[3]),
+			"fstype": fields[separator + 1],
+			"read_only": fields[5].split(",").has("ro"),
+		})
+	return mounts
+
+
+## The mount that holds `path`: the one with the longest mount point at or
+## above it and, among equals, the last listed, which is mounted on top.
+## {} when none is.
+static func mount_for_path(mounts: Array[Dictionary], path: String) -> Dictionary:
+	var target := path.simplify_path()
+	var found := {}
+	for mount in mounts:
+		var mount_point := str(mount["mount_point"])
+		if (
+			path_is_within(mount_point, target)
+			and mount_point.length() >= str(found.get("mount_point", "")).length()
+		):
+			found = mount
+	return found
+
+
+## The mount that contradicts a grant covering `path`: a file written there
+## would stay inside the sandbox, or cannot be written. {} when the mounts
+## bear the grant out.
+##
+## An ungranted location lies on the tmpfs Flatpak builds the sandbox on, and
+## so does a granted one that did not exist when the app started. A denial
+## hides a location behind an empty tmpfs of its own, as Flatpak hides
+## `~/.var/app`. Such a tmpfs is recognised by being mounted whole (root `/`)
+## below `boundary`, the directory the caller knows to be the host's: the home
+## directory for a path inside it, else the granted directory. A tmpfs at or
+## above the boundary is the host's own (`/home` on a tmpfs) and blocks
+## nothing. Neither does an empty mount list, so unreadable evidence leaves
+## the grant's answer standing.
+static func flatpak_blocking_mount(
+	mounts: Array[Dictionary], path: String, boundary: String
+) -> Dictionary:
+	var mount := mount_for_path(mounts, path)
+	if mount.is_empty():
+		return {}
+	var mount_point := str(mount["mount_point"])
+	var host_directory := boundary.simplify_path()
+	if mount["fstype"] == "tmpfs" and (
+		mount_point == "/"
+		or (
+			mount["root"] == "/"
+			and mount_point != host_directory
+			and path_is_within(host_directory, mount_point)
+		)
+	):
+		return mount
+	return mount if mount["read_only"] else {}
+
+
+## `path` as the kernel names it, with every symlink among its existing
+## components replaced by its target. Mount points are listed by real path, so
+## a home spelled `/home/<user>` on an ostree system has to be read as
+## `/var/home/<user>` before it can be looked up among them. Components that do
+## not exist are kept as written. A chain longer than `_MAX_LINK_HOPS` returns
+## the path unchanged.
+static func real_path(path: String) -> String:
+	if OS.get_name() == "Windows":
+		return path
+	var remaining := path.simplify_path().split("/", false)
+	var current := "/"
+	var hops := 0
+	while not remaining.is_empty():
+		var parent := DirAccess.open(current)
+		if parent == null:
+			break
+		var candidate := current.path_join(remaining[0])
+		remaining.remove_at(0)
+		if not parent.is_link(candidate):
+			current = candidate
+			continue
+		var target := parent.read_link(candidate)
+		if hops >= _MAX_LINK_HOPS or target.is_empty():
+			return path
+		hops += 1
+		if not target.is_absolute_path():
+			target = current.path_join(target)
+		var target_parts := target.simplify_path().split("/", false)
+		target_parts.append_array(remaining)
+		remaining = target_parts
+		current = "/"
+	return current if remaining.is_empty() else current.path_join("/".join(remaining))
+
+
+## Undo the octal escapes mountinfo uses for space, tab, newline and backslash.
+static func _unescape_mount_path(field: String) -> String:
+	if not field.contains("\\"):
+		return field
+	var path := ""
+	var index := 0
+	while index < field.length():
+		var code := field.substr(index + 1, 3)
+		if field[index] == "\\" and _is_octal_byte(code):
+			path += char(int(code[0]) * 64 + int(code[1]) * 8 + int(code[2]))
+			index += 4
+		else:
+			path += field[index]
+			index += 1
+	return path
+
+
+static func _is_octal_byte(code: String) -> bool:
+	if code.length() != 3:
+		return false
+	for digit in code:
+		if digit not in "01234567":
+			return false
+	return true
+
+
+static func _flatpak_info() -> String:
+	if not FileAccess.file_exists(_FLATPAK_INFO_PATH):
+		return ""
+	var file := FileAccess.open(_FLATPAK_INFO_PATH, FileAccess.READ)
+	return "" if file == null else file.get_as_text()