feat(multiplayer): Phase 3 tasks 3.1/3.2/3.5 - input redundancy + server jitter buffer

Client now sends the last 4 ticks' actions per packet (newest-first,
already-supported by net_codec's wire format from Phase 1) instead of a
single action with no redundancy. Server gains a real per-slot ring
buffer (new InputJitterBuffer class, scripts/input_jitter_buffer.gd) that
consumes exactly one sequence number per physics tick: repeats the last
action on a starve, zeroes only after a sustained 500ms stall, and
reports real input_buffer_depth/last_input_seq/echo_client_send_ms in
every snapshot instead of the hardcoded zeros Phase 2 shipped with.

InputJitterBuffer is a standalone, scene-free RefCounted (same pattern as
net_codec.gd/net_interpolator.gd) specifically so it's unit-testable
against scripted arrival traces (tests/cases/test_input_jitter_buffer.gd):
sequential consumption, redundancy surviving a 3-packet burst loss (3.1's
own acceptance criterion), starvation repeat-then-zero timing, stale/
reordered packet handling, buffered-depth reporting, and ring-wraparound
slot-tagging safety.

One real bug found wiring this into a live match: the server's ring
buffer started counting its own "expected sequence" from 0 the instant a
player's slot was created - well before that player's first real packet
could possibly have arrived (connection handshake, arena/ship spawn all
take real time first). Since both sides only ever advance monotonically
with no resync mechanism, that gap between the server's arbitrary local
counter and the client's actual from-1 sequence numbers never closed,
so the ship simply never received the client's input (0m movement in a
two-process test). Fixed by seeding the buffer's expected-sequence
counter from the client's own numbering on first real ingest, rather
than assuming a shared from-zero baseline.

Verified with real two-process runs: clean baseline movement restored,
zero starvation observed under 25% random simulated input loss (well
above what redundancy-4 needs to fully absorb), and correct starve-then-
stall behaviour confirmed under 100% loss as a sanity check that the
mechanism isn't a silent no-op. Full regression suite, including the
net-sim-latency milestone gate, re-run clean.

Game/scripts/input_jitter_buffer.gd

