fix(multiplayer): resolve composition regression from second adversarial review

A second adversarial review of the previous fix commit found two of its
nine fixes silently defeated each other: the seq-range guard (fix for a
MEDIUM epoch-mismatch finding) capped the exact variable the ring-overflow
resync (fix for the original CRITICAL finding) depends on, making the
resync unreachable in production and recreating permanent input death at
a lower failure threshold, reachable via ordinary server tick loss alone.

- CRITICAL: rebind the seq-range guard to InputJitterBuffer's own
  highest_ingested_seq (now public) instead of the consumer-side
  last_applied_seq, so it tracks the client's send epoch rather than a
  value that can lag arbitrarily far behind during a stall.
- HIGH: InputLeadController's release logic still ANDed the old
  `lead > LEAD_MIN` gate onto the new depth-driven condition, so a
  backlog the controller never caused still couldn't drain. Split into
  two independent decisions: the seq-duplicate action follows real
  depth alone; lead's own bookkeeping separately never drops below its
  floor.
- MEDIUM: widen the CI driver's movement/stalled sampling margin
  (run_seconds - 2.0, was - 0.5) and assert the peer is still in
  multiplayer.get_peers() at sample time, since the old margin let the
  check pass on residual starvation grace after a bot had already
  disconnected.
- LOW: measure horizontal-only displacement in the human smoke test's
  movement check — the old 3D-distance bar was beatable by pure
  gravity settling with fully dead input.
- LOW: fix a real "clean stderr" violation (match_net.gd broadcasting
  a departure notice to a peer whose ENet channels are already torn
  down, including a second peer disconnecting in the same poll batch)
  by deferring the notification to the next idle frame.
- Wire the server's per-slot stalled bit into the client debug overlay
  for real — a prior commit message claimed this already reached the
  overlay when only the CI gate actually read it.

Re-verified end-to-end against the real production RPC path (not just
unit tests in isolation, which is how the composition bug got past the
first round): a 2-bot CI match with a 1.5s host SIGSTOP freeze injected
mid-run, well past the 0.6s threshold the review reproduced the bug at,
now recovers cleanly on repeated runs with zero stderr noise.

Game/scripts/input_jitter_buffer.gd

@@ -42,7 +42,19 @@ var _seeded := false
 # it, a backlog bigger than RING_SIZE (a host stall, or persistent client/
 # server clock drift) permanently zeroed a connected player's input for the
 # rest of the match.
-var _highest_ingested_seq := -1
+#
+# Deliberately public (no underscore), same as last_applied_seq: the
+# networked_match.gd caller's seq-range guard (§3.1 step 4) must bound
+# against THIS, not against last_applied_seq. A second adversarial review
+# found that bounding against last_applied_seq caps every accepted seq at
+# last_applied_seq + RING_SIZE, which in turn caps this field at the same
+# ceiling — making the resync condition below (which needs this field to
+# reach expected + RING_SIZE) arithmetically unreachable on the only call
+# path that exists in production. The two fixes looked independent but
+# shared a variable and silently cancelled each other out. highest_ingested
+# tracks the client's own send epoch instead, which the guard can safely
+# let run ahead of a lagging consumer.
+var highest_ingested_seq := -1
 
 
 func _init() -> void:
@@ -72,8 +84,8 @@ func ingest(newest_seq: int, actions: Array) -> void:
 		# "expected" with reality the moment real data first exists.
 		last_applied_seq = newest_seq - actions.size()
 		_seeded = true
-	if newest_seq > _highest_ingested_seq:
-		_highest_ingested_seq = newest_seq
+	if newest_seq > highest_ingested_seq:
+		highest_ingested_seq = newest_seq
 	for i in actions.size():
 		var seq: int = newest_seq - i
 		if seq <= last_applied_seq:
@@ -110,7 +122,7 @@ func consume() -> ShipAction:
 	# ticks of not-yet-consumed data at once — if the caller has fallen
 	# further behind the newest data actually arriving than that (a host
 	# stall, or persistent client/server clock drift), every tick between
-	# "expected" and "_highest_ingested_seq - RING_SIZE" has already been
+	# "expected" and "highest_ingested_seq - RING_SIZE" has already been
 	# irrecoverably overwritten by more recent arrivals landing on the same
 	# ring slots. Waiting for it tick-by-tick would starve — and, past
 	# STARVE_ZERO_TICKS, zero this player's ship — for the ENTIRE gap even
