feat(multiplayer): Phase 5 tasks 5.6-5.10 - disconnects, spectators, replay log

Completes Phase 5's implementation. Every task is verified at 1v1; the
3v3 phase gate itself has not been run and remains outstanding.

5.6/5.7 disconnects: a ship is never despawned. The slot keeps it and
swaps the controller (--fill-bots gives it a bot, the default leaves it
inert per §1.4), sets `stalled` immediately so the nameplate greys out
rather than waiting ~500ms for the abandoned jitter buffer to starve,
and reserves the slot for 30s keyed by player name so a reconnect gets
the same ship back.

5.7 was a real bug, found by the test rather than by review:
SlotInfo.controller was declared RLShipController, but the takeover
swaps in an AIShipController or the base controller - the narrower type
makes that assignment fail its type check, leaving the field pointing at
the controller set_controller() just queue_free()d. It surfaced as
controller_valid=false on the first run. The per-tick action write is
now also gated on `is RLShipController`, since a disconnected slot's bot
drives itself and overwriting it from a starving buffer would pin it to
the departed player's last input.

§6.4's two rules conflict: reserve for 30s, but abort when the last
human leaves. Applied naively the abort wins instantly in a 1v1 and the
reservation can never be redeemed, making reconnect unreachable exactly
when it matters. Abort now waits for no connections AND no outstanding
reservations.

5.8 spectators: a slotless peer spawns no ship and receives the same
snapshot broadcast. HUDController.spectator_mode keeps the clock, score
and goal celebration and hides only the ship instrument cluster - it
previously push_error'd and bailed, leaving a spectator with a dead HUD.
Camera cycles ships in slot order then the ball. --max-spectators caps
it, counted from the live peer list so a dropped spectator cannot leak a
unit of the cap.

5.9 escape respawn: new GameMode._on_bodies_respawned() virtual;
NetworkedMatch bumps reset_gen through Phase 2's deferred path so the
bump and the respawned pose land in the same broadcast. Single-player
modes are unaffected - the base is a no-op.

5.10 replay log: scripts/replay_log.gd, --replay-log=<path>, storing the
wire bytes verbatim in both directions rather than re-serialising - a
re-encode would launder away precisely the malformed payload being
chased. A live 6s match recorded 1115 records (557 inputs / 558
snapshots) and a stored snapshot decodes back to server_tick=100
match_state=WARMUP bodies=2.

Note for future work: --check-only --script is the only thing that
catches a parse error in networked_match.gd, because the unit runner
never loads it. Two separate breakages passed the full unit suite while
breaking every two-process run. A new class_name also needs --import
before it resolves.

Test surface: --role=host-disconnect (three-process 5.6/5.7 scenario),
--match-length=<s>, --replay-log, --fill-bots/--no-fill-bots,
--max-spectators. The ball-contact scenario now steers at the ball with
closed-loop real input instead of a hand-tuned fixed heading, which 5.3
broke by adding KICKOFF_YAW_JITTER; thrusting while turning took it from
2/3 to 5/5.

Regression: 87 unit tests; free-flight LAN p99 0.094m with 0 hard snaps;
transition gate 0.00%; ball contact 5/5; lifecycle goal cycle and full
match to RESULTS/LOBBY; disconnect+reconnect; two-bot CI.

Game/tests/cases/test_replay_log.gd

