extends Node

# Test-only helper (tests/networked_match_smoke.gd). Not a project autoload
# — production code never references this. Same reason as
# tests/lobby_test_hooks.gd: networked_match.tscn is loaded via
# change_scene_to_file(), which frees whatever node initiated the load, so
# a driver can't keep orchestrating from a node that just got freed. The
# smoke test add_child()s this directly under get_tree().root instead (a
# sibling of current_scene, not a descendant of it), so it survives the swap.
#
# Uses preload(), not the bare `NetworkedMatch` class_name, and leaves
# `match_scene` itself untyped (Node) throughout — same global-script-class-
# cache-timing reason as tests/test_case.gd, plus every member access off an
# untyped Node returns Variant, which then needs explicit `: Type`
# annotations wherever `:=` would otherwise fail to infer one.

const NetworkedMatchScript = preload("res://scripts/networked_match.gd")
const BALL_BLEND_ACCEPTANCE_MS := 170 # 150ms contract + one rendered-frame allowance


func _is_networked_match(node) -> bool:
	# is_instance_valid FIRST, and the parameter is untyped for the same
	# reason: at RESULTS both peers change scene to the lobby (§6.2 step 10),
	# which frees the match scene while these hooks — deliberately parented
	# outside it so they survive scene swaps — are still holding a reference.
	# A typed Node parameter throws on a freed object before the body even
	# runs, which hung both processes for the full 5-minute timeout.
	return is_instance_valid(node) and node.get_script() == NetworkedMatchScript


func run_host_check(lifetime_seconds: float, force_goal: bool = false) -> void:
	await get_tree().create_timer(lifetime_seconds * 0.4).timeout
	var match_scene := get_tree().current_scene
	var ok := _is_networked_match(match_scene)
	var ship_count := 0
	var ball_ok := false
	var arena_name := "null"
	if ok:
		ship_count = match_scene.ships.size()
		ball_ok = is_instance_valid(match_scene.ball)
		if match_scene.arena:
			arena_name = match_scene.arena.name
	print("SMOKE INFO: host is_networked_match=%s ship_count=%d ball_ok=%s arena=%s" % [
		str(ok), ship_count, str(ball_ok), arena_name
	])
	var success := ok and ship_count == 1 and ball_ok
	print("SMOKE %s: host spawn check (ship_count=%d, ball_ok=%s)" % ["PASS" if success else "FAIL", ship_count, str(ball_ok)])

	if force_goal and ok:
		# Drive a real PLAYING -> GOAL_PAUSE -> WARMUP -> PLAYING cycle so the
		# client has a goal transition to follow. Teleporting the ball into a
		# goal is the same deterministic trick the CI driver and Phase 2's
		# goal-reset-ordering fix both use — two low-skill peers scoring
		# naturally inside a short run is not reliable enough to gate on.
		var goals: Array = match_scene.arena.get_goals() if match_scene.arena else []
		if is_instance_valid(match_scene.ball) and not goals.is_empty():
			match_scene.ball.linear_velocity = Vector3.ZERO
			match_scene.ball.global_position = goals[0].global_position
			print("SMOKE INFO: host forced a goal to exercise the GOAL_PAUSE transition")

	await get_tree().create_timer(lifetime_seconds * 0.6).timeout
	if not _is_networked_match(match_scene):
		# The match ended and the server returned itself to the lobby.
		print("SMOKE PASS: host ran the match to completion and left the match scene")
		NetworkManager.shutdown()
		get_tree().quit(0)
		return
	if force_goal and _is_networked_match(match_scene):
		print("SMOKE INFO: host final match_state=%s" % MatchState.to_name(match_scene.match_state))
	if _is_networked_match(match_scene) and not match_scene.ships.is_empty():
		var ship: Ship = match_scene.ships[0]
		print("SMOKE INFO: host ship final position=%s action=%s (spawned, driven by client input if any arrived)" % [str(ship.global_position), str(ship.get_current_action_copy().thrust)])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