@@ -0,0 +1,113 @@
+class_name InputJitterBuffer
+extends RefCounted
+
+# Per-player server-side input state (multiplayer-todo.md §3, task 3.2).
+# Deliberately a standalone RefCounted with no scene/RPC dependency — same
+# reason net_codec.gd and net_interpolator.gd are pure classes — so task
+# 3.5's unit tests can drive it with scripted arrival traces with no live
+# match. NetworkedMatch owns one instance per connected slot and is the only
+# thing that talks to the network layer; this class only knows about
+# sequence numbers and ShipActions.
+#
+# Ring is fixed-size and slot-tagged (§3.1 step 5's "a client can never make
+# the server allocate"): ingest() writes seq % RING_SIZE regardless of how
+# large or malicious seq is, and consume() only ever trusts a slot whose
+# stored seq exactly matches the one it expects — a stale or wrapped-around
+# entry is indistinguishable from an empty one. Range/rate validation of seq
+# against the current server tick is the CALLER's job (task 3.4), not this
+# class's, since only the caller knows the current server tick.
+
+const RING_SIZE := 32
+# 500ms at 60Hz (multiplayer-todo.md §3.2's own numbers) — a duration, not a
+# tick-rate-derived constant, so left as a literal rather than pulling in
+# SimConstants for one number.
+const STARVE_ZERO_TICKS := 30
+
+var last_applied_seq := -1   # -1: consume() has never been called yet
+var last_action := ShipAction.new()
+var starved_ticks := 0
+var stalled := false
+
+var _ring_action: Array = []
+var _ring_seq: PackedInt32Array = PackedInt32Array()
+# True once ingest() has ever been called for real. Consumption is a no-op
+# (no starvation counted, no advancement) until then — see ingest()'s own
+# comment for why an un-seeded buffer would otherwise never converge with
+# what the client is actually sending.
+var _seeded := false
+
+
+func _init() -> void:
+	_ring_action.resize(RING_SIZE)
+	_ring_seq.resize(RING_SIZE)
+	for i in RING_SIZE:
+		_ring_seq[i] = -1
+
+
+# newest_seq/actions match NetCodec.unpack_input's own "seq"/"actions"
+# fields directly: actions[i] is the action for sequence (newest_seq - i),
+# newest-first. Already-consumed or stale entries are silently discarded
+# (§3.1 step 5) — this is what makes redundant re-delivery of an already-
+# applied tick harmless.
+func ingest(newest_seq: int, actions: Array) -> void:
+	if not _seeded:
+		# The server starts calling consume() every tick the instant this
+		# slot exists — well before this player's first packet has had time
+		# to arrive (connection handshake, arena/ship spawn, first
+		# _physics_process tick on the client all take real time first). An
+		# un-seeded last_applied_seq of -1 would have consume() "expecting"
+		# sequence 0, 1, 2, ... via pure starvation the whole time, racing
+		# arbitrarily far ahead of whatever the client's own from-1
+		# numbering has actually reached by the time real packets show up —
+		# and since both sides only ever advance monotonically with no
+		# resync mechanism, that gap would never close. Seed to align
+		# "expected" with reality the moment real data first exists.
+		last_applied_seq = newest_seq - actions.size()
+		_seeded = true
+	for i in actions.size():
+		var seq: int = newest_seq - i
+		if seq <= last_applied_seq:
+			continue
+		var idx := seq % RING_SIZE
+		_ring_seq[idx] = seq
+		_ring_action[idx] = actions[i]
+
+
+# Contiguous run of not-yet-applied entries starting right after
+# last_applied_seq — reported as input_buffer_depth in every snapshot
+# (§3.3) and consumed client-side by the input_lead control loop (task 3.3).
+func depth() -> int:
+	if not _seeded or last_applied_seq < 0:
+		return 0
+	var d := 0
+	var seq := last_applied_seq + 1
+	while d < RING_SIZE and _ring_seq[seq % RING_SIZE] == seq:
+		d += 1
+		seq += 1
+	return d
+
+
+# Called once per server physics tick, before the step (§3.2). A no-op
+# (returns the zero-initialized last_action, no starvation counted) until
+# this player's first real packet has ever arrived — see ingest()'s comment.
+func consume() -> ShipAction:
+	if not _seeded:
+		return last_action
+	var expected := last_applied_seq + 1
+	var idx := expected % RING_SIZE
+	if _ring_seq[idx] == expected:
+		last_action = _ring_action[idx]
+		starved_ticks = 0
+		stalled = false
+	else:
+		# Repeat-last, not zero: inputs are heavily autocorrelated at 60Hz,
+		# and the client already predicted with the real input either way,
+		# so repeating minimises expected divergence (§3.2). Only zero after
+		# a sustained stall, so a disconnecting player's ship doesn't fly
+		# into a wall at full throttle forever.
+		starved_ticks += 1
+		if starved_ticks > STARVE_ZERO_TICKS:
+			last_action = ShipAction.new()
+			stalled = true
+	last_applied_seq = expected
+	return last_action

Game/scripts/networked_match.gd

@@ -27,6 +27,7 @@ signal score_changed(score: Dictionary)
 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 HUD_SCENE = preload("res://scenes/HUD.tscn")
 
 # Minimum plausible interpolation delay even on a same-machine/LAN link —
@@ -72,6 +73,8 @@ class SlotInfo:
 	var spawn_index: int
 	var ship: Ship
 	var controller: RLShipController      # server only
+	var jitter_buffer := InputJitterBuffer.new()   # server only (§3.2)
+	var last_client_send_ms := 0          # server only: echoed back per-peer next snapshot (§2.4)
 	var interpolator := NetInterpolator.new()   # client only
 
 
@@ -80,6 +83,11 @@ var _my_slot: SlotInfo = null             # client only
 var _ball_interpolator := NetInterpolator.new()   # client only
 var _local_input_sampler := PlayerShipController.new()   # client only: reads local input each tick to forward; never added to a Ship, never in the tree — get_action() only touches the global Input singleton
 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 _last_received_snapshot_tick := 0     # client only: echoed back as ack_snapshot_tick
 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
@@ -196,11 +204,8 @@ func _start_server() -> void:
 func _on_input_received(peer_id: int, decoded: Dictionary) -> void:
 	for slot in _slots:
 		if slot.peer_id == peer_id:
