feat(multiplayer): Phase 4 tasks 4.1/4.2 - local prediction history ring

Adds LocalPredictionHistory, a client-owned seq-tagged ring recording
predicted ship state per input sequence, plus wiring in
NetworkedMatch to record predictions on send and compare them against
authoritative snapshots on arrival. Ships stay frozen/interpolated
until 4.3 lands actual correction logic; this round only builds the
comparison machinery and its data.

Includes fixes from two review rounds: resync_required now
self-clears once acknowledgements catch back up (mirrors
InputJitterBuffer's stalled flag), NetBodyState gained a copy()
method to stop diagnostic accessors aliasing ring-owned state, and
corrected comments that had described the local ship as being
force-simulated pre-4.3 when it is still driven by interpolated
transform writes.

Game/scripts/local_prediction_history.gd

@@ -0,0 +1,173 @@
+class_name LocalPredictionHistory
+extends RefCounted
+
+const NetBodyState = preload("res://scripts/net_body_state.gd")
+
+# Client-owned local-ship prediction history (multiplayer-todo.md §4.3).
+# This is deliberately independent of NetworkedMatch and the scene tree so
+# sequence/ring behaviour can be tested from scripted traces. Each entry is
+# tagged with its full sequence number: an old value in a wrapped slot is
+# never accepted as a prediction for a newer sequence.
+#
+# Acknowledge and record are separate producer/consumer clocks. The input
+# sender can continue producing while snapshots stop arriving, so record()
+# explicitly marks overflow once more than RING_SIZE unacknowledged sequence
+# positions exist. It still retains the newest representable window, but
+# callers can see that an authoritative resync is required instead of
+# mistaking a wrapped overwrite for a valid comparison.
+#
+# resync_required is a live condition, NOT a latch: compare_authoritative()
+# clears it again once acknowledgements have genuinely caught back up (see
+# that method). This mirrors input_jitter_buffer.gd's `stalled`, which
+# likewise drops back to false the moment a normal tick is consumed again.
+# A latched flag would mean one transient ~2s stall anywhere in a match
+# permanently pinned every later tick into "needs a hard resync", which is
+# exactly the behaviour soft correction exists to avoid — and it would also
+# cap overflow_count at 1 forever, since a second episode could never
+# observe the flag going false again.
+#
+# Two things a "matched" result does NOT guarantee, flagged for whoever
+# builds task 4.3's actual correction logic on top of this:
+#
+# 1. A "matched" result can still be reporting stale data. The slot-tag
+#    equality check in get_prediction() guarantees a match's payload
+#    genuinely belongs to the queried seq (never wrong-seq data mislabeled
+#    as right), but nothing in the "matched" status itself says HOW OLD
+#    that entry is. Under sparse recording (record() is not called with
+#    strictly consecutive seqs — see the record() comment below), an entry
+#    from well over RING_SIZE ticks ago can still report "matched" for a
+#    query landing on its untouched residue. resync_required correctly
+#    stays true in that case (the span guard below is exact), but the
+#    comparison payload itself carries no matched_stale/age distinction. A
+#    caller wanting to reject "matched but ancient" needs to separately
+#    check newest_recorded_seq - seq itself.
+#
+# 2. record() can be called twice for the same seq with a DIFFERENT action,
+#    when input_lead_controller's release path resends a duplicated seq
+#    (delta == 0) — the later call silently overwrites the ring slot, so
+#    the stored action becomes whichever of the two calls happened last.
+#    This matches what the WIRE ends up sending for that seq (the resend
+#    replaces the redundancy history's front entry — see
+#    networked_match.gd's _send_local_input), but if the SERVER had
+#    already consumed the seq from the first packet before the resend
+#    arrived, the server's applied action and this ring's stored action for
+#    that same seq can disagree. Narrow (release only fires after 120 ticks
+#    of sustained surplus depth, when the server is least likely to be
+#    right on the edge of consuming that exact seq) but real; a future
+#    replay-based catch-up (task 4.5) built on this history should not
+#    assume the stored action is provably what the server actually applied.
+
+const RING_SIZE := 128
+
+var _ring_seq: PackedInt32Array = PackedInt32Array()
+var _ring_entry: Array = []
+var _has_recorded := false
+
+var newest_recorded_seq := -1
+var last_acknowledged_seq := 0
+var overflow_count := 0
+var resync_required := false
+
+
+func _init() -> void:
+	_ring_seq.resize(RING_SIZE)
+	_ring_entry.resize(RING_SIZE)
+	for i in RING_SIZE:
+		_ring_seq[i] = -1
+
+
+# Stores a private copy of both action and state. Returns true when this
+# record crossed the unacknowledged-capacity boundary; the caller does not
+# need that return today, but it makes the eviction event observable rather
+# than silent when reconciliation starts applying corrections in Phase 4.3.
+func record(seq: int, action: ShipAction, state: NetBodyState) -> bool:
+	var overflowed_now := false
+	if not _has_recorded or seq > newest_recorded_seq:
+		if seq - last_acknowledged_seq > RING_SIZE:
+			# Only the LEADING edge of an episode counts: resync_required is
+			# still true for every subsequent tick of the same stall, and
+			# counting those would report one outage as hundreds. Because
+			# compare_authoritative() can now clear the flag, a genuinely
+			# separate later episode does increment this again.
+			overflowed_now = not resync_required
+			resync_required = true
+			if overflowed_now:
+				overflow_count += 1
+		newest_recorded_seq = seq
+		_has_recorded = true
+	var idx := posmod(seq, RING_SIZE)
+	_ring_seq[idx] = seq
+	_ring_entry[idx] = {
+		"action": action.copy(),
+		"state": state.copy(),
+	}
+	return overflowed_now
+
+
+# Returns independent copies so diagnostic/reconciliation consumers cannot
+# mutate a retained prediction by accident.
+func get_prediction(seq: int) -> Dictionary:
+	var idx := posmod(seq, RING_SIZE)
+	if _ring_seq[idx] != seq:
+		return {}
+	var entry: Dictionary = _ring_entry[idx]
+	return {
+		"seq": seq,
+		"action": (entry["action"] as ShipAction).copy(),
+		"state": (entry["state"] as NetBodyState).copy(),
+	}
+
+
+# Produces comparison data only. Applying a snap, teleport, velocity delta,
+# or visual offset belongs to later Phase 4 tasks.
+func compare_authoritative(seq: int, authoritative: NetBodyState) -> Dictionary:
+	if seq > last_acknowledged_seq:
+		last_acknowledged_seq = seq
+	var prediction := get_prediction(seq)
+	if prediction.is_empty():
+		return {
+			"status": _missing_status(seq),
+			"seq": seq,
+			"authoritative_state": authoritative.copy(),
+		}
+
+	# A successful match is the only evidence that the acknowledgement clock
+	# has genuinely caught back up, so it is the only thing allowed to clear
+	# resync_required — a "missing_evicted"/"missing_not_recorded" result
+	# proves the opposite, and must leave the flag alone.
+	#
+	# The extra span check is not redundant. record() is not guaranteed to be
+	# called with consecutive sequences: input_lead_controller.update() can
+	# return 0 or up to 1+3, so the client's seq can skip forward, leaving a
+	# ring slot holding a tag OLDER than newest_recorded_seq - RING_SIZE
+	# (its residue was simply never rewritten). get_prediction() would still
+	# report that as "matched", so matching alone does not imply the
+	# outstanding window is back within capacity. Gate on the exact inverse
+	# of record()'s own trip inequality instead, which holds regardless of
+	# how sparsely sequences were recorded.
+	if newest_recorded_seq - last_acknowledged_seq <= RING_SIZE:
+		resync_required = false
+
+	var predicted_state: NetBodyState = prediction["state"]
+	var position_error := authoritative.position - predicted_state.position
+	var rotation_error_radians := predicted_state.rotation.angle_to(authoritative.rotation)
+	return {
+		"status": "matched",
+		"seq": seq,
+		"action": prediction["action"],
+		"predicted_state": predicted_state,
+		"authoritative_state": authoritative.copy(),
+		"position_error": position_error,
+		"position_error_magnitude": position_error.length(),
+		"rotation_error_radians": rotation_error_radians,
+		"rotation_error_degrees": rad_to_deg(rotation_error_radians),
+		"linear_velocity_error": authoritative.linear_velocity - predicted_state.linear_velocity,
+		"angular_velocity_error": authoritative.angular_velocity - predicted_state.angular_velocity,
+	}
+
+
+func _missing_status(seq: int) -> String:
+	if _has_recorded and seq <= newest_recorded_seq - RING_SIZE:
+		return "missing_evicted"
+	return "missing_not_recorded"
+

Game/scripts/net_body_state.gd

@@ -21,3 +21,27 @@ var turbo := false
 var thrust_z := 0.0          # -1..1; re-quantised to a 3-bit bin on the wire
 var stalled := false
 var avel_range := 4.0        # NetCodec.SHIP_AVEL_RANGE; set to BALL_AVEL_RANGE for the ball
+
+# Self-referential preload, not get_script().new() — this file deliberately
+# has no class_name (same cache-timing reason as test_case.gd and other
+# path-`extends`d files in this project), and get_script().new() throws
+# "Nonexistent function 'new' in base 'GDScript'" from within the script's
+# own body in this Godot version.
+const _NetBodyState = preload("res://scripts/net_body_state.gd")
+
+
+# Same contract as ShipAction.copy() (see its own comment): a distinct
+# instance with equal fields, for callers that hold onto a state past the
+# tick/comparison it was returned in.
+func copy() -> RefCounted:
+	var c := _NetBodyState.new()
+	c.position = position
+	c.rotation = rotation
+	c.linear_velocity = linear_velocity
+	c.angular_velocity = angular_velocity
+	c.frozen = frozen
+	c.turbo = turbo
+	c.thrust_z = thrust_z
+	c.stalled = stalled
+	c.avel_range = avel_range
+	return c

Game/scripts/networked_match.gd

@@ -8,6 +8,9 @@ extends GameMode
 # 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.
 #
 # 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
@@ -29,6 +32,7 @@ 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 LocalPredictionHistory = preload("res://scripts/local_prediction_history.gd")
 const HUD_SCENE = preload("res://scenes/HUD.tscn")
 
 # Minimum plausible interpolation delay even on a same-machine/LAN link —
@@ -101,6 +105,20 @@ var _input_seq := 0                       # client only
 # 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 _last_local_prediction_comparison: Dictionary = {}
 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
@@ -478,6 +496,12 @@ func _send_local_input() -> void:
 	# 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
@@ -535,6 +559,15 @@ 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 _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])
 	_update_tick_bias(server_tick)
 	for i in _slots.size():
 		if i < bodies.size():