func run_client_check(settle_seconds: float, drive_seconds: float, exercise_ball_contact: bool = false, exercise_free_flight: bool = false, warmup_seconds: float = 0.0, exercise_input_transitions: bool = false, exercise_match_state: bool = false) -> void:
	# Subscribed BEFORE the settle wait, not after: the server leaves LOADING
	# and enters WARMUP as soon as _start_server() finishes, and PLAYING 90
	# ticks later — both would already be history by the time a post-settle
	# listener attached, and the test would silently observe nothing.
	var observed_states: Array[int] = []
	var observed_ticks: Array[int] = []
	if exercise_match_state:
		# change_scene_to_file is deferred, and so is this call — current_scene
		# is still the smoke driver for the first few frames, so connecting
		# immediately silently observes nothing at all (it did: empty list).
		# Poll until the real scene exists, bounded so a genuine failure to
		# load reports as an empty observation rather than hanging.
		var deadline := Time.get_ticks_msec() + int(settle_seconds * 1000.0)
		while Time.get_ticks_msec() < deadline and not _is_networked_match(get_tree().current_scene):
			await get_tree().process_frame
		var state_scene := get_tree().current_scene
		if _is_networked_match(state_scene):
			# Seed with whatever the client has already converged to. The
			# server may legitimately have reached PLAYING before this client
			# finished loading — that is the snapshot-byte catch-up path doing
			# its job, not a missed transition.
			observed_states.append(state_scene.match_state)
			observed_ticks.append(state_scene.match_state_since_tick)
			state_scene.match_state_changed.connect(func(s: int, at_tick: int) -> void:
				observed_states.append(s)
				observed_ticks.append(at_tick)
			)

	await get_tree().create_timer(settle_seconds).timeout

	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: current_scene is not NetworkedMatch after %.1fs" % settle_seconds)
		get_tree().quit(1)
		return

	var slots_ok: bool = match_scene._slots.size() == 1
	var ball_ok: bool = is_instance_valid(match_scene.ball)
	var my_slot = match_scene._my_slot
	var my_slot_ok: bool = my_slot != null and is_instance_valid(my_slot.ship)
	var camera_ok: bool = is_instance_valid(match_scene._camera_rig)
	var hud_ok: bool = is_instance_valid(match_scene.hud)
	var start_position := Vector3.ZERO
	if my_slot_ok:
		start_position = my_slot.ship.global_position

	print("SMOKE INFO: client slots_ok=%s ball_ok=%s my_slot_ok=%s camera_ok=%s hud_ok=%s start_pos=%s" % [
		str(slots_ok), str(ball_ok), str(my_slot_ok), str(camera_ok), str(hud_ok), str(start_position)
	])

	if not (slots_ok and ball_ok and my_slot_ok and camera_ok and hud_ok):
		print("SMOKE FAIL: spawn/wiring check failed")
		get_tree().quit(1)
		return
	# Bodies are frozen during the kickoff countdown and the goal pause (tasks
	# 5.3/5.4), so every assertion below — "not frozen", "a controller drives
	# it", "it moved" — is only meaningful once play is actually live. Before
	# 5.3 the match was live the instant it loaded and this wait did not exist;
	# sampling during WARMUP now reports a legitimately frozen ship as a
	# prediction failure.
	var live_deadline := Time.get_ticks_msec() + 15000
	while Time.get_ticks_msec() < live_deadline and not MatchState.is_live(match_scene.match_state):
		await get_tree().physics_frame
	if not MatchState.is_live(match_scene.match_state):
		print("SMOKE FAIL: match never reached a live state (stuck in %s)" % MatchState.to_name(match_scene.match_state))
		get_tree().quit(1)
		return
	match_scene._local_ship_predictor.clear_metrics()

	# Drive forward thrust (a real, held key state — exercises the actual
	# client input path, not a synthetic RPC call) and confirm the ship
	# the CLIENT renders (its interpolated $Visual, not a raw snapshot
	# value) actually moved — proving input reached the server, the server
	# applied real thruster force, broadcast it back, and the client's
	# interpolator produced smooth motion from it.
	if exercise_ball_contact:
		# Steer at the ball with real input rather than a fixed-heading burst.
		# This used to be "forward + right for 1.1s", tuned by hand against the
		# spawn orientation — which task 5.3's kickoff broke, because
		# reset_ships() applies KICKOFF_YAW_JITTER (task 0.7) and the ship no
		# longer starts on a known heading. The old burst then flew past the
		# ball every time (0 contacts in 3/3 runs). Closing the loop on the
		# actual bearing keeps this exercising the real input path while being
		# indifferent to how the kickoff happened to orient the ship.
		await _drive_at_ball(my_slot.ship, match_scene.ball, 8.0)
		# Leave a >150ms observation window before the normal drive so a
		# subsequent goal reset cannot mask blend-back.
		Input.action_release("move_forward")
		await get_tree().create_timer(0.35).timeout
	if not exercise_free_flight and not exercise_input_transitions:
		Input.action_press("move_forward")
	if warmup_seconds > 0.0:
		await get_tree().create_timer(warmup_seconds).timeout
		match_scene._local_ship_predictor.clear_metrics()
	var free_flight_peak_distance := 0.0
	if exercise_input_transitions:
		# Deliberate action-sequence-label probe. A HELD input cannot falsify
		# the history's seq labelling: while thrust is constant, "the intent
		# from this tick" and "the action the server consumes for seq S" carry
		# the same value whichever seq the state is filed under, so the action
		# marker reports 0 mismatches under a correct AND an incorrect label.
		# Only a transition exposes the difference, and it exposes it for
		# roughly input_lead ticks per edge. Toggle forward thrust on a short
		# period so the run is mostly edges.
		await _run_input_transition_trace(drive_seconds)
	elif exercise_free_flight:
		# A straight 60-second forward trace reaches the goal/wall in seconds
		# and turns the supposed free-flight QA run into a contact test. Hover
		# in the open volume with alternating vertical thrust and yaw instead:
		# it remains a sustained real thrust/turn/airborne trace without ever
		# manufacturing a wall or goal contact.
		free_flight_peak_distance = await _run_free_flight_trace(my_slot.ship, start_position, drive_seconds)
	else:
		await get_tree().create_timer(drive_seconds * 0.5).timeout

	# Phase 4.3: own ship is a genuine unfrozen local simulation. Its slot
	# intentionally receives no NetInterpolator samples; a controller attached
	# to the body supplies the one action used by this tick's physics step.
	# Split into structural and live halves. The structural half holds at every
	# instant. The freeze/thrust half only means anything while play is live:
	# tasks 5.3/5.4 freeze the local ship for the kickoff countdown and the
	# goal pause, and a goal can land anywhere in a drive, so asserting
	# unconditionally reports a correctly-frozen ship as a prediction failure.
	# The match scene can be freed underneath this — a lost server sends the
	# client back to the main menu (§6.4), same class of teardown as RESULTS.
	if not _is_networked_match(match_scene) or not is_instance_valid(my_slot.ship):
		print("SMOKE INFO: match scene torn down mid-drive (server lost?)")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var live_now: bool = MatchState.is_live(match_scene.match_state)
	var structure_ok: bool = my_slot.ship.controller != null \
		and my_slot.ship.controller.get_parent() == my_slot.ship \
		and not my_slot.interpolator.has_samples()
	var driving_ok: bool = not live_now or (not my_slot.ship.freeze \
		and (my_slot.ship.get_current_action_copy().thrust.z > 0.5 or absf(my_slot.ship.get_current_action_copy().thrust.y) > 0.5))
	var local_prediction_ok: bool = structure_ok and driving_ok
	print("SMOKE INFO: local_prediction=%s state=%s structure_ok=%s driving_ok=%s freeze=%s controller_attached=%s local_interpolator_samples=%s" % [
		str(local_prediction_ok), MatchState.to_name(match_scene.match_state), str(structure_ok), str(driving_ok),
		str(my_slot.ship.freeze), str(my_slot.ship.controller != null and my_slot.ship.controller.get_parent() == my_slot.ship), str(my_slot.interpolator.has_samples())
	])

	if not exercise_free_flight and not exercise_input_transitions:
		await get_tree().create_timer(drive_seconds * 0.5).timeout
	Input.action_release("move_forward")
	Input.action_release("move_up")
	Input.action_release("move_down")
	Input.action_release("turn_left")
	Input.action_release("turn_right")
	if exercise_ball_contact:
		# Leave enough wall time for the bounded RTT window plus the 150ms
		# handoff blend to finish before inspecting lifecycle telemetry.
		await get_tree().create_timer(0.35).timeout

	# The match can legitimately END during a drive (§6.2 step 10: FULL_TIME ->
	# RESULTS -> LOBBY tears this scene down). Every assertion below reads the
	# match scene, so finish on the lifecycle evidence instead of dereferencing
	# freed objects.
	if not _is_networked_match(match_scene) or not is_instance_valid(my_slot.ship):
		var completed_ok := true
		if exercise_match_state:
			var seq: Array[String] = []
			for s in observed_states:
				seq.append(MatchState.to_name(s))
			completed_ok = MatchState.State.RESULTS in observed_states and MatchState.State.LOBBY in observed_states
			for i in observed_states.size() - 1:
				if not MatchState.can_transition(observed_states[i], observed_states[i + 1]):
					print("SMOKE FAIL: illegal transition %s -> %s" % [seq[i], seq[i + 1]])
					completed_ok = false
			print("SMOKE %s: match ran to completion and returned to the lobby (%s)" % [
				"PASS" if completed_ok else "FAIL", " -> ".join(seq)
			])
		else:
			print("SMOKE INFO: match scene torn down before the drive finished")
		await get_tree().create_timer(0.3).timeout
		NetworkManager.shutdown()
		get_tree().quit(0 if completed_ok else 1)
		return

	var end_position: Vector3 = my_slot.ship.global_position
	var prediction_stats: Dictionary = match_scene.get_net_debug_stats().get("prediction", {})
	var net_stats: Dictionary = match_scene.get_net_debug_stats()
	print("SMOKE INFO: prediction samples=%s raw_p95=%.3f raw_p99=%.3f free_samples=%s raw_free_p95=%.3f raw_free_p99=%.3f visual_free_p95=%.3f visual_free_p99=%.3f hard_snaps=%s free_hard_snaps=%s rate=%.2f/min hard_reasons=%s cohorts=%s marker=%s/%s replay=%s lead=%s target=%s buffer=%s ball_contacts=%s latest_error=%s latest_velocity_error=%s" % [
		str(prediction_stats.get("sample_count", 0)), prediction_stats.get("position_error_p95", 0.0), prediction_stats.get("position_error_p99", 0.0),
		str(prediction_stats.get("free_flight_sample_count", 0)), prediction_stats.get("free_flight_position_error_p95", 0.0), prediction_stats.get("free_flight_position_error_p99", 0.0), prediction_stats.get("free_flight_visual_correction_p95", 0.0), prediction_stats.get("free_flight_visual_correction_p99", 0.0),
		str(prediction_stats.get("hard_snap_count", 0)), str(prediction_stats.get("hard_snap_cohorts", {}).get("free_flight", 0)), prediction_stats.get("hard_snap_rate_per_min", 0.0), str(prediction_stats.get("hard_snap_reasons", {})), str(prediction_stats.get("cohorts", {})), str(net_stats.get("action_marker_mismatches", 0)), str(net_stats.get("action_marker_samples", 0)), str(prediction_stats.get("last_replay_count", 0)),
		str(net_stats.get("input_lead", "-")), str(net_stats.get("input_target_depth", "-")), str(net_stats.get("input_buffer_depth", "-")), str(net_stats.get("ball_prediction_contacts", 0)),
		str(net_stats.get("latest_prediction_error", Vector3.ZERO)), str(net_stats.get("latest_prediction_velocity_error", Vector3.ZERO))
	])
	var moved := start_position.distance_to(end_position)
	# Horizontal-only (XZ), not full 3D distance: an adversarial review
	# found a 1.2s window of completely dead input still registers ~1.07m
	# of pure gravity settling on the Y axis alone (spawn height dropping
	# to the floor), which sat ABOVE the old moved > 1.0 bar — only
	# thrust_z_ok caught that failure, not moved. Forward thrust is a
	# horizontal force (see ship.gd), so measuring XZ displacement can't
	# be satisfied by gravity alone, regardless of spawn height or timing.
	var moved_horizontal := Vector2(end_position.x, end_position.z).distance_to(Vector2(start_position.x, start_position.z))
	var verification_movement := free_flight_peak_distance if exercise_free_flight else moved_horizontal
	print("SMOKE INFO: client ship moved %.2fm (%.2fm horizontal) (start=%s end=%s) while holding forward thrust for %.1fs" % [
		moved, moved_horizontal, str(start_position), str(end_position), drive_seconds
	])

	# thrust_power 150 / mass 5 = 30 m/s^2 nominal acceleration (see ship.gd) —
	# over 2s even with drag/ramp-up this should clear a couple of metres.
	# A generous, not-tuned-to-the-decimal bound: this is a wiring smoke
	# test, not a physics-accuracy test (net_codec's own tests already cover
	# quantisation precision).
	# The contact path intentionally includes a goal/reset in this trace, so
	# its expected authoritative snaps are reported separately rather than
	# contaminating the contact-free prediction gate.
	# The scheduled local timeline now predicts the same command stream the
	# server consumes, so both the same-sequence raw residual and the exposed
	# render discontinuity are meaningful free-flight gates. Hard corrections
	# remain separately gated by cohort.
	# Transport health, printed alongside the quality numbers and asserted
	# separately below. Without this a p95 failure is undiagnosable: "the
	# predictor got worse" and "the client never received the data" look
	# identical in a percentile. An adversarial review hit exactly that — a
	# 3-process run failed at p95 0.688 with roughly a third of snapshots
	# missing, and it could not be told apart from a real regression.
	var snapshot_loss_pct := float(net_stats.get("snapshot_loss_pct", 0.0))
	var snapshot_age_ms := float(net_stats.get("snapshot_age_ms", 0.0))
	print("SMOKE INFO: transport snapshot_loss=%.1f%% snapshot_age=%.1fms rtt=%.1fms" % [
		snapshot_loss_pct, snapshot_age_ms, NetworkManager.rtt_ms
	])

	var raw_quality_p95: float = float(prediction_stats.get("free_flight_position_error_p95", INF))
	var raw_quality_p99: float = float(prediction_stats.get("free_flight_position_error_p99", INF))
	var raw_rotation_p95: float = float(prediction_stats.get("free_flight_rotation_error_p95", INF))
	var raw_rotation_p99: float = float(prediction_stats.get("free_flight_rotation_error_p99", INF))
	var quality_p95: float = float(prediction_stats.get("free_flight_visual_correction_p95", INF))
	var quality_p99: float = float(prediction_stats.get("free_flight_visual_correction_p99", INF))
	var quality_samples := int(prediction_stats.get("free_flight_sample_count", 0))
	var free_flight_hard_snaps := int(prediction_stats.get("hard_snap_cohorts", {}).get("free_flight", 99))
	# The action-sequence-label gate. Every other mode here holds its inputs
	# steady or near-steady, and a steady input CANNOT falsify the history's
	# seq labelling: while the commanded action is constant, "the intent from
	# this tick" and "the action the server consumes for seq S" carry the same
	# value under a correct and an incorrect label alike, so the action marker
	# reads 0/N either way. That is precisely how a real mislabelling survived
	# every earlier Phase 4 gate. Only this mode's forced edges expose it, so
	# only this mode asserts on the marker.
	#
	# Measured separation is wide, not marginal: labelling post-step state at
	# the server-consumption estimate reported 9.3% mismatch on LAN
	# (input_lead 1) and 24% at 80±20ms (input_lead 3) — it scales with the
	# lead, as the mechanism predicts — against 0-1.3% once filed under the
	# issuing sequence. The residual is seq-delta events (an attack issues
	# several sequences for one local physics step, a release duplicates one),
	# which relabelling does not claim to fix.
	var marker_samples := int(net_stats.get("action_marker_samples", 0))
	var marker_mismatches := int(net_stats.get("action_marker_mismatches", 99))
	var marker_rate := float(marker_mismatches) / float(maxi(marker_samples, 1))
	# The sample floor scales with the run, and that is load-bearing rather than
	# tidiness. An adversarial review reproduced a total, permanent input
	# blackout (a 3.5s host freeze) that this gate reported as PASS at 3.76%:
	# once reconciliation is suppressed, _record_metrics stops being called, so
	# the marker stops sampling entirely — the WORSE the outage, the FEWER
	# samples and the LOWER the reported mismatch rate. A flat ">= 200" is
	# satisfied by the handful of acks either side of the outage. Snapshots ack
	# at ~60Hz, so require half of nominal and a run this short is provably
	# still exchanging input for most of its length.
	var marker_sample_floor := maxi(200, int(drive_seconds * 30.0))
	var marker_samples_ok := marker_samples >= marker_sample_floor
	# Server-side starvation bit, round-tripped over the wire. A client flying
	# on pure prediction with the server ignoring it satisfies every other
	# assertion here, because all of them read the CLIENT's own action and
	# position.
	var server_stalled := bool(net_stats.get("server_stalled", false))
	# 5%, not 3%: the irreducible residual is seq-delta events and it scales
	# with input_lead, reaching 2.22% at lead 3 under 80±20ms — too close to a
	# 3% line for a CI gate. Separation from a genuinely mislabelled build is
	# 10-20x either way (control runs measure 24-50%), so the extra headroom
	# costs no real detection power. Tighten this only alongside recording
	# predictions for an attack's filled gap sequences.
	const MAX_ACTION_MARKER_MISMATCH_RATE := 0.05
	var action_label_ok: bool = marker_samples_ok and not server_stalled and marker_rate < MAX_ACTION_MARKER_MISMATCH_RATE
	# The 0.5/2.0 free-flight bounds belong to --exercise-free-flight and ONLY
	# to it, because that mode is the only one that produces the profile they
	# were calibrated on. _run_free_flight_trace exists precisely because, in
	# its own words, "a straight forward trace reaches the goal/wall in seconds
	# and turns the supposed free-flight QA run into a contact test" — yet the
	# plain role went on asserting the open-volume bounds against whatever
	# free-flight samples that contact-heavy drive happened to leave behind.
	#
	# Measured over 8 plain-role runs on an idle machine: the free-flight
	# cohort ranged from 12 to 257 samples and its p95 from 0.275 to 0.726,
	# failing the 0.5 bound in 3 of 8 — a ~37% flake rate with no defect
	# present. That is the p95 0.688 an adversarial review reported and I first
	# mis-attributed to three-process CPU contention: it reproduces on two
	# processes, on an idle box, with 0.0% snapshot loss. The mechanism is not
	# noise — error near the arena's surface-pull field is genuinely several
	# times higher than in open air (--exercise-free-flight measures 0.084-0.111
	# on the same build) — but a gate that fires a third of the time is worse
	# than no gate, and calibrating one bound for both profiles cannot work.
	#
	# So the plain role asserts the ALL-COHORT percentiles instead. They are
	# always well-sampled (545-696 samples across those same runs, versus a
	# free-flight cohort that can collapse to 12) and much tighter in spread:
	# raw_p95 0.354-0.609, raw_p99 0.362-0.742. The bounds below sit ~2x above
	# the worst observed. A free-flight-cohort regression still cannot hide:
	# free_flight_hard_snaps is asserted in both modes, and anything past 2.0m
	# IS a hard snap by definition.
	#
	# The 100-sample floor is not the plain drive's number (545-696) but
	# --exercise-match-state's: its forced goal suspends prediction for the
	# whole GOAL_PAUSE, so an 8s run yields ~153. Still five times what the
	# old free-flight floor accepted.
	const NEAR_SURFACE_P95 := 1.2
	const NEAR_SURFACE_P99 := 2.0
	var overall_p95: float = float(prediction_stats.get("position_error_p95", INF))
	var overall_p99: float = float(prediction_stats.get("position_error_p99", INF))
	var overall_samples := int(prediction_stats.get("sample_count", 0))
	var prediction_quality_ok: bool = action_label_ok if exercise_input_transitions else \
		prediction_stats.get("hard_snap_count", 99) < 4 if exercise_ball_contact else \
		quality_samples >= 30 \
		and raw_quality_p95 < 0.5 \
		and raw_quality_p99 < 2.0 \
		and raw_rotation_p95 < 5.0 \
		and raw_rotation_p99 < 15.0 \
		and quality_p95 < 0.5 \
		and quality_p99 < 2.0 \
		and free_flight_hard_snaps == 0 \
		if exercise_free_flight else \
		overall_samples >= 100 \
		and overall_p95 < NEAR_SURFACE_P95 \
		and overall_p99 < NEAR_SURFACE_P99 \
		and raw_rotation_p95 < 5.0 \
		and raw_rotation_p99 < 15.0 \
		and free_flight_hard_snaps == 0
	if not exercise_free_flight and not exercise_input_transitions and not exercise_ball_contact:
		print("SMOKE INFO: near-surface profile — asserting all-cohort p95=%.3f/p99=%.3f (bounds %.1f/%.1f, %d samples); free-flight cohort p95=%.3f/p99=%.3f over %d samples is REPORTED, NOT ASSERTED (see --exercise-free-flight for the calibrated gate)" % [
			overall_p95, overall_p99, NEAR_SURFACE_P95, NEAR_SURFACE_P99, overall_samples,
			raw_quality_p95, raw_quality_p99, quality_samples,
		])
	# ball_proxy_moved_before_authority counts ticks where the predicted proxy
	# had visibly moved BEFORE the next authoritative ball state arrived. That
	# is only a meaningful — or even achievable — claim when there is real RTT
	# to mask: snapshots land every ~16.7ms at 60Hz, so on a loopback LAN the
	# whole pre-authority window is about one physics tick and whether it is
	# observed is a coin flip on arrival timing. Measured 2 failures in 5 LAN
	# runs, versus 5/5 passes (count 2-3) at --net-sim-latency=80, with the
	# same-frame reveal itself correct in every single run either way.
	#
	# So require it only when the link actually has latency to hide, and let
	# the same-frame reveal carry the gate on LAN — that is the real claim
	# ("your own touches register on contact, not ~RTT later") and it is not
	# racy. Runs asserting the masking behaviour should pass --net-sim-latency.
	var rtt_ms := NetworkManager.rtt_ms
	var rtt_masks_authority := rtt_ms >= 20.0
	var proxy_motion_ok: bool = not rtt_masks_authority or int(net_stats.get("ball_proxy_moved_before_authority_count", 0)) > 0
	var ball_contact_ok := not exercise_ball_contact or (int(net_stats.get("ball_prediction_contacts", 0)) > 0 \
		and int(net_stats.get("ball_contact_frame", -1)) == int(net_stats.get("ball_reveal_frame", -2)) \
		and int(net_stats.get("ball_blend_complete_count", 0)) > 0 \
		and int(net_stats.get("ball_blend_max_duration_ms", BALL_BLEND_ACCEPTANCE_MS)) <= BALL_BLEND_ACCEPTANCE_MS \
		and proxy_motion_ok)
	# §6.1 task 5.1: the client must FOLLOW the server's machine, not run its
	# own. Assert three separable things — that transitions arrived at all,
	# that every consecutive pair is legal per the shared table (so the client
	# never lands somewhere the server could not have sent it), and that the
	# specific documented sequence for this scenario was observed.
	var match_state_ok := true
	if exercise_match_state:
		var names: Array[String] = []
		for s in observed_states:
			names.append(MatchState.to_name(s))
		for i in observed_states.size() - 1:
			if not MatchState.can_transition(observed_states[i], observed_states[i + 1]):
				print("SMOKE FAIL: client observed an illegal transition %s -> %s" % [names[i], names[i + 1]])
				match_state_ok = false
		# Ticks are absolute and monotonic; a transition attributed to an
		# earlier tick than its predecessor means the at_tick plumbing is wrong.
		for i in observed_ticks.size() - 1:
			if observed_ticks[i + 1] < observed_ticks[i]:
				print("SMOKE FAIL: transition ticks went backwards: %s" % str(observed_ticks))
				match_state_ok = false
		# Which lifecycle path a run takes depends on its own timing: a short
		# --match-length reaches FULL_TIME before the forced goal lands, a
		# longer one exercises the goal cycle instead. Assert what the run
		# actually did rather than hardcoding one shape — but require it did
		# at least ONE of them, so a match that merely sat in PLAYING the
		# whole time cannot quietly pass.
		var reached_playing := MatchState.State.PLAYING in observed_states
		var saw_goal_pause := MatchState.State.GOAL_PAUSE in observed_states
		var saw_full_time := MatchState.State.FULL_TIME in observed_states
		# A goal must lead back to a kickoff, not leave the match parked.
		var resumed_after_goal := false
		for i in observed_states.size() - 1:
			if observed_states[i] == MatchState.State.GOAL_PAUSE and observed_states[i + 1] == MatchState.State.WARMUP:
				resumed_after_goal = true
		# Full time must resolve: sudden death on a draw, results otherwise.
		var full_time_resolved := false
		for i in observed_states.size() - 1:
			if observed_states[i] == MatchState.State.FULL_TIME and observed_states[i + 1] in [MatchState.State.OVERTIME_WARMUP, MatchState.State.RESULTS]:
				full_time_resolved = true
		var goal_cycle_ok: bool = not saw_goal_pause or resumed_after_goal
		var full_time_ok: bool = not saw_full_time or full_time_resolved
		if not (reached_playing and goal_cycle_ok and full_time_ok and (saw_goal_pause or saw_full_time)):
			match_state_ok = false
		# The snapshot's match_state byte must carry the real state too, not a
		# hardcoded 0. Everything above is driven by the reliable state_change
		# RPC and would pass identically with a dead byte — which is exactly
		# how Phase 4's mislabelled prediction history survived every gate.
		# The byte is the only channel a late joiner or a client that missed a
		# transition has (§6.3), so assert it independently.
		var wire_state := int(net_stats.get("snapshot_match_state", -1))
		var live_state := int(net_stats.get("match_state", -1))
		if wire_state != live_state or not MatchState.is_valid(wire_state):
			print("SMOKE FAIL: snapshot match_state byte is %s but the client is in %s" % [
				MatchState.to_name(wire_state), MatchState.to_name(live_state)
			])
			match_state_ok = false
		if wire_state == MatchState.State.LOBBY:
			print("SMOKE FAIL: snapshot match_state byte reads LOBBY (0) mid-match — likely never populated")
			match_state_ok = false
		print("SMOKE %s: client followed the server's match state (%s; playing=%s goal_pause=%s resumed=%s full_time=%s resolved=%s ticks=%s)" % [
			"PASS" if match_state_ok else "FAIL", " -> ".join(names),
			str(reached_playing), str(saw_goal_pause), str(resumed_after_goal),
			str(saw_full_time), str(full_time_resolved), str(observed_ticks),
		])

	var success := verification_movement > 1.0 and local_prediction_ok and prediction_quality_ok and ball_contact_ok and match_state_ok
	# A run starved of snapshots has not measured prediction quality at all, so
	# say so explicitly instead of blaming the predictor. Deliberately does NOT
	# convert the failure into a pass — a client that cannot receive snapshots
	# is still a failed run, just a differently-diagnosed one.
	if not prediction_quality_ok and snapshot_loss_pct > 20.0:
		print("SMOKE FAIL: transport-starved, not a prediction regression (snapshot_loss=%.1f%%) — check host CPU contention before suspecting the predictor" % snapshot_loss_pct)
	print("SMOKE %s: client locally predicted %.2fm horizontal, local_prediction_ok=%s prediction_quality_ok=%s" % [
		"PASS" if success else "FAIL", moved_horizontal, str(local_prediction_ok), str(prediction_quality_ok)
	])
	if exercise_input_transitions:
		print("SMOKE %s: action-sequence labelling under forced input transitions (marker=%d/%d = %.2f%% mismatch, want <%.0f%%; samples %d/%d required; server_stalled=%s; input_lead=%s)" % [
			"PASS" if action_label_ok else "FAIL", marker_mismatches, marker_samples, marker_rate * 100.0, MAX_ACTION_MARKER_MISMATCH_RATE * 100.0,
			marker_samples, marker_sample_floor, str(server_stalled), str(net_stats.get("input_lead", "-")),
		])
	if exercise_ball_contact:
		print("SMOKE %s: local dynamic ball proxy registered a same-frame reveal (contact_frame=%s reveal_frame=%s pre_authority_motion=%s hard_handoffs=%s blend_started=%s blends=%s blend_max_ms=%s ends=%s missing_shadow=%s reset_cancels=%s reset_trace=%s)" % [
			"PASS" if ball_contact_ok else "FAIL", str(net_stats.get("ball_contact_frame", -1)), str(net_stats.get("ball_reveal_frame", -1)), str(net_stats.get("ball_proxy_moved_before_authority_count", 0)),
			str(net_stats.get("ball_hard_handoff_count", 0)), str(net_stats.get("ball_blend_started_count", 0)), str(net_stats.get("ball_blend_complete_count", 0)), str(net_stats.get("ball_blend_max_duration_ms", 0)),
			str(net_stats.get("ball_prediction_window_end_count", 0)), str(net_stats.get("ball_prediction_missing_shadow_count", 0)), str(net_stats.get("ball_prediction_reset_cancel_count", 0)), str(net_stats.get("ball_reset_trace", [])),
		])
		print("SMOKE INFO: ball pre-authority motion %s (rtt=%.1fms; asserted only at >=20ms, see proxy_motion_ok)" % [
			"asserted and met" if rtt_masks_authority else "not asserted on this near-zero-RTT link", rtt_ms,
		])
	await get_tree().create_timer(0.3).timeout
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# Toggles forward thrust every TOGGLE_TICKS physics frames for the requested
# duration, then leaves it pressed so the caller's own local_prediction_ok
# check still sees a live commanded action. Yaw alternates alongside it purely
# to keep the ship from driving straight into a wall and turning a labelling
# probe into a contact test.
# Samples the server's own score every physics frame for `seconds`, appending
# each distinct value. Lets the comparison below check a client's recorded
# score against a state the server genuinely passed through, rather than
# against whatever it happens to hold seconds later.
#
# `path_lengths`, when pre-seeded with peer_id -> 0.0, also accumulates how far
# each slot's ship actually travelled while this wait runs. Any caller that
# forces a goal must measure movement this way rather than by start-to-end
# displacement: a goal's kickoff teleports every ship back to spawn
# (_begin_kickoff -> reset_ships in networked_match.gd, which calls the teleport
# its own comment describes), so displacement from a position sampled before the
# goal measures only the distance covered since the last reset — a window whose
# length depends on when the sample lands relative to the kickoff rather than on
# whether input was flowing at all. Summing per-tick steps is reset-proof and
# strictly stronger: it asserts input kept arriving across the whole wait, not
# merely that the ship finished somewhere other than where it started.
func _await_recording_score(match_scene, seconds: float, history: Array[String], path_lengths: Dictionary = {}) -> void:
	# Ship.max_speed (35 m/s) is hard-clamped every tick in Ship._integrate_forces,
	# so at 60Hz no ship can legitimately cover more than ~0.58m between physics
	# frames. A step past this is the kickoff teleport, not travel, and must not
	# count toward the total.
	const MAX_TICK_TRAVEL := 2.0
	var deadline := Time.get_ticks_msec() + int(seconds * 1000.0)
	var previous_positions: Dictionary = {}
	while Time.get_ticks_msec() < deadline:
		var current := JSON.stringify(match_scene.score)
		if history[history.size() - 1] != current:
			history.append(current)
		for slot in match_scene._slots:
			if not path_lengths.has(slot.peer_id) or not is_instance_valid(slot.ship):
				continue
			var position: Vector3 = slot.ship.global_position
			if previous_positions.has(slot.peer_id):
				var step: float = (previous_positions[slot.peer_id] as Vector3).distance_to(position)
				if step < MAX_TICK_TRAVEL:
					path_lengths[slot.peer_id] += step
			previous_positions[slot.peer_id] = position
		await get_tree().physics_frame


