feat(multiplayer): Phase 4 prediction correctness + two input-death fixes

Closes Phase 4's outstanding action-sequence-correctness invariant, then
fixes two server-side bugs an adversarial review of that work uncovered.
Server simulation, bot observations, collision resources and tick rate are
unchanged: the server_physics_parity trace is byte-for-byte identical to
HEAD across 360 ticks including both ships' full observation vectors.

4.11 - prediction history filed under the ISSUING sequence

_send_local_input filed each post-step predicted state under the timeline's
estimate of the sequence the server would consume this tick, trailing
issuance by input_lead. The body had integrated the intent issued under
_input_seq, so predicted[S] held "state after the intent from now" while
the server's authority for S is "state after action(S)". They agree only
while the stick is still. Filing under _input_seq costs nothing: which
action the ship uses is decided in LocalNetShipController.get_action() and
is untouched.

Every prior Phase 4 gate held its input steady, and a steady input cannot
falsify a sequence label - the 60s runs honestly reported marker=0/3784.
New --exercise-input-transitions role toggles thrust every 6 ticks; it is
the only gate that can catch a label regression. Verified non-vacuous: the
old label fails it at 50%.

4.12 - issued-but-unsimulated sequences, and the release path

An attack (delta > 1) issues and sends several sequences for one local
physics step. Those gap sequences had no recorded prediction, so a server
ack of one reported missing_not_recorded - indistinguishable from ring
loss, costing a teleport and resync suppression several times a minute.
They are now recorded stateless via record_unsimulated() and answered with
a new "skip" decision mode. Free-flight hard snaps: 25/8/4 -> 0/0/0.

A release (delta == 0) re-recorded at the unchanged _input_seq, filing the
current intent under a sequence that went out carrying a different action;
LocalInputTimeline deliberately refuses to mutate an issued sequence, so
the ring contradicted the wire. Recording is now skipped on release ticks.

4.13 - two Phase 3 bugs silently killing player input

(a) InputJitterBuffer.consume() advanced last_applied_seq on every tick
including a starve. Since ingest() discards seq <= last_applied_seq, one
starve on a sequence the client had not sent yet stranded the stream one
ahead of arrivals permanently - both sides advancing in lockstep, every
honest packet discarded on arrival. The client's own input_lead release is
enough to trigger it, so input died for ~30 ticks roughly every 6.5s on a
clean LAN. Now only gives up on a sequence once strictly newer data proves
it lost. Silent-client stall and ring-overflow resync are unchanged.

(b) The seq-range guard bounded incoming seq against highest_ingested_seq,
which only advances inside ingest(), which that guard gates. After a ~2s
host hitch every packet was rejected forever with no diagnostic (600+
consecutive rejections reproduced via SIGSTOP). Third iteration of this
guard; each previous version bounded against a value only the accepted
path could advance. Adds an escape after 10 consecutive rejections, which
grants an attacker nothing the rate limiter does not already bound.

(c) The transitions gate reported PASS at 3.76% while input was completely
dead, because suppression stops _record_metrics - a worse outage yields
fewer samples and a LOWER rate. Now scales the required sample count with
run length and asserts the wire's server_stalled bit. Reverting both fixes
makes it fail at samples 292/600, server_stalled=true, input_lead=12.

Fixing (a) also explained a residual the review had already traced: 151 of
151 action-marker mismatches were the server repeating a stale action on a
starve, not a prediction defect. Marker is now 0.00% in all three
conditions (was 1.7-2.5%), and free-flight p99 improved to
0.141/0.168/0.154m from 0.170/0.176/0.184m.

Two pre-existing test defects fixed alongside: the ball gate asserted
RTT-masking on a link with no RTT (flaked 2 in 5; now asserted only at
rtt >= 20ms, 5/5 under latency), and the two-bot CI compared scores across
a 3-5s window (now polls the scores the server actually held; note
score_changed is emitted only on the client path).

QA: 72 unit tests; 60s free-flight at LAN/80+-20ms/5% loss; transition
gate in all three; 2.0s and 3.5s host-freeze recovery; ball contact x5;
two-bot CI x3; all three abuse roles; net/match_net/clock/lobby smokes.

Phase 4 sign-off still pending a human playtest at ~100ms RTT - the
milestone asks how it feels, which no gate here answers.

Game/scripts/networked_match.gd

@@ -1,16 +1,10 @@
 class_name NetworkedMatch
 extends GameMode
 
