CosmicClash/Game/scripts/networked_match.gd

class_name NetworkedMatch
extends GameMode

# Server-authoritative simulation with client-side local-ship prediction.
# The server simulates every slot via RLShipController and broadcasts 60Hz
# snapshots. A client simulates exactly its own unfrozen slot with one real
# controller; every remote slot and the ball stay frozen/interpolated.
#
# No HUD/Arena child in networked_match.tscn — both are built in code, once
# the arena is actually known (the server picks one; the client learns it
# from match_config), which is why this overrides _ready() completely
# rather than relying on GameMode's default (arena-required-synchronously)
# flow.

# Only score_changed is actually emitted in Phase 2 — Phase 5 owns the match
# lifecycle state machine (timer, kickoff countdown, overtime, results), so
# those signals get declared there, alongside real emission. Declaring one
# here without emitting it isn't harmless: HUDController gates the timer
# widget's visibility purely on has_signal("timer_updated"), so a declared-
# but-dead signal shows a permanently frozen timer rather than correctly
# hiding it the way free_play.gd's total absence of the signal does.
signal score_changed(score: Dictionary)
# §6.1 task 5.1. Emitted on BOTH peers — server-side when it drives a
# transition, client-side when it follows one — so HUD/camera work can bind to
# one signal regardless of which process it runs in.
signal match_state_changed(state: int, at_tick: int)
# §6.2's closing note: HUDController duck-types on all five of these
# (HUDController.gd:65, 88, 100, 103, 106) and silently omits a row when one
# is missing. They are declared here AND genuinely emitted from the lifecycle
# handlers below — Phase 2 learned that declaring a signal that never fires is
# worse than not declaring it (has_signal("timer_updated") was true, so the
# HUD showed a permanently frozen timer instead of correctly hiding it).
signal timer_updated(minutes: int, seconds: int)
signal match_ended(winning_team: int, score: Dictionary)
signal kickoff_countdown(count: int)
signal overtime_started

const NetCodec = preload("res://scripts/net_codec.gd")
const NetBodyState = preload("res://scripts/net_body_state.gd")
const NetInterpolator = preload("res://scripts/net_interpolator.gd")
const InputJitterBuffer = preload("res://scripts/input_jitter_buffer.gd")
const InputLeadController = preload("res://scripts/input_lead_controller.gd")
const AdaptiveInputDepthController = preload("res://scripts/adaptive_input_depth_controller.gd")
const LocalPredictionHistory = preload("res://scripts/local_prediction_history.gd")
const NetShipPredictor = preload("res://scripts/net_ship_predictor.gd")
const LocalInputTimeline = preload("res://scripts/local_input_timeline.gd")
const LocalNetShipController = preload("res://scripts/local_net_ship_controller.gd")
const HUD_SCENE = preload("res://scenes/HUD.tscn")

# Minimum plausible interpolation delay even on a same-machine/LAN link —
# §4.6's INTERP_DELAY clamp floor. The full formula (one_way + snapshot
# interval*1.5 + 2.5*jitter_ewma) is simplified here to one_way + interval*1.5
# with no jitter term yet (no jitter EWMA is tracked before Phase 3) — close
# enough for Phase 2's "smooth, not exactly latency-optimal" bar.
const INTERP_DELAY_MIN_MS := 25.0
const INTERP_DELAY_MAX_MS := 200.0
const SNAPSHOT_INTERVAL_MS := 1000.0 / 60.0
const STARVATION_ADVERTISEMENT_TICKS := 4 # ignores expected connection/startup transit
# Consecutive seq-guard rejections before the guard resyncs to the client's
# epoch instead of latching shut forever. Well above any honest transient
# (a legitimate client never trips the bound at all) and far below the
# hundreds of rejections an unrecoverable run produced.
const SEQ_REJECT_RESYNC_LIMIT := 10

# Phase 4.6: a client only predicts the ball immediately after its own ship
# touches it. Authority remains buffered throughout the short window.
@export var local_ball_prediction_enabled := true
# The present-time path passed the two-bot A/B residual gate (<0.3m/<5deg).
# Keep delayed interpolation available through the runtime debug toggle for
# comparison and regression diagnosis.
@export var remote_visual_present_time_enabled := true
const BALL_PREDICTION_MAX_MS := 250
const BALL_HARD_SNAP_DISTANCE := 3.0
const BALL_VISUAL_BLEND_MS := 150
const BALL_RECONTACT_COOLDOWN_MS := BALL_PREDICTION_MAX_MS + BALL_VISUAL_BLEND_MS

# NetInterpolator.to_tick() assumes Time.get_ticks_msec() == physics_frame *
# TICK_MS on the SERVER, i.e. that physics frame 0 happened at process-start
# wall time. It doesn't: real startup work (autoloads, asset loading) elapses
# before the first physics step, and any dropped tick widens the gap further
# — it only ever grows. An adversarial review found this was NOT a rounding
# error: it measured a steady +45-50ms bias on a real run, meaning EVERY
# to_tick(get_server_time_estimate_ms()) call landed 3+ ticks past the
# newest buffered sample, so sample_at() took the extrapolation branch 100%
# of the time — zero real interpolation ever happened, on LAN or under
# simulated latency alike, silently defeating the entire interpolation
# buffer this phase was built around.
#
# Fix: this bias is a property of the server's clock, not of any one body,
# so track ONE shared estimate here (not per-interpolator) from every
# snapshot's own server_tick versus this client's server-time estimate at
# receipt. Take the MINIMUM over a rolling window — same rationale as
# NetworkManager's own min-RTT filtering (network_manager.gd): the sample
# with the least one-way transit delay best isolates the constant epoch
# bias from per-packet network noise, and a rolling (not all-time) window
# lets a real increase in the bias — the server dropping more ticks later
# in the match — still get picked up rather than staying pinned to a
# now-stale historical minimum.
const TICK_BIAS_WINDOW_SEC := 5.0

var _tick_bias_samples: Array[Dictionary] = []   # [{t_ms:int, bias_ms:float}], client only
var _tick_bias_ms := 0.0                          # best current estimate; 0.0 until the first snapshot


class SlotInfo:
	var peer_id: int
	var team: int
	var spawn_index: int
	var ship: Ship
	var controller: RLShipController      # server only
	var jitter_buffer := InputJitterBuffer.new()   # server only (§3.2)
	# Server only. Consecutive packets rejected by the seq-range guard, reset by
	# any accepted one. The guard's bound is derived from a value only an
	# ACCEPTED packet can advance, so without an escape hatch it latches shut
	# permanently — see the guard's own comment in _on_input_received.
	var consecutive_seq_rejects := 0
	var last_client_send_ms := 0          # server only: echoed back per-peer next snapshot (§2.4)
	var interpolator := NetInterpolator.new()   # client only
	var visual_smoother_reset := true
	var visual_position_offset := Vector3.ZERO
	var visual_rotation_offset := Quaternion.IDENTITY


var _slots: Array[SlotInfo] = []
var _my_slot: SlotInfo = null             # client only
var _local_prediction_ready := false      # client waits for its first authoritative pose
var _ball_interpolator := NetInterpolator.new()   # client only
var _ball_shadow_state: NetBodyState = null       # newest authority for the frozen remote shadow
var _local_ball_proxy: Ball = null                 # client-only dynamic collision/prediction body
var _ball_prediction_until_ms := -1
var _ball_recontact_cooldown_until_ms := -1
var _ball_prediction_contact_count := 0
var _ball_visual_blend_from := Transform3D.IDENTITY
var _ball_visual_blend_started_ms := -1
var _last_ball_prediction_error := 0.0
var _ball_contact_frame := -1
var _ball_reveal_frame := -1
var _ball_blend_complete_count := 0
var _ball_blend_started_count := 0
var _ball_blend_max_duration_ms := 0
var _ball_hard_handoff_count := 0
var _ball_prediction_window_end_count := 0
var _ball_prediction_missing_shadow_count := 0
var _ball_prediction_reset_cancel_count := 0
var _ball_reset_trace: Array[String] = []
var _ball_proxy_contact_position := Vector3.ZERO
var _ball_proxy_moved_before_authority := false
var _ball_proxy_moved_before_authority_count := 0
var _ball_shadow_position_on_contact := Vector3.ZERO
var _ball_authority_changed_since_contact := false
var _remote_position_residuals: Array[float] = []
var _remote_rotation_residuals: Array[float] = []
var _ball_visual_smoother_reset := true
var _ball_visual_position_offset := Vector3.ZERO
var _ball_visual_rotation_offset := Quaternion.IDENTITY
const REMOTE_VISUAL_SMOOTH_RATE := 14.0
const REMOTE_VISUAL_HARD_DISTANCE := 2.0
const REMOTE_VISUAL_MAX_OFFSET := 0.4
const REMOTE_VISUAL_MAX_ROTATION_DEGREES := 15.0
const REMOTE_METRIC_CAPACITY := 3600
# --test-bot (task 3.6): CI/regression driver mode, an automated player via
# the existing AIShipController instead of a human — see CLAUDE.md's testing
# section. Read once in _ready(), consumed in _on_match_config_received.
var _test_bot_model_path := ""   # client only; non-empty means --test-bot mode is active
var _local_input_timeline: LocalInputTimeline = null
var _local_net_controller: LocalNetShipController = null
var _input_seq := 0                       # client only
# Redundancy (§3.1): newest-first, capped at NetCodec.MAX_REDUNDANCY, so a
# 3-packet burst loss still recovers every tick's action via a later
# packet's history. Client only.
var _input_history: Array[ShipAction] = []
var _local_prediction_history := LocalPredictionHistory.new()   # client only; 128-entry seq-tagged history (§4.3)
var _local_ship_predictor := NetShipPredictor.new()              # client only; reconciliation policy (§4.4)
# Latest raw comparison retained for diagnostics. NetShipPredictor consumes
# the same result immediately to apply the reconciliation decision.
var _last_local_prediction_comparison: Dictionary = {}
var _action_marker_samples := 0
var _action_marker_mismatches := 0
var _pending_local_reconciliation: Dictionary = {} # newest snapshot only; consumed once per physics tick
var _last_local_reset_gen := -1
var _last_received_snapshot_tick := 0     # client only: echoed back as ack_snapshot_tick
var _input_lead_controller := InputLeadController.new()   # client only (§3.3)
var _last_known_input_buffer_depth := -1  # client only: -1 = no snapshot with this field yet
var _has_received_healthy_buffer_depth := false
var _adaptive_input_depth := AdaptiveInputDepthController.new()
# Loss estimate (task 3.7's debug overlay), client only: snapshots go out
# at a steady one-tick cadence, so a server_tick that jumps by more than 1
# since the last received one is direct evidence of a dropped or reordered
# snapshot on the unreliable channel. EWMA over each reception's own
# "missed / (missed + 1)" fraction rather than a flat drop-count, so it
# reads as a live percentage and decays naturally once loss stops.
const SNAPSHOT_LOSS_EWMA_ALPHA := 1.0 / 16.0
var _snapshot_loss_ewma := 0.0
var _expected_next_snapshot_tick := -1
# An adversarial review found _snapshot_loss_ewma only updates on receipt —
# during a TOTAL outage, exactly when this metric matters most, it freezes
# at its last (probably low/healthy) value instead of climbing toward
# 100%. Track wall-clock receipt time so get_net_debug_stats() can report
# honestly once too long has passed with nothing arriving at all.
var _last_snapshot_wall_ms := -1
const SNAPSHOT_STALE_MS := 500.0   # ~30 ticks with nothing at all — treat as total loss, not "still fine"
var _unknown_sender_input_count := 0      # server only, observability (§3.1 step 1)
var _reset_gen := 0                       # server only: bumped on every kickoff/goal reset so the client hard-snaps instead of interpolating across the teleport
# Server only. _on_goal_scored's reset_ball()/reset_ships() only QUEUE
# teleports (task 0.15's queue_teleport — applied on each body's next
# _integrate_forces), but _broadcast_snapshot runs later in the SAME frame
# _on_goal_scored fires in, before that teleport lands. Bumping _reset_gen
# immediately would tag the still-pre-teleport snapshot with the new
# generation: the client clears its buffer expecting a hard snap, then
# keeps exactly that stale in-goal sample and lerps a full-arena slide to
# the next, genuinely-post-teleport sample — an adversarial review measured
# a 26.8m ball slide from this.
#
# A plain "bump on the next _physics_process" boolean flag turned out NOT
# to fix it: the goal Area's body_entered signal (and so _on_goal_scored)
# fires as part of physics tick N's OWN step processing, before tick N's
# _physics_process callback — so a flag set there is already true by the
# time that SAME tick's _physics_process checks it, consuming on tick N
# instead of N+1 as intended (empirically confirmed: with a boolean flag,
# gen still bumped on the same tick the stale position was broadcast).
# The queued teleport, by contrast, isn't applied until tick N+1's
# _integrate_forces. So the two must be compared by TICK NUMBER, not by
# "next callback": only bump once the current tick is strictly later than
# the tick the goal was detected on, which guarantees at least one full
# _integrate_forces has run — and therefore the queued teleport has
# landed — since the flag was set.
var _pending_reset_gen_bump := false
var _pending_reset_gen_bump_tick := -1