@@ -119,8 +131,8 @@ func consume() -> ShipAction:
 	# host freeze permanently zeroed a connected player's input for the
 	# rest of the match, with no self-recovery). Skip the unrecoverable
 	# span and resync directly to what the ring can still actually provide.
-	if _highest_ingested_seq - expected >= RING_SIZE:
-		last_applied_seq = _highest_ingested_seq - RING_SIZE
+	if highest_ingested_seq - expected >= RING_SIZE:
+		last_applied_seq = highest_ingested_seq - RING_SIZE
 		expected = last_applied_seq + 1
 		idx = expected % RING_SIZE
 

Game/scripts/input_lead_controller.gd

@@ -78,21 +78,31 @@ func update(input_buffer_depth: int) -> int:
 		return 1
 
 	# Release must react to the ACTUAL server-reported depth, not to this
-	# controller's own memory of past attacks. An adversarial review found
-	# the original gate here was `lead > LEAD_MIN` — a self-tracked counter
-	# of this controller's own past decisions — so any backlog it did NOT
-	# itself create (a server hitch, persistent client/server clock drift,
-	# a burst re-delivery) was never drained: `lead` stayed at its starting
-	# value the whole time even while `input_buffer_depth` sat well above
-	# target, permanently adding latency with the control loop reporting
-	# itself perfectly healthy. Gate on the real signal instead.
+	# controller's own memory of past attacks. A first pass at this fix
+	# added the depth check above but left the OLD gate, `lead > LEAD_MIN`,
+	# still ANDed onto the final condition below — so a backlog this
+	# controller did NOT itself cause (a server hitch, persistent client/
+	# server clock drift, a ring resync) still could never be drained:
+	# with lead pinned at its starting floor, that clause always failed
+	# even while input_buffer_depth sat well above target. A second
+	# adversarial review caught it, confirmed by this file's own
+	# test_release_drains_a_backlog_it_never_caused_itself, whose original
+	# assertion text literally said "lead cannot release below its own
+	# floor even under large surplus" as if that were correct.
+	#
+	# The fix splits the one gate into two separate decisions: whether to
+	# duplicate this tick's seq (the only thing that actually narrows real
+	# buffered depth) follows the real signal alone, below; whether to
+	# keep decrementing `lead`'s own bookkeeping below its documented
+	# floor is a separate, cosmetic-only choice made inside that branch.
 	if input_buffer_depth > TARGET_DEPTH:
 		_clean_surplus_ticks += 1
 	else:
 		_clean_surplus_ticks = 0
 
-	if _clean_surplus_ticks >= CLEAN_SURPLUS_TICKS and _ticks_since_change >= RELEASE_INTERVAL_TICKS and lead > LEAD_MIN:
-		lead -= 1
+	if _clean_surplus_ticks >= CLEAN_SURPLUS_TICKS and _ticks_since_change >= RELEASE_INTERVAL_TICKS:
+		if 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/match_net.gd

@@ -98,7 +98,35 @@ func _remove_player(peer_id: int) -> void:
 		return
 	roster.erase(peer_id)
 	player_left.emit(peer_id)
-	_player_left.rpc(peer_id)
+	# rpc() broadcasts to every peer in multiplayer.get_peers() — including,
+	# transiently, the very peer that just disconnected: this fires from
+	# NetworkManager's client_disconnected signal, and empirically that
+	# peer's own ENetConnection can still be momentarily present in the
+	# broadcast's target set with its channels already torn down, which
+	# logs "Unable to send packet on channel 0, max channels: 0" on every
+	# single disconnect (found by a second adversarial review — harmless to
+	# the game, since the departing peer obviously doesn't need to hear
+	# about its own departure, but it meant "clean stderr" wasn't actually
+	# clean for any test in this project).
+	#
+	# A first attempt filtered the broadcast down to rpc_id() calls that
+	# explicitly skip `peer_id`. That's necessary but not sufficient: when
+	# two peers disconnect within the same poll() batch (both bots quitting
+	# at the end of a CI run land within the same tick), get_peers() here
+	# can still list the SECOND peer as connected while its own disconnect
+	# event just hasn't been dispatched yet in this same batch — sending to
+	# it hits the identical error, one hop later. Defer the whole
+	# notification to the next idle frame instead of sending synchronously
+	# from inside signal-handling: by then poll() has fully returned, every
+	# disconnect event in this batch has been dispatched, and get_peers()
+	# reflects the settled, genuinely-still-connected set.
+	call_deferred("_broadcast_player_left", peer_id)
+
+
+func _broadcast_player_left(peer_id: int) -> void:
+	for other_peer_id in multiplayer.get_peers():
+		if other_peer_id != peer_id:
+			_player_left.rpc_id(other_peer_id, peer_id)
 
 
 # Balances a new joiner onto whichever team currently has fewer players