-# Phase 2: server-authoritative simulation, dumb client (multiplayer-todo.md
-# §7 Phase 2). The server runs the real physics for every ship — via
-# RLShipController, fed by each connected player's forwarded input — and
-# the ball, and broadcasts NetCodec snapshots at 60Hz. The client renders
-# everything, including its own ship, from the interpolation buffer; there
-# is no local prediction yet (that's Phase 4), so every body on the client
-# is FREEZE_MODE_KINEMATIC and driven entirely by incoming snapshots.
-# Tasks 4.1/4.2 add the seq-tagged recording and comparison plumbing that
-# Phase 4 will need (LocalPredictionHistory below), but deliberately stop
-# short of unfreezing or locally simulating anything — that is task 4.3.
+# 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
@@ -32,7 +26,11 @@ 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 —
@@ -43,6 +41,24 @@ const HUD_SCENE = preload("res://scenes/HUD.tscn")
 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
@@ -79,49 +95,80 @@ class SlotInfo:
 	var ship: Ship
 	var controller: RLShipController      # server only
 	var jitter_buffer := InputJitterBuffer.new()   # server only (§3.2)
+	# Server only. Consecutive packets rejected by the seq-range guard, reset by
+	# any accepted one. The guard's bound is derived from a value only an
+	# ACCEPTED packet can advance, so without an escape hatch it latches shut
+	# permanently — see the guard's own comment in _on_input_received.
+	var consecutive_seq_rejects := 0
 	var last_client_send_ms := 0          # server only: echoed back per-peer next snapshot (§2.4)
 	var interpolator := NetInterpolator.new()   # client only
+	var visual_smoother_reset := true
+	var visual_position_offset := Vector3.ZERO
+	var visual_rotation_offset := Quaternion.IDENTITY
 
 
 var _slots: Array[SlotInfo] = []
 var _my_slot: SlotInfo = null             # client only
+var _local_prediction_ready := false      # client waits for its first authoritative pose
 var _ball_interpolator := NetInterpolator.new()   # client only
-# client only: reads local input each tick to forward. Normally a
-# PlayerShipController that's deliberately never added to a Ship/the tree —
-# get_action() only touches the global Input singleton, so it needs no
-# scene context. --test-bot mode (task 3.6) swaps this for a real
-# AIShipController once the client's own ship is known (see
-# _on_match_config_received) — unlike PlayerShipController, AIShipController
-# DOES need real scene context (get_parent() as Ship, plus ball/teammate/
-# opponent discovery via groups), so it's parented onto _my_slot.ship via
-# Ship.set_controller() rather than left floating.
-var _local_input_sampler: ShipController = PlayerShipController.new()
+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)
-# Latest raw result from LocalPredictionHistory.compare_authoritative(). This
-# pass records and compares only; Phase 4.3 will consume it to choose and
-# apply the actual reconciliation correction.
-#
-# Read its error fields with the caveat documented on
-# _local_ship_prediction_state(): until task 4.3 unfreezes and locally
-# simulates the local ship, the "predicted" side of every comparison is an
-# interpolated past-snapshot pose, not a forward simulation. The
-# position_error / rotation_error_radians / *_velocity_error numbers
-# therefore measure interpolation-vs-authoritative drift, and are NOT
-# prediction error. Expect them to be small and largely uninformative, and
-# do not calibrate any snap/blend threshold against them yet.
+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
@@ -180,6 +227,14 @@ func _ready() -> void:
 				_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)
@@ -205,18 +260,8 @@ func _owns_world_simulation() -> bool:
 	return multiplayer.is_server()
 
 
-# _local_input_sampler is a plain Node (PlayerShipController extends
-# ShipController extends Node) that's deliberately never added to the tree
-# — dropping the last reference to it does not free it. An adversarial
-# review traced the "3 resources still in use at exit" warning on every
-# Phase 2 test run directly to this: --verbose named the leaked script
-# chain (player_ship_controller.gd, ship_controller.gd, ship_action.gd)
-# exactly, and adding this cleanup made the warning disappear. Runs
-# unconditionally (not just client-side) since the field is initialized
-# unconditionally too, despite its "client only" comment.
 func _exit_tree() -> void:
-	if is_instance_valid(_local_input_sampler):
-		_local_input_sampler.free()
+	pass
 
 
 # ============================================================
@@ -301,10 +346,36 @@ func _on_input_received(peer_id: int, decoded: Dictionary) -> void:
 			# 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:
-				return
+				slot.consecutive_seq_rejects += 1
+				if slot.consecutive_seq_rejects < SEQ_REJECT_RESYNC_LIMIT:
+					return
+				# Fall through and accept: this is the escape hatch, not a
+				# missing `return`.
+			slot.consecutive_seq_rejects = 0
 			jb.ingest(seq, decoded["actions"])
 			slot.last_client_send_ms = decoded["client_send_ms"]
 			return
