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
	# Base type, NOT RLShipController: §6.4's takeover swaps in either an
	# AIShipController (--fill-bots) or the inert base controller, and a
	# narrower declared type makes that assignment fail its type check — which
	# leaves this field pointing at the controller set_controller() just
	# queue_free()d. Exactly task 5.7's dangling reference, and it showed up as
	# controller_valid=false the first time the disconnect test ran.
	var controller: ShipController        # 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)
	# §6.4 (tasks 5.6/5.7). A ship is NEVER despawned on disconnect — the slot
	# keeps its ship and swaps the controller, so body order (and therefore
	# every snapshot index) stays stable for the whole match.
	var player_name := ""                 # identity key for reconnect; peer_id changes across a reconnect
	var disconnected := false
	var reserved_until_tick := -1         # server only: slot held for this player until here
	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
# Ticks of regulation left, banked whenever the clock stops. Authoritative
# while _clock_running is false; end_tick is rebased from it on resume.
var _clock_remaining_ticks := -1
var _last_emitted_second := -1
@export var match_length_seconds := 150.0

# §6.4's --fill-bots takeover controller. Mirrors match_mode.gd's exports so a
# server operator configures the replacement bot exactly as a single-player
# match configures its opponent, rather than through a second parallel scheme.
@export_group("Disconnect fill bot")
@export_file("*.json") var bot_model_path: String = ""
@export_range(1, 60) var bot_reaction_ticks: int = 8
@export_range(0.0, 1.0) var bot_action_noise: float = 0.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
# §1.4: public servers default to leaving an abandoned ship inert rather than
# handing it to a bot, so a disconnect cannot change the competitive balance
# of a match in progress. --fill-bots opts in.
var _fill_bots := false
# Task 5.10, server only. null unless --replay-log= was passed.
var _replay_log: ReplayLog = null
# Server only: kept so match_config can be rebuilt after a slot's peer_id
# changes on reconnect (see _rebroadcast_match_config).
var _arena_path := ""
# Client only: the authoritative resume tick while a goal cinematic is playing,
# read by _goal_pause_seconds(). -1 when no goal window is open.
var _client_goal_resume_tick := -1
# §6.3 (task 5.8), client only.
var _is_spectator := false
var _spectator_target_index := 0
# §6.3, server only. Peers that joined mid-match with no slot to reclaim, in
# arrival order, waiting for the next kickoff to hand them a vacated slot.
var _late_joiners: Array[Dictionary] = []
# Task 6.5, server only. Set by ServerMatchLoop immediately before it switches
# to this scene; static because the loop cannot hold a reference to a node that
# does not exist yet, and consumed on read so it cannot leak into a later match.
static var server_arena_override := ""
# §6.3's "cap with --max-spectators". Server only; 0 disables spectating
# entirely, negative means unlimited.
var _max_spectators := -1
var _last_emitted_countdown := -1
var _in_overtime := false
var _match_over := false
# Dedicated-export smoke hook (task 6.2). It is parsed only by the authoritative
# server, cannot be triggered by an RPC, and defaults to disabled.
var _smoke_force_goal_tick := -1
var _smoke_goal_forced := 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():
		# Task 6.3: the same declaration server_boot.gd validated, re-read here
		# LENIENTLY — this scene is one consumer of an argv the smoke harnesses
		# also fill with --role=, --drive-seconds= and client-side flags. The
		# strict pass at the process entry point already rejected any typo in a
		# server flag, so nothing is lost by ignoring what is not ours.
		var config := ServerConfig.parse(OS.get_cmdline_user_args(), false)
		_fill_bots = bool(config.get_value("fill-bots"))
		var spectator_cap := int(config.get_value("max-spectators"))
		_max_spectators = spectator_cap if spectator_cap < 0 else maxi(0, spectator_cap)
		_slot_reservation_seconds = maxf(0.0, float(config.get_value("slot-reservation-seconds")))
		# 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, float(config.get_value("match-length")))
		var smoke_after := float(config.get_value("smoke-force-goal-after"))
		if smoke_after >= 0.0:
			_smoke_force_goal_tick = -2 # arm when PLAYING begins; -1 remains disabled
		var replay_path := String(config.get_value("replay-log"))
		if not replay_path.is_empty():
			# Task 5.10. Diagnostic only: a log that cannot be opened must
			# never stop the server serving the match.
			_replay_log = ReplayLog.new()
			var replay_err := _replay_log.open_for_write(replay_path)
			if replay_err != OK:
				push_warning("NetworkedMatch: could not open replay log %s (%s)" % [replay_path, error_string(replay_err)])
				_replay_log = null
			else:
				print("NetworkedMatch: recording replay log to %s" % replay_path)
		_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)
		MatchSim.match_bootstrap_received.connect(_on_match_bootstrap_received)
		MatchSim.slot_assigned_received.connect(_on_slot_assigned)
		# lobby.gd does this; the match scene never did. Without it a client
		# whose host exits stays in a dead match forever, emitting thousands of
		# "multiplayer instance isn't currently active" / "RPC via a peer which
		# is not connected" errors per run — it only ever left because a test
		# timer happened to fire.
		NetworkManager.disconnected_from_server.connect(_on_disconnected_from_server)
		_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:
	# Task 5.10. Freeing the RefCounted would close the file anyway, but only
	# implicitly and only whenever the last reference happens to go — and it
	# would never print the summary, which is the one line that tells whoever
	# collected the log whether it is complete. Leaving the match scene is the
	# real end of the recording, so end it here explicitly.
	if _replay_log != null:
		_replay_log.close()
		print("NetworkedMatch: replay log closed — %d records, %d bytes, %d dropped%s; uncapped reject totals %s" % [
			_replay_log.records_written, _replay_log.bytes_written, _replay_log.records_dropped,
			" (WRITE FAILED — log is truncated)" if _replay_log.write_failed else "",
			MatchSim.get_reject_totals(),
		])
		_replay_log = null


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

