Thread live-status through the spawn path

The spawn path queried `_probe_live_server_status_for_port` from up to
five independent decision points (compatibility check, strong-recovery
proof, post-kill incompatible-server message, recovery proof again,
recovery name-only fallback). On the occupied-port path each probe
costs a ~500 ms localhost HTTPClient poll loop, so a user-reported
trace measured five back-to-back probes totalling ~2.6 s — dominating
the dock's first-paint window.

Replace the redundant probes with explicit data flow: probe once at
the top of `_start_server` and thread the snapshot down through the
proof and incompatible-server helpers. Each helper now takes an
optional `live: Dictionary = {}` and falls back to its historical
fresh probe when the caller doesn't have one (preserves behavior for
the dock's UI buttons and the version-handshake mismatch path).

Also fixes a latent staleness bug in the no-drift + legacy_pidfile
kill path: the original code reused the pre-kill `live` for the
incompatible-server diagnostic even when a kill had been attempted,
so the diagnostic could describe the killed occupant instead of the
current one. The unified failure arm now re-probes after every
recovery failure — costs one extra probe in the rare no-targets sub-
case, fixes the bug structurally.

Tests: five new unit tests in test_plugin_lifecycle.gd assert that
each helper threads `live` correctly and that omitting `live` still
triggers the historical probe. Counter on `_ProofPlugin` (probe_calls)
backs the assertions.

Author: claude

plugin/addons/godot_ai/plugin.gd

@@ -581,24 +581,24 @@ func _start_server() -> void:
 			## kill the process currently holding the port.
 			print("MCP | managed server v%s does not match plugin v%s, restarting"
 				% [record_version, current_version])
-			if not _recover_strong_port_occupant(port, 3.0):
-				_server_started_this_session = true
-				_set_incompatible_server(_probe_live_server_status_for_port(port), current_version, port)
-				push_warning(_server_status_message)
-				return
-			## Fall through to spawn.
-		else:
-			if not _recover_strong_port_occupant(port, 3.0):
-				## No recorded version drift and the live status probe did not
-				## verify an exact/current-compatible godot-ai server. A stale
-				## matching record alone is not enough ownership proof to kill
-				## the current port owner, and opening the WebSocket could route
-				## clients to an incompatible HTTP/MCP tool surface.
-				_server_started_this_session = true
-				_set_incompatible_server(live, current_version, port)
-				push_warning(_server_status_message)
-				return
-			## Fall through to spawn.
+		## No recorded drift case: a stale matching record alone is not
+		## enough ownership proof to kill the current port owner, and
+		## opening the WebSocket could route clients to an incompatible
+		## HTTP/MCP tool surface — `_recover_strong_port_occupant` only
+		## acts on strong proof (managed_record / pidfile_listener /
+		## status_matches_record) so this branch is structurally safe.
+		##
+		## `live` is forwarded so the recovery path's proof helpers reuse
+		## the snapshot we already paid for. The kill itself invalidates
+		## that snapshot, so the failure-mode arm below re-probes before
+		## handing data to `_set_incompatible_server`.
+		if not _recover_strong_port_occupant(port, 3.0, live):
+			_server_started_this_session = true
+			var post_recovery_live := _probe_live_server_status_for_port(port)
+			_set_incompatible_server(post_recovery_live, current_version, port)
+			push_warning(_server_status_message)
+			return
+		## Fall through to spawn.
 
 	_set_resolved_ws_port(_resolve_ws_port())
 	ws_port = _resolved_ws_port