Game/scripts/net_debug_overlay.gd

@@ -43,15 +43,18 @@ func _process(_delta: float) -> void:
 			# task 3.7: RTT, jitter, loss, buffer depth, snapshot age,
 			# bandwidth all live here now. Prediction error is intentionally
 			# absent — there is no client-side prediction until Phase 4, so
-			# there is nothing honest to show for it yet.
+			# there is nothing honest to show for it yet. STALLED shows the
+			# server's own InputJitterBuffer.stalled bit for this client's
+			# slot, round-tripped through the wire.
 			var stats := {}
 			var game := get_tree().get_first_node_in_group("game")
 			if game and game.has_method("get_net_debug_stats"):
 				stats = game.get_net_debug_stats()
-			_label.text = "NET: client  RTT %.1fms  jitter %.1fms  offset %.1fms\nbuf depth %s  lead %s  loss %.1f%%  snap age %.1fms\nout %s  in %s" % [
+			var stalled_suffix := "  STALLED" if stats.get("server_stalled", false) else ""
+			_label.text = "NET: client  RTT %.1fms  jitter %.1fms  offset %.1fms\nbuf depth %s  lead %s  loss %.1f%%  snap age %.1fms%s\nout %s  in %s" % [
 				NetworkManager.rtt_ms, NetworkManager.jitter_ms, NetworkManager.clock_offset_ms,
 				str(stats.get("input_buffer_depth", -1)), str(stats.get("input_lead", "-")),
-				stats.get("snapshot_loss_pct", 0.0), stats.get("snapshot_age_ms", 0.0),
+				stats.get("snapshot_loss_pct", 0.0), stats.get("snapshot_age_ms", 0.0), stalled_suffix,
 				_format_kbps(MatchSim.get_bytes_sent_per_sec()), _format_kbps(MatchSim.get_bytes_received_per_sec()),
 			]
 	else:

Game/scripts/networked_match.gd

@@ -256,16 +256,35 @@ func _on_input_received(peer_id: int, decoded: Dictionary) -> void:
 			# ring" rationale wasn't actually achieved), and could silently
 			# drop an honest client's input forever the moment accumulated
 			# server tick loss closed whatever accidental head-start margin
-			# existed. Bound against this slot's own last_applied_seq
-			# instead — the client's own epoch, which the ring is already
-			# anchored to — using the ring's own capacity as the bound,
-			# exactly matching what InputJitterBuffer.consume()'s own
-			# overflow-resync logic treats as "unrecoverably far ahead"
-			# anyway. Falls back to seq itself (never rejects) before the
-			# buffer has ever been seeded — there's no baseline yet to
-			# bound against.
+			# existed.
+			#
+			# A first rebound bounded against this slot's own
+			# last_applied_seq — the CONSUMER's position — using the ring's
+			# capacity as the bound. A second adversarial review found this
+			# broke the ring-overflow resync it was landed alongside: capping
+			# every accepted seq at last_applied_seq + RING_SIZE also caps
+			# jb.highest_ingested_seq at that same ceiling, so
+			# consume()'s resync condition (which needs highest_ingested_seq
+			# to reach expected + RING_SIZE) could never fire in production —
+			# silently recreating the exact permanent-input-death bug this
+			# whole guard-rebound was part of fixing, at an even LOWER
+			# freeze threshold, reachable via ordinary server tick loss alone
+			# with no external trigger.
+			#
+			# Bound against jb.highest_ingested_seq instead — the highest
+			# seq this slot has ever actually been ALLOWED to ingest, i.e.
+			# the client's own send epoch — using the ring's own capacity as
+			# the bound, same as before. An honest client's consecutive
+			# packets differ by only a few seq (redundancy + a bounded
+			# input_lead skip), so this bound tracks a well-behaved client
+			# regardless of how far the CONSUMER has fallen behind, while
+			# still rejecting a single garbage-far-future jump: an attacker
+			# can only walk highest_ingested_seq forward at the rate the
+			# 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.
 			var jb := slot.jitter_buffer
-			var seq_bound: int = (jb.last_applied_seq if jb.last_applied_seq >= 0 else seq) + InputJitterBuffer.RING_SIZE
+			var seq_bound: int = (jb.highest_ingested_seq if jb.highest_ingested_seq >= 0 else seq) + InputJitterBuffer.RING_SIZE
 			if seq > seq_bound:
 				return
 			jb.ingest(seq, decoded["actions"])
@@ -343,8 +362,9 @@ func _ship_to_net_body_state(ship: Ship, stalled: bool) -> NetBodyState:
 	# §3.2: InputJitterBuffer.stalled was computed all along but never
 	# reached the wire — an adversarial review found this was the exact
 	# signal that would have made the ring-overflow bug (this session's
-	# critical fix) visible to the client, the debug overlay, and the CI
-	# gate, and its absence is part of why none of them ever noticed.
+	# critical fix) visible to the client and the CI gate, and its absence
+	# is part of why neither ever noticed. get_net_debug_stats() below is
+	# what actually surfaces it to the debug overlay now.
 	s.stalled = stalled
 	return s
 
@@ -588,11 +608,25 @@ func get_net_debug_stats() -> Dictionary:
 	# long has passed with nothing arriving at all.
 	var is_stale := _last_snapshot_wall_ms >= 0 and Time.get_ticks_msec() - _last_snapshot_wall_ms > SNAPSHOT_STALE_MS
 	var snapshot_loss_pct := 100.0 if is_stale else _snapshot_loss_ewma * 100.0
+	# The server's jitter_buffer.stalled bit for THIS client's own slot,
+	# round-tripped through NetBodyState onto the wire (§3.2) — added by the
+	# first adversarial-review fix round, but a second review found nothing
+	# actually read it client-side (net_interpolator.gd only passed it
+	# through lerp/extrapolate), so the commit's claim that it made the
+	# server-side starvation state "visible to the client, the debug
+	# 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
 	return {
 		"input_buffer_depth": _last_known_input_buffer_depth,
 		"input_lead": _input_lead_controller.lead,
 		"snapshot_age_ms": snapshot_age_ms,
 		"snapshot_loss_pct": snapshot_loss_pct,
+		"server_stalled": server_stalled,
 	}
 
 

Game/tests/cases/test_input_lead_controller.gd

@@ -67,12 +67,18 @@ func test_release_requires_both_clean_surplus_and_its_own_interval() -> void:
 
 func test_release_stops_at_minimum() -> void:
 	var c := InputLeadController.new()
-	# Sustained surplus depth, but lead is already at LEAD_MIN — must never
-	# push it below the floor regardless of how much surplus is reported.
+	# Sustained surplus depth with lead already at LEAD_MIN: `lead` itself
+	# must never drop below the floor, but release must still fire
+	# (duplicate a seq) once its own timing conditions are met, since a
+	# real reported surplus at floor lead is exactly the "backlog this
+	# controller never caused" case — capping `lead` is cosmetic, it must
+	# not also block the seq-duplicate action that drains real depth.
+	var released := false
 	for i in InputLeadController.CLEAN_SURPLUS_TICKS * 3:
-		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")
+		if c.update(InputLeadController.TARGET_DEPTH + 1) == 0:
+			released = true
+		assert_eq(c.lead, InputLeadController.LEAD_MIN, "lead never drops below the floor, tick %d" % i)
+	assert_true(released, "release still fires (duplicates a seq) even though lead itself is pinned at minimum")
 
 
 func test_starve_resets_clean_surplus_counter() -> void:
@@ -99,14 +105,18 @@ func test_starve_resets_clean_surplus_counter() -> void:
 	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 >
+# A first attempt at fixing this gated the whole release branch on `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.
+# even while the server kept reporting a deep, real backlog, and — because
+# that gate blocked the seq-duplicate action too, not just lead's own
+# bookkeeping — the actual buffered depth was never drained either. A
+# second adversarial review caught that the depth check added alongside
+# it didn't remove the old gate, just sat next to it. This reproduces the
+# scenario directly: lead never attacks (depth is never reported as a
+# starve, <= 0), yet release must still fire from sustained real surplus
+# alone, even while lead itself stays pinned at its floor throughout.
 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")
@@ -114,9 +124,12 @@ func test_release_drains_a_backlog_it_never_caused_itself() -> void:
 	# 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.
+	var released_at_floor := false
 	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")
+		if c.update(10) == 0:
+			released_at_floor = true
+	assert_eq(c.lead, InputLeadController.LEAD_MIN, "lead's own bookkeeping never drops below its floor")
+	assert_true(released_at_floor, "release still fires (duplicates a seq, actually draining real depth) even while lead is pinned at the floor")
 
 	# Raise it above the floor via one real attack, then confirm sustained
 	# external surplus (not self-caused) still drains it back down.

Game/tests/networked_match_test_hooks.gd

@@ -98,8 +98,16 @@ func run_client_check(settle_seconds: float, drive_seconds: float) -> void:
 
 	var end_position: Vector3 = my_slot.ship.visual.global_position
 	var moved := start_position.distance_to(end_position)
-	print("SMOKE INFO: client ship moved %.2fm (start=%s end=%s) while holding forward thrust for %.1fs" % [
-		moved, str(start_position), str(end_position), drive_seconds
+	# Horizontal-only (XZ), not full 3D distance: an adversarial review
+	# found a 1.2s window of completely dead input still registers ~1.07m
+	# of pure gravity settling on the Y axis alone (spawn height dropping
+	# to the floor), which sat ABOVE the old moved > 1.0 bar — only
+	# thrust_z_ok caught that failure, not moved. Forward thrust is a
+	# horizontal force (see ship.gd), so measuring XZ displacement can't
+	# be satisfied by gravity alone, regardless of spawn height or timing.
+	var moved_horizontal := Vector2(end_position.x, end_position.z).distance_to(Vector2(start_position.x, start_position.z))
+	print("SMOKE INFO: client ship moved %.2fm (%.2fm horizontal) (start=%s end=%s) while holding forward thrust for %.1fs" % [
+		moved, moved_horizontal, str(start_position), str(end_position), drive_seconds
 	])
 
 	# thrust_power 150 / mass 5 = 30 m/s^2 nominal acceleration (see ship.gd) —
@@ -107,9 +115,9 @@ func run_client_check(settle_seconds: float, drive_seconds: float) -> void:
 	# A generous, not-tuned-to-the-decimal bound: this is a wiring smoke
 	# test, not a physics-accuracy test (net_codec's own tests already cover
 	# quantisation precision).
-	var success := moved > 1.0 and thrust_z_ok
-	print("SMOKE %s: client observed %.2fm of server-authoritative movement via interpolation, thrust_z_ok=%s" % [
-		"PASS" if success else "FAIL", moved, str(thrust_z_ok)
+	var success := moved_horizontal > 1.0 and thrust_z_ok
+	print("SMOKE %s: client observed %.2fm horizontal of server-authoritative movement via interpolation, thrust_z_ok=%s" % [
+		"PASS" if success else "FAIL", moved_horizontal, str(thrust_z_ok)
 	])
 	await get_tree().create_timer(0.3).timeout
 	NetworkManager.shutdown()
@@ -268,20 +276,33 @@ func run_ci_host_check(run_seconds: float) -> void:
 	# 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)
+	# overflow bug this check exists to catch if sampled too late. A first
+	# attempt used a 0.5s margin (run_seconds - 0.5); a second adversarial
+	# review instrumented multiplayer.get_peers() at sample time and found
+	# it was already EMPTY — both bots had legitimately disconnected before
+	# the sample ran, and the check was only passing on the ~200ms of
+	# residual STARVE_ZERO_TICKS starvation grace, not because it was
+	# genuinely still connected as this print used to claim. Widen the
+	# margin AND assert connectivity directly at sample time, rather than
+	# inferring it from timing, so a future regression in either direction
+	# (margin too tight again, or client run_seconds changing) fails loudly
+	# here instead of silently passing on residual grace.
+	var movement_check_delay := maxf(1.0, run_seconds - 2.0)
 	await get_tree().create_timer(movement_check_delay).timeout
+	var connected_peers := multiplayer.get_peers()
 	var input_reached_server := true
 	for slot in match_scene._slots:
+		var still_connected: bool = slot.peer_id in connected_peers
+		if not still_connected:
+			input_reached_server = false
+			print("SMOKE FAIL: peer %d already disconnected at movement-sample time (connected_peers=%s) — margin too tight" % [slot.peer_id, str(connected_peers)])
 		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)])
+		print("SMOKE INFO: peer %d moved %.2fm server-side (connected=%s), stalled=%s" % [slot.peer_id, moved, str(still_connected), str(stalled)])
 		if moved <= 0.5 or stalled:
 			input_reached_server = false