# §6.1 task 5.1. Authoritative on the server; on a client this mirrors what
# the server last told us, via state_change (prompt, carries at_tick) or the
# snapshot's match_state byte (the catch-up path — see _apply_match_state).
var match_state := MatchState.State.LOADING
var match_state_since_tick := 0
# Client only: the match_state byte of the most recently decoded snapshot.
# Distinct from `match_state` on purpose — it is what the WIRE said, so a test
# can prove the byte is genuinely populated rather than passing on the
# reliable state_change RPC alone.
var _last_snapshot_match_state := -1
# Server only: the tick the current state's own timer expires on, or -1 when
# the state has no timer (PLAYING ends on a goal or the clock, not a deadline).
var _state_deadline_tick := -1
# Placeholder durations. Task 5.3 replaces the WARMUP one with the real
# broadcast kickoff (reset transforms + a countdown derived from server_tick),
# and 5.4 replaces the GOAL_PAUSE one with _goal_pause_seconds() and the
# client-cinematic split. They exist here only so 5.1 drives REAL transitions
# to verify against, rather than a state machine nothing ever moves.
const WARMUP_TICKS := 3 * SimConstants.TICK_HZ    # 3s kickoff countdown (§6.2 step 6)
const RESULTS_TICKS := 8 * SimConstants.TICK_HZ   # how long RESULTS holds before returning to the lobby

# §6.2 step 9. Tick-derived, never a Timer: `remaining = end_tick - now`.
# -1 until the first kickoff arms it.
var _end_tick := -1
var _clock_running := false
var _last_emitted_second := -1
@export var match_length_seconds := 150.0

# §6.2 step 6. The tick play resumes on — the countdown's own end. Both peers
# derive the displayed count from this and their own server-tick estimate, so
# nothing depends on a local Timer staying in step.
var _kickoff_resume_tick := -1
# Client only: a kickoff that arrived before _slots existed (see
# _on_kickoff_received), replayed once match_config lands.
var _pending_kickoff := {}
# Freeze is applied on a strictly later tick than the kickoff teleport that
# precedes it — see _apply_kickoff. -1 when nothing is pending.
var _pending_freeze_tick := -1
var _last_emitted_countdown := -1
var _in_overtime := false
var _match_over := false


func _ready() -> void:
	add_to_group("game")
	Engine.max_physics_steps_per_frame = 4
	if kickoff_rng_seed == 0:
		_kickoff_rng.randomize()
	if multiplayer.is_server():
		for arg: String in OS.get_cmdline_user_args():
			if arg.begins_with("--match-length="):
				# Regulation is 150s; a smoke test cannot wait that long to see
				# FULL_TIME/OVERTIME/RESULTS, so allow an override. Server-side
				# only — a client cannot shorten anyone's match.
				match_length_seconds = maxf(1.0, arg.get_slice("=", 1).to_float())
		_start_server()
	else:
		for arg: String in OS.get_cmdline_user_args():
			if arg == "--test-bot":
				_test_bot_model_path = "res://bots/promoted/medium.json"
			elif arg.begins_with("--test-bot-model="):
				_test_bot_model_path = arg.get_slice("=", 1)
			elif arg == "--remote-present-time":
				# Explicit A/B opt-in remains useful even though present time is
				# now the default; it also makes test intent visible in logs.
				remote_visual_present_time_enabled = true
			elif arg == "--remote-delayed":
				# A/B control: preserves the former delayed-interpolation render
				# path exactly, with no present-time residual offset applied.
				remote_visual_present_time_enabled = false
		MatchSim.match_config_received.connect(_on_match_config_received)
		MatchSim.snapshot_received.connect(_on_snapshot_received)
		MatchSim.score_update_received.connect(_on_score_update_received)
		MatchSim.state_change_received.connect(_on_state_change_received)
		MatchSim.kickoff_received.connect(_on_kickoff_received)
		MatchSim.goal_scored_received.connect(_on_goal_scored_received)
		MatchSim.clock_state_received.connect(_on_clock_state_received)
		_request_match_config_until_received()


# The one-shot server broadcast in _start_server() is racy against however
# long this client's own scene load took to reach this line — it may have
# already fired into a MatchSim with no listener connected yet, or the
# server may not have even started the match yet. Keep asking until
# _on_match_config_received actually populates _slots.
func _request_match_config_until_received() -> void:
	while _slots.is_empty() and is_inside_tree():
		MatchSim.request_match_config()
		await get_tree().create_timer(0.5).timeout


func _owns_goal_logic() -> bool:
	return multiplayer.is_server()


func _owns_world_simulation() -> bool:
	return multiplayer.is_server()


func _exit_tree() -> void:
	pass


# ============================================================
# Server
# ============================================================

func _start_server() -> void:
	var arena_path := ArenaRegistry.random_path()
	arena = (load(arena_path) as PackedScene).instantiate()
	add_child(arena)
	for goal in arena.get_goals():
		goal.goal_scored.connect(_handle_goal_scored)

	spawn_ball()

	var peer_ids := PackedInt32Array()
	var teams := PackedInt32Array()
	var spawn_indices := PackedInt32Array()
	var team_counts := {0: 0, 1: 0}
	var sorted_peer_ids: Array = MatchNet.roster.keys()
	sorted_peer_ids.sort()
	for peer_id in sorted_peer_ids:
		var info: MatchNet.PlayerInfo = MatchNet.roster[peer_id]
		var spawn_index: int = team_counts.get(info.team, 0)
		team_counts[info.team] = spawn_index + 1
		var slot := SlotInfo.new()
		slot.peer_id = peer_id
		slot.team = info.team
		slot.spawn_index = spawn_index
		slot.controller = RLShipController.new()
		slot.ship = spawn_ship(info.team, spawn_index, slot.controller)
		_slots.append(slot)
		peer_ids.append(peer_id)
		teams.append(info.team)
		spawn_indices.append(spawn_index)

	MatchSim.send_match_config(arena_path, peer_ids, teams, spawn_indices)
	MatchSim.input_received.connect(_on_input_received)

	# §6.1: the arena, ball and every slot's ship now exist and match_config is
	# out, so LOADING is genuinely over. Task 5.3 gates this on the clients'
	# own scene_ready (with a 10s timeout) instead of leaving immediately —
	# there is no scene_ready message yet, and inventing half of one here
	# would be worse than the honest placeholder.
	_apply_match_state(MatchState.State.LOADING, Engine.get_physics_frames())
	_set_match_state(MatchState.State.WARMUP)
	# The clock covers regulation only and is armed once; the goal-pause
	# extension below (§6.2 step 9) adjusts end_tick rather than restarting it.
	_arm_clock(int(match_length_seconds * SimConstants.TICK_HZ) + WARMUP_TICKS)
	_begin_kickoff()


func _on_input_received(peer_id: int, decoded: Dictionary) -> void:
	for slot in _slots:
		if slot.peer_id == peer_id:
			var seq: int = decoded["seq"]
			# §3.1 step 4, rebound after an adversarial review found the
			# original check (seq > Engine.get_physics_frames() + 20)
			# compared two unrelated epochs: get_physics_frames() counts
			# from the SERVER PROCESS's own start, while a client's
			# _input_seq starts at 0 when ITS match scene loads —
			# input_jitter_buffer.gd's own seeding logic exists specifically
			# because these share no baseline (see its header comment).
			# Bounding against server uptime meant this guard could never
			# fire on a long-running dedicated server (no real protection —
			# the stated "keeps garbage-far-future seq values out of the
			# ring" rationale wasn't actually achieved), and could silently
			# drop an honest client's input forever the moment accumulated
			# server tick loss closed whatever accidental head-start margin
			# existed.
			#
			# A first rebound bounded against this slot's own
			# last_applied_seq — the CONSUMER's position — using the ring's
			# capacity as the bound. A second adversarial review found this
			# broke the ring-overflow resync it was landed alongside: capping
			# every accepted seq at last_applied_seq + RING_SIZE also caps
			# jb.highest_ingested_seq at that same ceiling, so
			# consume()'s resync condition (which needs highest_ingested_seq
			# to reach expected + RING_SIZE) could never fire in production —
			# silently recreating the exact permanent-input-death bug this
			# whole guard-rebound was part of fixing, at an even LOWER
			# freeze threshold, reachable via ordinary server tick loss alone
			# with no external trigger.
			#
			# Bound against jb.highest_ingested_seq instead — the highest
			# seq this slot has ever actually been ALLOWED to ingest, i.e.
			# the client's own send epoch — using the ring's own capacity as
			# the bound, same as before. An honest client's consecutive
			# packets differ by only a few seq (redundancy + a bounded
			# input_lead skip), so this bound tracks a well-behaved client
			# regardless of how far the CONSUMER has fallen behind, while
			# still rejecting a single garbage-far-future jump: an attacker
			# can only walk highest_ingested_seq forward at the rate the
			# packet-rate limiter (§3.4) already allows. Falls back to seq
			# itself (never rejects) before the buffer has ever been
			# seeded — there's no baseline yet to bound against.
			# THIRD rebound, and the first one that cannot latch. Every previous
			# version bounded `seq` against a value that only an ACCEPTED packet
			# can advance (server uptime, then last_applied_seq, then
			# highest_ingested_seq) — which makes the guard a one-way door: once
			# a client's live sequence gets far enough ahead, every packet is
			# rejected, the bound can never move again, and that player's input
			# is dead for the rest of the match with no diagnostic. An
			# adversarial review reproduced exactly that with a 2s SIGSTOP host
			# freeze: 600+ consecutive rejections, the server applying zero
			# thrust for 1300 sequences while the client's wire carried full
			# thrust throughout, unrecoverable.
			#
			# Keep the bound (it still rejects a single garbage-far-future jump
			# on the spot) but give it an escape: after SEQ_REJECT_RESYNC_LIMIT
			# consecutive rejections the client is evidently not a one-off
			# glitch but a real peer whose epoch has genuinely run away from
			# ours, so accept the packet and let ingest()/consume()'s existing
			# resync machinery re-establish the baseline. This grants an
			# attacker nothing new: walking the epoch forward by sustained
			# rejection costs the same packets as walking it forward by
			# acceptance, and §3.4's rate limiter already bounds that rate.
			var jb := slot.jitter_buffer
			var seq_bound: int = (jb.highest_ingested_seq if jb.highest_ingested_seq >= 0 else seq) + InputJitterBuffer.RING_SIZE
			if seq > seq_bound:
				slot.consecutive_seq_rejects += 1
				if slot.consecutive_seq_rejects < SEQ_REJECT_RESYNC_LIMIT:
					return
				# Fall through and accept: this is the escape hatch, not a
				# missing `return`.
			slot.consecutive_seq_rejects = 0
			jb.ingest(seq, decoded["actions"])
			slot.last_client_send_ms = decoded["client_send_ms"]
			return
	# A connected-but-not-yet-slotted peer (or one whose slot somehow
	# vanished) sending input — harmless (the packet is simply dropped,
	# same as always), but worth counting for observability (§3.1 step 1)
	# rather than silently discarding with no trace at all.
	_unknown_sender_input_count += 1


# --- §6.1 match state machine (task 5.1) -----------------------------------
#
# Deliberately does NOT gate physics, freezing or input this task. Tasks 5.3
# and 5.4 own freeze/unfreeze at kickoff and goal, and doing it here would
# both duplicate that work and silently change the conditions every Phase 4
# prediction gate was measured under. 5.1's job is the machine, the broadcast
# and the client following it.
func _set_match_state(new_state: int) -> void:
	if not multiplayer.is_server():
		push_error("NetworkedMatch: only the server may drive match state")
		return
	if new_state == match_state:
		return
	if not MatchState.can_transition(match_state, new_state):
		# Loud, not silent: this is a server logic error, and the symptom it
		# produces otherwise (clients faithfully following into a state the
		# server's own code never meant to reach) is near-impossible to
		# diagnose from a field report.
		push_error("NetworkedMatch: illegal match state transition %s -> %s" % [
			MatchState.to_name(match_state), MatchState.to_name(new_state)
		])
		return
	var at_tick := Engine.get_physics_frames()
	_apply_match_state(new_state, at_tick)
	MatchSim.send_state_change(new_state, at_tick)


# The one place either peer's state actually changes, so the signal and the
# bookkeeping cannot drift apart between the server and client paths.
func _apply_match_state(new_state: int, at_tick: int) -> void:
	if new_state == match_state:
		return
	match_state = new_state
	match_state_since_tick = at_tick
	_state_deadline_tick = -1
	if multiplayer.is_server():
		match new_state:
			MatchState.State.RESULTS:
				_state_deadline_tick = at_tick + RESULTS_TICKS
			MatchState.State.GOAL_PAUSE:
				# Owned here rather than assigned by the caller: _apply_match_state
				# resets _state_deadline_tick on every transition, so a deadline
				# set BEFORE _set_match_state was silently wiped and the match sat
				# in GOAL_PAUSE forever. at_tick is the goal tick, so this matches
				# the resume_tick already broadcast to clients.
				_state_deadline_tick = at_tick + int(_goal_pause_seconds() * SimConstants.TICK_HZ)
			MatchState.State.PLAYING, MatchState.State.OVERTIME:
				# Kickoff is over: bodies move again, and the clock resumes.
				_pending_freeze_tick = -1
				_set_bodies_frozen(false)
	# The clock only advances during live play (§6.2 step 9). Derived here
	# rather than tracked separately so it cannot disagree with the state.
	_clock_running = MatchState.is_live(new_state) and not _match_over
	if new_state == MatchState.State.LOBBY and not multiplayer.is_server():
		# §6.2 step 10: both sides return to the LOBBY, not the main menu.
		# Deferred because this runs from an RPC handler mid-tree-traversal
		# (gotcha 27: change_scene_to_file must not be called synchronously
		# from inside a node's own callback chain).
		get_tree().change_scene_to_file.call_deferred(ScenePaths.LOBBY)
	match_state_changed.emit(new_state, at_tick)


# --- §6.2 step 6: kickoff (task 5.3) ---------------------------------------

# Server: reset every body, then broadcast the RESULTING transforms. §1's
# locked decision — never a shared RNG seed, because shared-seed determinism
# needs both sides to consume the stream in identical order forever and the
# first randf() anyone adds to the reset path desyncs kickoff silently.
func _begin_kickoff() -> void:
	reset_ball()
	reset_ships()
	# Bump before the broadcast so the kickoff and the reset_gen it announces
	# describe the same world. This deliberately does NOT use Phase 2's
	# deferred _pending_reset_gen_bump path: that exists because the GOAL
	# sensor fires mid-tick, before the queued teleport lands. Here we are
	# the ones issuing the teleport, and we send the transforms explicitly
	# rather than relying on a snapshot taken after they apply.
	_reset_gen = (_reset_gen + 1) % 256
	_pending_reset_gen_bump = false
	var countdown_start_tick := Engine.get_physics_frames()
	var positions := PackedVector3Array()
	var rotations := PackedFloat32Array()
	for slot in _slots:
		var t: Transform3D = slot.ship.global_transform if is_instance_valid(slot.ship) else Transform3D.IDENTITY
		_append_kickoff_body(positions, rotations, _pending_teleport_or_current(slot.ship, t))
	if is_instance_valid(ball):
		_append_kickoff_body(positions, rotations, _pending_teleport_or_current(ball, ball.global_transform))
	MatchSim.send_kickoff(positions, rotations, countdown_start_tick, _reset_gen)
	_apply_kickoff(positions, rotations, countdown_start_tick, _reset_gen)


# reset_ball()/reset_ships() QUEUE a teleport applied in the body's own next
# _integrate_forces (task 0.15), so global_transform still reads the PRE-reset
# pose right now. Broadcasting that would send every client the old position
# and then correct it a tick later — the same class of bug as Phase 2's 27m
# goal slide. Read the queued target instead when there is one.
func _pending_teleport_or_current(body: Node, fallback: Transform3D) -> Transform3D:
	if is_instance_valid(body) and body.has_method("get_pending_teleport"):
		var pending = body.call("get_pending_teleport")
		if pending != null:
			return pending
	return fallback


func _append_kickoff_body(positions: PackedVector3Array, rotations: PackedFloat32Array, t: Transform3D) -> void:
	positions.append(t.origin)
	var q := t.basis.get_rotation_quaternion().normalized()
	rotations.append_array(PackedFloat32Array([q.x, q.y, q.z, q.w]))


# Both peers. Places bodies exactly, freezes them, and arms the countdown.
func _apply_kickoff(positions: PackedVector3Array, rotations: PackedFloat32Array, countdown_start_tick: int, reset_gen: int) -> void:
	_reset_gen = reset_gen
	_kickoff_resume_tick = countdown_start_tick + WARMUP_TICKS
	_last_emitted_countdown = -1
	var index := 0
	for slot in _slots:
		if index < positions.size() and is_instance_valid(slot.ship):
			_place_body(slot.ship, positions[index], _quat_at(rotations, index))
		index += 1
	if index < positions.size() and is_instance_valid(ball):
		_place_body(ball, positions[index], _quat_at(rotations, index))
	# Freeze on a LATER tick, not now. _place_body queues the teleport into the
	# body's next _integrate_forces, but a frozen body never runs one — and
	# set_deferred("freeze", true) lands at the end of this idle frame, before
	# that next physics step. Freezing immediately therefore strands the
	# teleport and leaves every body exactly where the goal left it. This is
	# the same "queued teleport lands a tick later" hazard Phase 2 hit with
	# _pending_reset_gen_bump_tick, and the same fix: gate on a strictly later
	# tick so the teleport has provably applied.
	_pending_freeze_tick = Engine.get_physics_frames() + 1
	# A client's prediction history describes the pre-kickoff world. Starting a
	# fresh epoch is the same contract §4.4 already specifies for a reset_gen
	# change; doing it here too means a kickoff that arrives before the first
	# post-kickoff snapshot cannot be reconciled against stale history.
	if not multiplayer.is_server() and _local_prediction_history != null:
		_local_prediction_history.begin_epoch()
		_last_local_reset_gen = reset_gen
	# §6.2's explicit late-arrival case: a kickoff delayed past its own resume
	# tick (ENet RTO can stretch a lifecycle burst to ~600ms on a lossy link)
	# must apply the reset immediately and SKIP the countdown, never schedule
	# it into the past and render a negative number.
	if _current_server_tick() >= _kickoff_resume_tick:
		_kickoff_resume_tick = -1
		_pending_freeze_tick = -1   # never freeze for a countdown already over
		kickoff_countdown.emit(0)
		_set_bodies_frozen(false)


func _quat_at(rotations: PackedFloat32Array, index: int) -> Quaternion:
	var base := index * 4
	if base + 3 >= rotations.size():
		return Quaternion.IDENTITY
	return Quaternion(rotations[base], rotations[base + 1], rotations[base + 2], rotations[base + 3]).normalized()


func _place_body(body: Node, position: Vector3, rotation: Quaternion) -> void:
	var target := Transform3D(Basis(rotation), position)
	# A body that is ALREADY frozen never runs _integrate_forces, so a queued
	# teleport would sit unapplied until something unfroze it — which on a
	# client is never, for the permanently-kinematic remote bodies. Those are
	# transform-driven by design (_apply_collider_state does exactly this), so
	# write directly. Anything still simulating goes through the Jolt-safe
	# queue instead (task 0.15): writing state.transform outside the body's own
	# _integrate_forces races the physics step.
	if body is RigidBody3D and (body as RigidBody3D).freeze:
		(body as RigidBody3D).global_transform = target
		(body as RigidBody3D).linear_velocity = Vector3.ZERO
		(body as RigidBody3D).angular_velocity = Vector3.ZERO
	else:
		body.call("queue_teleport_with_velocity", target, Vector3.ZERO, Vector3.ZERO)
	if body is Ship:
		var ship := body as Ship
		ship.net_visual_offset = Vector3.ZERO
		ship.net_visual_rotation_offset = Quaternion.IDENTITY
		if is_instance_valid(ship.visual):
			ship.visual.position = Vector3.ZERO
			ship.visual.basis = Basis.IDENTITY


func _set_bodies_frozen(frozen: bool) -> void:
	# set_deferred, matching match_mode.gd's own _set_frozen: `freeze` is a
	# physics-server-backed property and writing it mid-step is unsafe.
	#
	# ASYMMETRIC BY NECESSITY. On the server every body is a real dynamic
	# simulation and all of them freeze. On a CLIENT, `freeze` is already
	# load-bearing for something else: remote ships and the ball are
	# permanently FREEZE_MODE_KINEMATIC and driven purely by transform writes
	# from the interpolator, and only the local ship is unfrozen so Phase 4 can
	# predict it. Freezing "all bodies" on a client therefore UNFREEZES the
	# remote ones on the way back out — they immediately start falling under
	# gravity while the interpolator fights them for the transform. That is
	# what it did: 210 hard snaps and an infinite p99 in the first run.
	# A client only ever freezes the one body it actually simulates; the
	# remote ones already stop moving because the server's snapshots stop
	# changing.
	if multiplayer.is_server():
		if is_instance_valid(ball):
			ball.set_deferred("freeze", frozen)
		for slot in _slots:
			if is_instance_valid(slot.ship):
				slot.ship.set_deferred("freeze", frozen)
		return
	if _my_slot != null and is_instance_valid(_my_slot.ship):
		_my_slot.ship.set_deferred("freeze", frozen)


func _apply_pending_freeze() -> void:
	if _pending_freeze_tick < 0 or Engine.get_physics_frames() <= _pending_freeze_tick:
		return
	_pending_freeze_tick = -1
	_set_bodies_frozen(true)


func _on_kickoff_received(positions: PackedVector3Array, rotations: PackedFloat32Array, countdown_start_tick: int, reset_gen: int) -> void:
	# match_config and kickoff are both reliable channel-0 messages, but a
	# client that is still loading its scene can receive the kickoff before it
	# has built _slots — and body order is slot order, so applying it early
	# placed the BALL at positions[0], i.e. exactly on top of the first ship.
	# The visible symptom was the ball-cam spamming "target vector can't be
	# zero" because its look-from and look-at had become the same point.
	# Hold it until the roster exists, then apply.
	if _slots.size() + 1 != positions.size():
		_pending_kickoff = {
			"positions": positions, "rotations": rotations,
			"countdown_start_tick": countdown_start_tick, "reset_gen": reset_gen,
		}
		return
	_apply_kickoff(positions, rotations, countdown_start_tick, reset_gen)


func _apply_pending_kickoff() -> void:
	if _pending_kickoff.is_empty():
		return
	var k := _pending_kickoff
	_pending_kickoff = {}
	if _slots.size() + 1 != (k["positions"] as PackedVector3Array).size():
		# Still inconsistent (a roster change between the two messages). The
		# snapshot stream carries authoritative poses every tick regardless, so
		# dropping a stale kickoff is safe — it only costs the countdown.
		push_warning("NetworkedMatch: dropping a kickoff whose body count never matched the roster")
		return
	_apply_kickoff(k["positions"], k["rotations"], int(k["countdown_start_tick"]), int(k["reset_gen"]))


# Both peers, once per physics tick. Emits the countdown from absolute ticks
# so the two sides agree without either running a local Timer.
func _update_kickoff_countdown() -> void:
	if _kickoff_resume_tick < 0:
		return
	var remaining_ticks := _kickoff_resume_tick - _current_server_tick()
	if remaining_ticks <= 0:
		_kickoff_resume_tick = -1
		_last_emitted_countdown = 0
		kickoff_countdown.emit(0)
		if not multiplayer.is_server():
			# The server unfreezes via its own PLAYING/OVERTIME transition;
			# a client does it here so it never waits a round trip to move.
			_set_bodies_frozen(false)
		return
	var count := int(ceil(float(remaining_ticks) / float(SimConstants.TICK_HZ)))
	if count != _last_emitted_countdown:
		_last_emitted_countdown = count
		kickoff_countdown.emit(count)


# --- §6.2 step 8: goals (task 5.4) -----------------------------------------

func _on_goal_scored_received(scoring_team: int, new_score: Dictionary, goal_tick: int, resume_tick: int) -> void:
	# Client. Authoritative score first, then presentation — a client must
	# never derive the score from its own sensor.
	score = new_score.duplicate()
	score_changed.emit(score.duplicate())
	_set_bodies_frozen(true)
	# The cinematic is bounded by [goal_tick, resume_tick] (§6.2 step 8), and
	# is presentation only: it never gates when play resumes, which is what
	# kept the server resetting while clients were mid-celebration.
	_play_goal_celebration(scoring_team, 1 - scoring_team)


# --- §6.2 step 9: clock (task 5.2) -----------------------------------------

func _current_server_tick() -> int:
	if multiplayer.is_server():
		return Engine.get_physics_frames()
	# Before the clock has synced this estimate is meaningless (Phase 2 fix
	# (5)); match_state_since_tick is the best bound available until then.
	if NetworkManager.rtt_ms < 0.0:
		return match_state_since_tick
	return _estimated_tick(NetworkManager.get_server_time_estimate_ms())


func _arm_clock(length_ticks: int) -> void:
	_end_tick = Engine.get_physics_frames() + length_ticks
	_broadcast_clock_state()


func _broadcast_clock_state() -> void:
	MatchSim.send_clock_state(_clock_running, _end_tick, Engine.get_physics_frames())


func _on_clock_state_received(running: bool, end_tick: int, _at_tick: int) -> void:
	_clock_running = running
	_end_tick = end_tick


# Both peers. Emits timer_updated only when the displayed second changes, the
# same threshold pattern Ship uses for its telemetry signals.
func _update_clock() -> void:
	if _end_tick < 0:
		return
	var remaining_ticks := maxi(0, _end_tick - _current_server_tick())
	var remaining_seconds := int(ceil(float(remaining_ticks) / float(SimConstants.TICK_HZ)))
	if remaining_seconds != _last_emitted_second:
		_last_emitted_second = remaining_seconds
		timer_updated.emit(remaining_seconds / 60, remaining_seconds % 60)


# --- §6.2 step 10: full time, overtime, results (task 5.5) -----------------

func _enter_results(winning_team: int) -> void:
	_match_over = true
	_clock_running = false
	_set_bodies_frozen(true)
	match_ended.emit(winning_team, score.duplicate())
	_set_match_state(MatchState.State.RESULTS)


func _winning_team() -> int:
	if score[0] == score[1]:
		return -1
	return 0 if score[0] > score[1] else 1


# Server only, once per physics tick. Advances the states that end on their
# own timer; goal- and clock-driven exits are pushed in from their own events.
func _update_match_state() -> void:
	var now := Engine.get_physics_frames()
	# Full time is checked before the deadline switch below so a clock expiry
	# during PLAYING is acted on the tick it happens, not one state later.
	if _clock_running and _end_tick >= 0 and now >= _end_tick:
		if match_state == MatchState.State.PLAYING:
			_set_match_state(MatchState.State.FULL_TIME)
			return
	# A kickoff countdown ending is what starts play; the resume tick is
	# authoritative, not a separate deadline, so the two cannot drift apart.
	if _kickoff_resume_tick >= 0 and now >= _kickoff_resume_tick:
		if match_state == MatchState.State.WARMUP:
			_set_match_state(MatchState.State.PLAYING)
			_broadcast_clock_state()
			return
		if match_state == MatchState.State.OVERTIME_WARMUP:
			_set_match_state(MatchState.State.OVERTIME)
			_broadcast_clock_state()
			return
	if match_state == MatchState.State.FULL_TIME:
		# §6.2 step 10. A draw goes to sudden death; anything else is decided.
		if _winning_team() < 0:
			_in_overtime = true
			overtime_started.emit()
			_set_match_state(MatchState.State.OVERTIME_WARMUP)
			_begin_kickoff()
		else:
			_enter_results(_winning_team())
		return
	if _state_deadline_tick < 0 or now < _state_deadline_tick:
		return
	match match_state:
		MatchState.State.GOAL_PAUSE:
			if _in_overtime:
				# Golden goal: the first score after a draw ends it outright.
				_enter_results(_winning_team())
				return
			_set_match_state(MatchState.State.WARMUP)
			_begin_kickoff()
		MatchState.State.RESULTS:
			# §6.2 step 10: clients return to the LOBBY, never the main menu —
			# a community server that empties every 2.5 minutes is dead on
			# arrival. The state change is what moves both sides; the server
			# then leaves the match scene itself.
			_set_match_state(MatchState.State.LOBBY)
			get_tree().change_scene_to_file.call_deferred(ScenePaths.LOBBY)


func _on_state_change_received(state: int, at_tick: int) -> void:
	# Client path. MatchSim already rejected an unknown state value, and the
	# server is the only peer allowed to send this (rpc "authority").
	_apply_match_state(state, at_tick)


func _on_goal_registered(conceding_team: int) -> void:
	# §6.2 step 8. Immediate and authoritative at sensor time, before any
	# presentation — a last-second goal must count even though the cinematic
	# and the reset happen later.
	var scoring_team := 1 - conceding_team
	_record_goal(scoring_team)
	MatchSim.send_score_update(score.duplicate())
	if not multiplayer.is_server() or not MatchState.is_live(match_state):
		return
	var goal_tick := Engine.get_physics_frames()
	var resume_tick := goal_tick + int(_goal_pause_seconds() * SimConstants.TICK_HZ)
	# The clock stops for the celebration and resumes after it — expressed as
	# a shift of the absolute end tick (§5.2's own formula), never as pausing
	# a Timer, so no float drift accumulates across ten goals.
	if _end_tick >= 0 and not _in_overtime:
		_end_tick += resume_tick - goal_tick
	MatchSim.send_goal_scored(scoring_team, score.duplicate(), goal_tick, resume_tick)
	_set_bodies_frozen(true)
	_set_match_state(MatchState.State.GOAL_PAUSE)
	_broadcast_clock_state()


func _on_goal_scored(_conceding_team: int) -> void:
	# Deliberately empty. Before task 5.4 this reset the world the instant the
	# sensor fired, which is precisely the "server reset fires while clients
	# are mid-celebration" failure §5.4 exists to remove. The reset is now the
	# KICKOFF's job at resume_tick (_update_match_state -> _begin_kickoff), so
	# bodies stay frozen exactly where the goal happened for the whole
	# celebration window and every peer sees the same thing.
	pass


func _broadcast_snapshot() -> void:
	var server_tick := Engine.get_physics_frames()
	var bodies: Array[NetBodyState] = []
	# Always one entry per slot, even for a momentarily-invalid ship
	# (placeholder zero state), so the ball always lands at the fixed index
	# _slots.size() the client assumes in _on_snapshot_received — skipping
	# invalid ships entirely would shift every later index. "No ship is ever
	# despawned" (§6.4) means this is unreachable today, but it's a silent
	# total-garbage failure mode the moment that stops being true, and the
	# fix costs nothing.
	for slot in _slots:
		bodies.append(_ship_to_net_body_state(slot.ship, slot.jitter_buffer.stalled) if is_instance_valid(slot.ship) else NetBodyState.new())
	if is_instance_valid(ball):
		bodies.append(_ball_to_net_body_state(ball))
	var segment := NetCodec.pack_snapshot_body_segment(server_tick, match_state, _reset_gen, bodies)
	# Building the shared body segment once and reusing it per peer (rather
	# than re-encoding per client) is the whole reason §2.4 splits the wire
	# format into a per-client header + a shared body segment in the first
	# place — see pack_snapshot_body_segment's own doc comment. The per-
	# client header (last_input_seq/input_buffer_depth/echo_client_send_ms)
	# is genuinely per-peer, built fresh below from each slot's own
	# InputJitterBuffer (§3.2) — last_applied_seq of -1 (nothing consumed
	# yet) encodes as 0 on the wire, which is safe: the client's own seq
	# numbering starts at 1, so 0 never collides with a real seq.
	# "No ship is ever despawned" (§6.4) means _slots outlives a disconnect —
	# a real one will be handled by Phase 5's reconnect/controller-swap
	# logic, but sending an RPC to a peer_id ENet no longer knows about
	# (found via the smoke test: a client that exits mid-match spammed
	# "Attempt to call RPC with unknown peer ID" every tick for the rest of
	# the host's run) throws instead of silently no-op'ing. Guard against it.
	var connected_peers := multiplayer.get_peers()
	for slot in _slots:
		if connected_peers.has(slot.peer_id):
			var last_input_seq := maxi(slot.jitter_buffer.last_applied_seq, 0)
			# -1 is reserved for client "not established" state. -2 reports a
			# genuine sustained server starvation event.
			var advertised_depth := -2 if slot.jitter_buffer.starved_ticks >= STARVATION_ADVERTISEMENT_TICKS else slot.jitter_buffer.depth()
			var bytes := NetCodec.pack_snapshot(last_input_seq, advertised_depth, slot.last_client_send_ms, segment)
			MatchSim.send_snapshot(slot.peer_id, bytes)