@@ -668,7 +668,12 @@ func _set_incompatible_server(live: Dictionary, expected_version: String, port:
 	_server_actual_version = _live_version_for_message(live)
 	_server_dev_version_mismatch_allowed = false
 	_server_status_message = _incompatible_server_message(live, expected_version, port, _resolved_ws_port)
-	var proof := _evaluate_recovery_port_occupant_proof(port)
+	## `live` is the caller's most-current snapshot — pass it through to
+	## the recovery proof helper so it doesn't fire another probe of the
+	## same port. The `_set_incompatible_server` contract is "use exactly
+	## this live", so threading it down keeps the proof determination
+	## consistent with the diagnostic message we just built.
+	var proof := _evaluate_recovery_port_occupant_proof(port, live)
 	var proof_name := str(proof.get("proof", ""))
 	_can_recover_incompatible = not proof_name.is_empty()
 	print("MCP | proof: %s" % (proof_name if _can_recover_incompatible else "(none)"))
@@ -1320,7 +1325,16 @@ func _find_managed_pid(port: int) -> int:
 	return _find_pid_on_port(port)
 
 
-func _evaluate_strong_port_occupant_proof(port: int) -> Dictionary:
+## `live` is the result of a prior `_probe_live_server_status_for_port`
+## call that the caller already has on hand. When non-empty it short-
+## circuits the internal probe at the bottom of this helper, so a single
+## `_start_server` invocation that probes once at the top can thread the
+## same snapshot through compatibility check + recovery without paying
+## for a second ~500 ms localhost HTTPClient poll loop. Default `{}`
+## preserves the historical behavior for callers outside the spawn flow
+## (`can_recover_incompatible_server`, the dock's UI buttons), where a
+## fresh probe is the right thing.
+func _evaluate_strong_port_occupant_proof(port: int, live: Dictionary = {}) -> Dictionary:
 	var result := {"proof": "", "pids": []}
 	var listener_pids := _find_all_pids_on_port(port)
 	if listener_pids.is_empty():
@@ -1338,31 +1352,40 @@ func _evaluate_strong_port_occupant_proof(port: int) -> Dictionary:
 	if not legacy_targets.is_empty():
 		return {"proof": "pidfile_listener", "pids": legacy_targets}
 
-	var live := _probe_live_server_status_for_port(port)
+	var current_live: Dictionary = live if not live.is_empty() else _probe_live_server_status_for_port(port)
 	if (
-		_live_status_identifies_godot_ai(live)
+		_live_status_identifies_godot_ai(current_live)
 		and not record_version.is_empty()
-		and str(live.get("version", "")) == record_version
+		and str(current_live.get("version", "")) == record_version
 	):
 		return {"proof": "status_matches_record", "pids": listener_pids}
 
 	return result
 
 
-func _evaluate_recovery_port_occupant_proof(port: int) -> Dictionary:
-	var proof := _evaluate_strong_port_occupant_proof(port)
+## See `_evaluate_strong_port_occupant_proof` for the `live` contract.
+## Threads `live` through the strong-proof delegate so neither helper
+## probes when the caller already knows the port-owner status.
+func _evaluate_recovery_port_occupant_proof(port: int, live: Dictionary = {}) -> Dictionary:
+	var proof := _evaluate_strong_port_occupant_proof(port, live)
 	if not str(proof.get("proof", "")).is_empty():
 		return proof
 
-	var live := _probe_live_server_status_for_port(port)
-	if _live_status_identifies_godot_ai(live):
+	var current_live: Dictionary = live if not live.is_empty() else _probe_live_server_status_for_port(port)
+	if _live_status_identifies_godot_ai(current_live):
 		return {"proof": "status_name", "pids": _find_all_pids_on_port(port)}
 
 	return {"proof": "", "pids": []}
 
 
-func _recover_strong_port_occupant(port: int, wait_s: float) -> bool:
-	var proof := _evaluate_strong_port_occupant_proof(port)
+## `pre_kill_live` is forwarded to `_evaluate_strong_port_occupant_proof`
+## so the proof helper doesn't re-probe a port the caller already
+## probed. The kill itself invalidates that snapshot — by the time this
+## function returns, the caller must re-probe before consuming any
+## live-status data. Default `{}` lets non-startup callers (none today)
+## still use the historical fresh-probe behavior.
+func _recover_strong_port_occupant(port: int, wait_s: float, pre_kill_live: Dictionary = {}) -> bool:
+	var proof := _evaluate_strong_port_occupant_proof(port, pre_kill_live)
 	var targets: Array[int] = []
 	targets.assign(proof.get("pids", []))
 	if targets.is_empty():

test_project/tests/test_plugin_lifecycle.gd

@@ -27,6 +27,7 @@ class _ProofPlugin extends GodotAiPlugin:
 	var killed_targets: Array[int] = []
 	var cleared_record_calls := 0
 	var waited_calls := 0
+	var probe_calls := 0
 
 	func _find_all_pids_on_port(_port: int) -> Array[int]:
 		var pids: Array[int] = []
@@ -46,6 +47,7 @@ class _ProofPlugin extends GodotAiPlugin:
 		return branded_pids.has(pid)
 
 	func _probe_live_server_status_for_port(_port: int) -> Dictionary:
+		probe_calls += 1
 		return live_status.duplicate()
 
 	func _is_port_in_use(_port: int) -> bool:
@@ -1134,6 +1136,98 @@ func test_parse_lsof_pids_ignores_non_numeric_lines() -> void:
 	assert_eq(pids[0], 32696)
 
 
+## --- Live-status threading ------------------------------------------
+##
+## `_start_server` probes once at the top of the spawn body and threads
+## that snapshot through `_recover_strong_port_occupant`,
+## `_evaluate_*_port_occupant_proof`, and `_set_incompatible_server` so
+## downstream decision points reuse the result instead of re-probing.
+## Each probe costs a ~500 ms localhost HTTPClient poll loop, so a user-
+## reported occupied-port trace measured five back-to-back probes
+## dominating the dock's first-paint window.
+
+func test_strong_proof_uses_provided_live_without_probing() -> void:
+	var plugin := _ProofPlugin.new()
+	plugin.listener_pids = [13579] as Array[int]
+	plugin.managed_record = {"pid": 0, "version": "2.1.0", "ws_port": 9500}
+	var caller_live := {"name": "godot-ai", "version": "2.1.0", "ws_port": 9500, "status_code": 200}
+
+	var proof := plugin._evaluate_strong_port_occupant_proof(TEST_PORT, caller_live)
+	var probe_calls := plugin.probe_calls
+	plugin.free()
+
+	assert_eq(str(proof.get("proof", "")), "status_matches_record")
+	assert_eq(probe_calls, 0, "passing live must skip the internal probe")
+
+
+func test_strong_proof_probes_when_live_omitted() -> void:
+	var plugin := _ProofPlugin.new()
+	plugin.listener_pids = [13579] as Array[int]
+	plugin.managed_record = {"pid": 0, "version": "2.1.0", "ws_port": 9500}
+	plugin.live_status = {"name": "godot-ai", "version": "2.1.0", "ws_port": 9500, "status_code": 200}
+
+	var proof := plugin._evaluate_strong_port_occupant_proof(TEST_PORT)
+	var probe_calls := plugin.probe_calls
+	plugin.free()
+
+	assert_eq(str(proof.get("proof", "")), "status_matches_record")
+	assert_eq(probe_calls, 1, "omitting live must trigger the internal probe (preserves historical behavior)")
+
+
+func test_recovery_proof_threads_live_through_strong_call() -> void:
+	## `_evaluate_recovery_port_occupant_proof` first delegates to
+	## `_evaluate_strong_port_occupant_proof` (one probe site) and on
+	## empty proof probes again itself (second probe site). Passing
+	## `live` must skip both.
+	var plugin := _ProofPlugin.new()
+	plugin.listener_pids = [13579] as Array[int]
+	plugin.managed_record = {"pid": 0, "version": "", "ws_port": 0}
+	var caller_live := {"name": "godot-ai", "version": "2.1.0", "ws_port": 9500, "status_code": 200}
+
+	var proof := plugin._evaluate_recovery_port_occupant_proof(TEST_PORT, caller_live)
+	var probe_calls := plugin.probe_calls
+	plugin.free()
+
+	assert_eq(str(proof.get("proof", "")), "status_name", "fall-through path uses status name proof")
+	assert_eq(probe_calls, 0, "threading live must skip both internal probes")
+
+
+func test_recover_strong_port_occupant_threads_live_to_proof() -> void:
+	## `_recover_strong_port_occupant` uses `pre_kill_live` for the
+	## ownership-proof determination only. Anything after the kill must
+	## re-probe at the caller; the function itself never reuses
+	## `pre_kill_live` past `_kill_processes_and_windows_spawn_children`.
+	var plugin := _ProofPlugin.new()
+	plugin.listener_pids = [13579] as Array[int]
+	plugin.managed_record = {"pid": 13579, "version": "2.1.0", "ws_port": 9500}
+	plugin.alive_pids = [13579] as Array[int]
+	plugin.port_in_use_sequence = [false] as Array[bool]
+	var caller_live := {"name": "godot-ai", "version": "2.1.0", "ws_port": 9500, "status_code": 200}
+
+	var ok := plugin._recover_strong_port_occupant(TEST_PORT, 0.1, caller_live)
+	var probe_calls := plugin.probe_calls
+	plugin.free()
+
+	assert_true(ok, "managed-record proof should recover when the port frees")
+	assert_eq(probe_calls, 0, "_recover_strong_port_occupant must thread live to its proof helper")
+
+
+func test_set_incompatible_server_threads_live_to_recovery_proof() -> void:
+	## `_set_incompatible_server` already accepts `live`; with the
+	## refactor it forwards that snapshot to its internal recovery-proof
+	## call so the proof helper doesn't re-probe.
+	var plugin := _ProofPlugin.new()
+	plugin.listener_pids = [13579] as Array[int]
+	plugin.managed_record = {"pid": 0, "version": "", "ws_port": 0}
+	var caller_live := {"name": "godot-ai", "version": "2.1.0", "ws_port": 9500, "status_code": 200}
+
+	plugin._set_incompatible_server(caller_live, "2.3.1", TEST_PORT)
+	var probe_calls := plugin.probe_calls
+	plugin.free()
+
+	assert_eq(probe_calls, 0, "_set_incompatible_server must reuse the caller's live for its proof determination")
+
+
 func _seed_managed_record(pid: int, version: String) -> void:
 	var es := EditorInterface.get_editor_settings()
 	if es == null: