fix(multiplayer): adversarial review fixes for Phase 3

An Opus subagent's adversarial review of Phase 3 found a critical, silent,
permanent bug plus eight smaller real issues, all empirically verified
with real two- and three-process runs:

CRITICAL: InputJitterBuffer's 32-entry ring permanently bricked a
player's input once the un-consumed backlog exceeded the ring's
capacity - a fresh arrival would land in the exact slot consume() was
still waiting on, and since both counters only ever advance, the gap
never closed. Reproduced with a real SIGSTOP/SIGCONT host freeze:
client movement dropped from ~26m to 0.00m at ~0.7s, worse under real
loss (a lossy link lowered the fatal threshold to ~400ms), and
reachable via ordinary clock drift with no external trigger at all.
Fixed by tracking the highest seq ever ingested and having consume()
jump directly to what the ring can still provide once the gap exceeds
capacity, instead of starving through an unrecoverable span. Re-verified
with a 3s freeze (well past the original threshold): full recovery.

HIGH: InputLeadController's release logic was gated on its own past
attacks (lead > LEAD_MIN) rather than the real server-reported depth, so
a backlog it didn't itself cause was never drained. Fixed to gate on
actual depth vs target.

MEDIUM-HIGH: the rate limiter's "N consecutive over-budget seconds"
streak hard-reset to 0 on any clean window, letting a duty-cycled flood
(burst, one clean window, repeat) sustain ~33x budget indefinitely with
zero warnings. Replaced with a leaky-bucket accumulator immune to the
same evasion by construction.