@@ -0,0 +1,120 @@
+extends "res://tests/test_case.gd"
+
+# Task 5.10. The log's whole value is that a recorded match can be replayed
+# faithfully enough to reproduce a reported snap, so what matters is that the
+# bytes come back BYTE-IDENTICAL and correctly framed — not merely that
+# something was written.
+
+const ReplayLogScript = preload("res://scripts/replay_log.gd")
+
+
+func _temp_path(suffix: String) -> String:
+	return "user://test_replay_%s_%d.ccrp" % [suffix, Time.get_ticks_usec()]
+
+
+func test_records_round_trip_byte_for_byte() -> void:
+	var path := _temp_path("roundtrip")
+	var log_writer = ReplayLogScript.new()
+	assert_eq(log_writer.open_for_write(path), OK, "opens for write")
+
+	var input_payload := PackedByteArray([0x01, 0xFF, 0x00, 0x7F, 0x80])
+	var snapshot_payload := PackedByteArray([0xDE, 0xAD, 0xBE, 0xEF])
+	log_writer.record_input(120, 4242, input_payload)
+	log_writer.record_snapshot(121, snapshot_payload)
+	log_writer.close()
+
+	var read := ReplayLogScript.read_all(path)
+	assert_eq(read.get("version", -1), ReplayLogScript.FORMAT_VERSION, "version round-trips")
+	assert_eq(read.get("tick_hz", -1), SimConstants.TICK_HZ, "tick rate is recorded so a reader need not guess")
+	var records: Array = read.get("records", [])
+	assert_eq(records.size(), 2, "both records read back")
+
+	assert_eq(records[0]["kind"], ReplayLogScript.RecordKind.INPUT, "first is an input")
+	assert_eq(records[0]["tick"], 120, "input tick")
+	assert_eq(records[0]["peer_id"], 4242, "input peer")
+	assert_eq(records[0]["payload"], input_payload, "input payload is byte-identical")
+
+	assert_eq(records[1]["kind"], ReplayLogScript.RecordKind.SNAPSHOT, "second is a snapshot")
+	assert_eq(records[1]["tick"], 121, "snapshot tick")
+	assert_eq(records[1]["payload"], snapshot_payload, "snapshot payload is byte-identical")
+	DirAccess.remove_absolute(ProjectSettings.globalize_path(path))
+
+
+func test_a_real_packet_survives_the_round_trip() -> void:
+	# The payloads above are hand-made. Use a genuine NetCodec input packet so
+	# a framing bug that only shows up at real packet sizes cannot hide.
+	var path := _temp_path("realpacket")
+	var action := ShipAction.new()
+	action.thrust = Vector3(0.5, -0.25, 1.0)
+	action.turbo = true
+	var packet := NetCodec.pack_input(77, 55, 1234, [action, action, action])
+
+	var log_writer = ReplayLogScript.new()
+	log_writer.open_for_write(path)
+	log_writer.record_input(500, 7, packet)
+	log_writer.close()
+
+	var records: Array = ReplayLogScript.read_all(path).get("records", [])
+	assert_eq(records.size(), 1, "one record")
+	assert_eq(records[0]["payload"], packet, "a real input packet round-trips unchanged")
+	# And it must still decode as the packet it was.
+	var decoded := NetCodec.unpack_input(records[0]["payload"])
+	assert_eq(decoded["seq"], 77, "the replayed packet still decodes to its own sequence")
+	DirAccess.remove_absolute(ProjectSettings.globalize_path(path))
+
+
+func test_empty_log_reads_back_as_no_records() -> void:
+	var path := _temp_path("empty")
+	var log_writer = ReplayLogScript.new()
+	log_writer.open_for_write(path)
+	log_writer.close()
+	var read := ReplayLogScript.read_all(path)
+	assert_eq(read.get("records", [-1]).size(), 0, "a header-only log has no records")
+	assert_eq(read.get("tick_hz", -1), SimConstants.TICK_HZ, "but still reports its header")
+	DirAccess.remove_absolute(ProjectSettings.globalize_path(path))
+
+
+func test_a_non_replay_file_is_rejected_rather_than_misread() -> void:
+	# FileAccess zero-fills past EOF exactly as StreamPeerBuffer does, so
+	# without a magic check an arbitrary file decodes as an endless run of
+	# zero-length records instead of failing.
+	var path := _temp_path("garbage")
+	var f := FileAccess.open(path, FileAccess.WRITE)
+	f.store_string("this is definitely not a replay log")
+	f.close()
+	assert_true(ReplayLogScript.read_all(path).is_empty(), "a foreign file is refused")
+	DirAccess.remove_absolute(ProjectSettings.globalize_path(path))
+
+
+func test_a_truncated_log_yields_its_intact_records() -> void:
+	# A server killed mid-write is the NORMAL way one of these ends, so a
+	# partial final record must not discard the whole session.
+	var path := _temp_path("truncated")
+	var log_writer = ReplayLogScript.new()
+	log_writer.open_for_write(path)
+	log_writer.record_input(1, 1, PackedByteArray([1, 2, 3, 4]))
+	log_writer.record_input(2, 1, PackedByteArray([5, 6, 7, 8]))
+	log_writer.close()
+
+	var whole := FileAccess.get_file_as_bytes(path)
+	var cut := whole.slice(0, whole.size() - 3)   # lop off part of the last payload
+	var f := FileAccess.open(path, FileAccess.WRITE)
+	f.store_buffer(cut)
+	f.close()
+
+	var records: Array = ReplayLogScript.read_all(path).get("records", [])
+	assert_eq(records.size(), 1, "the intact record survives a truncated tail")
+	assert_eq(records[0]["payload"], PackedByteArray([1, 2, 3, 4]), "and is still correct")
+	DirAccess.remove_absolute(ProjectSettings.globalize_path(path))
+
+
+func test_writing_to_an_unopened_log_is_a_no_op() -> void:
+	# --replay-log is optional, so every record_* call happens behind a null
+	# check in production — but the class must not corrupt or crash if that
+	# check is ever missed.
+	var log_writer = ReplayLogScript.new()
+	assert_true(not log_writer.is_open(), "starts closed")
+	log_writer.record_input(1, 1, PackedByteArray([1]))
+	log_writer.record_snapshot(1, PackedByteArray([1]))
+	assert_eq(log_writer.records_written, 0, "nothing was recorded")
+	log_writer.close()