func _start_server() -> void:
	# Task 6.5: the server match loop hands the arena down so rotation is a
	# rotation rather than a coincidence. Consumed once, so a match started any
	# other way (a test harness, a future lobby button) still picks at random.
	var arena_path := server_arena_override if not server_arena_override.is_empty() else ArenaRegistry.random_path()
	server_arena_override = ""
	_arena_path = arena_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.player_name = info.player_name
		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)
	if _replay_log != null:
		MatchSim.input_rejected.connect(_on_input_rejected)
	NetworkManager.client_disconnected.connect(_on_client_disconnected)
	# Piggyback live state on the existing retry loop, so a peer that missed
	# the join-time bootstrap gets one every time it re-asks for config.
	MatchSim.match_config_requested.connect(_send_match_bootstrap)
	MatchNet.player_joined.connect(_on_player_joined_midmatch)

	# §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))
	_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:
					# Recorded, not just counted: this is the drop that used to
					# be permanent input death, and a log that shows only what
					# the server accepted cannot distinguish "the client stopped
					# sending" from "the server refused everything it sent".
					if _replay_log != null:
						_replay_log.record_rejected_input(
							ReplayLog.RecordKind.REJECTED_SEQ_GUARD,
							Engine.get_physics_frames(), peer_id, decoded.get("raw", PackedByteArray())
						)
					return
				# Fall through and accept: this is the escape hatch, not a
				# missing `return`.
			slot.consecutive_seq_rejects = 0
			if _replay_log != null:
				_replay_log.record_input(Engine.get_physics_frames(), peer_id, decoded.get("raw", PackedByteArray()))
			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


# Task 5.10, server only, connected only when a replay log is open. MatchSim
# rejects at the protocol layer and knows nothing about the replay format, so
# the reason-to-record-kind mapping lives here.
func _on_input_rejected(peer_id: int, reason: int, bytes: PackedByteArray) -> void:
	if _replay_log == null:
		return
	var kind := ReplayLog.RecordKind.REJECTED_MALFORMED
	if reason == MatchSim.InputRejectReason.RATE_LIMIT:
		kind = ReplayLog.RecordKind.REJECTED_RATE_LIMIT
	_replay_log.record_rejected_input(kind, Engine.get_physics_frames(), peer_id, bytes)