MEDIUM: the seq > server_tick + 20 guard compared two unrelated clock
epochs (server process uptime vs. client's own from-zero seq numbering),
so it never actually protected anything on a long-running server and
could silently drop an honest client's input forever. Bound against the
buffer's own last_applied_seq instead.

MEDIUM: InputJitterBuffer.stalled was computed but never reached the
wire - the one signal that would have made the ring-overflow bug visible
anywhere. Now wired through _ship_to_net_body_state.

MEDIUM: task 3.6's CI driver's assertions didn't depend on client input
reaching the server at all, so it kept passing with the ring-overflow
bug actively triggered. Added real ship-movement and non-stalled checks,
sampled while bots are still connected (an initial attempt sampled after
their own legitimate disconnect, which starves identically to the bug).

LOW-MEDIUM: a lead change silently mislabelled _input_history's older
entries, since the wire format has no per-entry seq field. Fixed by
handling each delta case (ordinary/release/attack) on its own terms.

LOW: bandwidth and snapshot-loss overlay metrics froze at their last
value during a total outage instead of decaying - exactly when they
matter most. Both now report honest post-outage values.

LOW: a guard comment on NetworkManager._ping misdescribed the actual
disconnect_peer() arguments in use. Corrected.

New permanent regression tests: test_ring_overflow_resyncs_to_fresh_data
_instead_of_starving_forever, test_release_drains_a_backlog_it_never_
caused_itself, and client-abuse-flood-dutycycle (reproduces the exact
duty-cycle evasion). Full regression suite, including the net-sim-latency
milestone gate, all abuse roles, and the CI driver, re-run clean after
every fix.

Game/tests/cases/test_input_jitter_buffer.gd

@@ -110,3 +110,44 @@ func test_ring_wraparound_does_not_confuse_a_stale_slot_with_a_fresh_one() -> vo
 	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")
+
+
+# The under-full direction (above) was covered before an adversarial review
+# found the OVER-full direction was not: a backlog bigger than RING_SIZE
+# (a host stall, or persistent client/server clock drift) made consume()
+# starve — and, past STARVE_ZERO_TICKS, zero the player's ship — forever,
+# because both last_applied_seq and the client's own seq only ever advance
+# with no resync, so the gap never closed even though fresh, real input
+# kept arriving the whole time.
+func test_ring_overflow_resyncs_to_fresh_data_instead_of_starving_forever() -> void:
+	var buf := InputJitterBuffer.new()
+	buf.ingest(0, [_action(0.0)])
+	buf.consume()   # last_applied_seq = 0
+
+	# A burst of packets arriving all at once, exactly what poll() delivers
+	# in one batch once a stalled server resumes — the client kept sending
+	# normally the whole time (a real packet every tick, last-4 redundancy,
+	# newest-first), nothing consumed in between. 50 ticks' worth, well
+	# past one full lap of the 32-entry ring.
+	for seq in range(1, 51):
+		var window: Array = []
+		for k in 4:
+			window.append(_action(float(seq - k) * 0.01))
+		buf.ingest(seq, window)
+	assert_eq(buf.last_applied_seq, 0, "nothing consumed yet, only ingested")
+
+	# The gap (50 - 1 = 49) exceeds RING_SIZE (32): everything older than
+	# "50 - RING_SIZE" has already been irrecoverably overwritten by more
+	# recent arrivals landing on the same ring slots. A single consume()
+	# must resync directly to the oldest data the ring can still actually
+	# provide, not starve through the entire abandoned span.
+	var a := buf.consume()
+	var expected_resync_seq := 50 - InputJitterBuffer.RING_SIZE + 1
+	assert_eq(buf.last_applied_seq, expected_resync_seq, "resynced to exactly RING_SIZE behind the newest data")
+	assert_almost_eq(a.thrust.z, float(expected_resync_seq) * 0.01, 0.0001, "recovered the resynced tick's real action from the ring, not a stale ghost or a zeroed one")
+	assert_eq(buf.starved_ticks, 0, "resyncing to real data is not starvation")
+	assert_true(not buf.stalled, "a recovered player must not be reported as stalled")
+
+	# Normal sequential consumption resumes correctly from the resync point.
+	var next := buf.consume()
+	assert_almost_eq(next.thrust.z, float(expected_resync_seq + 1) * 0.01, 0.0001, "next tick continues in order from the resync point")

Game/tests/cases/test_input_lead_controller.gd

@@ -51,25 +51,26 @@ func test_release_requires_both_clean_surplus_and_its_own_interval() -> void:
 	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.
+	# Fewer than CLEAN_SURPLUS_TICKS of surplus depth (above TARGET_DEPTH):
+	# must not release yet.
 	for i in InputLeadController.CLEAN_SURPLUS_TICKS - 1:
-		c.update(1)
+		c.update(InputLeadController.TARGET_DEPTH + 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
+	# One more surplus 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)
+	var delta := c.update(InputLeadController.TARGET_DEPTH + 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.
+	# Sustained surplus depth, but lead is already at LEAD_MIN — must never
+	# push it below the floor regardless of how much surplus is reported.
 	for i in InputLeadController.CLEAN_SURPLUS_TICKS * 3:
-		var delta := c.update(1)
+		var delta := c.update(InputLeadController.TARGET_DEPTH + 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")
 
@@ -85,14 +86,49 @@ func test_starve_resets_clean_surplus_counter() -> void:
 	# 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(InputLeadController.TARGET_DEPTH + 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)
+		c.update(InputLeadController.TARGET_DEPTH + 1)
 	assert_eq(c.lead, lead_after_attack, "the starve interruption forced a fresh 2s clean window, so no release yet")
-	c.update(1)
+	c.update(InputLeadController.TARGET_DEPTH + 1)
 	assert_eq(c.lead, lead_after_attack - 1, "release finally fires once a full fresh clean window has elapsed since the interruption")
+
+
+# An adversarial review found the original release gate was `lead >
+# LEAD_MIN` — this controller's own memory of past attacks — so a backlog
+# it did NOT itself create (a server hitch, persistent client/server clock
+# drift, a burst re-delivery) was never drained: lead stayed at 1 forever
+# even while the server kept reporting a deep, real backlog. This
+# reproduces that scenario directly: lead never attacks (depth is never
+# reported as a starve, <= 0), yet release must still fire from sustained
+# real surplus alone.
+func test_release_drains_a_backlog_it_never_caused_itself() -> void:
+	var c := InputLeadController.new()
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "starts at minimum, never attacked")
+
+	# A large, externally-caused surplus (e.g. right after the server's own
+	# ring-overflow resync) reported for well over 2s — lead never moves
+	# via attack since depth is never <= 0.
+	for i in InputLeadController.CLEAN_SURPLUS_TICKS + InputLeadController.RELEASE_INTERVAL_TICKS:
+		c.update(10)
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "lead cannot release below its own floor even under large surplus")
+
+	# Raise it above the floor via one real attack, then confirm sustained
+	# external surplus (not self-caused) still drains it back down.
+	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, "attack raised lead")
+
+	var released := false
+	for i in InputLeadController.CLEAN_SURPLUS_TICKS + InputLeadController.RELEASE_INTERVAL_TICKS:
+		if c.update(10) == 0:
+			released = true
+			break
+	assert_true(released, "sustained externally-caused surplus (depth=10) must eventually trigger a release")
+	assert_true(c.lead < lead_after_attack, "lead actually decreased in response to real depth, not just internal bookkeeping")

Game/tests/networked_match_smoke.gd

@@ -39,7 +39,7 @@ func _ready() -> void:
 				return
 			print("SMOKE: joining ...")
 			MatchNet.welcomed.connect(_on_client_welcomed)
-		"client-abuse-malformed", "client-abuse-flood":
+		"client-abuse-malformed", "client-abuse-flood", "client-abuse-flood-dutycycle":
 			# task 3.4's disconnect-abusive-peer paths: joins normally (so
 			# it's a real connected peer, exactly like a hostile custom
 			# client would be — the validation doesn't get to assume
@@ -93,5 +93,7 @@ func _on_abuser_welcomed() -> void:
 	get_tree().root.add_child.call_deferred(hooks)
 	if _role == "client-abuse-malformed":
 		hooks.run_malformed_abuse_check.call_deferred()
+	elif _role == "client-abuse-flood-dutycycle":
+		hooks.run_duty_cycle_flood_abuse_check.call_deferred()
 	else:
 		hooks.run_rate_limit_abuse_check.call_deferred()

Game/tests/networked_match_test_hooks.gd

@@ -148,12 +148,13 @@ func run_malformed_abuse_check() -> void:
 	get_tree().quit(0 if disconnected[0] else 1)
 
 
-# task 3.4: MatchSim._recv_input must rate-limit and disconnect after
-# RATE_LIMIT_OVER_BUDGET_SECONDS_TO_DISCONNECT (3) consecutive seconds over
-# RATE_LIMIT_PACKETS_PER_SEC (110/s). Every packet here is individually
-# well-formed (a real NetCodec.pack_input payload) — only the SEND RATE is
-# abusive, confirming the rate limiter fires independently of the malformed-
-# packet counter, not as a side effect of it.
+# task 3.4: MatchSim._recv_input must rate-limit and disconnect a sustained
+# continuous flood well above RATE_LIMIT_PACKETS_PER_SEC (110/s) via the
+# leaky-bucket excess accumulator (RATE_LIMIT_EXCESS_PACKETS_TO_DISCONNECT).
+# Every packet here is individually well-formed (a real NetCodec.pack_input
+# payload) — only the SEND RATE is abusive, confirming the rate limiter
+# fires independently of the malformed-packet counter, not as a side
+# effect of it.
 func run_rate_limit_abuse_check() -> void:
 	await get_tree().create_timer(1.0).timeout
 	var disconnected := [false]   # see run_malformed_abuse_check's comment on why not a plain bool
@@ -179,6 +180,55 @@ func run_rate_limit_abuse_check() -> void:
 	get_tree().quit(0 if disconnected[0] else 1)
 
 
+# Regression test for a real bug an adversarial review found and this
+# session fixed: the ORIGINAL rate limiter tracked "N consecutive
+# over-budget seconds" and hard-reset that streak to 0 on any single clean
+# window — so a burst-then-idle duty cycle (flood hard, go quiet for one
+# window, repeat) evaded it indefinitely. Reproduced against the real
+# MatchSim._recv_input: ~33x the packet budget sustained for 28.5s with
+# zero disconnect warnings. The fix (a leaky-bucket excess accumulator
+# that grows by the window's actual total and drains by only one window's
+# worth of budget, every window) doesn't care how the excess is
+# distributed in time. This test reproduces the exact attack shape.
+func run_duty_cycle_flood_abuse_check() -> void:
+	await get_tree().create_timer(1.0).timeout
+	var disconnected := [false]
+	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)
+
+	var net_codec := preload("res://scripts/net_codec.gd")
+	var ship_action_script := preload("res://scripts/ship_action.gd")
+	var bytes: PackedByteArray = net_codec.pack_input(1, 0, Time.get_ticks_msec(), [ship_action_script.new()])
+
+	const CYCLE_SECONDS := 3.0
+	const BURST_SECONDS := 0.35
+	const TEST_SECONDS := 6.0   # the leaky bucket trips within the first cycle; no need for a long soak
+	const TRICKLE_HZ := 60   # legitimate-shaped background rate, well under budget alone
+
+	var deadline_ms := Time.get_ticks_msec() + int(TEST_SECONDS * 1000.0)
+	var cycle_start_ms := Time.get_ticks_msec()
+	while Time.get_ticks_msec() < deadline_ms and not disconnected[0]:
+		var t_in_cycle := float(Time.get_ticks_msec() - cycle_start_ms) / 1000.0
+		if t_in_cycle >= CYCLE_SECONDS:
+			cycle_start_ms = Time.get_ticks_msec()
+			t_in_cycle = 0.0
+		if t_in_cycle < BURST_SECONDS:
+			for i in 200:   # a hard burst, far above budget
+				MatchSim._recv_input.rpc_id(1, bytes)
+		else:
+			for i in maxi(1, TRICKLE_HZ / 60):   # ~60/s trickle, keeps the window rolling and stays under budget alone
+				MatchSim._recv_input.rpc_id(1, bytes)
+		NetworkManager.poll()
+		await get_tree().process_frame
+	await get_tree().create_timer(0.5).timeout
+	NetworkManager.poll()
+
+	print("SMOKE %s: duty-cycled flood (burst %.2fs / cycle %.1fs) %s" % [
+		"PASS" if disconnected[0] else "FAIL", BURST_SECONDS, CYCLE_SECONDS,
+		"resulted in disconnect" if disconnected[0] else "evaded rate limiting entirely",
+	])
+	get_tree().quit(0 if disconnected[0] else 1)
+
+
 # task 3.6, host role: waits for both bots' scenes to settle, forces a
 # deterministic goal (bot-vs-bot scoring isn't reliable enough within a
 # short CI run to gate on), then compares the server's own final score
