feat(multiplayer): Phase 3 task 3.3 - client-owned input_lead control loop

New InputLeadController (scripts/input_lead_controller.gd, standalone and
unit-tested like input_jitter_buffer.gd): fast attack (+3 immediately,
debounced to once per 30 ticks) on any server-reported starve, slow
release (-1 per 60 ticks, gated behind a one-time 2s clean-surplus bar)
otherwise, clamped [1, 12]. Deliberately the only thing that adapts
buffer depth - the server (InputJitterBuffer) stays a pure reporter, per
§3.3's explicit warning that multiple control loops acting on one plant
(buffer occupancy) oscillate and present as unattributable sticky
controls.

Wired into the client's per-tick input send: a lead change is realized as
extra distance between the client's outgoing sequence numbers and what
the server has consumed - an attack skips extra sequence numbers, a
release duplicates the current one (sent again, unincremented). The
server's ring buffer needs no special handling for either: a skipped seq
is an ordinary drop, a duplicated one is a same-seq resend already
discarded by the existing "already consumed" check.

Verified with real two-process runs: on a clean LAN, one early attack
(a momentary hiccup during connection setup) recovers via two releases
within the test's own ~4s window, settling back near minimum. Under
sustained 30% simulated loss, lead climbs to 7 via repeated attacks and
never releases while genuine loss continues - confirming the debounce,
attack, and release gates all fire on real conditions, not just in
isolated unit tests. Full regression suite, including the net-sim-latency
milestone gate, re-run clean.

Game/scripts/input_lead_controller.gd

@@ -0,0 +1,80 @@
+class_name InputLeadController
+extends RefCounted
+
+# Client-owned input_lead control loop (multiplayer-todo.md §3.3, task 3.3).
+# Standalone RefCounted, same reason as input_jitter_buffer.gd — scene-free
+# so it's directly unit-testable against scripted depth traces.
+#
+# §3.3's own rationale for why this is the CLIENT's job alone, not shared
+# with any server-side adaptation: three control loops acting on one plant
+# (buffer occupancy) with different time constants is a textbook
+# oscillation, and on a jittery link it presents to the player as
+# intermittent sticky controls that are nearly impossible to attribute.
+# The server (InputJitterBuffer, §3.2) only ever reports input_buffer_depth
+# — it does nothing adaptive with it.
+#
+# "Lead" is realized concretely as extra distance between this client's own
+# outgoing sequence numbers and what the server has actually consumed:
+# skipping a sequence number (jumping the client's own seq counter by more
+# than 1 for one tick) buys the server one more tick of buffered depth
+# before it would starve; duplicating one (not incrementing the seq counter
+# for one tick — the same seq gets sent again) narrows that margin by one
+# tick of latency. The server's own ring buffer doesn't need to know this
+# happened: a skipped seq just means "the redundant copies of it never
+# existed, it's an ordinary drop" (already handled), and a duplicated seq
+# is a same-seq resend, already discarded harmlessly once consumed
+# (InputJitterBuffer.ingest()'s "seq <= last_applied_seq" check).
+#
+# Fast attack, slow release — a symmetric ±1-per-N-ticks slew would take
+# two full seconds to absorb a single wifi spike, during which the player
+# steers and the ship does not turn, "the most rage-inducing failure mode
+# in any netcode" per §3.3's own words.
+
+const LEAD_MIN := 1
+const LEAD_MAX := 12
+# "Never change it more than once per 30 ticks" (§3.3) — the floor that
+# binds the fast-attack side; slow-release's own 60-tick cadence already
+# exceeds it, so this one constant covers both.
+const MIN_CHANGE_INTERVAL_TICKS := 30
+const RELEASE_INTERVAL_TICKS := 60
+const CLEAN_SURPLUS_TICKS := 120   # 2s at 60Hz
+
+var lead := LEAD_MIN
+
+var _ticks_since_change := 0
+var _clean_surplus_ticks := 0
+
+
+# Call once per client physics tick with the most recently known server-
+# reported input_buffer_depth for THIS client's own slot (echoed in every
+# snapshot, §3.2) — or -1 if no snapshot carrying that field has arrived
+# yet. Returns the seq delta the caller should add for this tick's
+# outgoing packet: ordinarily 1 (ship normally increments its send
+# sequence by exactly one tick's worth), or 1+N / 0 on a tick where a lead
+# change actually fires (skip N extra / duplicate the current one).
+func update(input_buffer_depth: int) -> int:
+	_ticks_since_change += 1
+	if input_buffer_depth < 0:
+		return 1
+
+	if input_buffer_depth <= 0:
+		# A starve: the server's ring was empty for this player when it
+		# built that snapshot. React immediately, not after 2 seconds of
+		# evidence like release requires — but still debounced against
+		# MIN_CHANGE_INTERVAL_TICKS so a burst of consecutive starve
+		# reports doesn't compound into repeated, overlapping jumps.
+		_clean_surplus_ticks = 0
+		if _ticks_since_change >= MIN_CHANGE_INTERVAL_TICKS and lead < LEAD_MAX:
+			var new_lead := mini(lead + 3, LEAD_MAX)
+			var delta := new_lead - lead
+			lead = new_lead
+			_ticks_since_change = 0
+			return 1 + delta
+		return 1
+
+	_clean_surplus_ticks += 1
+	if _clean_surplus_ticks >= CLEAN_SURPLUS_TICKS and _ticks_since_change >= RELEASE_INTERVAL_TICKS and lead > LEAD_MIN:
+		lead -= 1
+		_ticks_since_change = 0
+		return 0   # duplicate this tick's seq — one tick of latency recovered
+	return 1