func _run_input_transition_trace(duration_seconds: float) -> void:
	const TOGGLE_TICKS := 6   # ~100ms at 60Hz: several edges per second
	var frames := int(duration_seconds * 60.0)
	var pressed := false
	var yaw_left := false
	for frame in frames:
		if frame % TOGGLE_TICKS == 0:
			pressed = not pressed
			if pressed:
				Input.action_press("move_forward")
			else:
				Input.action_release("move_forward")
		if frame % (TOGGLE_TICKS * 4) == 0:
			yaw_left = not yaw_left
			Input.action_release("turn_right" if yaw_left else "turn_left")
			Input.action_press("turn_left" if yaw_left else "turn_right")
		await get_tree().physics_frame
	Input.action_release("turn_left")
	Input.action_release("turn_right")
	Input.action_press("move_forward")
	await get_tree().physics_frame


# Closed-loop steering: yaw toward the ball, thrust once roughly aligned, and
# stop as soon as we are close enough that contact is imminent. Uses only real
# Input actions, so the client input -> server -> snapshot path under test is
# exercised exactly as a player would.
func _drive_at_ball(ship: Ship, ball_body: Node3D, timeout_seconds: float) -> void:
	const ALIGNED_RADIANS := 0.25
	var deadline := Time.get_ticks_msec() + int(timeout_seconds * 1000.0)
	while Time.get_ticks_msec() < deadline:
		if not is_instance_valid(ship) or not is_instance_valid(ball_body):
			break
		var to_ball := ball_body.global_position - ship.global_position
		if to_ball.length() < 1.2:
			break   # touching distance; momentum carries it the rest of the way
		# Deliberately keeps steering all the way in rather than breaking off
		# early and coasting: breaking at 3m let the ship sail past the ball
		# without ever touching it (0 contacts in 1 run of 3).
		# Bearing in the ship's own frame: -Z is forward, +X is right.
		var local := ship.global_transform.basis.inverse() * to_ball
		var yaw_error := atan2(local.x, -local.z)
		Input.action_release("turn_left")
		Input.action_release("turn_right")
		if absf(yaw_error) > ALIGNED_RADIANS:
			Input.action_press("turn_right" if yaw_error > 0.0 else "turn_left")
		# Thrust whenever the ball is anywhere ahead, not only once perfectly
		# aligned. Cutting thrust to turn made the ship hover and burn the
		# window without closing distance, which is why this reached the ball
		# only 2 runs in 3; turning under power converges much faster.
		if absf(yaw_error) < PI * 0.5:
			Input.action_press("move_forward")
		else:
			Input.action_release("move_forward")
		# Vertical alignment matters too — the ball sits above the floor and a
		# ship that is climbing sails straight over it.
		Input.action_release("move_up")
		Input.action_release("move_down")
		if local.y > 1.0:
			Input.action_press("move_up")
		elif local.y < -1.0:
			Input.action_press("move_down")
		await get_tree().physics_frame
	Input.action_release("turn_left")
	Input.action_release("turn_right")
	Input.action_release("move_up")
	Input.action_release("move_down")
	Input.action_press("move_forward")