@@ -194,17 +244,51 @@ func run_ci_host_check(run_seconds: float) -> void:
 		return
 	print("SMOKE INFO: host ship_count=%d slot_count=%d" % [match_scene.ships.size(), match_scene._slots.size()])
 
+	# An adversarial review found this driver's original checks (snapshot
+	# count, a server-FORCED goal's cross-peer score agreement) don't
+	# depend on client input ever reaching the server at all — it kept
+	# reporting PASS with the input pipeline completely dead (verified by
+	# injecting the ring-overflow bug this session's critical fix
+	# addresses, mid-run). Record each ship's starting position now, before
+	# anything moves, so real server-side movement over the run can be
+	# checked directly — the same signal run_client_check already uses for
+	# a human client, applied here per-bot instead of just for "my own ship".
+	var start_positions: Dictionary = {}
+	for slot in match_scene._slots:
+		if is_instance_valid(slot.ship):
+			start_positions[slot.peer_id] = slot.ship.global_position
+
 	var goals: Array = match_scene.arena.get_goals() if match_scene.arena else []
 	if is_instance_valid(match_scene.ball) and not goals.is_empty():
 		match_scene.ball.linear_velocity = Vector3.ZERO
 		match_scene.ball.global_position = goals[0].global_position
 		print("SMOKE INFO: host forced a goal for the cross-peer score agreement check")
 