func _ship_to_net_body_state(ship: Ship, stalled: bool) -> NetBodyState:
	var s := NetBodyState.new()
	s.position = ship.global_position
	s.rotation = ship.global_transform.basis.get_rotation_quaternion()
	s.linear_velocity = ship.linear_velocity
	s.angular_velocity = ship.angular_velocity
	# Was hardcoded false. NetShipPredictor.decide() hard-corrects on
	# `authoritative.frozen != local_frozen`, which is exactly the mechanism
	# that keeps a client's predicted ship from drifting during a kickoff
	# freeze — it only works if the wire tells the truth.
	s.frozen = ship.freeze
	s.turbo = ship.is_turbo_active()
	# Matches Ship._update_movement_vfx's own read of thrust.z: only positive
	# forward thrust drives the visible flame (see task 2.6).
	s.thrust_z = clampf(maxf(ship.controller.get_action().thrust.z if ship.controller else 0.0, 0.0), 0.0, 1.0)
	s.avel_range = NetCodec.SHIP_AVEL_RANGE
	# §3.2: InputJitterBuffer.stalled was computed all along but never
	# reached the wire — an adversarial review found this was the exact
	# signal that would have made the ring-overflow bug (this session's
	# critical fix) visible to the client and the CI gate, and its absence
	# is part of why neither ever noticed. get_net_debug_stats() below is
	# what actually surfaces it to the debug overlay now.
	s.stalled = stalled
	return s


func _ball_to_net_body_state(b: RigidBody3D) -> NetBodyState:
	var s := NetBodyState.new()
	s.position = b.global_position
	s.rotation = b.global_transform.basis.get_rotation_quaternion()
	s.linear_velocity = b.linear_velocity
	s.angular_velocity = b.angular_velocity
	s.avel_range = NetCodec.BALL_AVEL_RANGE
	return s


# ============================================================
# Client
# ============================================================

func _on_match_config_received(arena_path: String, peer_ids: PackedInt32Array, teams: PackedInt32Array, spawn_indices: PackedInt32Array) -> void:
	if not _slots.is_empty():
		# Not idempotent by accident: the original broadcast from
		# _start_server() and a reply to this client's own
		# request_match_config() (see _request_match_config_until_received)
		# can both legitimately arrive — the retry loop exists specifically
		# because either one alone isn't reliably delivered, so seeing both
		# is expected, not a protocol error. Processing this twice would
		# double-spawn the whole match (found via the two-process smoke
		# test: two arenas, two ships, two HUDs, _slots.size() == 2 instead
		# of 1). Once is enough.
		return
	var known := false
	for a in ArenaRegistry.ARENAS:
		if a["path"] == arena_path:
			known = true
			break
	if not known:
		push_error("NetworkedMatch: server sent unknown arena path '%s', refusing match_config" % arena_path)
		return

	arena = (load(arena_path) as PackedScene).instantiate()
	add_child(arena)
	# _owns_goal_logic() is false here, so GameMode's usual goal-signal wiring
	# never happens — a client's local (interpolated, laggy) Goal sensor must
	# never be allowed to decide a score, only the server's real one can.

	spawn_ball()
	ball.freeze = true
	ball.freeze_mode = RigidBody3D.FREEZE_MODE_KINEMATIC
	# The authority shadow is presentation-only on clients. Its collider must
	# not steal an impulse from the dynamic client-only proxy below.
	ball.collision_layer = 0
	ball.collision_mask = 0
	if is_instance_valid((ball as Ball).visual):
		(ball as Ball).visual.physics_interpolation_mode = Node.PHYSICS_INTERPOLATION_MODE_OFF
	_spawn_local_ball_proxy()

	var my_id := multiplayer.get_unique_id()
	for i in peer_ids.size():
		var slot := SlotInfo.new()
		slot.peer_id = peer_ids[i]
		slot.team = teams[i]
		slot.spawn_index = spawn_indices[i]
		slot.ship = spawn_ship(slot.team, slot.spawn_index, null)
		var is_local := slot.peer_id == my_id
		# Do not let the local dynamic body fall or collide during the
		# match_config→first-snapshot gap. Prediction starts from a genuine
		# server pose below, not from an unsynchronised spawn approximation.
		slot.ship.freeze = true
		slot.ship.freeze_mode = RigidBody3D.FREEZE_MODE_KINEMATIC
		# §4.6: manual, per-render-frame $Visual updates must not fight
		# Godot's own built-in physics interpolation.
		if not is_local and is_instance_valid(slot.ship.visual):
			slot.ship.visual.physics_interpolation_mode = Node.PHYSICS_INTERPOLATION_MODE_OFF
		_slots.append(slot)
		if is_local:
			_my_slot = slot

	_spawn_hud()
	if is_instance_valid(_my_slot) and is_instance_valid(_my_slot.ship):
		spawn_camera_rig(_my_slot.ship)
		_my_slot.ship.ball_contact.connect(_on_local_ball_contact)
		# Headless training ships intentionally do not install Ship's render-side
		# body_entered signal. Attach this client-only callback only to the
		# locally predicted match ship so contact QA sees the same event without
		# changing training instances.
		if DisplayServer.get_name() == "headless":
			_my_slot.ship.body_entered.connect(_on_local_ship_body_entered)
		if not _test_bot_model_path.is_empty():
			# --test-bot (task 3.6): attach a real
			# AIShipController. Unlike PlayerShipController, this one needs
			# real scene context (get_parent() as Ship for itself, plus
			# ball/teammate/opponent discovery via groups) — Ship.set_controller()
			# parents it correctly, satisfying that. Known limitation: this
			# local bot controller to the genuinely simulated local ship.
			var bot := AIShipController.new()
			bot.model_path = _test_bot_model_path
			_my_slot.ship.add_child(bot)
			_local_input_timeline = LocalInputTimeline.new()
			_local_net_controller = LocalNetShipController.new(bot, _local_input_timeline)
			_my_slot.ship.set_controller(_local_net_controller)
		else:
			var player := PlayerShipController.new()
			_local_input_timeline = LocalInputTimeline.new()
			_local_net_controller = LocalNetShipController.new(player, _local_input_timeline)
			_local_net_controller.add_child(player)
			_my_slot.ship.set_controller(_local_net_controller)
	# The roster now exists, so a kickoff that raced ahead of match_config can
	# finally be placed against the right bodies.
	_apply_pending_kickoff()


func _spawn_hud() -> void:
	hud = HUD_SCENE.instantiate()
	add_child(hud)


func _send_local_input(record_prediction: bool = true) -> void:
	if _slots.is_empty():
		return   # match_config hasn't arrived yet
	if not _local_prediction_ready or _my_slot == null or not is_instance_valid(_my_slot.ship):
		return
	# Client-owned input_lead control loop (§3.3): ordinarily +1 (ship
	# increments its send sequence by exactly one tick's worth), but a lead
	# change this tick skips extra sequence numbers (attack, more server-
	# side buffer margin) or duplicates the current one (release, delta 0 —
	# one tick of latency recovered).
	_update_adaptive_input_target()
	var reported_depth := -2 if _last_known_input_buffer_depth == -2 else (_last_known_input_buffer_depth if _has_received_healthy_buffer_depth else -1)
	var delta := _input_lead_controller.update(reported_depth, _current_input_target_depth())
	if _local_input_timeline == null or _local_net_controller == null:
		return
	var applied_action := _my_slot.ship.get_current_action_copy()
	var previous_issued_seq := _input_seq
	_input_seq = _local_input_timeline.issue(delta, _local_net_controller.last_sampled_intent)
	# The body used the raw action immediately, and that action was issued under
	# _input_seq this tick — so _input_seq is the sequence whose post-step state
	# this is. Label it there.
	#
	# This deliberately does NOT delay local control: which action the ship uses
	# is decided in LocalNetShipController.get_action() (still the raw current
	# intent, still immediate) and is untouched by which seq its resulting state
	# is filed under. The previous label, _local_net_controller.last_applied_seq,
	# was the timeline's ESTIMATE of the sequence the server would consume this
	# tick — input_lead ticks behind issuance — so predicted[S] held "state after
	# integrating the intent from now" while the server's authoritative state for
	# S is "state after integrating action(S)", sampled input_lead ticks earlier.
	# Those agree only while the stick is still, which is why a held-input trace
	# could never falsify it and a transition-heavy one reports ~9% action-marker
	# mismatch.
	var history_seq := _input_seq
	if delta > 0 and record_prediction:
		# An attack (delta > 1) issues and SENDS several sequences for this one
		# local physics step; only the newest carries the action the body just
		# integrated. The skipped ones are real outstanding sequences the server
		# will acknowledge, but the client never simulated them, so they are
		# recorded stateless rather than left absent — absent is indistinguishable
		# from genuine ring loss, and cost a teleport plus resync suppression
		# every time the lead controller attacked.
		for gap_seq in range(previous_issued_seq + 1, history_seq):
			if gap_seq <= 0:
				continue
			var gap_action = _local_input_timeline.action_for(gap_seq)
			if gap_action != null:
				_local_prediction_history.record_unsimulated(gap_seq, gap_action)
		if history_seq > 0:
			_local_prediction_history.record(history_seq, applied_action, _local_ship_prediction_state(_my_slot.ship, applied_action), _my_slot.ship.net_prediction_contact_window)
	# delta <= 0 is a release: the timeline deliberately does NOT mutate an
	# already-issued sequence, so re-recording here would file the CURRENT intent
	# under a sequence that went out carrying a different action — the ring would
	# then contradict the wire, and the action marker would (correctly) report a
	# mismatch whenever the server had already consumed the original. The existing
	# predicted[S] is right; leave it alone. The extra unlabelled local step is
	# precisely the tick of latency the release exists to recover.
	_input_history.clear()
	for packet_action in _local_input_timeline.packet_actions(NetCodec.MAX_REDUNDANCY):
		_input_history.append(packet_action)
	if _input_history.is_empty():
		return
	var bytes := NetCodec.pack_input(_input_seq, _last_received_snapshot_tick, Time.get_ticks_msec(), _input_history)
	MatchSim.send_input(bytes)