Game/scripts/networked_match.gd

@@ -28,6 +28,7 @@ const NetCodec = preload("res://scripts/net_codec.gd")
 const NetBodyState = preload("res://scripts/net_body_state.gd")
 const NetInterpolator = preload("res://scripts/net_interpolator.gd")
 const InputJitterBuffer = preload("res://scripts/input_jitter_buffer.gd")
+const InputLeadController = preload("res://scripts/input_lead_controller.gd")
 const HUD_SCENE = preload("res://scenes/HUD.tscn")
 
 # Minimum plausible interpolation delay even on a same-machine/LAN link —
@@ -88,6 +89,8 @@ var _input_seq := 0                       # client only
 # 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 _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 _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
@@ -350,7 +353,18 @@ func _send_local_input() -> void:
 	if _slots.is_empty():
 		return   # match_config hasn't arrived yet
 	var action := _local_input_sampler.get_action().copy()
-	_input_seq += 1
+	# 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). A duplicated tick can, in the narrow
+	# case where an older redundant copy hasn't been superseded yet, smear
+	# one of _input_history's older backup slots by one position — the
+	# PRIMARY (freshest, most-recently-relevant) value for every seq is
+	# unaffected, so this only ever degrades a backup copy, never the real
+	# per-tick record; §3.3 itself only promises "skip or duplicate a
+	# sequence number," not frame-perfect bookkeeping under a lead change.
+	_input_seq += _input_lead_controller.update(_last_known_input_buffer_depth)
 	# 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
@@ -369,6 +383,11 @@ func _on_snapshot_received(decoded: Dictionary) -> void:
 	var reset_gen: int = decoded["reset_gen"]
 	var bodies: Array = decoded["bodies"]
 	_last_received_snapshot_tick = server_tick
+	# Per-client header (§2.4): unlike the shared body segment, this is
+	# genuinely this recipient's own — input_buffer_depth is THIS client's
+	# own slot's server-side InputJitterBuffer.depth() at send time, which
+	# is exactly what the input_lead control loop (§3.3) needs.
+	_last_known_input_buffer_depth = decoded["input_buffer_depth"]
 	_update_tick_bias(server_tick)
 	for i in _slots.size():
 		if i < bodies.size():

Game/tests/cases/test_input_lead_controller.gd