-	# Extra buffer beyond run_seconds: clients start ~1.5s after the host
-	# (established two-process test convention) and run for their own
-	# run_seconds measured from THEIR start, so waiting only run_seconds
-	# here would race their score files not being written yet.
-	await get_tree().create_timer(run_seconds + 5.0).timeout
+	# Movement/stalled must be checked WHILE clients are still actively
+	# connected and playing, not after their run finishes — a client's own
+	# (legitimate, expected) disconnect at the end of its run naturally
+	# starves its jitter buffer too, which looks identical to the ring-
+	# overflow bug this check exists to catch if sampled too late. Clients
+	# start ~1.5s after the host and finish their own run_seconds shortly
+	# before disconnecting, so sample just ahead of that, not after.
+	var movement_check_delay := maxf(1.0, run_seconds - 0.5)
+	await get_tree().create_timer(movement_check_delay).timeout
+	var input_reached_server := true
+	for slot in match_scene._slots:
+		if not is_instance_valid(slot.ship) or not start_positions.has(slot.peer_id):
+			input_reached_server = false
+			print("SMOKE FAIL: peer %d has no valid ship to check movement on" % slot.peer_id)
+			continue
+		var moved: float = start_positions[slot.peer_id].distance_to(slot.ship.global_position)
+		var stalled: bool = slot.jitter_buffer.stalled
+		print("SMOKE INFO: peer %d moved %.2fm server-side (while still connected), stalled=%s" % [slot.peer_id, moved, str(stalled)])
+		if moved <= 0.5 or stalled:
+			input_reached_server = false
+
+	# Extra buffer beyond run_seconds: clients run for their own run_seconds
+	# measured from THEIR (later) start, so waiting only run_seconds here
+	# would race their score files not being written yet.
+	await get_tree().create_timer(run_seconds + 5.0 - movement_check_delay).timeout
 	print("SMOKE INFO: host final score=%s" % str(match_scene.score))
 
 	var slots_ok: bool = match_scene._slots.size() == 2
@@ -225,9 +309,9 @@ func run_ci_host_check(run_seconds: float) -> void:
 			print("SMOKE FAIL: peer %d saw score %s, server has %s" % [slot.peer_id, client_score, expected])
 			scores_agree = false
 
-	var success: bool = slots_ok and scores_agree and scores_seen == 2
-	print("SMOKE %s: CI host run (slots_ok=%s scores_agree=%s scores_seen=%d/2)" % [
-		"PASS" if success else "FAIL", str(slots_ok), str(scores_agree), scores_seen,
+	var success: bool = slots_ok and scores_agree and scores_seen == 2 and input_reached_server
+	print("SMOKE %s: CI host run (slots_ok=%s scores_agree=%s scores_seen=%d/2 input_reached_server=%s)" % [
+		"PASS" if success else "FAIL", str(slots_ok), str(scores_agree), scores_seen, str(input_reached_server),
 	])
 	NetworkManager.shutdown()
 	get_tree().quit(0 if success else 1)