# --- §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.
	var was_running := _clock_running
	_clock_running = MatchState.is_live(new_state) and not _match_over
	if _end_tick >= 0 and multiplayer.is_server():
		if was_running and not _clock_running:
			# Stopping: bank whatever is left. This replaces the old
			# per-goal `end_tick += resume_tick - goal_tick` arithmetic, which
			# only ever compensated for the CELEBRATION and silently ate the
			# kickoff countdown that follows it.
			_clock_remaining_ticks = maxi(0, _end_tick - at_tick)
			_broadcast_clock_state()
		elif not was_running and _clock_running:
			# Resuming: rebase the absolute end tick off the banked remainder,
			# so every stoppage costs exactly zero regulation time regardless
			# of how long it lasted.
			_end_tick = at_tick + maxi(0, _clock_remaining_ticks)
			_broadcast_clock_state()
	if not multiplayer.is_server():
		# HUDController duck-types on these two, and both previously emitted
		# ONLY inside server-side logic — so a client froze and returned to the
		# lobby without ever showing a result, and its timer never switched to
		# overtime. Derive them from replicated state instead of adding two
		# more RPCs: the client already has the authoritative score, and the
		# state transition itself is the event.
		# Bodies stop on the server at FULL_TIME/RESULTS but the client only
		# ever froze at kickoff and on a goal — so a player flew around for the
		# whole 8s results screen while every other peer saw their ship parked.
		if new_state in [MatchState.State.FULL_TIME, MatchState.State.RESULTS, MatchState.State.LOBBY]:
			_pending_freeze_tick = -1
			_set_bodies_frozen(true)
		if new_state == MatchState.State.OVERTIME_WARMUP:
			_in_overtime = true
			overtime_started.emit()
		elif new_state == MatchState.State.RESULTS:
			_match_over = true
			match_ended.emit(_winning_team(), score.duplicate())
	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:
	# §6.3: before the reset, so a promoted player's ship is placed by this very
	# kickoff rather than left wherever its previous owner abandoned it.
	_promote_late_joiners()
	ServerLog.debug("kickoff", {"reset_gen": (_reset_gen + 1) % 256, "slots": _slots.size()})
	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.
	#
	# resume_tick is used, not just received. A reliable-channel retransmit can
	# deliver this hundreds of ms after goal_tick, and starting a fresh
	# fixed-length timer on ARRIVAL would then run the celebration past the
	# server's own window and overlap the next kickoff. _goal_pause_seconds()
	# below reads this and returns the time actually remaining.
	_client_goal_resume_tick = resume_tick
	_play_goal_celebration(scoring_team, 1 - scoring_team)
	_client_goal_resume_tick = -1


# Overrides GameMode's virtual. On a client during a goal, the pause is
# whatever is LEFT of the authoritative window, not a fresh full duration.
func _goal_pause_seconds() -> float:
	if _client_goal_resume_tick < 0:
		return super()
	var remaining := float(_client_goal_resume_tick - _current_server_tick()) / float(SimConstants.TICK_HZ)
	# Clamp: a window that already elapsed must not produce a negative timer
	# (Godot's create_timer asserts on <= 0), and a wildly future tick from a
	# corrupt packet must not hang the celebration open.
	return clampf(remaining, 0.05, super())


# --- §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:
	# Bank the full regulation length AND set an end tick. The clock is stopped
	# during the opening kickoff, so the banked value is what is displayed
	# until play starts, and the resume path rebases end_tick off it. WARMUP is
	# deliberately NOT folded into end_tick any more: doing so made a 14s match
	# open its HUD at 0:17.
	_clock_remaining_ticks = length_ticks
	_end_tick = Engine.get_physics_frames() + length_ticks
	_broadcast_clock_state()


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


func _on_disconnected_from_server() -> void:
	# Deferred: this arrives from inside NetworkManager's poll, and gotcha 27
	# requires change_scene_to_file never run synchronously from a callback
	# mid-traversal.
	get_tree().change_scene_to_file.call_deferred(ScenePaths.MAIN_MENU)


func _on_match_bootstrap_received(state: int, at_tick: int, new_score: Dictionary, end_tick: int, clock_running: bool, reset_gen: int, remaining_ticks: int) -> void:
	score = new_score.duplicate()
	score_changed.emit(score.duplicate())
	_end_tick = end_tick
	_clock_running = clock_running
	_clock_remaining_ticks = remaining_ticks
	_reset_gen = reset_gen
	_last_local_reset_gen = reset_gen
	_apply_match_state(state, at_tick)


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