@@ -0,0 +1,98 @@
+extends "res://tests/test_case.gd"
+
+const InputLeadController = preload("res://scripts/input_lead_controller.gd")
+
+
+func test_starts_at_minimum() -> void:
+	var c := InputLeadController.new()
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "initial lead")
+
+
+func test_unknown_depth_is_a_normal_tick() -> void:
+	var c := InputLeadController.new()
+	assert_eq(c.update(-1), 1, "no snapshot info yet -> ordinary +1 seq increment")
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "lead unchanged with no info")
+
+
+func test_healthy_depth_is_a_normal_tick_and_no_immediate_release() -> void:
+	var c := InputLeadController.new()
+	for i in 10:
+		assert_eq(c.update(1), 1, "healthy depth -> ordinary +1 tick %d" % i)
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "release needs 2s clean, not 10 ticks")
+
+
+# §3.3: "on any starve, increase by up to 3 immediately" — but debounced by
+# MIN_CHANGE_INTERVAL_TICKS so it isn't literally same-tick.
+func test_starve_triggers_fast_attack_after_debounce_floor() -> void:
+	var c := InputLeadController.new()
+	var deltas: Array[int] = []
+	for i in InputLeadController.MIN_CHANGE_INTERVAL_TICKS:
+		deltas.append(c.update(0))
+	# Every tick before the debounce floor is an ordinary +1 (no jump yet).
+	for i in InputLeadController.MIN_CHANGE_INTERVAL_TICKS - 1:
+		assert_eq(deltas[i], 1, "no lead change before the debounce floor, tick %d" % i)
+	assert_eq(deltas[InputLeadController.MIN_CHANGE_INTERVAL_TICKS - 1], 4, "attack fires on the debounce-floor tick: +1 ordinary + 3 skip")
+	assert_eq(c.lead, InputLeadController.LEAD_MIN + 3, "lead jumped by 3")
+
+
+func test_repeated_starvation_climbs_toward_max_and_clamps() -> void:
+	var c := InputLeadController.new()
+	# Enough sustained starvation to trigger several attack steps.
+	for i in InputLeadController.MIN_CHANGE_INTERVAL_TICKS * 6:
+		c.update(0)
+	assert_eq(c.lead, InputLeadController.LEAD_MAX, "clamps at LEAD_MAX under sustained starvation, never exceeds it")
+
+
+func test_release_requires_both_clean_surplus_and_its_own_interval() -> void:
+	var c := InputLeadController.new()
+	# Force lead above minimum first via one attack step.
+	for i in InputLeadController.MIN_CHANGE_INTERVAL_TICKS:
+		c.update(0)
+	var lead_after_attack := c.lead
+	assert_true(lead_after_attack > InputLeadController.LEAD_MIN, "lead raised above minimum before testing release")
+
+	# Fewer than CLEAN_SURPLUS_TICKS of healthy depth: must not release yet.
+	for i in InputLeadController.CLEAN_SURPLUS_TICKS - 1:
+		c.update(1)
+	assert_eq(c.lead, lead_after_attack, "no release before 2s of clean surplus has elapsed")
+
+	# One more healthy tick crosses the clean-surplus threshold AND the
+	# release interval (both are already satisfied by now since the
+	# debounce timer has been running the whole time) -> releases by 1.
+	var delta := c.update(1)
+	assert_eq(delta, 0, "release tick duplicates rather than incrementing seq")
+	assert_eq(c.lead, lead_after_attack - 1, "lead released by exactly 1")
+
+
+func test_release_stops_at_minimum() -> void:
+	var c := InputLeadController.new()
+	# Never starve — with lead already at LEAD_MIN, sustained health must
+	# never push it below the floor.
+	for i in InputLeadController.CLEAN_SURPLUS_TICKS * 3:
+		var delta := c.update(1)
+		assert_true(delta == 1, "lead already at minimum, never duplicates a seq trying to release further, tick %d" % i)
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "stays at minimum")
+
+
+func test_starve_resets_clean_surplus_counter() -> void:
+	var c := InputLeadController.new()
+	for i in InputLeadController.MIN_CHANGE_INTERVAL_TICKS:
+		c.update(0)   # raise lead above minimum via one attack step
+	var lead_after_attack := c.lead
+
+	# Some, but not all, of a clean surplus window — and well under the
+	# 30-tick attack debounce floor too, so the interrupting starve below
+	# can't accidentally retrigger a second attack step of its own.
+	var partial_clean_ticks := 10
+	for i in partial_clean_ticks:
+		c.update(1)
+	c.update(0)   # a lone starve tick, resetting _clean_surplus_ticks
+	assert_eq(c.lead, lead_after_attack, "the lone starve tick was too soon after the last change to trigger another attack")
+
+	# A full clean window from this fresh starting point is required before
+	# release fires — one tick short must not be enough.
+	for i in InputLeadController.CLEAN_SURPLUS_TICKS - 1:
+		c.update(1)
+	assert_eq(c.lead, lead_after_attack, "the starve interruption forced a fresh 2s clean window, so no release yet")
+	c.update(1)
+	assert_eq(c.lead, lead_after_attack - 1, "release finally fires once a full fresh clean window has elapsed since the interruption")