@@ -361,7 +432,10 @@ func _broadcast_snapshot() -> void:
 	for slot in _slots:
 		if connected_peers.has(slot.peer_id):
 			var last_input_seq := maxi(slot.jitter_buffer.last_applied_seq, 0)
-			var bytes := NetCodec.pack_snapshot(last_input_seq, slot.jitter_buffer.depth(), slot.last_client_send_ms, segment)
+			# -1 is reserved for client "not established" state. -2 reports a
+			# genuine sustained server starvation event.
+			var advertised_depth := -2 if slot.jitter_buffer.starved_ticks >= STARVATION_ADVERTISEMENT_TICKS else slot.jitter_buffer.depth()
+			var bytes := NetCodec.pack_snapshot(last_input_seq, advertised_depth, slot.last_client_send_ms, segment)
 			MatchSim.send_snapshot(slot.peer_id, bytes)
 
 
@@ -431,6 +505,13 @@ func _on_match_config_received(arena_path: String, peer_ids: PackedInt32Array, t
 	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():
@@ -439,45 +520,49 @@ func _on_match_config_received(arena_path: String, peer_ids: PackedInt32Array, t
 		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 is_instance_valid(slot.ship.visual):
+		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 slot.peer_id == my_id:
+		if is_local:
 			_my_slot = slot
 
 	_spawn_hud()
 	if is_instance_valid(_my_slot) and is_instance_valid(_my_slot.ship):
 		spawn_camera_rig(_my_slot.ship)
+		_my_slot.ship.ball_contact.connect(_on_local_ball_contact)
+		# Headless training ships intentionally do not install Ship's render-side
+		# body_entered signal. Attach this client-only callback only to the
+		# locally predicted match ship so contact QA sees the same event without
+		# changing training instances.
+		if DisplayServer.get_name() == "headless":
+			_my_slot.ship.body_entered.connect(_on_local_ship_body_entered)
 		if not _test_bot_model_path.is_empty():
-			# --test-bot (task 3.6): swap the human input sampler for a real
+			# --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
-			# client's ships are all FREEZE_MODE_KINEMATIC and driven purely by
-			# transform writes (§4.1/§4.6) — nothing here ever writes
-			# linear_velocity/angular_velocity onto them, so ShipObservations
-			# always sees every ship (including this one's own) as
-			# stationary. The policy still produces well-formed, bounded
-			# actions from that degraded input (PolicyNetwork's output layer
-			# is bounded regardless of input quality) — good enough for a CI
-			# traffic generator, which is this task's actual job, not bot
-			# skill.
+			# local bot controller to the genuinely simulated local ship.
 			var bot := AIShipController.new()
 			bot.model_path = _test_bot_model_path
-			_my_slot.ship.set_controller(bot)
-			# Reassigning _local_input_sampler would orphan the original
-			# PlayerShipController it pointed to — the exact same leak class
-			# an adversarial review already caught once for this same field
-			# (it's a plain Node, never in the tree, so nothing else would
-			# ever free it). It's never parented, so free() is safe directly.
-			if is_instance_valid(_local_input_sampler):
-				_local_input_sampler.free()
-			_local_input_sampler = bot
+			_my_slot.ship.add_child(bot)
+			_local_input_timeline = LocalInputTimeline.new()
+			_local_net_controller = LocalNetShipController.new(bot, _local_input_timeline)
+			_my_slot.ship.set_controller(_local_net_controller)
+		else:
+			var player := PlayerShipController.new()
+			_local_input_timeline = LocalInputTimeline.new()
+			_local_net_controller = LocalNetShipController.new(player, _local_input_timeline)
+			_local_net_controller.add_child(player)
+			_my_slot.ship.set_controller(_local_net_controller)
 
 
 func _spawn_hud() -> void:
@@ -488,57 +573,65 @@ func _spawn_hud() -> void:
 func _send_local_input() -> void:
 	if _slots.is_empty():
 		return   # match_config hasn't arrived yet
-	var action := _local_input_sampler.get_action().copy()
+	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).
-	var delta := _input_lead_controller.update(_last_known_input_buffer_depth)
-	_input_seq += delta
-	# Record this tick's (seq, action, local-ship state) triple. action is
-	# sampled exactly once above; record() makes its own copy for the
-	# longer-lived prediction history. See _local_ship_prediction_state() for
-	# what the "state" half does and does not currently mean.
-	if _my_slot != null and is_instance_valid(_my_slot.ship):
-		_local_prediction_history.record(_input_seq, action, _local_ship_prediction_state(_my_slot.ship, action))
-	# Redundancy (§3.1): carry the last MAX_REDUNDANCY ticks' actions,
-	# newest-first, so a burst of up to (MAX_REDUNDANCY - 1) consecutive
-	# packet losses still lets the server recover every dropped tick's
-	# action from a later packet — InputJitterBuffer.ingest() discards
-	# whichever of these the server already applied, so re-sending old
-	# ticks every packet is harmless, not just tolerated. NetCodec's wire
-	# format has no per-entry seq field — actions[i] is implicitly
-	# "seq - i" — so _input_history must actually BE that many consecutive
-	# ticks, not just "the last few samples taken". A plain push_front on
-	# every tick regardless of delta broke that: an adversarial review
-	# found a lead change silently relabelled older entries (a duplicated
-	# tick shifts everything back by one position without a matching seq
-	# change, and a skip-ahead makes the whole history discontiguous with
-	# the new seq), causing the server to replay already-applied ticks or
-	# apply the wrong redundant copy for a given seq. Handle each case on
-	# its own terms instead of always pushing.
-	if delta == 1:
-		_input_history.push_front(action)
-		if _input_history.size() > NetCodec.MAX_REDUNDANCY:
-			_input_history.resize(NetCodec.MAX_REDUNDANCY)
-	elif delta == 0:
-		# Release: seq didn't advance, so this tick's freshest sample
-		# REPLACES the front entry (still "seq") rather than pushing
-		# everything else back a position under a label that no longer
-		# matches what's actually there.
-		if _input_history.is_empty():
-			_input_history.push_front(action)
-		else:
-			_input_history[0] = action
-	else:
-		# Attack: seq jumped ahead by more than one, so nothing previously
-		# in history is contiguous with the new seq any more — the skipped
-		# range was never sent, by design (that's what "buys more server-
-		# side buffer margin" means). Reset the redundancy window to just
-		# this tick's sample; it rebuilds naturally over the next few
-		# ticks, the same way it does at connection start.
-		_input_history = [action]
+	_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:
+		# 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)
 
@@ -559,19 +652,45 @@ func _on_snapshot_received(decoded: Dictionary) -> void:
 	# 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"]
-	# Compare the server state for this client's own fixed slot against the
-	# entry tagged with the exact input sequence the server applied. Do not
-	# correct the body here yet: this result is intentionally inspection data
-	# for the later snap/blend pass, and (per
-	# _local_ship_prediction_state()) is not yet true prediction error.
+	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():
-			_last_local_prediction_comparison = _local_prediction_history.compare_authoritative(decoded["last_input_seq"], bodies[my_index])
+			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():
-			_slots[i].interpolator.add_sample(server_tick, bodies[i], reset_gen)
+			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
@@ -580,38 +699,29 @@ func _on_snapshot_received(decoded: Dictionary) -> void:
 		# — 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_interpolator.add_sample(server_tick, ball_state, reset_gen)
+		_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)
 
 
-# NOT a prediction yet, despite the name — the name is for task 4.3, which
-# is what will make it true. Pre-4.3 EVERY ship on the client, including this
-# client's own, is freeze = true / FREEZE_MODE_KINEMATIC (see _apply_match_config,
-# which sets that uniformly with no exception for _my_slot) and is moved only
-# by NetInterpolator transform writes derived from ALREADY-RECEIVED, past
-# server snapshots. Nothing locally simulates the local ship, and nothing ever
-# writes linear_velocity/angular_velocity onto it.
-#
-# So what this samples is "wherever the interpolator had smoothed the ship to
-# at packet-send time", NOT "where the action sampled this tick will put the
-# ship". The consequences for anyone reading the comparison output:
-#   - linear_velocity/angular_velocity here are NOT zero — a first pass at
-#     this comment claimed they were, but FREEZE_MODE_KINEMATIC derives a
-#     body's velocity from its own consecutive transform writes, so these
-#     fields genuinely reflect the interpolator's implied motion (confirmed
-#     live: non-zero, direction-correct velocities while driving). What they
-#     are NOT is the result of locally simulating the sampled action's
-#     thrust/rotation through the ship's own force formulas.
-#   - the resulting position_error / rotation_error_radians measure how far
-#     an interpolated PAST pose (and its implied velocity) sits from the
-#     later-arriving authoritative pose for that sequence. That is
-#     interpolation lag, not prediction error, and on a clean link it will
-#     read small and largely uninformative.
-#   - do not calibrate a snap-vs-blend threshold, or benchmark "prediction
-#     quality", against these numbers.
-# They only become genuine prediction error once task 4.3's net_ship_predictor.gd
-# unfreezes the local ship and steps it forward locally (multiplayer-todo.md
-# §4 / §7 tasks 4.3 and 4.5). The recording/matching plumbing is landed first,
-# on purpose, so 4.3 has a tested ring to build on.
+# 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
@@ -625,10 +735,97 @@ func _local_ship_prediction_state(ship: Ship, action: ShipAction) -> NetBodyStat
 	return state
 
 