# 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
	# While the clock is STOPPED the remaining time is frozen, not derived from
	# the current tick. Deriving it regardless meant regulation time drained
	# during every goal pause and kickoff countdown: measured 180 ticks — one
	# whole WARMUP — lost per goal, plus the pre-kickoff display opening at
	# 0:17 for a 14s match. _clock_remaining_ticks is the authority whenever
	# _clock_running is false.
	var remaining_ticks := maxi(0, _end_tick - _current_server_tick()) if _clock_running else maxi(0, _clock_remaining_ticks)
	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())
	ServerLog.info("match_ended", {"score_0": score.get(0, 0), "score_1": score.get(1, 0), "overtime": _in_overtime})
	_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
	ServerLog.info("goal", {
		"team": scoring_team, "score_0": score.get(0, 0), "score_1": score.get(1, 0),
		"tick": Engine.get_physics_frames(),
	})
	var goal_tick := Engine.get_physics_frames()
	var resume_tick := goal_tick + int(_goal_pause_seconds() * SimConstants.TICK_HZ)
	# No end_tick arithmetic here any more: entering GOAL_PAUSE banks the
	# remaining ticks and leaving it rebases end_tick (see _apply_match_state),
	# which covers the celebration AND the kickoff countdown after it. Still
	# tick-derived, so no float drift accumulates across ten goals.
	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 _maybe_force_smoke_goal() -> void:
	if _smoke_goal_forced or _smoke_force_goal_tick == -1 or match_state != MatchState.State.PLAYING:
		return
	if _smoke_force_goal_tick == -2:
		var config := ServerConfig.parse(OS.get_cmdline_user_args(), false)
		_smoke_force_goal_tick = Engine.get_physics_frames() + int(maxf(0.0, float(config.get_value("smoke-force-goal-after"))) * SimConstants.TICK_HZ)
		ServerLog.info("smoke_goal_armed", {"tick": _smoke_force_goal_tick})
		return
	if Engine.get_physics_frames() < _smoke_force_goal_tick:
		return
	_smoke_goal_forced = true
	ServerLog.info("smoke_goal_forced", {"tick": Engine.get_physics_frames()})
	_on_goal_registered(0)


# Task 5.9. Server-only by construction: _respawn_escaped_bodies() is gated on
# _owns_world_simulation(). The bump uses Phase 2's deferred path because the
# respawn only QUEUES a teleport — bumping now would broadcast the new
# generation alongside the still-escaped position, which is exactly the 27m
# slide that fix exists to prevent.
# --- §6.4 disconnects and reconnects (tasks 5.6/5.7) -----------------------

const SLOT_RESERVATION_SECONDS := 30.0
# Server only, --slot-reservation-seconds=. §6.3's promotion can only happen at
# a kickoff AFTER the departed player's reservation lapses, so a smoke test of
# it would otherwise have to run for over half a minute before the interesting
# moment. Same rationale and same shape as --match-length: a server-side
# override, never something a client can shorten for anyone.
var _slot_reservation_seconds := SLOT_RESERVATION_SECONDS


func _on_client_disconnected(peer_id: int) -> void:
	if not multiplayer.is_server():
		return
	# A spectator waiting for a slot can leave too, and a queue entry for a
	# departed peer would hand the next free slot to nobody.
	_forget_late_joiner(peer_id)
	for slot in _slots:
		if slot.peer_id != peer_id or slot.disconnected:
			continue
		slot.disconnected = true
		slot.reserved_until_tick = Engine.get_physics_frames() + int(_slot_reservation_seconds * SimConstants.TICK_HZ)
		_swap_slot_controller(slot, _build_takeover_controller())
		print("NetworkedMatch: peer %d (%s) disconnected; ship kept, slot reserved for %.0fs" % [
			peer_id, slot.player_name, _slot_reservation_seconds
		])
		break
	_abort_if_abandoned()


# §6.4 has two rules that pull against each other: reserve a departed player's
# slot for 30s, and abort to the lobby once the last human leaves. Applied
# naively the abort wins instantly in a 1v1 — the moment the only player drops,
# the match is torn down and their reservation can never be redeemed, which
# makes the reconnect path unreachable exactly when it matters most (a single
# player whose connection blipped). The reservation therefore takes precedence:
# abort only once nobody is connected AND nobody is still expected back.
func _abort_if_abandoned() -> void:
	if MatchState.is_terminal(match_state):
		return
	var now := Engine.get_physics_frames()
	for slot in _slots:
		if not slot.disconnected:
			return   # somebody is still playing
		if slot.reserved_until_tick >= 0 and now <= slot.reserved_until_tick:
			return   # somebody may still come back
	# §6.3's queue counts as "somebody is still here" for the same reason the
	# reservation does. Without this, a spectator waiting for the slot that just
	# opened up is dumped back to the lobby at the exact moment they were about
	# to get it — and they are a connected human watching a live match, which is
	# not what "abandoned" means.
	for entry in _late_joiners:
		if int(entry["peer_id"]) in multiplayer.get_peers():
			return
	print("NetworkedMatch: no players left and no reservations outstanding, aborting to lobby")
	_set_match_state(MatchState.State.LOBBY)
	get_tree().change_scene_to_file.call_deferred(ScenePaths.LOBBY)