-			var actions: Array = decoded["actions"]
-			# Newest-first; no redundancy handling yet (task 3.x) — just take
-			# the newest one every time a packet arrives.
-			if not actions.is_empty():
-				slot.controller.action = actions[0]
+			slot.jitter_buffer.ingest(decoded["seq"], decoded["actions"])
+			slot.last_client_send_ms = decoded["client_send_ms"]
 			return
 
 
@@ -231,12 +236,15 @@ func _broadcast_snapshot() -> void:
 	if is_instance_valid(ball):
 		bodies.append(_ball_to_net_body_state(ball))
 	var segment := NetCodec.pack_snapshot_body_segment(server_tick, 0, _reset_gen, bodies)
-	# Per-client header fields (last_input_seq/input_buffer_depth/echo) aren't
-	# tracked yet — that's the jitter-buffer work in Phase 3 (tasks 3.1-3.2).
 	# 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.
+	# 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
@@ -246,7 +254,9 @@ func _broadcast_snapshot() -> void:
 	var connected_peers := multiplayer.get_peers()
 	for slot in _slots:
 		if connected_peers.has(slot.peer_id):
-			MatchSim.send_snapshot(slot.peer_id, NetCodec.pack_snapshot(0, 0, 0, segment))
+			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)
+			MatchSim.send_snapshot(slot.peer_id, bytes)
 
 
 func _ship_to_net_body_state(ship: Ship) -> NetBodyState:
@@ -341,7 +351,16 @@ func _send_local_input() -> void:
 		return   # match_config hasn't arrived yet
 	var action := _local_input_sampler.get_action().copy()
 	_input_seq += 1