Game/tests/cases/test_replay_log.gd.uid

@@ -0,0 +1 @@
+uid://b2e71m5byxbiy

Game/tests/networked_match_smoke.gd

@@ -42,6 +42,14 @@ func _ready() -> void:
 			_warmup_seconds = maxf(0.0, arg.get_slice("=", 1).to_float())
 
 	match _role:
+		"host-disconnect":
+			var derr := NetworkManager.host(PORT)
+			if derr != OK:
+				print("SMOKE FAIL: host() failed: %s" % error_string(derr))
+				get_tree().quit(1)
+				return
+			print("SMOKE: hosting (disconnect/reconnect scenario) on port %d ..." % PORT)
+			MatchNet.player_joined.connect(_on_disconnect_host_player_joined)
 		"host":
 			var err := NetworkManager.host(PORT)
 			if err != OK:
@@ -109,6 +117,15 @@ func _on_client_welcomed() -> void:
 	hooks.run_client_check.call_deferred(_settle_seconds, _drive_seconds, _exercise_ball_contact, _exercise_free_flight, _warmup_seconds, _exercise_input_transitions, _exercise_match_state)
 
 
+func _on_disconnect_host_player_joined(_peer_id: int, _name: String) -> void:
+	MatchNet.player_joined.disconnect(_on_disconnect_host_player_joined)
+	print("SMOKE: host loading networked_match.tscn (disconnect scenario) ...")
+	get_tree().change_scene_to_file.call_deferred("res://scenes/networked_match.tscn")
+	var hooks := preload("res://tests/networked_match_test_hooks.gd").new()
+	get_tree().root.add_child.call_deferred(hooks)
+	hooks.run_disconnect_host_check.call_deferred(_drive_seconds)
+
+
 func _on_abuser_welcomed() -> void:
 	MatchNet.welcomed.disconnect(_on_abuser_welcomed)
 	var hooks := preload("res://tests/networked_match_test_hooks.gd").new()

Game/tests/networked_match_test_hooks.gd