# Task 5.7. Ship.set_controller() calls queue_free() on the OUTGOING
# controller, so slot.controller is a dangling reference the instant the swap
# happens — and _physics_process writes slot.controller.action every single
# tick. Rebinding must therefore happen in the same transaction as the swap,
# never as a follow-up statement that an early return or an await could skip.
func _swap_slot_controller(slot: SlotInfo, replacement: ShipController) -> void:
	if not is_instance_valid(slot.ship):
		slot.controller = null
		return
	slot.ship.set_controller(replacement)
	slot.controller = replacement


func _build_takeover_controller() -> ShipController:
	if _fill_bots:
		return _build_opponent(bot_model_path, bot_reaction_ticks, bot_action_noise, "NetworkedMatch")
	# §6.4's default for public servers: inert but still simulated, exactly the
	# placeholder GameMode already uses for an unfilled slot. An abandoned ship
	# that keeps flying on its last input would be worse than one that coasts.
	return ShipController.new()


# Called when a peer joins while this match is already running. Returns true if
# it reclaimed a reserved slot (§6.4's 30s identity-keyed reservation).
func _try_reclaim_slot(peer_id: int, player_name: String) -> bool:
	if not multiplayer.is_server():
		return false
	var now := Engine.get_physics_frames()
	for slot in _slots:
		if not slot.disconnected or slot.player_name == "" or slot.player_name != player_name:
			continue
		if slot.reserved_until_tick >= 0 and now > slot.reserved_until_tick:
			continue   # reservation lapsed; this is a fresh joiner, not a return
		slot.peer_id = peer_id
		slot.disconnected = false
		slot.reserved_until_tick = -1
		# Reset the input pipeline: the returning client starts its sequence
		# numbering from scratch, and the old buffer's cursor belongs to a
		# different epoch entirely (input_jitter_buffer.gd's seeding comment).
		slot.jitter_buffer = InputJitterBuffer.new()
		slot.consecutive_seq_rejects = 0
		_swap_slot_controller(slot, RLShipController.new())
		# CRITICAL, and the reason a reconnect silently became a spectator: the
		# slot's peer_id just changed, but MatchSim caches the last
		# match_config and replays THAT to anyone who asks. A reconnecting
		# client in a fresh process requests config, receives the pre-
		# disconnect peer-id array, cannot find itself in it, leaves
		# _my_slot null and falls through to the spectator path — no ship, no
		# input, for the rest of the match. Re-broadcast so the cache and the
		# roster agree again.
		_rebroadcast_match_config()
		_send_match_bootstrap(peer_id)
		print("NetworkedMatch: peer %d reclaimed %s's reserved slot" % [peer_id, player_name])
		return true
	return false


# §6.3/§6.4. A peer joining while this match runs is either a returning player
# claiming their reserved slot, or a late joiner — who spectates until the next
# kickoff, because swapping a controller at a kickoff boundary is free and
# mid-play it is not.
func _on_player_joined_midmatch(peer_id: int, player_name: String) -> void:
	if not multiplayer.is_server() or _slots.is_empty():
		return
	if _try_reclaim_slot(peer_id, player_name):
		return
	if _max_spectators >= 0 and _spectator_count() > _max_spectators:
		print("NetworkedMatch: spectator cap (%d) reached, disconnecting peer %d" % [_max_spectators, peer_id])
		# Same call the abuse paths use (match_sim.gd:285, match_net.gd:207) —
		# default force=false, so ENet flushes cleanly rather than leaving the
		# server's own peer bookkeeping inconsistent (§9 gotcha on force=true).
		multiplayer.multiplayer_peer.disconnect_peer(peer_id)
		return
	# §6.3: a spectator/late joiner reconstructs from this, since match_config
	# carries arena and roster only — no score, clock or match state.
	_send_match_bootstrap(peer_id)
	# "Spectate now, take the slot at the next kickoff" — queued here, acted on
	# in _promote_late_joiners(). Queued in arrival order and consumed from the
	# front, so waiting is first-come-first-served rather than whichever slot
	# index happens to free up first.
	_late_joiners.append({"peer_id": peer_id, "player_name": player_name})
	print("NetworkedMatch: peer %d (%s) joined mid-match; spectating until the next kickoff" % [peer_id, player_name])