func _on_snapshot_received(decoded: Dictionary) -> void:
	var server_tick: int = decoded["server_tick"]
	var reset_gen: int = decoded["reset_gen"]
	var bodies: Array = decoded["bodies"]
	if _expected_next_snapshot_tick >= 0:
		var missed := maxi(0, server_tick - _expected_next_snapshot_tick)
		var sample := float(missed) / float(missed + 1)
		_snapshot_loss_ewma += (sample - _snapshot_loss_ewma) * SNAPSHOT_LOSS_EWMA_ALPHA
	_expected_next_snapshot_tick = server_tick + 1
	_last_received_snapshot_tick = server_tick
	_last_snapshot_wall_ms = Time.get_ticks_msec()
	# match_state catch-up (§6.1). state_change is reliable, so this is not a
	# loss-recovery path — it covers the cases reliability cannot: a client
	# that joined mid-match and has not been sent a transition yet, and the
	# window between scene load and the first state_change arriving. Snapshots
	# carry no at_tick for the transition, so attribute it to this snapshot's
	# own server_tick, which is the tightest bound available and is never
	# later than the true transition tick.
	var snapshot_state: int = decoded["match_state"]
	_last_snapshot_match_state = snapshot_state
	# The tick guard is load-bearing, not defensive padding. state_change is
	# reliable on channel 0 while snapshots are unreliable_ordered on channel
	# 2, and ordering is only guaranteed WITHIN a channel — so a state_change
	# for tick N routinely arrives before a snapshot that was sent at tick
	# N-2 and is still in flight. Without this the client would apply the new
	# state, then be dragged straight back by the older snapshot's byte, and
	# oscillate on every single transition. Observed exactly that while
	# testing a deliberately-broken byte: LOADING -> WARMUP -> LOBBY ->
	# PLAYING -> LOBBY -> ... Only accept a byte at least as new as whatever
	# told us the current state.
	if snapshot_state != match_state and MatchState.is_valid(snapshot_state) and server_tick >= match_state_since_tick:
		_apply_match_state(snapshot_state, server_tick)
	# Per-client header (§2.4): unlike the shared body segment, this is
	# genuinely this recipient's own — input_buffer_depth is THIS client's
	# own slot's server-side InputJitterBuffer.depth() at send time, which
	# is exactly what the input_lead control loop (§3.3) needs.
	_last_known_input_buffer_depth = decoded["input_buffer_depth"]
	if _last_known_input_buffer_depth >= 0:
		_has_received_healthy_buffer_depth = true
	# Compare against the same input sequence then reconcile the genuinely
	# locally-simulated ship.  The predictor owns the snap-vs-soft decision.
	if _my_slot != null:
		var my_index := _slots.find(_my_slot)
		if my_index >= 0 and my_index < bodies.size():
			if not _local_prediction_ready:
				var initial: NetBodyState = bodies[my_index]
				if _local_input_timeline != null:
					var one_way_ms := maxf(NetworkManager.rtt_ms * 0.5, 0.0)
					var label_delay_ticks := ceili(one_way_ms / SNAPSHOT_INTERVAL_MS) + _current_input_target_depth()
					_local_input_timeline.configure_initial_delay(label_delay_ticks)
				_my_slot.ship.queue_teleport_with_velocity(Transform3D(Basis(initial.rotation), initial.position), initial.linear_velocity, initial.angular_velocity)
				_my_slot.ship.freeze = false
				_local_prediction_ready = true
			else:
				# Receipt can run from both process callbacks. Stage immutable wire
				# data only: comparison mutates acknowledgement/history state and
				# must happen atomically with the correction below.
				_pending_local_reconciliation = {
					"ack_seq": decoded["last_input_seq"],
					"authoritative": (bodies[my_index] as NetBodyState).copy(),
					"reset_gen": reset_gen,
				}
	_update_tick_bias(server_tick)
	for i in _slots.size():
			if i < bodies.size():
				if _slots[i] != _my_slot:
					var slot := _slots[i]
					var accepts_remote_tick := slot.interpolator.accepts_tick(server_tick)
					var remote_reset := accepts_remote_tick and slot.interpolator.reset_gen != -1 and reset_gen != slot.interpolator.reset_gen
					if remote_reset:
						slot.visual_smoother_reset = true
						slot.visual_position_offset = Vector3.ZERO
						slot.visual_rotation_offset = Quaternion.IDENTITY
					elif accepts_remote_tick:
						_accumulate_remote_residual(slot.interpolator, server_tick, bodies[i], slot)
					slot.interpolator.add_sample(server_tick, bodies[i], reset_gen)
	if bodies.size() > _slots.size():
		var ball_state: NetBodyState = bodies[_slots.size()]
		# unpack_snapshot() decodes every body's angular_velocity assuming
		# SHIP_AVEL_RANGE; the ball was quantised at BALL_AVEL_RANGE
		# (_ball_to_net_body_state), so it decodes 8x too small without this
		# — dormant today (nothing reads decoded angular_velocity yet) but
		# silently wrong the moment ball-spin VFX or Phase 4 prediction does.
		NetCodec.rescale_avel(ball_state, NetCodec.BALL_AVEL_RANGE)
		_ball_shadow_state = ball_state.copy()
		if _ball_prediction_until_ms >= 0 and ball_state.position.distance_to(_ball_shadow_position_on_contact) > 0.01:
			_ball_authority_changed_since_contact = true
		var accepts_ball_tick := _ball_interpolator.accepts_tick(server_tick)
		var ball_was_reset := accepts_ball_tick and _ball_interpolator.reset_gen != -1 and reset_gen != _ball_interpolator.reset_gen
		if accepts_ball_tick and not ball_was_reset:
			_accumulate_ball_residual(_ball_interpolator, server_tick, ball_state)
		var ball_reset := _ball_interpolator.add_sample(server_tick, ball_state, reset_gen)
		if ball_reset:
			_ball_reset_trace.append("%d:%d" % [server_tick, reset_gen])
			if _ball_reset_trace.size() > 12:
				_ball_reset_trace.pop_front()
			_ball_visual_smoother_reset = true
			_ball_visual_position_offset = Vector3.ZERO
			_ball_visual_rotation_offset = Quaternion.IDENTITY
			_cancel_ball_prediction_for_reset(ball_state)
		if is_instance_valid(_local_ball_proxy) and _ball_prediction_until_ms < 0:
			_local_ball_proxy.queue_teleport_with_velocity(Transform3D(Basis(ball_state.rotation), ball_state.position), ball_state.linear_velocity, ball_state.angular_velocity)


# Called from NetworkedMatch._physics_process after Ship._integrate_forces,
# so this is the genuine post-step state caused by the local controller's one
# action pull. _send_local_input then pairs it with the copied wire action.
func _local_ship_prediction_state(ship: Ship, action: ShipAction) -> NetBodyState:
	var state := NetBodyState.new()
	state.position = ship.global_position
	state.rotation = ship.global_transform.basis.get_rotation_quaternion()
	state.linear_velocity = ship.linear_velocity
	state.angular_velocity = ship.angular_velocity
	state.frozen = ship.freeze
	state.turbo = action.turbo
	state.thrust_z = action.thrust.z
	state.avel_range = NetCodec.SHIP_AVEL_RANGE
	return state


func _on_local_ball_contact(_intensity: float, _world_position: Vector3) -> void:
	if not local_ball_prediction_enabled or multiplayer.is_server() or not is_instance_valid(ball) or not is_instance_valid(_local_ball_proxy):
		return
	# body_entered can fire repeatedly while the proxy remains in a manifold.
	# One touch owns one bounded RTT window; extending it per callback can keep
	# speculation alive indefinitely and prevents the required blend-back.
	var now_ms := Time.get_ticks_msec()
	if _ball_prediction_until_ms >= 0 or now_ms < _ball_recontact_cooldown_until_ms:
		return
	var prediction_window_ms := int(minf(maxf(NetworkManager.rtt_ms, SNAPSHOT_INTERVAL_MS), BALL_PREDICTION_MAX_MS))
	_ball_prediction_until_ms = now_ms + prediction_window_ms
	_ball_recontact_cooldown_until_ms = now_ms + max(BALL_RECONTACT_COOLDOWN_MS, prediction_window_ms + BALL_VISUAL_BLEND_MS)
	_ball_visual_blend_started_ms = -1
	_ball_contact_frame = Engine.get_physics_frames()
	_ball_reveal_frame = Engine.get_physics_frames()
	(ball as Ball).visual.visible = false
	_local_ball_proxy.visual.visible = true
	_local_ball_proxy.set_visual_speed(-1.0)
	_ball_prediction_contact_count += 1
	_ball_proxy_contact_position = _local_ball_proxy.global_position
	_ball_proxy_moved_before_authority = false
	_ball_shadow_position_on_contact = _ball_shadow_state.position if _ball_shadow_state != null else _local_ball_proxy.global_position
	_ball_authority_changed_since_contact = false


func _on_local_ship_body_entered(body: Node) -> void:
	if body is Ball:
		_on_local_ball_contact(0.0, (body as Ball).global_position)


func _finish_ball_prediction() -> void:
	if _ball_prediction_until_ms >= 0 and not _ball_authority_changed_since_contact and is_instance_valid(_local_ball_proxy) and _local_ball_proxy.global_position.distance_to(_ball_proxy_contact_position) > 0.01:
		if not _ball_proxy_moved_before_authority:
			_ball_proxy_moved_before_authority = true
			_ball_proxy_moved_before_authority_count += 1
	if _ball_prediction_until_ms < 0 or Time.get_ticks_msec() < _ball_prediction_until_ms:
		return
	_ball_prediction_until_ms = -1
	_ball_prediction_window_end_count += 1
	if not is_instance_valid(ball) or not is_instance_valid(_local_ball_proxy):
		return
	(ball as Ball).visual.visible = true
	_local_ball_proxy.visual.visible = false
	if _ball_shadow_state == null:
		_ball_prediction_missing_shadow_count += 1
		return
	_last_ball_prediction_error = _local_ball_proxy.global_position.distance_to(_ball_shadow_state.position)
	if _last_ball_prediction_error > BALL_HARD_SNAP_DISTANCE:
		# A large disagreement is dishonest to hide. Resume the authoritative
		# shadow immediately, then re-seed the invisible proxy on next arrival.
		_ball_visual_blend_started_ms = -1
		_ball_hard_handoff_count += 1
		return
	_ball_visual_blend_from = _local_ball_proxy.visual.global_transform
	_ball_visual_blend_started_ms = Time.get_ticks_msec()
	_ball_blend_started_count += 1
	# Never push the speculative result into authority; only presentation
	# blends over to the continuously-buffered shadow.


func _cancel_ball_prediction_for_reset(authoritative: NetBodyState) -> void:
	if _ball_prediction_until_ms >= 0 or _ball_visual_blend_started_ms >= 0:
		_ball_prediction_reset_cancel_count += 1
	_ball_prediction_until_ms = -1
	_ball_recontact_cooldown_until_ms = -1
	_ball_visual_blend_started_ms = -1
	_ball_proxy_moved_before_authority = false
	_ball_authority_changed_since_contact = false
	_last_ball_prediction_error = 0.0
	if is_instance_valid(ball):
		(ball as Ball).visual.visible = true
	if is_instance_valid(_local_ball_proxy):
		_local_ball_proxy.visual.visible = false
		_local_ball_proxy.queue_teleport_with_velocity(Transform3D(Basis(authoritative.rotation), authoritative.position), authoritative.linear_velocity, authoritative.angular_velocity)


func _spawn_local_ball_proxy() -> void:
	if multiplayer.is_server() or not local_ball_prediction_enabled:
		return
	_local_ball_proxy = ball_scene.instantiate() as Ball
	_local_ball_proxy.name = "LocalBallPredictionProxy"
	_local_ball_proxy.remove_from_group("ball")
	add_child(_local_ball_proxy)
	_local_ball_proxy.global_transform = ball.global_transform
	_local_ball_proxy.visual.visible = false
	# This body keeps normal ball-vs-ship/arena collision settings, but exists
	# only in this client process. It therefore receives the contact impulse on
	# the same local physics frame without altering server or training physics.