func _run_free_flight_trace(ship: Ship, start_position: Vector3, duration_seconds: float) -> float:
	var elapsed := 0.0
	var peak_distance := 0.0
	while elapsed < duration_seconds:
		Input.action_release("move_down")
		Input.action_press("move_up")
		var up_seconds := minf(0.7, duration_seconds - elapsed)
		await get_tree().create_timer(up_seconds).timeout
		elapsed += up_seconds
		peak_distance = maxf(peak_distance, ship.global_position.distance_to(start_position))
		if elapsed >= duration_seconds:
			break
		Input.action_release("move_up")
		Input.action_press("move_down")
		var down_seconds := minf(0.3, duration_seconds - elapsed)
		await get_tree().create_timer(down_seconds).timeout
		elapsed += down_seconds
		peak_distance = maxf(peak_distance, ship.global_position.distance_to(start_position))
	return peak_distance


# task 3.4: MatchSim._recv_input must count malformed packets and disconnect
# after MALFORMED_LIMIT_TO_DISCONNECT (20) of them. Calls the RPC directly
# with garbage bytes rather than going through networked_match.gd's own
# honest encoder — this IS what a hostile custom client sending raw ENet
# packets would look like, so bypassing the normal send path is the point,
# not a shortcut.
# §6.4 (tasks 5.6/5.7), host side. Watches its own slots across a client's
# disconnect and reconnect and asserts the documented contract: the ship is
# never despawned, the controller is swapped rather than left dangling, the
# slot is reserved by identity, and a returning player gets it back.
# hold_after_reclaim_seconds is not padding. The original version ticked 60
# physics frames (1.0s) after the reclaim and then shut the server down, which
# meant the reconnecting client — whose own wiring check waits a 2.0s settle
# before it looks at anything — had its peer torn out from under it every time
# and reported "current_scene is not NetworkedMatch after 2.0s". The server side
# passed throughout, so the harness looked green from the only side anyone read.
# The reconnecting player is half of what §6.4 promises; it gets a real window.
func run_disconnect_host_check(lifetime_seconds: float, hold_after_reclaim_seconds: float = 8.0) -> void:
	await get_tree().create_timer(2.0).timeout
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: host scene is not NetworkedMatch")
		get_tree().quit(1)
		return

	var slots_before: int = match_scene._slots.size()
	if slots_before == 0:
		print("SMOKE FAIL: host has no slots — the client never made it into the roster")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var ship_before = match_scene._slots[0].ship
	var name_before: String = match_scene._slots[0].player_name
	print("SMOKE INFO: host has %d slot(s), player_name=%s" % [slots_before, name_before])

	# Wait for the client to drop. Guarded on the scene still existing: §6.4's
	# abort can tear the match down underneath this loop.
	var drop_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0)
	while Time.get_ticks_msec() < drop_deadline and _is_networked_match(match_scene) and not match_scene._slots[0].disconnected:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match aborted during the disconnect window — the reservation should have held it open")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var saw_disconnect: bool = match_scene._slots[0].disconnected
	var ship_survived: bool = match_scene._slots.size() == slots_before and is_instance_valid(match_scene._slots[0].ship) and match_scene._slots[0].ship == ship_before
	var controller_valid: bool = is_instance_valid(match_scene._slots[0].controller)
	var reserved: bool = match_scene._slots[0].reserved_until_tick > Engine.get_physics_frames()
	print("SMOKE INFO: after disconnect saw_disconnect=%s ship_survived=%s controller_valid=%s reserved=%s" % [
		str(saw_disconnect), str(ship_survived), str(controller_valid), str(reserved)
	])

	# Then for it to come back and reclaim the slot.
	var back_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0)
	while Time.get_ticks_msec() < back_deadline and _is_networked_match(match_scene) and match_scene._slots[0].disconnected:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match aborted before the player could reconnect")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var reclaimed: bool = not match_scene._slots[0].disconnected
	var same_ship: bool = is_instance_valid(match_scene._slots[0].ship) and match_scene._slots[0].ship == ship_before
	# Ticking on past the swap proves task 5.7: _physics_process writes
	# slot.controller.action every tick, so a dangling reference from
	# set_controller()'s queue_free() would have crashed by now. It also keeps
	# the server alive long enough for the reconnected client to run its own
	# checks and actually play — see this function's header.
	var reclaim_position := Vector3.ZERO
	if is_instance_valid(match_scene._slots[0].ship):
		reclaim_position = match_scene._slots[0].ship.global_position
	# The reconnected player's input must reach the server and move the ship the
	# server owns. Every other assertion here is about slot bookkeeping and
	# would hold identically for a client whose input pipeline came back dead —
	# which is the failure §6.4's reservation exists to prevent.
	#
	# Both the position and the connection state are sampled WHILE the peer is
	# still connected, not once at the end of the hold. The client finishes its
	# own checks and leaves on its own schedule, so an end-of-hold sample reads
	# a legitimately departed peer and reports "still_connected=false" for a
	# perfectly good run — the same mis-timed sampling a Phase 3 review caught
	# in the CI gate.
	var saw_connected := false
	var last_connected_position := reclaim_position
	var hold_deadline := Time.get_ticks_msec() + int(hold_after_reclaim_seconds * 1000.0)
	while Time.get_ticks_msec() < hold_deadline and _is_networked_match(match_scene):
		if match_scene._slots[0].peer_id in multiplayer.get_peers():
			saw_connected = true
			if is_instance_valid(match_scene._slots[0].ship):
				last_connected_position = match_scene._slots[0].ship.global_position
		await get_tree().physics_frame

	var still_live := _is_networked_match(match_scene)
	var server_side_movement := Vector2(
		last_connected_position.x - reclaim_position.x,
		last_connected_position.z - reclaim_position.z
	).length()
	var drove_after_reclaim := saw_connected and server_side_movement > 1.0
	print("SMOKE INFO: reconnected player moved %.2fm horizontally server-side while connected, over a %.1fs hold (saw_connected=%s)" % [
		server_side_movement, hold_after_reclaim_seconds, str(saw_connected),
	])

	var success := saw_disconnect and ship_survived and controller_valid and reserved and reclaimed and same_ship \
		and still_live and drove_after_reclaim and is_instance_valid(match_scene._slots[0].controller)
	print("SMOKE %s: disconnect kept the ship and the reconnect reclaimed the slot (disconnect=%s ship_kept=%s reserved=%s reclaimed=%s same_ship=%s still_live=%s drove_after_reclaim=%s)" % [
		"PASS" if success else "FAIL", str(saw_disconnect), str(ship_survived), str(reserved), str(reclaimed), str(same_ship),
		str(still_live), str(drove_after_reclaim),
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# §6.3 (task 5.8). A peer that joins mid-match with a name nobody reserved is a
# spectator: no slot, no ship, but it MUST still receive the snapshot stream
# and follow the lifecycle. An adversarial review found spectators received no
# snapshots at all, because _broadcast_snapshot unicasts per SLOT.
func run_spectator_check(run_seconds: float) -> void:
	var snapshot_count := [0]
	MatchSim.snapshot_received.connect(func(_d: Dictionary) -> void: snapshot_count[0] += 1)

	var deadline := Time.get_ticks_msec() + 10000
	while Time.get_ticks_msec() < deadline and not _is_networked_match(get_tree().current_scene):
		await get_tree().process_frame
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: spectator never loaded the match scene")
		get_tree().quit(1)
		return

	await get_tree().create_timer(run_seconds).timeout
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match scene torn down during the spectator run")
		get_tree().quit(1)
		return

	var is_spectator: bool = match_scene._my_slot == null
	var got_snapshots: bool = snapshot_count[0] > int(run_seconds * 20.0)
	var camera_ok: bool = is_instance_valid(match_scene._camera_rig)
	var state_ok: bool = MatchState.is_valid(match_scene.match_state) and match_scene.match_state != MatchState.State.LOBBY
	# Bootstrap: a late joiner must know the live clock, not wait for a goal.
	var stats: Dictionary = match_scene.get_net_debug_stats()
	var wire_state := int(stats.get("snapshot_match_state", -1))
	var wire_ok: bool = wire_state == int(stats.get("match_state", -2))
	# Cycling must be safe and must never hand the camera a non-Ship.
	match_scene.cycle_spectator_target(1)
	match_scene.cycle_spectator_target(1)
	match_scene.cycle_spectator_target(-1)
	var cycle_ok: bool = is_instance_valid(match_scene._camera_rig) and (match_scene._camera_rig.target == null or match_scene._camera_rig.target is Ship)

	print("SMOKE INFO: spectator is_spectator=%s snapshots=%d camera_ok=%s state=%s wire_state=%s cycle_ok=%s" % [
		str(is_spectator), snapshot_count[0], str(camera_ok), MatchState.to_name(match_scene.match_state),
		MatchState.to_name(wire_state), str(cycle_ok)
	])
	var success := is_spectator and got_snapshots and camera_ok and state_ok and wire_ok and cycle_ok
	print("SMOKE %s: spectator received the snapshot stream and followed the match (snapshots=%d, want > %d)" % [
		"PASS" if success else "FAIL", snapshot_count[0], int(run_seconds * 20.0)
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# §6.3's "free slot mid-match → spectate now, take the slot at the next
# kickoff", server side. The sequence this drives: a player leaves, their §6.4
# reservation lapses (run with --slot-reservation-seconds= small, or this waits
# 30 real seconds for the interesting moment), a goal is forced to produce a
# kickoff, and the waiting spectator must be holding the slot afterwards.
#
# The forced goal is the same deterministic trick the CI driver and the
# match-state check use — waiting for two peers to score naturally inside a
# short run is not something to gate on.
func run_late_joiner_host_check(lifetime_seconds: float) -> void:
	await get_tree().create_timer(2.0).timeout
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: host scene is not NetworkedMatch")
		get_tree().quit(1)
		return
	if match_scene._slots.is_empty():
		print("SMOKE FAIL: host has no slots — the first client never joined")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var original_peer: int = match_scene._slots[0].peer_id
	var original_name: String = match_scene._slots[0].player_name
	var ship_before = match_scene._slots[0].ship

	# Wait for the seated player to drop.
	var drop_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0)
	while Time.get_ticks_msec() < drop_deadline and _is_networked_match(match_scene) and not match_scene._slots[0].disconnected:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene) or not match_scene._slots[0].disconnected:
		print("SMOKE FAIL: the seated player never dropped")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	# A spectator must be queued by now, or the rest of this proves nothing.
	var queued: int = match_scene._late_joiners.size()

	# Then for the reservation to lapse. Until it does, the slot belongs to the
	# player who left — §6.4 outranks §6.3, and taking it early would quietly
	# break the reconnect promise.
	var lapse_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0) + 32000
	while Time.get_ticks_msec() < lapse_deadline and _is_networked_match(match_scene) \
			and match_scene._slots[0].reserved_until_tick >= 0 \
			and Engine.get_physics_frames() <= match_scene._slots[0].reserved_until_tick:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match aborted while the spectator was waiting for the slot")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	# Nothing may have promoted yet: the reservation lapsing is not a kickoff.
	var promoted_before_kickoff: bool = not match_scene._slots[0].disconnected
	var goals: Array = match_scene.arena.get_goals() if match_scene.arena else []
	if is_instance_valid(match_scene.ball) and not goals.is_empty():
		match_scene.ball.linear_velocity = Vector3.ZERO
		match_scene.ball.global_position = goals[0].global_position
		print("SMOKE INFO: host forced a goal to produce a kickoff")

	var promote_deadline := Time.get_ticks_msec() + 10000
	while Time.get_ticks_msec() < promote_deadline and _is_networked_match(match_scene) and match_scene._slots[0].disconnected:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match aborted before the kickoff could promote anyone")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return

	var slot = match_scene._slots[0]
	var took_slot: bool = not slot.disconnected and slot.peer_id != original_peer
	var renamed: bool = slot.player_name != original_name and slot.player_name != ""
	var same_ship: bool = is_instance_valid(slot.ship) and slot.ship == ship_before
	var controller_valid: bool = is_instance_valid(slot.controller)
	# Not load-bearing on its own: the queue also empties when a waiting peer
	# gives up and leaves, which is exactly what a control run with a long
	# reservation showed. took_slot plus the name change is the real evidence.
	var queue_drained: bool = match_scene._late_joiners.is_empty()
	print("SMOKE INFO: late joiner queued=%d promoted_before_kickoff=%s took_slot=%s new_name=%s same_ship=%s queue_drained=%s" % [
		queued, str(promoted_before_kickoff), str(took_slot), slot.player_name, str(same_ship), str(queue_drained)
	])

	# It must be a real seat, not just a relabelled one: hold on and require
	# the new owner's input to move the ship the server owns, sampled while
	# they are still connected.
	var start_position: Vector3 = slot.ship.global_position if is_instance_valid(slot.ship) else Vector3.ZERO
	var last_connected_position := start_position
	var saw_connected := false
	var hold_deadline := Time.get_ticks_msec() + 8000
	while Time.get_ticks_msec() < hold_deadline and _is_networked_match(match_scene):
		if slot.peer_id in multiplayer.get_peers():
			saw_connected = true
			if is_instance_valid(slot.ship):
				last_connected_position = slot.ship.global_position
		await get_tree().physics_frame
	var moved := Vector2(last_connected_position.x - start_position.x, last_connected_position.z - start_position.z).length()
	var drove: bool = saw_connected and moved > 1.0

	var success := queued > 0 and not promoted_before_kickoff and took_slot and renamed \
		and same_ship and controller_valid and queue_drained and drove
	print("SMOKE %s: late joiner took the vacated slot at the kickoff (queued=%d waited_for_kickoff=%s took_slot=%s same_ship=%s drove=%.2fm)" % [
		"PASS" if success else "FAIL", queued, str(not promoted_before_kickoff), str(took_slot), str(same_ship), moved
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# The same promotion from the SPECTATOR's side. It must start with no slot,
# gain one without reloading the scene, and be able to fly it — the client's
# _is_spectator was assigned once at match_config time and never revisited,
# so "the server promoted me" and "I can actually play" are separate claims.
func run_late_joiner_client_check(lifetime_seconds: float) -> void:
	var load_deadline := Time.get_ticks_msec() + 10000
	while Time.get_ticks_msec() < load_deadline and not _is_networked_match(get_tree().current_scene):
		await get_tree().process_frame
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: late joiner never loaded the match scene")
		get_tree().quit(1)
		return
	await get_tree().create_timer(1.0).timeout
	var started_spectating: bool = match_scene._my_slot == null and match_scene._is_spectator
	if not started_spectating:
		print("SMOKE FAIL: late joiner was given a slot immediately — it should spectate until a kickoff (my_slot=%s is_spectator=%s)" % [
			str(match_scene._my_slot != null), str(match_scene._is_spectator)
		])
		NetworkManager.shutdown()
		get_tree().quit(1)
		return

	var promote_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0) + 42000
	while Time.get_ticks_msec() < promote_deadline and _is_networked_match(match_scene) and match_scene._my_slot == null:
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match scene torn down before the late joiner was promoted")
		get_tree().quit(1)
		return
	var promoted: bool = match_scene._my_slot != null and not match_scene._is_spectator
	if not promoted:
		print("SMOKE FAIL: late joiner never got a slot (my_slot=%s is_spectator=%s)" % [
			str(match_scene._my_slot != null), str(match_scene._is_spectator)
		])
		NetworkManager.shutdown()
		get_tree().quit(1)
		return

	# Wait for live play — a promotion lands at a kickoff, so the very next
	# thing is a countdown with every body frozen.
	var live_deadline := Time.get_ticks_msec() + 15000
	while Time.get_ticks_msec() < live_deadline and _is_networked_match(match_scene) and not MatchState.is_live(match_scene.match_state):
		await get_tree().physics_frame
	var my_slot = match_scene._my_slot
	var owns_slot: bool = my_slot != null and my_slot.peer_id == multiplayer.get_unique_id()
	var ship_ok: bool = my_slot != null and is_instance_valid(my_slot.ship)
	# The promoted ship was a REMOTE body a moment ago: frozen kinematic and fed
	# by the interpolator. Promotion deliberately does NOT unfreeze it on the
	# spot — it waits for the first authoritative pose, exactly as a fresh
	# client does — so this WAITS for prediction to start rather than sampling
	# at whichever frame the state happened to go live. Sampling immediately is
	# a race the run loses about half the time, reporting predicting=false on a
	# client that then flew 45m perfectly well.
	# Poll the whole condition, not _local_prediction_ready alone. Unfreezing is
	# QUEUED and applied on the body's own next _integrate_forces (task 0.15),
	# so there is a real window where the state is PLAYING and the flag is set
	# but ship.freeze has not flipped yet — sampling on that frame reported
	# predicting=false for a client that then flew 45m, twice in five runs.
	var predicting := false
	var predict_deadline := Time.get_ticks_msec() + 5000
	while Time.get_ticks_msec() < predict_deadline and _is_networked_match(match_scene):
		predicting = ship_ok and match_scene._local_prediction_ready \
			and not my_slot.ship.freeze and not my_slot.interpolator.has_samples()
		if predicting:
			break
		await get_tree().physics_frame
	var controller_ok: bool = ship_ok and my_slot.ship.controller != null and my_slot.ship.controller.get_parent() == my_slot.ship

	var start_position: Vector3 = my_slot.ship.global_position if ship_ok else Vector3.ZERO
	Input.action_press("move_forward")
	await get_tree().create_timer(3.0).timeout
	Input.action_release("move_forward")
	if not _is_networked_match(match_scene) or not (ship_ok and is_instance_valid(my_slot.ship)):
		print("SMOKE FAIL: promoted client lost its ship or scene mid-drive")
		get_tree().quit(1)
		return
	var end_position: Vector3 = my_slot.ship.global_position
	var moved := Vector2(end_position.x - start_position.x, end_position.z - start_position.z).length()
	var moved_ok := moved > 1.0

	print("SMOKE INFO: promotion spectated_first=%s owns_slot=%s ship_ok=%s predicting=%s (ready=%s frozen=%s interp_samples=%s state=%s) controller_ok=%s moved=%.2fm" % [
		str(started_spectating), str(owns_slot), str(ship_ok), str(predicting),
		str(match_scene._local_prediction_ready), str(ship_ok and my_slot.ship.freeze),
		str(ship_ok and my_slot.interpolator.has_samples()), MatchState.to_name(match_scene.match_state),
		str(controller_ok), moved
	])
	var success := started_spectating and promoted and owns_slot and ship_ok and predicting and controller_ok and moved_ok
	print("SMOKE %s: spectator was promoted to player and can fly the slot it inherited (moved=%.2fm)" % [
		"PASS" if success else "FAIL", moved
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# §6.4's reconnect, graded from the RECONNECTING PLAYER's side. The
# host-disconnect scenario already asserts the server's bookkeeping — slot
# reserved, ship kept, reclaimed by name — but every one of those assertions
# holds identically for a client that came back as a spectator, or came back
# owning a slot whose input pipeline is dead. Both have happened: a stale
# _last_match_config made a reconnecting player a spectator, and that bug was
# visible in this scenario's own logs while it reported PASS.
#
# So this asserts what the returning player actually cares about: I am a
# player and not a spectator, I own a slot with a real ship, the match I
# rejoined is live with a clock already running (§6.2 step 2's bootstrap — a
# reconnecting player must not have to wait for the next goal to learn the
# score), and my input still moves my ship.
func run_reconnect_client_check(settle_seconds: float, drive_seconds: float) -> void:
	var deadline := Time.get_ticks_msec() + int(maxf(settle_seconds, 2.0) * 1000.0)
	while Time.get_ticks_msec() < deadline and not _is_networked_match(get_tree().current_scene):
		await get_tree().process_frame
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: reconnecting client never loaded the match scene")
		get_tree().quit(1)
		return

	# Play may legitimately be paused for a kickoff or a goal when a client
	# rejoins, and a frozen ship cannot be driven — wait for live rather than
	# grading the reconnect on whichever moment it happened to land in.
	var live_deadline := Time.get_ticks_msec() + 15000
	while Time.get_ticks_msec() < live_deadline and _is_networked_match(match_scene) and not MatchState.is_live(match_scene.match_state):
		await get_tree().physics_frame
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: match scene torn down before the reconnecting client could play")
		get_tree().quit(1)
		return

	var my_slot = match_scene._my_slot
	var is_player: bool = my_slot != null and not match_scene._is_spectator
	var ship_ok: bool = is_player and is_instance_valid(my_slot.ship)
	var owns_slot: bool = is_player and my_slot.peer_id == multiplayer.get_unique_id()
	# Reported before the drive, not after. Coming back as a spectator is the
	# specific bug this role exists to catch (a stale _last_match_config caused
	# exactly that), and falling through to the drive would report it as the
	# generic "lost its ship mid-drive" — which is what a control run, rejoining
	# while the slot was still occupied, actually printed.
	if not (is_player and ship_ok and owns_slot):
		print("SMOKE FAIL: reconnecting player did NOT reclaim a slot — is_player=%s owns_slot=%s ship_ok=%s (came back as a spectator?)" % [
			str(is_player), str(owns_slot), str(ship_ok)
		])
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	var state_ok: bool = MatchState.is_live(match_scene.match_state)
	# The bootstrap half (§6.2 step 2). _end_tick stays -1 on a client nobody
	# told about the clock, so this is exactly "did my rejoin carry the live
	# match with it" — a reconnecting player that has to wait for the next goal
	# to learn the clock and score has not really rejoined the match.
	var clock_ok: bool = match_scene._end_tick >= 0

	var start_position: Vector3 = my_slot.ship.global_position if ship_ok else Vector3.ZERO
	Input.action_press("move_forward")
	await get_tree().create_timer(drive_seconds).timeout
	Input.action_release("move_forward")

	if not _is_networked_match(match_scene) or not (ship_ok and is_instance_valid(my_slot.ship)):
		print("SMOKE FAIL: reconnecting client lost its ship or scene mid-drive")
		get_tree().quit(1)
		return
	var end_position: Vector3 = my_slot.ship.global_position
	# Horizontal only: forward thrust is a horizontal force, and full 3D
	# distance is satisfiable by gravity alone from the spawn height.
	var moved := Vector2(end_position.x - start_position.x, end_position.z - start_position.z).length()
	var moved_ok := moved > 1.0

	print("SMOKE INFO: reconnect is_player=%s owns_slot=%s ship_ok=%s state=%s end_tick=%d moved=%.2fm" % [
		str(is_player), str(owns_slot), str(ship_ok), MatchState.to_name(match_scene.match_state),
		match_scene._end_tick, moved,
	])
	var success := is_player and owns_slot and ship_ok and state_ok and clock_ok and moved_ok
	print("SMOKE %s: reconnecting player rejoined as a player and can still drive (player=%s owns_slot=%s clock=%s moved=%.2fm)" % [
		"PASS" if success else "FAIL", str(is_player), str(owns_slot), str(clock_ok), moved,
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


func run_malformed_abuse_check() -> void:
	await get_tree().create_timer(1.0).timeout
	# A single-element Array, not a plain bool: GDScript lambdas capture
	# outer local variables BY VALUE at creation time, not by reference, so
	# `disconnected = true` inside the lambda below would silently mutate
	# only the lambda's own captured copy — invisible to this function's
	# own `disconnected` if it were a plain bool. Mutating an Array's
	# CONTENTS from inside the lambda works because the Array object
	# itself (not a copy of it) is what got captured.
	var disconnected := [false]
	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)

	for i in 25:
		MatchSim._recv_input.rpc_id(1, PackedByteArray([1, 2, 3]))   # far too short to even hold a header
		NetworkManager.poll()
		await get_tree().physics_frame
	await get_tree().create_timer(1.0).timeout
	NetworkManager.poll()

	print("SMOKE %s: 25 malformed packets %s" % [
		"PASS" if disconnected[0] else "FAIL",
		"resulted in disconnect" if disconnected[0] else "did NOT disconnect the abusive peer",
	])
	get_tree().quit(0 if disconnected[0] else 1)


# task 3.4: MatchSim._recv_input must rate-limit and disconnect a sustained
# continuous flood well above RATE_LIMIT_PACKETS_PER_SEC (110/s) via the
# leaky-bucket excess accumulator (RATE_LIMIT_EXCESS_PACKETS_TO_DISCONNECT).
# Every packet here is individually well-formed (a real NetCodec.pack_input
# payload) — only the SEND RATE is abusive, confirming the rate limiter
# fires independently of the malformed-packet counter, not as a side
# effect of it.
func run_rate_limit_abuse_check() -> void:
	await get_tree().create_timer(1.0).timeout
	var disconnected := [false]   # see run_malformed_abuse_check's comment on why not a plain bool
	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)

	var net_codec := preload("res://scripts/net_codec.gd")
	var ship_action_script := preload("res://scripts/ship_action.gd")
	var bytes: PackedByteArray = net_codec.pack_input(1, 0, Time.get_ticks_msec(), [ship_action_script.new()])

	var deadline_ms := Time.get_ticks_msec() + 4000
	while Time.get_ticks_msec() < deadline_ms and not disconnected[0]:
		for i in 40:   # well above 110/s once summed across a frame's worth of iterations
			MatchSim._recv_input.rpc_id(1, bytes)
		NetworkManager.poll()
		await get_tree().process_frame
	await get_tree().create_timer(0.5).timeout
	NetworkManager.poll()

	print("SMOKE %s: sustained packet flood %s" % [
		"PASS" if disconnected[0] else "FAIL",
		"resulted in disconnect" if disconnected[0] else "did NOT disconnect the abusive peer",
	])
	get_tree().quit(0 if disconnected[0] else 1)


# Regression test for a real bug an adversarial review found and this
# session fixed: the ORIGINAL rate limiter tracked "N consecutive
# over-budget seconds" and hard-reset that streak to 0 on any single clean
# window — so a burst-then-idle duty cycle (flood hard, go quiet for one
# window, repeat) evaded it indefinitely. Reproduced against the real
# MatchSim._recv_input: ~33x the packet budget sustained for 28.5s with
# zero disconnect warnings. The fix (a leaky-bucket excess accumulator
# that grows by the window's actual total and drains by only one window's
# worth of budget, every window) doesn't care how the excess is
# distributed in time. This test reproduces the exact attack shape.
func run_duty_cycle_flood_abuse_check() -> void:
	await get_tree().create_timer(1.0).timeout
	var disconnected := [false]
	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)

	var net_codec := preload("res://scripts/net_codec.gd")
	var ship_action_script := preload("res://scripts/ship_action.gd")
	var bytes: PackedByteArray = net_codec.pack_input(1, 0, Time.get_ticks_msec(), [ship_action_script.new()])

	const CYCLE_SECONDS := 3.0
	const BURST_SECONDS := 0.35
	const TEST_SECONDS := 6.0   # the leaky bucket trips within the first cycle; no need for a long soak
	const TRICKLE_HZ := 60   # legitimate-shaped background rate, well under budget alone

	var deadline_ms := Time.get_ticks_msec() + int(TEST_SECONDS * 1000.0)
	var cycle_start_ms := Time.get_ticks_msec()
	while Time.get_ticks_msec() < deadline_ms and not disconnected[0]:
		var t_in_cycle := float(Time.get_ticks_msec() - cycle_start_ms) / 1000.0
		if t_in_cycle >= CYCLE_SECONDS:
			cycle_start_ms = Time.get_ticks_msec()
			t_in_cycle = 0.0
		if t_in_cycle < BURST_SECONDS:
			for i in 200:   # a hard burst, far above budget
				MatchSim._recv_input.rpc_id(1, bytes)
		else:
			for i in maxi(1, TRICKLE_HZ / 60):   # ~60/s trickle, keeps the window rolling and stays under budget alone
				MatchSim._recv_input.rpc_id(1, bytes)
		NetworkManager.poll()
		await get_tree().process_frame
	await get_tree().create_timer(0.5).timeout
	NetworkManager.poll()

	print("SMOKE %s: duty-cycled flood (burst %.2fs / cycle %.1fs) %s" % [
		"PASS" if disconnected[0] else "FAIL", BURST_SECONDS, CYCLE_SECONDS,
		"resulted in disconnect" if disconnected[0] else "evaded rate limiting entirely",
	])
	get_tree().quit(0 if disconnected[0] else 1)


# task 3.6, host role: waits for both bots' scenes to settle, forces a
# deterministic goal (bot-vs-bot scoring isn't reliable enough within a
# short CI run to gate on), then compares the server's own final score
# against what each client independently wrote to disk (run_ci_client_check
# below) — genuine cross-peer agreement, not just "the server thinks so".
func run_ci_host_check(run_seconds: float) -> void:
	await get_tree().create_timer(2.0).timeout
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: host scene is not NetworkedMatch")
		NetworkManager.shutdown()
		get_tree().quit(1)
		return
	print("SMOKE INFO: host ship_count=%d slot_count=%d" % [match_scene.ships.size(), match_scene._slots.size()])

	# Every score the SERVER has actually held, in order. The comparison below
	# used to check each client's recorded score against the server's score at
	# READ time — but the clients write their files several seconds earlier
	# (they wait run_seconds from their own later start, then the host waits
	# run_seconds + 5 more), so any goal scored in that window failed the run
	# with both bots agreeing perfectly with each other and only "disagreeing"
	# with a future they could not have seen. It was latent until the input
	# blackout fix (§3.2) made the bots effective enough to reliably score a
	# SECOND goal: reproduced 2 of 3 runs, and each failure had server=2 vs
	# both clients=1. Cross-peer agreement is the real claim here, so assert
	# that both clients agree with each other AND that what they saw is a
	# state the server genuinely passed through.
	# Polled, not signal-driven: score_changed is emitted only in
	# _on_score_update_received, i.e. the CLIENT path. The server mutates
	# `score` directly in _record_goal and never emits, so connecting here
	# silently recorded nothing but the initial 0-0 (verified — it made all
	# three runs fail with a one-entry history).
	var score_history: Array[String] = [JSON.stringify(match_scene.score)]

	# An adversarial review found this driver's original checks (snapshot
	# count, a server-FORCED goal's cross-peer score agreement) don't
	# depend on client input ever reaching the server at all — it kept
	# reporting PASS with the input pipeline completely dead (verified by
	# injecting the ring-overflow bug this session's critical fix
	# addresses, mid-run). Accumulate each ship's real server-side travel over
	# the run so the input pipeline is checked directly — the same signal
	# run_client_check already uses for a human client, applied here per-bot
	# instead of just for "my own ship". Seeding a peer_id here is what opts it
	# into the per-tick accumulation _await_recording_score performs; see that
	# function for why this cannot be a start-to-end displacement.
	var path_lengths: Dictionary = {}
	for slot in match_scene._slots:
		if is_instance_valid(slot.ship):
			path_lengths[slot.peer_id] = 0.0

	var goals: Array = match_scene.arena.get_goals() if match_scene.arena else []
	if is_instance_valid(match_scene.ball) and not goals.is_empty():
		match_scene.ball.linear_velocity = Vector3.ZERO
		match_scene.ball.global_position = goals[0].global_position
		print("SMOKE INFO: host forced a goal for the cross-peer score agreement check")

	# Movement/stalled must be checked WHILE clients are still actively
	# connected and playing, not after their run finishes — a client's own
	# (legitimate, expected) disconnect at the end of its run naturally
	# starves its jitter buffer too, which looks identical to the ring-
	# overflow bug this check exists to catch if sampled too late. A first
	# attempt used a 0.5s margin (run_seconds - 0.5); a second adversarial
	# review instrumented multiplayer.get_peers() at sample time and found
	# it was already EMPTY — both bots had legitimately disconnected before
	# the sample ran, and the check was only passing on the ~200ms of
	# residual STARVE_ZERO_TICKS starvation grace, not because it was
	# genuinely still connected as this print used to claim. Widen the
	# margin AND assert connectivity directly at sample time, rather than
	# inferring it from timing, so a future regression in either direction
	# (margin too tight again, or client run_seconds changing) fails loudly
	# here instead of silently passing on residual grace. Sample near the end
	# of the active run: a server begins timing as soon as both peers join,
	# whereas each bot needs to load the match and settle before its input can
	# accumulate meaningful motion. The bots remain connected for an extra
	# three seconds after their active run, leaving a generous live margin.
	var movement_check_delay := maxf(1.0, run_seconds - 0.5)
	await _await_recording_score(match_scene, movement_check_delay, score_history, path_lengths)
	var connected_peers := multiplayer.get_peers()
	var input_reached_server := true
	for slot in match_scene._slots:
		var still_connected: bool = slot.peer_id in connected_peers
		if not still_connected:
			input_reached_server = false
			print("SMOKE FAIL: peer %d already disconnected at movement-sample time (connected_peers=%s) — margin too tight" % [slot.peer_id, str(connected_peers)])
		if not is_instance_valid(slot.ship) or not path_lengths.has(slot.peer_id):
			input_reached_server = false
			print("SMOKE FAIL: peer %d has no valid ship to check movement on" % slot.peer_id)
			continue
		# Cumulative travel, so a live pipeline clears this by an enormous margin
		# (seconds of play at up to 35 m/s is tens of metres) while a dead one
		# reads ~0. Kept deliberately loose rather than tightened to match: the
		# check exists to catch a totally dead input path, and the printed value
		# is the evidence for raising it later if that is ever worth doing.
		const MIN_SERVER_TRAVEL := 0.5
		var travelled: float = path_lengths[slot.peer_id]
		var stalled: bool = slot.jitter_buffer.stalled
		print("SMOKE INFO: peer %d travelled %.2fm server-side (connected=%s), stalled=%s" % [slot.peer_id, travelled, str(still_connected), str(stalled)])
		if travelled <= MIN_SERVER_TRAVEL or stalled:
			input_reached_server = false

	# Extra buffer beyond run_seconds: clients run for their own run_seconds
	# measured from THEIR (later) start, so waiting only run_seconds here
	# would race their score files not being written yet.
	await _await_recording_score(match_scene, run_seconds + 5.0 - movement_check_delay, score_history)
	print("SMOKE INFO: host final score=%s (server held: %s)" % [str(match_scene.score), str(score_history)])

	var slots_ok: bool = match_scene._slots.size() == 2
	var scores_agree := true
	var scores_seen := 0
	var client_scores: Array[String] = []
	for slot in match_scene._slots:
		var path := "/tmp/cosmicclash_ci_score_%d.txt" % slot.peer_id
		if not FileAccess.file_exists(path):
			print("SMOKE FAIL: no score file from peer %d at %s" % [slot.peer_id, path])
			scores_agree = false
			continue
		var f := FileAccess.open(path, FileAccess.READ)
		var client_score := f.get_as_text()
		f.close()
		scores_seen += 1
		client_scores.append(client_score)
		if not score_history.has(client_score):
			print("SMOKE FAIL: peer %d saw score %s, which the server never held (history %s)" % [slot.peer_id, client_score, str(score_history)])
			scores_agree = false
	# The strong half: two independent peers must have reached the SAME view.
	for other in client_scores:
		if other != client_scores[0]:
			print("SMOKE FAIL: peers disagree with each other: %s" % str(client_scores))
			scores_agree = false
			break

	var success: bool = slots_ok and scores_agree and scores_seen == 2 and input_reached_server
	print("SMOKE %s: CI host run (slots_ok=%s scores_agree=%s scores_seen=%d/2 input_reached_server=%s)" % [
		"PASS" if success else "FAIL", str(slots_ok), str(scores_agree), scores_seen, str(input_reached_server),
	])
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)


# task 3.6, client-bot role: counts real snapshots received over run_seconds
# (proving steady traffic, not just a handshake) and writes this peer's own
# final server-authoritative score to a peer-id-keyed file for the host to
# compare against the other bot's (run_ci_host_check above).
func run_ci_client_check(run_seconds: float) -> void:
	var snapshot_count := [0]
	MatchSim.snapshot_received.connect(func(_decoded: Dictionary) -> void: snapshot_count[0] += 1)

	await get_tree().create_timer(1.0).timeout
	var match_scene := get_tree().current_scene
	if not _is_networked_match(match_scene):
		print("SMOKE FAIL: client-bot scene is not NetworkedMatch")
		get_tree().quit(1)
		return

	await get_tree().create_timer(run_seconds).timeout

	var slots_ok: bool = not match_scene._slots.is_empty()
	# 60Hz nominal; generous margin for connection/scene-load settle time
	# eaten out of run_seconds and for the odd dropped/simulated-lossy tick.
	var min_expected := int((run_seconds - 2.0) * 30.0)
	var snapshot_count_ok: bool = snapshot_count[0] >= min_expected
	var net_stats: Dictionary = match_scene.get_net_debug_stats()
	var present_time := bool(match_scene.remote_visual_present_time_enabled)
	var remote_position_p95 := float(net_stats.get("remote_residual_position_p95", INF))
	var remote_rotation_p95 := float(net_stats.get("remote_residual_rotation_p95", INF))
	var remote_position_p99 := float(net_stats.get("remote_residual_position_p99", INF))
	var remote_rotation_p99 := float(net_stats.get("remote_residual_rotation_p99", INF))
	# Two bars rather than one loose one. The tight bar moved to p95, which is
	# stable across runs; p99 over a few hundred samples is its worst handful,
	# so on a shared CI host it measures scheduling jitter as much as
	# interpolation. The p99 bar is the product's own tolerance: beyond
	# REMOTE_VISUAL_MAX_OFFSET the visual smoother stops absorbing a correction
	# in one step, so exceeding it is a real defect rather than a slow runner.
	#
	# The single p99 < 0.3 bar produced false failures: two clients in one run
	# reported 0.324 and 0.187 with everything else identical, and the same
	# commit passed and failed in the same minute.
	var remote_quality_ok := not present_time or (
		remote_position_p95 < 0.3 and remote_rotation_p95 < 5.0
		and remote_position_p99 < NetworkedMatch.REMOTE_VISUAL_MAX_OFFSET
		and remote_rotation_p99 < NetworkedMatch.REMOTE_VISUAL_MAX_ROTATION_DEGREES
	)

	var my_id := multiplayer.get_unique_id()
	var score_path := "/tmp/cosmicclash_ci_score_%d.txt" % my_id
	var f := FileAccess.open(score_path, FileAccess.WRITE)
	f.store_string(JSON.stringify(match_scene.score))
	f.close()

	print("SMOKE INFO: client-bot snapshot_count=%d (want >= %d) slots_ok=%s final_score=%s remote_present_time=%s residual_p95=%.3fm/%.3fdeg residual_p99=%.3fm/%.3fdeg (p95 bar %.2fm/%.1fdeg, p99 bar %.2fm/%.1fdeg)" % [
		snapshot_count[0], min_expected, str(slots_ok), str(match_scene.score), str(present_time),
		remote_position_p95, remote_rotation_p95, remote_position_p99, remote_rotation_p99,
		0.3, 5.0, NetworkedMatch.REMOTE_VISUAL_MAX_OFFSET, NetworkedMatch.REMOTE_VISUAL_MAX_ROTATION_DEGREES,
	])
	var success: bool = slots_ok and snapshot_count_ok and remote_quality_ok
	print("SMOKE %s: CI client-bot run" % ("PASS" if success else "FAIL"))
	# The host validates live server-side motion at `run_seconds - 2`. The
	# first client may have entered its scene before the second one joined,
	# so it otherwise can finish and disconnect just before that sample. Stay
	# connected long enough for the host to observe both real input streams.
	await get_tree().create_timer(3.0).timeout
	NetworkManager.shutdown()
	get_tree().quit(0 if success else 1)