-# Diagnostic accessor. Same caveat as _local_ship_prediction_state(): the
-# error fields are interpolation-vs-authoritative drift, not prediction error,
-# until task 4.3 lands.
-#
+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"/
@@ -685,7 +882,25 @@ func _estimated_tick(server_time_ms: float) -> float:
 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, INTERP_DELAY_MIN_MS, INTERP_DELAY_MAX_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"
@@ -713,19 +928,67 @@ func get_net_debug_stats() -> Dictionary:
 	# 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 is_instance_valid(_my_slot):
-		var latest := _my_slot.interpolator.latest()
-		if latest != null:
-			server_stalled = latest.stalled
+	if _last_local_prediction_comparison.get("authoritative_state", null) != null:
+		server_stalled = (_last_local_prediction_comparison["authoritative_state"] as NetBodyState).stalled
 	return {
 		"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) —
@@ -738,23 +1001,25 @@ func _physics_process(_delta: float) -> void:
 	if _owns_world_simulation():
 		_respawn_escaped_bodies()
 	if multiplayer.is_server():
-		# Once per tick, before the step (§3.2) — RLShipController reads
-		# .action lazily in the ship's own _integrate_forces, which for this
-		# tick already ran (physics step precedes _physics_process, §9
-		# gotcha 34), so this actually takes effect on the NEXT tick's step.
-		# That's the same one-tick input latency Phase 2 already had; this
-		# just replaces "read the newest packet naively" with a real
-		# sequence-tracked ring buffer that survives redundant/reordered/
-		# lost packets.
+		# _physics_process runs after this frame's _integrate_forces. Snapshot
+		# FIRST: the body state therefore still describes the sequence consumed
+		# on the prior callback. Sending after consume mislabeled that old state
+		# with NEXT tick's input sequence, making every client reconciliation
+		# comparison one action off and causing the Phase 4 snap cascade.
+		_broadcast_snapshot()
+		# The newly consumed action is deliberately installed for NEXT frame's
+		# integration. This preserves the existing one-tick server input delay
+		# while keeping snapshot.last_input_seq truthfully coupled to its body.
 		for slot in _slots:
 			slot.controller.action = slot.jitter_buffer.consume()
 		if _pending_reset_gen_bump and Engine.get_physics_frames() > _pending_reset_gen_bump_tick:
 			_reset_gen = (_reset_gen + 1) % 256
 			_pending_reset_gen_bump = false
-		_broadcast_snapshot()
 		return
 
 	_send_local_input()
+	_consume_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
@@ -767,12 +1032,36 @@ func _physics_process(_delta: float) -> void:
 	var server_time_est := NetworkManager.get_server_time_estimate_ms()
 	var collider_tick := _estimated_tick(server_time_est)
 	for slot in _slots:
-		if is_instance_valid(slot.ship) and slot.interpolator.has_samples():
+		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
@@ -795,13 +1084,26 @@ func _process(_delta: float) -> void:
 	if NetworkManager.rtt_ms < 0.0:
 		return
 	var server_time_est := NetworkManager.get_server_time_estimate_ms()
-	var visual_tick := _estimated_tick(server_time_est - _current_interp_delay_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 is_instance_valid(slot.ship) and slot.interpolator.has_samples():
-			_apply_ship_visual_state(slot.ship, slot.interpolator.sample_at(visual_tick))
+		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())
 
 
@@ -811,14 +1113,93 @@ func _apply_collider_state(body: RigidBody3D, state: NetBodyState) -> void:
 	body.global_transform = Transform3D(Basis(state.rotation), state.position)
 
 
-func _apply_ship_visual_state(ship: Ship, state: NetBodyState) -> void:
+func _apply_ship_visual_state(ship: Ship, state: NetBodyState, delta: float, slot: SlotInfo) -> void:
 	if state == null:
 		return
 	if is_instance_valid(ship.visual):
-		ship.visual.global_transform = Transform3D(Basis(state.rotation), state.position)
+		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())