# Diagnostic accessor.
# Dictionary.duplicate(true) recurses into Arrays/Dictionaries but copies
# Objects (RefCounted included) BY REFERENCE — an adversarial review caught
# that this returned a dict sharing its "action"/"predicted_state"/
# "authoritative_state" ShipAction/NetBodyState instances with the stored
# comparison, so a caller writing through the "copy" silently rewrote
# history. ShipAction.copy() and NetBodyState.copy() exist precisely so
# callers holding onto one past its own tick copy it (see ship_action.gd's
# own comment) — this accessor has to honor that contract itself, not just
# assume duplicate(true) does.
func get_last_local_prediction_comparison() -> Dictionary:
	var result := _last_local_prediction_comparison.duplicate(true)
	for key in ["action", "predicted_state", "authoritative_state"]:
		if result.has(key):
			result[key] = result[key].copy()
	return result


# See the class-level comment above _tick_bias_samples for why this exists.
# bias_ms is how much further ahead to_tick(server_time_est) lands than the
# server_tick this snapshot actually carries — mostly the server's own
# physics-frame/wall-clock startup skew, plus a little real one-way transit
# noise that the rolling minimum below filters back out.
func _update_tick_bias(server_tick: int) -> void:
	# get_server_time_estimate_ms() is meaningless before the first pong
	# lands (clock_offset_ms == 0.0 until then, per network_manager.gd's own
	# doc comment) — recording a bias sample from it during that window
	# produced a garbage value (~-1.1s, the client's own raw pre-sync
	# uptime standing in for a server-synced estimate) that the rolling-min
	# window then locked onto for the rest of a short test, since 5 real
	# seconds never fully elapsed before the test ended. Skip entirely
	# until the clock is actually synced.
	if NetworkManager.rtt_ms < 0.0:
		return
	var server_time_est := NetworkManager.get_server_time_estimate_ms()
	var bias_ms := server_time_est - float(server_tick) * NetInterpolator.TICK_MS
	var now_ms := Time.get_ticks_msec()
	_tick_bias_samples.append({"t_ms": now_ms, "bias_ms": bias_ms})
	var cutoff := now_ms - int(TICK_BIAS_WINDOW_SEC * 1000.0)
	_tick_bias_samples = _tick_bias_samples.filter(func(s: Dictionary) -> bool: return s["t_ms"] >= cutoff)
	var best: float = _tick_bias_samples[0]["bias_ms"]
	for sample: Dictionary in _tick_bias_samples:
		var sample_bias: float = sample["bias_ms"]
		if sample_bias < best:
			best = sample_bias
	_tick_bias_ms = best


# Bias-corrected replacement for NetInterpolator.to_tick(server_time_est) —
# use this instead of calling to_tick() directly on a server-time estimate.
func _estimated_tick(server_time_ms: float) -> float:
	return NetInterpolator.to_tick(server_time_ms - _tick_bias_ms)


func _current_interp_delay_ms() -> float:
	var rtt := NetworkManager.rtt_ms
	var one_way := (rtt / 2.0) if rtt >= 0.0 else INTERP_DELAY_MIN_MS
	return clampf(one_way + SNAPSHOT_INTERVAL_MS * 1.5 + 2.5 * NetworkManager.jitter_ms, INTERP_DELAY_MIN_MS, INTERP_DELAY_MAX_MS)


func _current_input_target_depth() -> int:
	# A clean LAN needs no intentionally buffered input tick. Preserve one
	# tick whenever measured RTT jitter crosses the small threshold; starvation
	# still triggers the controller's existing fast-attack path either way.
	# Keep the headless policy-driver protocol at its established depth: these
	# bots are regression/training tooling, not the human latency experiment.
	if not _test_bot_model_path.is_empty():
		return InputLeadController.TARGET_DEPTH
	return _adaptive_input_depth.target_depth


func _update_adaptive_input_target() -> void:
	if not _test_bot_model_path.is_empty():
		_adaptive_input_depth.target_depth = InputLeadController.TARGET_DEPTH
		return
	_adaptive_input_depth.update(NetworkManager.rtt_ms, NetworkManager.jitter_ms, _last_known_input_buffer_depth)


# Client-only stats for task 3.7's debug overlay, discovered via the "game"
# group the same way HUDController finds this node — no direct reference
# needed, and the overlay degrades gracefully (has_method check) against
# any mode that doesn't implement this at all.
func get_net_debug_stats() -> Dictionary:
	var snapshot_age_ms := 0.0
	if NetworkManager.rtt_ms >= 0.0:
		var estimated_now_tick := _estimated_tick(NetworkManager.get_server_time_estimate_ms())
		snapshot_age_ms = (estimated_now_tick - float(_last_received_snapshot_tick)) * NetInterpolator.TICK_MS
	# _snapshot_loss_ewma only updates on receipt, so during a TOTAL outage
	# — exactly when this matters most — it would otherwise freeze at
	# whatever it last read (probably low/healthy) instead of climbing
	# toward 100%, an adversarial review found. Report honestly once too
	# long has passed with nothing arriving at all.
	var is_stale := _last_snapshot_wall_ms >= 0 and Time.get_ticks_msec() - _last_snapshot_wall_ms > SNAPSHOT_STALE_MS
	var snapshot_loss_pct := 100.0 if is_stale else _snapshot_loss_ewma * 100.0
	# The server's jitter_buffer.stalled bit for THIS client's own slot,
	# round-tripped through NetBodyState onto the wire (§3.2) — added by the
	# first adversarial-review fix round, but a second review found nothing
	# actually read it client-side (net_interpolator.gd only passed it
	# through lerp/extrapolate), so the commit's claim that it made the
	# server-side starvation state "visible to the client, the debug
	# overlay" was false; only the CI gate read it, and only via the
	# server's own field directly, not the wire bit. Read it here for real.
	var server_stalled := false
	if _last_local_prediction_comparison.get("authoritative_state", null) != null:
		server_stalled = (_last_local_prediction_comparison["authoritative_state"] as NetBodyState).stalled
	return {
		"match_state": match_state,
		"snapshot_match_state": _last_snapshot_match_state,
		"input_buffer_depth": _last_known_input_buffer_depth,
		"input_lead": _input_lead_controller.lead,
		"input_target_depth": _current_input_target_depth(),
		"snapshot_age_ms": snapshot_age_ms,
		"snapshot_loss_pct": snapshot_loss_pct,
		"server_stalled": server_stalled,
		"prediction": _local_ship_predictor.get_metrics(),
		"ball_prediction_contacts": _ball_prediction_contact_count,
		"ball_prediction_active": _ball_prediction_until_ms >= 0,
		"ball_prediction_error": _last_ball_prediction_error,
		"ball_contact_frame": _ball_contact_frame,
		"ball_reveal_frame": _ball_reveal_frame,
		"ball_blend_complete_count": _ball_blend_complete_count,
		"ball_blend_started_count": _ball_blend_started_count,
		"ball_blend_max_duration_ms": _ball_blend_max_duration_ms,
		"ball_hard_handoff_count": _ball_hard_handoff_count,
		"ball_prediction_window_end_count": _ball_prediction_window_end_count,
		"ball_prediction_missing_shadow_count": _ball_prediction_missing_shadow_count,
		"ball_prediction_reset_cancel_count": _ball_prediction_reset_cancel_count,
		"ball_reset_trace": _ball_reset_trace.duplicate(),
		"ball_proxy_moved_before_authority": _ball_proxy_moved_before_authority_count > 0,
		"ball_proxy_moved_before_authority_count": _ball_proxy_moved_before_authority_count,
		"ball_authority_changed_since_contact": _ball_authority_changed_since_contact,
		"remote_residual_position_p99": _remote_percentile(_remote_position_residuals, 0.99),
		"remote_residual_rotation_p99": _remote_percentile(_remote_rotation_residuals, 0.99),
		"latest_prediction_error": _last_local_prediction_comparison.get("position_error", Vector3.ZERO),
		"latest_prediction_velocity_error": _last_local_prediction_comparison.get("linear_velocity_error", Vector3.ZERO),
		"action_marker_samples": _action_marker_samples,
		"action_marker_mismatches": _action_marker_mismatches,
	}


func adjust_prediction_tuning(position_delta: float = 0.0, decay_delta: float = 0.0, offset_delta: float = 0.0, toggle_present_time: bool = false) -> void:
	# Debug-only runtime knobs; this object is never instantiated by the server
	# for an interactive client and cannot change action, collision, or Jolt
	# simulation parameters.
	if multiplayer.is_server():
		return
	_local_ship_predictor.hard_position_error = clampf(_local_ship_predictor.hard_position_error + position_delta, 0.25, 5.0)
	_local_ship_predictor.max_visual_offset = clampf(_local_ship_predictor.max_visual_offset + offset_delta, 0.05, 2.0)
	if _my_slot != null and is_instance_valid(_my_slot.ship):
		_my_slot.ship.set_network_visual_tuning(_my_slot.ship.net_visual_offset_decay + decay_delta, _local_ship_predictor.max_visual_offset)
	if toggle_present_time:
		remote_visual_present_time_enabled = not remote_visual_present_time_enabled
		_reset_remote_visual_smoothers()


func _reset_remote_visual_smoothers() -> void:
	for slot in _slots:
		if slot != _my_slot:
			slot.visual_smoother_reset = true
			slot.visual_position_offset = Vector3.ZERO
			slot.visual_rotation_offset = Quaternion.IDENTITY
	_ball_visual_smoother_reset = true
	_ball_visual_position_offset = Vector3.ZERO
	_ball_visual_rotation_offset = Quaternion.IDENTITY