-	var bytes := NetCodec.pack_input(_input_seq, 0, Time.get_ticks_msec(), [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.
+	_input_history.push_front(action)
+	if _input_history.size() > NetCodec.MAX_REDUNDANCY:
+		_input_history.resize(NetCodec.MAX_REDUNDANCY)
+	var bytes := NetCodec.pack_input(_input_seq, _last_received_snapshot_tick, Time.get_ticks_msec(), _input_history)
 	MatchSim.send_input(bytes)
 
 
@@ -349,6 +368,7 @@ 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"]
+	_last_received_snapshot_tick = server_tick
 	_update_tick_bias(server_tick)
 	for i in _slots.size():
 		if i < bodies.size():
@@ -418,6 +438,16 @@ 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.
+		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

Game/tests/cases/test_input_jitter_buffer.gd

@@ -0,0 +1,112 @@
+extends "res://tests/test_case.gd"
+
+const InputJitterBuffer = preload("res://scripts/input_jitter_buffer.gd")
+const ShipAction = preload("res://scripts/ship_action.gd")
+
+
+func _action(thrust_z: float) -> ShipAction:
+	var a := ShipAction.new()
+	a.thrust = Vector3(0.0, 0.0, thrust_z)
+	return a
+
+
+func test_sequential_ingest_and_consume() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.1)])
+	assert_almost_eq(buf.consume().thrust.z, 0.1, 0.0001, "tick 0")
+	buf.ingest(1, [_action(0.2)])
+	assert_almost_eq(buf.consume().thrust.z, 0.2, 0.0001, "tick 1")
+	assert_eq(buf.last_applied_seq, 1, "last_applied_seq after 2 ticks")
+	assert_eq(buf.starved_ticks, 0, "no starvation on a clean sequential stream")
+
+
+# §3.1's own acceptance criterion: "a 3-packet burst loss produces no
+# starvation." Redundancy-4 means a single surviving packet after 3 losses
+# still carries all 4 of the most recent ticks' actions.
+func test_redundancy_survives_3_packet_burst_loss() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.0)])
+	assert_almost_eq(buf.consume().thrust.z, 0.0, 0.0001, "seq 0")
+
+	# Packets for seq 1, 2, 3 are "lost" (never ingested individually) — only
+	# the seq=4 packet, carrying seq 4,3,2,1 (newest-first, redundancy 4),
+	# actually arrives.
+	buf.ingest(4, [_action(0.4), _action(0.3), _action(0.2), _action(0.1)])
+
+	assert_almost_eq(buf.consume().thrust.z, 0.1, 0.0001, "seq 1 recovered from redundancy")
+	assert_eq(buf.starved_ticks, 0, "seq 1 was not a starve")
+	assert_almost_eq(buf.consume().thrust.z, 0.2, 0.0001, "seq 2 recovered from redundancy")
+	assert_almost_eq(buf.consume().thrust.z, 0.3, 0.0001, "seq 3 recovered from redundancy")
+	assert_almost_eq(buf.consume().thrust.z, 0.4, 0.0001, "seq 4 recovered from redundancy")
+	assert_eq(buf.starved_ticks, 0, "no starvation anywhere across the whole burst-loss window")
+
+
+func test_starvation_repeats_last_action_then_zeroes_after_500ms() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.7)])
+	buf.consume()
+
+	# Nothing else ever arrives — every consume() from here on starves.
+	for i in InputJitterBuffer.STARVE_ZERO_TICKS:
+		var a := buf.consume()
+		assert_almost_eq(a.thrust.z, 0.7, 0.0001, "repeat-last during starve, tick %d" % i)
+		assert_true(not buf.stalled, "not yet stalled at tick %d" % i)
+
+	# One more tick past STARVE_ZERO_TICKS (30 = 500ms at 60Hz) crosses the
+	# "> 30" threshold and zeroes rather than keeps repeating forever.
+	var stalled_action := buf.consume()
+	assert_almost_eq(stalled_action.thrust.z, 0.0, 0.0001, "zeroed after sustained stall")
+	assert_true(buf.stalled, "stalled flag set after 500ms of starvation")
+
+
+func test_late_stale_packet_is_discarded_harmlessly() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(5, [_action(0.5)])
+	buf.consume()   # seeded to 4 by ingest() (newest_seq - 1 action), one consume reaches 5
+	assert_eq(buf.last_applied_seq, 5, "consumed up through seq 5")
+
+	# A reordered/duplicated packet for an already-consumed seq arrives late.
+	buf.ingest(3, [_action(0.3)])
+	assert_eq(buf.depth(), 0, "a stale packet below last_applied_seq must not appear as buffered depth")
+
+	buf.ingest(6, [_action(0.6)])
+	assert_almost_eq(buf.consume().thrust.z, 0.6, 0.0001, "the genuinely-next seq still consumes correctly")
+
+
+func test_depth_reports_contiguous_buffered_run() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.0)])
+	buf.consume()   # last_applied_seq = 0
+
+	assert_eq(buf.depth(), 0, "nothing buffered ahead yet")
+	buf.ingest(3, [_action(0.3), _action(0.2), _action(0.1)])
+	assert_eq(buf.depth(), 3, "seq 1,2,3 all buffered and contiguous with last_applied_seq")
+
+	# A gap (seq 5 arrives but seq 4 never does) caps depth at the gap, not
+	# the highest seq seen.
+	buf.ingest(5, [_action(0.5)])
+	assert_eq(buf.depth(), 3, "seq 5 sits past a gap at seq 4, so it doesn't extend the contiguous run")
+
+
+func test_ring_wraparound_does_not_confuse_a_stale_slot_with_a_fresh_one() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.0)])
+	buf.consume()
+
+	# Advance last_applied_seq well past one full lap of the ring (32
+	# entries) purely via starvation, with nothing re-ingested — every
+	# ring slot's stored seq is now far behind "expected" at each step, so
+	# none of them should ever be misread as valid.
+	for i in InputJitterBuffer.RING_SIZE * 2:
+		buf.consume()
+	assert_eq(buf.last_applied_seq, InputJitterBuffer.RING_SIZE * 2, "advanced purely by starvation")
+	assert_true(buf.stalled, "long starvation run ends stalled")
+
+	# Now a fresh packet lands at the seq the ring slot for "expected" was
+	# LAST used for, one full lap ago — if slot-tagging didn't work, this
+	# would be misread as already-fresh data from the stale write.
+	var expected := buf.last_applied_seq + 1
+	buf.ingest(expected, [_action(0.9)])
+	var a := buf.consume()
+	assert_almost_eq(a.thrust.z, 0.9, 0.0001, "correctly reads the fresh same-slot-index seq, not a stale wraparound ghost")
+	assert_eq(buf.starved_ticks, 0, "starvation clears once fresh data resumes")