# §6.3's "free slot mid-match → spectate now, take the slot at the next
# kickoff". Called from _begin_kickoff BEFORE the reset transforms are read, so
# a promoted player's ship is placed by the same kickoff everyone else gets and
# the controller swap lands on an already-frozen body — which is the whole
# reason the spec puts it at a kickoff boundary rather than mid-play.
#
# A slot is available when its player has gone AND their 30s reservation has
# lapsed (§6.4). Taking a still-reserved slot would quietly break the reconnect
# promise, so the reservation always outranks the queue.
func _promote_late_joiners() -> void:
	if not multiplayer.is_server() or _late_joiners.is_empty():
		return
	var connected := multiplayer.get_peers()
	# A queued joiner may have left again while waiting. Drop them here rather
	# than handing a slot to a peer that no longer exists — which would look
	# exactly like an occupied slot nobody is playing.
	var waiting: Array[Dictionary] = []
	for entry in _late_joiners:
		if int(entry["peer_id"]) in connected:
			waiting.append(entry)
	_late_joiners = waiting

	var now := Engine.get_physics_frames()
	var promoted := false
	for index in _slots.size():
		if _late_joiners.is_empty():
			break
		var slot := _slots[index]
		if not slot.disconnected:
			continue
		if slot.reserved_until_tick >= 0 and now <= slot.reserved_until_tick:
			continue
		var joiner: Dictionary = _late_joiners.pop_front()
		var joiner_peer := int(joiner["peer_id"])
		slot.peer_id = joiner_peer
		slot.player_name = String(joiner["player_name"])
		slot.disconnected = false
		slot.reserved_until_tick = -1
		# Same reasoning as the reclaim path: the arriving client numbers its
		# input sequence from scratch, and the old cursor belongs to a different
		# epoch entirely (see input_jitter_buffer.gd's seeding comment).
		slot.jitter_buffer = InputJitterBuffer.new()
		slot.consecutive_seq_rejects = 0
		_swap_slot_controller(slot, RLShipController.new())
		MatchSim.send_slot_assigned(joiner_peer, index)
		promoted = true
		print("NetworkedMatch: peer %d (%s) took slot %d at the kickoff" % [joiner_peer, slot.player_name, index])
	if promoted:
		# Same cache hazard the reclaim path documents: MatchSim replays the
		# last match_config to anyone who asks, and it now names the wrong peer
		# for this slot.
		_rebroadcast_match_config()


func _forget_late_joiner(peer_id: int) -> void:
	for i in _late_joiners.size():
		if int(_late_joiners[i]["peer_id"]) == peer_id:
			_late_joiners.remove_at(i)
			return


# Connected peers that hold no slot. Counted from the live peer list rather
# than tracked incrementally, so a spectator that drops cannot leak a unit of
# the cap permanently.
func _spectator_count() -> int:
	var slotted := {}
	for slot in _slots:
		if not slot.disconnected:
			slotted[slot.peer_id] = true
	var count := 0
	for peer_id in multiplayer.get_peers():
		if not slotted.has(peer_id):
			count += 1
	return count


# Rebuilds match_config from the CURRENT slot list and re-sends it. Slot order
# (and therefore snapshot body order) is preserved because _slots itself is
# never reordered — only a slot's peer_id changes on reclaim.
func _rebroadcast_match_config() -> void:
	var peer_ids := PackedInt32Array()
	var teams := PackedInt32Array()
	var spawn_indices := PackedInt32Array()
	for slot in _slots:
		peer_ids.append(slot.peer_id)
		teams.append(slot.team)
		spawn_indices.append(slot.spawn_index)
	MatchSim.send_match_config(_arena_path, peer_ids, teams, spawn_indices)


# §6.2 step 2: give one peer the live state it cannot get from match_config.
func _send_match_bootstrap(peer_id: int) -> void:
	MatchSim.send_match_bootstrap(
		peer_id, match_state, match_state_since_tick, score.duplicate(),
		_end_tick, _clock_running, _reset_gen, _clock_remaining_ticks
	)


func _expire_slot_reservations() -> void:
	var now := Engine.get_physics_frames()
	for slot in _slots:
		if slot.disconnected and slot.reserved_until_tick >= 0 and now > slot.reserved_until_tick:
			slot.reserved_until_tick = -1
			print("NetworkedMatch: %s's slot reservation lapsed" % slot.player_name)
			# The abort was deferred while this reservation was live; now that
			# it has lapsed, re-check whether anyone is left at all.
			_abort_if_abandoned()