# Collider time: present-time estimate, applied once per physics tick.
func _physics_process(_delta: float) -> void:
	# Automatic multiplayer polling is disabled project-wide (task 1.3) —
	# every scene that sends/receives RPCs has to poll manually, and this
	# one is no exception. Missing this meant NOTHING sent after entering
	# this scene ever actually reached the wire in either direction
	# (queued but never flushed) — found via the two-process smoke test,
	# not by inspection.
	NetworkManager.poll()
	if _owns_world_simulation():
		_respawn_escaped_bodies()
	# ORDER IS LOAD-BEARING. _update_match_state() consumes _kickoff_resume_tick
	# to drive WARMUP -> PLAYING, and _update_kickoff_countdown() clears that
	# same field once it reaches zero. Running the countdown first meant the
	# server's transition condition was wiped before it was ever evaluated and
	# the match sat in WARMUP forever with every body frozen.
	if multiplayer.is_server():
		# Also before the broadcast, so a transition taken this tick ships in
		# this tick's own match_state byte rather than trailing it by one.
		_update_match_state()
	# Countdown and clock are derived from absolute ticks on both peers, so
	# these run on the client too.
	_apply_pending_freeze()
	_update_kickoff_countdown()
	_update_clock()
	if multiplayer.is_server():
		# _physics_process runs after this frame's _integrate_forces. Snapshot
		# FIRST: the body state therefore still describes the sequence consumed
		# on the prior callback. Sending after consume mislabeled that old state
		# with NEXT tick's input sequence, making every client reconciliation
		# comparison one action off and causing the Phase 4 snap cascade.
		_broadcast_snapshot()
		# The newly consumed action is deliberately installed for NEXT frame's
		# integration. This preserves the existing one-tick server input delay
		# while keeping snapshot.last_input_seq truthfully coupled to its body.
		for slot in _slots:
			slot.controller.action = slot.jitter_buffer.consume()
		if _pending_reset_gen_bump and Engine.get_physics_frames() > _pending_reset_gen_bump_tick:
			_reset_gen = (_reset_gen + 1) % 256
			_pending_reset_gen_bump = false
		return

	# Prediction and reconciliation are suspended while the match is not live.
	# During a kickoff countdown or a goal pause the local ship is frozen on
	# BOTH peers, so there is nothing to predict — but the reconciler still ran
	# its delta transport and visual-offset maths over those frozen states and
	# produced garbage: 200 hard snaps and a p95 position error of 2.4e10 m in
	# a single 12s run, while the instantaneous error stayed small. Input keeps
	# flowing so the server's jitter buffer does not starve into `stalled` and
	# the input_lead loop keeps its cadence; only the local prediction ring and
	# the correction step pause.
	var live := MatchState.is_live(match_state)
	_send_local_input(live)
	if live:
		_consume_local_reconciliation()
	else:
		# Anything queued from before the whistle describes the old world.
		_pending_local_reconciliation = {}
	_finish_ball_prediction()
	# get_server_time_estimate_ms() is meaningless before the first pong
	# lands (network_manager.gd's own doc comment says so explicitly) — an
	# adversarial review found this was used unguarded here, which against
	# a long-running dedicated server (clock_offset_ms == 0.0, so this
	# process's own short uptime is compared against the server's enormous
	# tick count) freezes every remote body at the oldest buffered pose for
	# the whole first second of every match.
	if NetworkManager.rtt_ms < 0.0:
		return
	var server_time_est := NetworkManager.get_server_time_estimate_ms()
	var collider_tick := _estimated_tick(server_time_est)
	for slot in _slots:
		if slot != _my_slot and is_instance_valid(slot.ship) and slot.interpolator.has_samples():
			_apply_collider_state(slot.ship, slot.interpolator.sample_at(collider_tick))
	if is_instance_valid(ball) and _ball_interpolator.has_samples():
		_apply_collider_state(ball, _ball_interpolator.sample_at(collider_tick))


func _consume_local_reconciliation() -> void:
	if _pending_local_reconciliation.is_empty() or _my_slot == null or not is_instance_valid(_my_slot.ship):
		return
	var pending := _pending_local_reconciliation
	_pending_local_reconciliation = {}
	var reset_gen: int = pending["reset_gen"]
	# Reset starts an isolated history epoch before its state is compared.
	if _last_local_reset_gen != -1 and _last_local_reset_gen != reset_gen:
		_local_prediction_history.begin_epoch()
	_last_local_reset_gen = reset_gen
	var comparison := _local_prediction_history.compare_authoritative(int(pending["ack_seq"]), pending["authoritative"])
	if comparison.get("status", "") == "matched":
		var action: ShipAction = comparison["action"]
		var authority: NetBodyState = comparison["authoritative_state"]
		_action_marker_samples += 1
		if absf(action.thrust.z - authority.thrust_z) > 0.26:
			_action_marker_mismatches += 1
	_last_local_prediction_comparison = comparison
	# Must match the clock _send_local_input files predictions under, since this
	# is the upper bound of the rebase range over retained history.
	var current_seq := _input_seq
	_local_ship_predictor.reconcile(comparison, _my_slot.ship, reset_gen, current_seq, _local_prediction_history)


# Visual time: present-minus-INTERP_DELAY, applied once per rendered frame —
# separate from the collider update above so a high-refresh client samples
# remote motion at true render rate instead of repeating the same 60Hz value
# several times in a row (§2.4's "240 distinct positions/s, not 60").
#
# Ball only gets the VFX half of this (trail speed), not a transform write:
# unlike Ship, Ball has no separate $Visual child to offset from its
# collider (task 0.2's Visual-node split was scoped to Ship only) — giving
# it one is a bigger structural change than Phase 2's remit, so for now the
# ball's rendered position is whatever _physics_process's present-time
# collider update leaves it at, one tick behind true dual-time smoothness.
func _process(_delta: float) -> void:
	# §7 task 1.3: poll for receive unconditionally at the top of both
	# _process and _physics_process, not just physics — a snapshot that
	# lands between ticks can be rendered immediately at high refresh rates
	# instead of waiting for the next physics step.
	NetworkManager.poll()
	if multiplayer.is_server() or _slots.is_empty():
		return
	if NetworkManager.rtt_ms < 0.0:
		return
	var server_time_est := NetworkManager.get_server_time_estimate_ms()
	var visual_time := server_time_est if remote_visual_present_time_enabled else server_time_est - _current_interp_delay_ms()
	var visual_tick := _estimated_tick(visual_time)
	for slot in _slots:
		if slot != _my_slot and is_instance_valid(slot.ship) and slot.interpolator.has_samples():
			_apply_ship_visual_state(slot.ship, slot.interpolator.sample_at(visual_tick), _delta, slot)
	if is_instance_valid(ball) and _ball_interpolator.has_samples():
		var state := _ball_interpolator.sample_at(visual_tick)
		if state != null:
			if _ball_prediction_until_ms < 0 and is_instance_valid((ball as Ball).visual):
				var target := Transform3D(Basis(state.rotation), state.position)
				if _ball_visual_blend_started_ms >= 0:
					var elapsed := Time.get_ticks_msec() - _ball_visual_blend_started_ms
					var t := clampf(float(elapsed) / float(BALL_VISUAL_BLEND_MS), 0.0, 1.0)
					(ball as Ball).visual.global_transform = _ball_visual_blend_from.interpolate_with(target, t)
					if t >= 1.0:
						_ball_blend_max_duration_ms = maxi(_ball_blend_max_duration_ms, elapsed)
						_ball_visual_blend_started_ms = -1
						_ball_blend_complete_count += 1
				else:
					_apply_ball_visual_state(target, _delta)
			(ball as Ball).set_visual_speed(state.linear_velocity.length())


func _apply_collider_state(body: RigidBody3D, state: NetBodyState) -> void:
	if state == null:
		return
	body.global_transform = Transform3D(Basis(state.rotation), state.position)


func _apply_ship_visual_state(ship: Ship, state: NetBodyState, delta: float, slot: SlotInfo) -> void:
	if state == null:
		return
	if is_instance_valid(ship.visual):
		var target := Transform3D(Basis(state.rotation), state.position)
		# Keep the delayed-interpolation A/B control genuinely unchanged. The
		# follower is only evaluating present-time rendering, never silently
		# adding a second lag source to the baseline path.
		if not remote_visual_present_time_enabled:
			ship.visual.global_transform = target
			slot.visual_smoother_reset = false
		elif slot.visual_smoother_reset:
			ship.visual.global_transform = target
			slot.visual_smoother_reset = false
		else:
			var t := clampf(1.0 - exp(-REMOTE_VISUAL_SMOOTH_RATE * delta), 0.0, 1.0)
			slot.visual_position_offset = slot.visual_position_offset.lerp(Vector3.ZERO, t)
			slot.visual_rotation_offset = slot.visual_rotation_offset.slerp(Quaternion.IDENTITY, t)
			ship.visual.global_transform = Transform3D(Basis(slot.visual_rotation_offset * state.rotation), target.origin + slot.visual_position_offset)
	ship.set_visual_action(state.thrust_z, state.turbo)


func _apply_ball_visual_state(target: Transform3D, delta: float) -> void:
	if not is_instance_valid(ball) or not is_instance_valid((ball as Ball).visual):
		return
	var visual := (ball as Ball).visual
	if not remote_visual_present_time_enabled:
		visual.global_transform = target
		_ball_visual_smoother_reset = false
	elif _ball_visual_smoother_reset:
		visual.global_transform = target
		_ball_visual_smoother_reset = false
	else:
		var t := clampf(1.0 - exp(-REMOTE_VISUAL_SMOOTH_RATE * delta), 0.0, 1.0)
		_ball_visual_position_offset = _ball_visual_position_offset.lerp(Vector3.ZERO, t)
		_ball_visual_rotation_offset = _ball_visual_rotation_offset.slerp(Quaternion.IDENTITY, t)
		visual.global_transform = Transform3D(Basis(_ball_visual_rotation_offset * target.basis.get_rotation_quaternion()), target.origin + _ball_visual_position_offset)


func _accumulate_remote_residual(interpolator: NetInterpolator, tick: int, authoritative: NetBodyState, slot: SlotInfo) -> void:
	if interpolator.has_samples():
		var predicted := interpolator.sample_at(tick)
		if predicted != null:
			var position_residual := authoritative.position - predicted.position
			_remote_position_residuals.append(position_residual.length())
			_remote_rotation_residuals.append(rad_to_deg(predicted.rotation.angle_to(authoritative.rotation)))
			if _remote_position_residuals.size() > REMOTE_METRIC_CAPACITY:
				_remote_position_residuals.pop_front()
				_remote_rotation_residuals.pop_front()
			if remote_visual_present_time_enabled:
				slot.visual_position_offset = (slot.visual_position_offset - position_residual).limit_length(REMOTE_VISUAL_MAX_OFFSET)
				var residual_rotation := (predicted.rotation * authoritative.rotation.inverse()).normalized()
				if rad_to_deg(Quaternion.IDENTITY.angle_to(residual_rotation)) <= REMOTE_VISUAL_MAX_ROTATION_DEGREES:
					slot.visual_rotation_offset = (residual_rotation * slot.visual_rotation_offset).normalized()
				else:
					slot.visual_rotation_offset = Quaternion.IDENTITY


func _accumulate_ball_residual(interpolator: NetInterpolator, tick: int, authoritative: NetBodyState) -> void:
	if not interpolator.has_samples():
		return
	var predicted := interpolator.sample_at(tick)
	if predicted == null:
		return
	var position_residual := authoritative.position - predicted.position
	_remote_position_residuals.append(position_residual.length())
	_remote_rotation_residuals.append(rad_to_deg(predicted.rotation.angle_to(authoritative.rotation)))
	if _remote_position_residuals.size() > REMOTE_METRIC_CAPACITY:
		_remote_position_residuals.pop_front()
		_remote_rotation_residuals.pop_front()
	if remote_visual_present_time_enabled:
		_ball_visual_position_offset = (_ball_visual_position_offset - position_residual).limit_length(REMOTE_VISUAL_MAX_OFFSET)
		var residual_rotation := (predicted.rotation * authoritative.rotation.inverse()).normalized()
		if rad_to_deg(Quaternion.IDENTITY.angle_to(residual_rotation)) <= REMOTE_VISUAL_MAX_ROTATION_DEGREES:
			_ball_visual_rotation_offset = (residual_rotation * _ball_visual_rotation_offset).normalized()
		else:
			_ball_visual_rotation_offset = Quaternion.IDENTITY


func _remote_percentile(samples: Array[float], fraction: float) -> float:
	if samples.is_empty():
		return 0.0
	var sorted := samples.duplicate()
	sorted.sort()
	return sorted[clampi(roundi((sorted.size() - 1) * fraction), 0, sorted.size() - 1)]


func _on_score_update_received(new_score: Dictionary) -> void:
	score = new_score.duplicate()
	score_changed.emit(score.duplicate())