@@ -159,7 +159,7 @@ func run_client_check(settle_seconds: float, drive_seconds: float, exercise_ball
 		# ball every time (0 contacts in 3/3 runs). Closing the loop on the
 		# actual bearing keeps this exercising the real input path while being
 		# indifferent to how the kickoff happened to orient the ship.
-		await _drive_at_ball(my_slot.ship, match_scene.ball, 3.0)
+		await _drive_at_ball(my_slot.ship, match_scene.ball, 8.0)
 		# Leave a >150ms observation window before the normal drive so a
 		# subsequent goal reset cannot mask blend-back.
 		Input.action_release("move_forward")
@@ -514,9 +514,14 @@ func _drive_at_ball(ship: Ship, ball_body: Node3D, timeout_seconds: float) -> vo
 		Input.action_release("turn_right")
 		if absf(yaw_error) > ALIGNED_RADIANS:
 			Input.action_press("turn_right" if yaw_error > 0.0 else "turn_left")
-			Input.action_release("move_forward")
-		else:
+		# Thrust whenever the ball is anywhere ahead, not only once perfectly
+		# aligned. Cutting thrust to turn made the ship hover and burn the
+		# window without closing distance, which is why this reached the ball
+		# only 2 runs in 3; turning under power converges much faster.
+		if absf(yaw_error) < PI * 0.5:
 			Input.action_press("move_forward")
+		else:
+			Input.action_release("move_forward")
 		# Vertical alignment matters too — the ball sits above the floor and a
 		# ship that is climbing sails straight over it.
 		Input.action_release("move_up")
@@ -560,6 +565,73 @@ func _run_free_flight_trace(ship: Ship, start_position: Vector3, duration_second
 # honest encoder — this IS what a hostile custom client sending raw ENet
 # packets would look like, so bypassing the normal send path is the point,
 # not a shortcut.
+# §6.4 (tasks 5.6/5.7), host side. Watches its own slots across a client's
+# disconnect and reconnect and asserts the documented contract: the ship is
+# never despawned, the controller is swapped rather than left dangling, the
+# slot is reserved by identity, and a returning player gets it back.
+func run_disconnect_host_check(lifetime_seconds: float) -> void:
+	await get_tree().create_timer(2.0).timeout
+	var match_scene := get_tree().current_scene
+	if not _is_networked_match(match_scene):
+		print("SMOKE FAIL: host scene is not NetworkedMatch")
+		get_tree().quit(1)
+		return
+
+	var slots_before: int = match_scene._slots.size()
+	if slots_before == 0:
+		print("SMOKE FAIL: host has no slots — the client never made it into the roster")
+		NetworkManager.shutdown()
+		get_tree().quit(1)
+		return
+	var ship_before = match_scene._slots[0].ship
+	var name_before: String = match_scene._slots[0].player_name
+	print("SMOKE INFO: host has %d slot(s), player_name=%s" % [slots_before, name_before])
+
+	# Wait for the client to drop. Guarded on the scene still existing: §6.4's
+	# abort can tear the match down underneath this loop.
+	var drop_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0)
+	while Time.get_ticks_msec() < drop_deadline and _is_networked_match(match_scene) and not match_scene._slots[0].disconnected:
+		await get_tree().physics_frame
+	if not _is_networked_match(match_scene):
+		print("SMOKE FAIL: match aborted during the disconnect window — the reservation should have held it open")
+		NetworkManager.shutdown()
+		get_tree().quit(1)
+		return
+	var saw_disconnect: bool = match_scene._slots[0].disconnected
+	var ship_survived: bool = match_scene._slots.size() == slots_before and is_instance_valid(match_scene._slots[0].ship) and match_scene._slots[0].ship == ship_before
+	var controller_valid: bool = is_instance_valid(match_scene._slots[0].controller)
+	var reserved: bool = match_scene._slots[0].reserved_until_tick > Engine.get_physics_frames()
+	print("SMOKE INFO: after disconnect saw_disconnect=%s ship_survived=%s controller_valid=%s reserved=%s" % [
+		str(saw_disconnect), str(ship_survived), str(controller_valid), str(reserved)
+	])
+
+	# Then for it to come back and reclaim the slot.
+	var back_deadline := Time.get_ticks_msec() + int(lifetime_seconds * 1000.0)
+	while Time.get_ticks_msec() < back_deadline and _is_networked_match(match_scene) and match_scene._slots[0].disconnected:
+		await get_tree().physics_frame
+	if not _is_networked_match(match_scene):
+		print("SMOKE FAIL: match aborted before the player could reconnect")
+		NetworkManager.shutdown()
+		get_tree().quit(1)
+		return
+	var reclaimed: bool = not match_scene._slots[0].disconnected
+	var same_ship: bool = is_instance_valid(match_scene._slots[0].ship) and match_scene._slots[0].ship == ship_before
+	# Ticking on past the swap proves task 5.7: _physics_process writes
+	# slot.controller.action every tick, so a dangling reference from
+	# set_controller()'s queue_free() would have crashed by now.
+	for i in 60:
+		if not _is_networked_match(match_scene):
+			break
+		await get_tree().physics_frame
+
+	var success := saw_disconnect and ship_survived and controller_valid and reserved and reclaimed and same_ship and is_instance_valid(match_scene._slots[0].controller)
+	print("SMOKE %s: disconnect kept the ship and the reconnect reclaimed the slot (disconnect=%s ship_kept=%s reserved=%s reclaimed=%s same_ship=%s)" % [
+		"PASS" if success else "FAIL", str(saw_disconnect), str(ship_survived), str(reserved), str(reclaimed), str(same_ship)
+	])
+	NetworkManager.shutdown()
+	get_tree().quit(0 if success else 1)
+
+
 func run_malformed_abuse_check() -> void:
 	await get_tree().create_timer(1.0).timeout
 	# A single-element Array, not a plain bool: GDScript lambdas capture