func _on_bodies_respawned() -> void:
	if not multiplayer.is_server():
		return
	_pending_reset_gen_bump = true
	_pending_reset_gen_bump_tick = Engine.get_physics_frames()


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:
		# §6.4: `stalled` is what greys out the nameplate, so a disconnected
		# player must set it immediately rather than waiting the ~500ms it
		# takes their abandoned jitter buffer to starve into the same state.
		bodies.append(_ship_to_net_body_state(slot.ship, slot.jitter_buffer.stalled or slot.disconnected) 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)
			if _replay_log != null:
				_replay_log.record_snapshot(server_tick, bytes)
			MatchSim.send_snapshot(slot.peer_id, bytes)
	# §6.3: "a spectator receives identical snapshots (the snapshot is already
	# a broadcast — zero extra server work)". That was only true of the SEGMENT:
	# the loop above unicasts one packet per SLOT, so a peer without a slot
	# received nothing at all — no poses, no reset_gen, no match_state byte.
	# An adversarial review caught it; spectating was entirely non-functional.
	# The body segment is shared, so this really is just one extra send per
	# spectator. The per-slot header fields are meaningless without a slot:
	# there is no acknowledged input sequence, and -1 is the codec's own
	# "client not established" value for buffer depth (§3.3).
	var spectator_bytes := PackedByteArray()
	for peer_id in connected_peers:
		var has_slot := false
		for slot in _slots:
			if slot.peer_id == peer_id:
				has_slot = true
				break
		if has_slot:
			continue
		if spectator_bytes.is_empty():
			spectator_bytes = NetCodec.pack_snapshot(0, -1, 0, segment)
		MatchSim.send_snapshot(peer_id, spectator_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

	# §6.3 (task 5.8): a peer with no slot is a spectator. It receives the
	# identical snapshot broadcast (zero extra server work), spawns no ship of
	# its own, and points a camera rig at somebody else's.
	_is_spectator = _my_slot == null
	_spawn_hud()
	if _is_spectator:
		_spectator_target_index = 0
		_point_spectator_camera()
		print("NetworkedMatch: no slot for this peer — spectating (%d ship(s) + ball)" % _slots.size())
	if is_instance_valid(_my_slot) and is_instance_valid(_my_slot.ship):
		spawn_camera_rig(_my_slot.ship)
		_take_local_ownership(_my_slot)
	# The roster now exists, so a kickoff that raced ahead of match_config can
	# finally be placed against the right bodies.
	_apply_pending_kickoff()


# Client only. Everything that makes one of the spawned ships THIS peer's own:
# contact hooks and the local input controller. Factored out of
# _on_match_config_received because §6.3's late-joiner promotion needs the
# identical setup at a completely different moment, and a second copy of it
# would be a copy that silently drifts.
func _take_local_ownership(slot: SlotInfo) -> void:
	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":
		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
		slot.ship.add_child(bot)
		_local_input_timeline = LocalInputTimeline.new()
		_local_net_controller = LocalNetShipController.new(bot, _local_input_timeline)
		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)
		slot.ship.set_controller(_local_net_controller)


# §6.3 (task 5.8), client only: the server has handed this peer a vacated slot
# at a kickoff. Broadcast, so every client runs the first half — their own copy
# of the slot list must name the new owner — and only the promoted peer runs
# the second.
func _on_slot_assigned(peer_id: int, slot_index: int) -> void:
	if multiplayer.is_server() or slot_index < 0 or slot_index >= _slots.size():
		return
	var slot := _slots[slot_index]
	slot.peer_id = peer_id
	if peer_id != multiplayer.get_unique_id() or not _is_spectator:
		return

	# This body has been a REMOTE one until now: driven by transform writes from
	# the interpolator, with Godot's own physics interpolation switched off so
	# those writes could not fight it (§4.6). Both have to be undone, and the
	# interpolator emptied — its buffered samples describe the previous owner's
	# flight and would otherwise be smoothed into the first predicted frames.
	_my_slot = slot
	_is_spectator = false
	slot.interpolator = NetInterpolator.new()
	slot.visual_smoother_reset = true
	slot.visual_position_offset = Vector3.ZERO
	slot.visual_rotation_offset = Quaternion.IDENTITY
	if is_instance_valid(slot.ship) and is_instance_valid(slot.ship.visual):
		slot.ship.visual.physics_interpolation_mode = Node.PHYSICS_INTERPOLATION_MODE_INHERIT
	# Stay frozen until the first authoritative pose arrives, exactly as a fresh
	# client does — _on_snapshot_received teleports to it, unfreezes, and starts
	# prediction. Unfreezing here instead would predict from whatever pose the
	# interpolator last wrote, which is a render-side approximation.
	_local_prediction_ready = false
	_input_seq = 0
	# Same call the reset path uses: everything recorded so far belongs to a
	# peer that was not simulating anything.
	_local_prediction_history.begin_epoch()
	_pending_local_reconciliation = {}
	_take_local_ownership(slot)
	# The HUD was built in spectator mode, which hides the ship instruments and
	# wires nothing to a ship. It reads spectator_mode once, a frame after
	# _ready, so flipping the flag on the live instance does nothing — rebuild.
	if is_instance_valid(hud):
		hud.queue_free()
	_spawn_hud()
	if is_instance_valid(_camera_rig):
		_camera_rig.target = slot.ship
		hud.ship = slot.ship
	else:
		spawn_camera_rig(slot.ship)
	print("NetworkedMatch: promoted from spectator to player in slot %d" % slot_index)