@@ -550,6 +583,69 @@ func _on_snapshot_received(decoded: Dictionary) -> void:
 		_ball_interpolator.add_sample(server_tick, ball_state, reset_gen)
 
 
+# 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.
+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
+
+
+# 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.
+#
+# 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

Game/tests/cases/test_local_prediction_history.gd

@@ -0,0 +1,161 @@
+extends "res://tests/test_case.gd"
+
+const LocalPredictionHistory = preload("res://scripts/local_prediction_history.gd")
+const ShipAction = preload("res://scripts/ship_action.gd")
+const NetBodyState = preload("res://scripts/net_body_state.gd")
+
+
+func _action(value: float) -> ShipAction:
+	var action := ShipAction.new()
+	action.thrust = Vector3(0.0, 0.0, value)
+	action.rotation = Vector3(value, 0.0, 0.0)
+	action.turbo = value > 0.5
+	return action
+
+
+func _state(value: float) -> NetBodyState:
+	var state := NetBodyState.new()
+	state.position = Vector3(value, value + 1.0, value + 2.0)
+	state.rotation = Quaternion(Vector3.UP, value * 0.1)
+	state.linear_velocity = Vector3(value * 2.0, 0.0, 0.0)
+	state.angular_velocity = Vector3(0.0, value * 3.0, 0.0)
+	return state
+
+
+func test_matches_authoritative_state_at_the_same_sequence() -> void:
+	var history := LocalPredictionHistory.new()
+	var predicted := _state(2.0)
+	history.record(17, _action(0.4), predicted)
+	var authoritative := _state(2.0)
+	authoritative.position += Vector3(1.0, -2.0, 0.5)
+	authoritative.linear_velocity += Vector3(0.25, 0.0, 0.0)
+
+	var result := history.compare_authoritative(17, authoritative)
+	assert_eq(result["status"], "matched", "same tagged sequence matches")
+	assert_almost_eq((result["position_error"] as Vector3).x, 1.0, 0.0001, "position delta x")
+	assert_almost_eq((result["position_error"] as Vector3).y, -2.0, 0.0001, "position delta y")
+	assert_almost_eq(result["position_error_magnitude"], 2.2912878, 0.0001, "position error magnitude")
+	assert_almost_eq((result["linear_velocity_error"] as Vector3).x, 0.25, 0.0001, "linear velocity delta")
+	assert_eq(history.last_acknowledged_seq, 17, "comparison advances acknowledgement epoch")
+
+
+func test_record_and_lookup_do_not_alias_action_or_state() -> void:
+	var history := LocalPredictionHistory.new()
+	var action := _action(0.25)
+	var state := _state(3.0)
+	history.record(4, action, state)
+	action.thrust.z = 9.0
+	state.position.x = 99.0
+
+	var prediction := history.get_prediction(4)
+	assert_almost_eq((prediction["action"] as ShipAction).thrust.z, 0.25, 0.0001, "stored action is copied")
+	assert_almost_eq((prediction["state"] as NetBodyState).position.x, 3.0, 0.0001, "stored state is copied")
+	(prediction["action"] as ShipAction).thrust.z = -5.0
+	(prediction["state"] as NetBodyState).position.x = -5.0
+	var second_lookup := history.get_prediction(4)
+	assert_almost_eq((second_lookup["action"] as ShipAction).thrust.z, 0.25, 0.0001, "lookup action cannot mutate ring")
+	assert_almost_eq((second_lookup["state"] as NetBodyState).position.x, 3.0, 0.0001, "lookup state cannot mutate ring")
+
+
+func test_slot_tags_reject_wrapped_stale_predictions() -> void:
+	var history := LocalPredictionHistory.new()
+	history.record(1, _action(0.1), _state(1.0))
+	history.record(1 + LocalPredictionHistory.RING_SIZE, _action(0.8), _state(8.0))
+
+	assert_true(history.get_prediction(1).is_empty(), "old same-index entry is not mistaken for current data")
+	assert_eq(history.compare_authoritative(1, _state(1.0))["status"], "missing_evicted", "stale acknowledgement is explicitly evicted")
+	assert_eq(history.compare_authoritative(1 + LocalPredictionHistory.RING_SIZE, _state(8.0))["status"], "matched", "current same-index entry still matches")
+
+
+func test_unacknowledged_overflow_is_explicit_and_keeps_newest_window() -> void:
+	var history := LocalPredictionHistory.new()
+	for seq in range(1, LocalPredictionHistory.RING_SIZE + 2):
+		history.record(seq, _action(float(seq)), _state(float(seq)))
+
+	assert_true(history.resync_required, "producer outrunning acknowledgements sets explicit resync state")
+	assert_eq(history.overflow_count, 1, "one continuous overflow episode is counted once")
+	assert_eq(history.compare_authoritative(1, _state(1.0))["status"], "missing_evicted", "oldest unacknowledged prediction was explicitly evicted")
+	var newest := history.compare_authoritative(LocalPredictionHistory.RING_SIZE + 1, _state(float(LocalPredictionHistory.RING_SIZE + 1)))
+	assert_eq(newest["status"], "matched", "newest prediction remains usable after overflow")
+
+
+# Records seq..seq+n until the unacknowledged window trips overflow, and
+# returns the newest sequence recorded. Mirrors the real producer, which
+# calls record() once per client physics tick with no acknowledgements
+# arriving during a stall.
+func _stall_until_overflow(history: LocalPredictionHistory, from_seq: int) -> int:
+	# Hard-bounded rather than `while not history.resync_required`: assert_true
+	# only records a failure, it cannot abort, so an unbounded loop here would
+	# hang the whole headless runner instead of failing if the trip condition
+	# ever regressed.
+	var newest := from_seq - 1
+	for seq in range(from_seq, from_seq + 4 * LocalPredictionHistory.RING_SIZE):
+		if history.resync_required:
+			break
+		history.record(seq, _action(float(seq)), _state(float(seq)))
+		newest = seq
+	assert_true(history.resync_required, "overflow must trip within a bounded number of ticks")
+	return newest
+
+
+func test_resync_required_clears_once_acknowledgements_catch_back_up() -> void:
+	var history := LocalPredictionHistory.new()
+	var newest := _stall_until_overflow(history, 1)
+	assert_true(history.resync_required, "transient stall trips explicit resync state")
+	assert_eq(history.overflow_count, 1, "first episode counted once")
+
+	# An acknowledgement that lands on an already-evicted sequence is not
+	# evidence of recovery and must leave the flag alone.
+	assert_eq(history.compare_authoritative(1, _state(1.0))["status"], "missing_evicted", "oldest sequence is gone")
+	assert_true(history.resync_required, "an evicted comparison does not count as catching up")
+
+	# A real, current, matched acknowledgement does.
+	var recovered := history.compare_authoritative(newest, _state(float(newest)))
+	assert_eq(recovered["status"], "matched", "newest prediction still matches")
+	assert_true(not history.resync_required, "resync state clears once acknowledgements are flowing again")
+	assert_eq(history.overflow_count, 1, "recovery does not retroactively change the episode count")
+
+
+func test_second_distinct_overflow_episode_is_counted_separately() -> void:
+	var history := LocalPredictionHistory.new()
+	var first_newest := _stall_until_overflow(history, 1)
+	history.compare_authoritative(first_newest, _state(float(first_newest)))
+	assert_true(not history.resync_required, "first episode recovered")
+	assert_eq(history.overflow_count, 1, "first episode counted")
+
+	var second_newest := _stall_until_overflow(history, first_newest + 1)
+	assert_true(history.resync_required, "a later separate stall trips resync state again")
+	assert_eq(history.overflow_count, 2, "a second distinct episode is counted separately, not capped at one")
+
+	history.compare_authoritative(second_newest, _state(float(second_newest)))
+	assert_true(not history.resync_required, "second episode also recovers")
+	assert_eq(history.overflow_count, 2, "episode count is cumulative across recoveries")
+
+
+func test_continuing_stall_does_not_recount_the_same_episode() -> void:
+	var history := LocalPredictionHistory.new()
+	var newest := _stall_until_overflow(history, 1)
+	for seq in range(newest + 1, newest + 1 + LocalPredictionHistory.RING_SIZE):
+		history.record(seq, _action(float(seq)), _state(float(seq)))
+	assert_true(history.resync_required, "the stall is still in progress")
+	assert_eq(history.overflow_count, 1, "one continuous outage stays one episode however long it lasts")
+
+
+# Sequences are not guaranteed consecutive: InputLeadController.update() can
+# return 0 or up to 1+3, so the client's seq can skip forward and leave a ring
+# slot holding a tag older than newest_recorded_seq - RING_SIZE. get_prediction()
+# reports such a stale-but-untouched slot as "matched", so matching alone must
+# not be enough to clear resync_required while the real backlog is still huge.
+func test_stale_matched_comparison_does_not_clear_a_live_backlog() -> void:
+	var history := LocalPredictionHistory.new()
+	history.record(1, _action(1.0), _state(1.0))
+	var far := LocalPredictionHistory.RING_SIZE + 2   # skips index 1, so seq 1's slot survives
+	history.record(far, _action(float(far)), _state(float(far)))
+	assert_true(history.resync_required, "the skip outran the acknowledgement window")
+
+	var stale := history.compare_authoritative(1, _state(1.0))
+	assert_eq(stale["status"], "matched", "seq 1's ring slot was never rewritten")
+	assert_true(history.resync_required, "a stale match while the backlog is still oversized must not clear resync state")
+
+	history.compare_authoritative(far, _state(float(far)))
+	assert_true(not history.resync_required, "acknowledging the newest sequence does clear it")