func _spawn_hud() -> void:
	hud = HUD_SCENE.instantiate()
	# BEFORE add_child: HUDController reads this in _initialize_hud(), which
	# runs one process frame after _ready(). Setting it afterwards would be a
	# race against that frame, and losing it means a spectator's HUD
	# push_error()s about a missing ship and wires up nothing at all.
	hud.spectator_mode = _is_spectator
	add_child(hud)


# §6.3's "points a camera rig at a chosen ship or the ball", plus the target
# cycling. Targets are every ship in slot order, then the ball.
# §6.3's "cycle targets". Bound to the existing `reset_ball` action, which is
# already a mode-level key and is meaningless to a spectator (it only fires in
# Free Play), rather than adding a new binding to project.godot for one mode.
func _unhandled_input(event: InputEvent) -> void:
	# super() is NOT optional here. GameMode._unhandled_input owns ui_cancel ->
	# main menu, and this override returned early for every non-spectator
	# without ever chaining, which silently killed Esc for every networked
	# player. Handle the spectator key, then always fall through.
	if _is_spectator and event.is_action_pressed("reset_ball"):
		cycle_spectator_target(1)
		get_viewport().set_input_as_handled()
		return
	super(event)


func _spectator_target_count() -> int:
	return _slots.size()


func _point_spectator_camera() -> void:
	var count := _spectator_target_count()
	if count == 0:
		return
	_spectator_target_index = posmod(_spectator_target_index, count)
	# SHIPS ONLY. ShipCameraRig.target is declared `var target: Ship`
	# (ship_camera.gd:37) and the rig reaches into ship-only API (`visual`,
	# `is_turbo_active`, `get_speed_ratio`), so assigning the ball here was a
	# type error waiting to fire the moment anyone cycled past the last ship.
	# The rig already has its own ball-cam MODE for watching the ball, which is
	# the supported way to do it — this cycles whose ship we follow.
	var target: Ship = null
	if _spectator_target_index < _slots.size():
		target = _slots[_spectator_target_index].ship
	if not is_instance_valid(target):
		return
	if not is_instance_valid(_camera_rig):
		# spawn_camera_rig types its parameter as Ship, so the ball can only
		# ever be a LATER target, never the one the rig is created with.
		var first_ship: Ship = null
		for slot in _slots:
			if is_instance_valid(slot.ship):
				first_ship = slot.ship
				break
		if first_ship == null:
			return
		spawn_camera_rig(first_ship)
	if is_instance_valid(_camera_rig):
		_camera_rig.target = target
		if is_instance_valid(hud):
			hud.ship = target


func cycle_spectator_target(step: int = 1) -> void:
	if not _is_spectator:
		return
	var count := _spectator_target_count()
	if count == 0:
		return
	_spectator_target_index = posmod(_spectator_target_index + step, count)
	_point_spectator_camera()


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()
		_maybe_force_smoke_goal()
		_expire_slot_reservations()
	# 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:
			# is_instance_valid, not a null check: set_controller() queue_free()s
			# the outgoing controller on every disconnect swap, and a freed
			# object is non-null right up until the frame it is collected.
			var consumed := slot.jitter_buffer.consume()
			# Only a live player's slot is driven by the wire. A slot whose
			# player disconnected now holds a bot or the inert base controller
			# (§6.4), which drives itself — overwriting its action every tick
			# from a permanently-starving jitter buffer would pin it to the
			# departed player's last input forever.
			if is_instance_valid(slot.controller) and slot.controller is RLShipController:
				(slot.controller as RLShipController).action = consumed
		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
	# The scene change on disconnect is deferred, so this can run one more time
	# against a torn-down peer — which throws from get_unique_id() rather than
	# returning anything.
	if multiplayer.multiplayer_peer == null or multiplayer.multiplayer_peer.get_connection_status() != MultiplayerPeer.CONNECTION_CONNECTED:
		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())