feat(multiplayer): task 2.8 net_sim.gd, close out Phase 2

New NetSim autoload: seeded, CLI-driven (--net-sim-latency/-jitter/-loss/-dup)
latency/jitter/loss/duplicate decorator, a true no-op passthrough unless a
flag is set. Wraps MatchSim.send_input/send_snapshot per the design doc's
scope, plus NetworkManager's ping/pong so the already-tested RTT/clock
measurement becomes the acceptance signal for "raises observed RTT" without
waiting on Phase 3's per-peer snapshot echo.

Two real bugs found while building and verifying this against Phase 2's own
milestone gate (a real match under --net-sim-latency 80 --net-sim-jitter
20, not just LAN): a timestamp captured inside a delayed RPC closure
silently ate that side's own added delay out of the round-trip
measurement instead of adding to it; and a delayed send whose target
disconnected (or whose own process had already shut down) during the hold
threw RPC errors, since the existing get_peers() filtering only checked
validity at schedule time. Fixed by capturing timestamps before handing
off to NetSim, and by having NetSim re-validate the target at fire time.

Phase 2's milestone gate now passes for real: a full 1v1 under simulated
80ms latency / 20ms jitter still shows clean server-authoritative
movement and zero RPC errors. Full Phase 1 + Phase 2 regression suite
re-verified clean with NetSim present but inactive.

Game/project.godot

@@ -44,6 +44,7 @@ GameSettings="*res://scripts/game_settings.gd"
 VideoSettings="*res://scripts/video_settings.gd"
 BackgroundFPS="*res://scripts/background_fps.gd"
 PerfOverlay="*res://scripts/perf_overlay.gd"
+NetSim="*res://scripts/net_sim.gd"
 NetworkManager="*res://scripts/network_manager.gd"
 MatchNet="*res://scripts/match_net.gd"
 MatchSim="*res://scripts/match_sim.gd"

Game/scripts/match_sim.gd

@@ -48,11 +48,13 @@ func request_match_config() -> void:
 
 
 func send_input(bytes: PackedByteArray) -> void:
-	_recv_input.rpc_id(1, bytes)
+	# bytes is already fully packed (any timestamps it carries are already
+	# fixed), so wrapping the dispatch itself is enough — task 2.8.
+	NetSim.send(func() -> void: _recv_input.rpc_id(1, bytes), 1)
 
 
 func send_snapshot(peer_id: int, bytes: PackedByteArray) -> void:
-	_snapshot.rpc_id(peer_id, bytes)
+	NetSim.send(func() -> void: _snapshot.rpc_id(peer_id, bytes), peer_id)
 
 
 func send_score_update(score: Dictionary) -> void:

Game/scripts/net_sim.gd

@@ -0,0 +1,116 @@
+extends Node
+
+# Autoload (project.godot [autoload] NetSim). Debug-only, seeded
+# latency/jitter/loss/duplicate decorator around outgoing RPC dispatch —
+# task 2.8. A pure passthrough (send() calls dispatch.call() immediately)
+# unless CLI flags are given, so every existing test and the real game are
+# byte-for-byte unaffected by this autoload merely existing.
+#
+# CLI (read once, in this process's own OS.get_cmdline_user_args()):
+#   --net-sim-latency=<ms>  one-way delay added before each wrapped send
+#   --net-sim-jitter=<ms>   extra uniform-random 0..jitter added per send
+#   --net-sim-loss=<0..1>   fraction of sends dropped entirely (never sent)
+#   --net-sim-dup=<0..1>    probability a send is ALSO sent a second time
+#   --net-sim-seed=<int>    RNG seed (default fixed, so a bad run reproduces
+#                            unless a CI/local run deliberately wants a
+#                            different one — same "seeded so failures
+#                            reproduce" bar as §11's testing section sets)
+#
+# "Asymmetric-capable" per §7 task 2.8 is not a separate feature: each
+# process reads only its own CLI args and only delays its own outgoing
+# sends, so running the host and client with different flags (e.g. a
+# lossy-upload client against a clean host) already produces asymmetric
+# behaviour with no extra plumbing.
+#
+# Call sites build a zero-argument Callable that performs the actual
+# rpc_id()/rpc() dispatch, so NetSim never needs to know per-call argument
+# shapes. IMPORTANT for callers that embed a timestamp in the call (e.g.
+# NetworkManager's _ping/_pong): capture Time.get_ticks_msec() *before*
+# calling send(), not inside the wrapped Callable — the delay is meant to
+# simulate wire transit *after* the packet is "sent", so a timestamp taken
+# inside the delayed closure would silently absorb this process's own
+# outbound leg out of any round-trip measurement built on top of it.
+#
+# Wraps MatchSim.send_input / send_snapshot per the doc's task 2.8 scope,
+# plus NetworkManager's _ping/_pong dispatch — the latter is a deliberate
+# addition beyond the literal task text: it's the only RTT measurement that
+# already exists and is already tested (tests/clock_smoke.gd, task 1.8), so
+# routing it through NetSim is what makes "`--net-sim-latency 80` measurably
+# raises observed RTT" (this task's own stated acceptance criterion)
+# checkable today, without waiting on Phase 3's per-peer snapshot echo.
+
+const DEFAULT_SEED := 20260820
+
+var latency_ms := 0.0
+var jitter_ms := 0.0
+var loss_fraction := 0.0
+var dup_fraction := 0.0
+
+var _rng := RandomNumberGenerator.new()   # owned instance — never the global RNG, task 0.7's rule
+
+
+func _ready() -> void:
+	var seed_value := DEFAULT_SEED
+	for arg: String in OS.get_cmdline_user_args():
+		if arg.begins_with("--net-sim-latency="):
+			latency_ms = maxf(0.0, arg.get_slice("=", 1).to_float())
+		elif arg.begins_with("--net-sim-jitter="):
+			jitter_ms = maxf(0.0, arg.get_slice("=", 1).to_float())
+		elif arg.begins_with("--net-sim-loss="):
+			loss_fraction = clampf(arg.get_slice("=", 1).to_float(), 0.0, 1.0)
+		elif arg.begins_with("--net-sim-dup="):
+			dup_fraction = clampf(arg.get_slice("=", 1).to_float(), 0.0, 1.0)
+		elif arg.begins_with("--net-sim-seed="):
+			seed_value = arg.get_slice("=", 1).to_int()
+	_rng.seed = seed_value
+
+
+func is_active() -> bool:
+	return latency_ms > 0.0 or jitter_ms > 0.0 or loss_fraction > 0.0 or dup_fraction > 0.0
+
+
+# target_peer_id: the specific remote peer this dispatch is addressed to
+# (rpc_id's target), or -1 for a broadcast / not a targeted send. Only used
+# to re-validate a delayed send right before it actually fires — see _fire.
+func send(dispatch: Callable, target_peer_id: int = -1) -> void:
+	if not is_active():
+		dispatch.call()
+		return
+	if _rng.randf() < loss_fraction:
+		return
+	_schedule(dispatch, target_peer_id, (latency_ms + _rng.randf() * jitter_ms) / 1000.0)
+	if _rng.randf() < dup_fraction:
+		_schedule(dispatch, target_peer_id, (latency_ms + _rng.randf() * jitter_ms) / 1000.0)
+
+
+func _schedule(dispatch: Callable, target_peer_id: int, delay_sec: float) -> void:
+	if delay_sec <= 0.0:
+		dispatch.call()
+		return
+	get_tree().create_timer(delay_sec, false).timeout.connect(func() -> void: _fire(dispatch, target_peer_id))
+
+
+# Re-validates the target right before a DELAYED send actually fires.
+# NetSim's whole point is to hold a packet in flight past the moment it was
+# queued, and in that window the target peer (or this process's own
+# connection) can legitimately be gone — a disconnect mid-match, or this
+# process's own shutdown() already having reset multiplayer_peer to a fresh
+# OfflineMultiplayerPeer. Firing anyway reproduced two real bugs while
+# building this task: "Attempt to call RPC with unknown peer ID" (stale
+# remote target — networked_match.gd's own get_peers() filter on
+# _broadcast_snapshot only checked validity at *schedule* time, and the
+# target had disconnected by the time the delayed send actually fired) and
+# "'_recv_input' on yourself is not allowed by selected mode" (this
+# process's own peer was already torn down, so peer id 1 now refers to
+# itself instead of the server). The synchronous (delay_sec <= 0 / NetSim
+# inactive) path is deliberately NOT re-validated here — nothing has had
+# time to change since the caller's own validation, and matching the
+# pre-NetSim behaviour exactly there is what keeps NetSim a true no-op when
+# no CLI flags are given.
+func _fire(dispatch: Callable, target_peer_id: int) -> void:
+	var peer := multiplayer.multiplayer_peer
+	if peer == null or peer is OfflineMultiplayerPeer:
+		return
+	if target_peer_id != -1 and target_peer_id not in multiplayer.get_peers():
+		return
+	dispatch.call()

Game/scripts/network_manager.gd

@@ -93,7 +93,10 @@ func _process(delta: float) -> void:
 	_ping_accum_sec += delta
 	if _ping_accum_sec >= PING_INTERVAL_SEC:
 		_ping_accum_sec = 0.0
-		_ping.rpc_id(1, Time.get_ticks_msec())
+		# Capture the timestamp now, before NetSim (task 2.8) can add any
+		# simulated delay — see net_sim.gd's header comment for why.
+		var send_ms := Time.get_ticks_msec()
+		NetSim.send(func() -> void: _ping.rpc_id(1, send_ms), 1)
 
 
 # Estimate of what the server's Time.get_ticks_msec() reads right now.
@@ -165,7 +168,13 @@ func shutdown() -> void:
 func _ping(client_send_ms: int) -> void:
 	if not multiplayer.is_server():
 		return
-	_pong.rpc_id(multiplayer.get_remote_sender_id(), client_send_ms, Time.get_ticks_msec())
+	# Same rule as the client's send above: read the server's clock now, at
+	# true receipt time, before NetSim can delay the reply — otherwise the
+	# server's own outbound leg would be silently absorbed out of both the
+	# RTT sample and the offset estimate instead of adding to them.
+	var server_now := Time.get_ticks_msec()
+	var sender_id := multiplayer.get_remote_sender_id()
+	NetSim.send(func() -> void: _pong.rpc_id(sender_id, client_send_ms, server_now), sender_id)
 
 
 @rpc("authority", "call_remote", "reliable")

Game/tests/net_sim_smoke.gd

@@ -0,0 +1,111 @@
+extends Node
+
+# Manual two-process smoke test for NetSim (task 2.8 acceptance:
+# "--net-sim-latency 80 measurably raises observed RTT"). Deliberately not
+# part of the pure-function suite — needs two real processes and real wall
+# time to observe a delayed pong.
+#
+# NetSim reads its own --net-sim-* flags directly from OS.get_cmdline_user_args()
+# (see net_sim.gd) — this driver only needs --role= and passes any
+# --net-sim-* flags straight through untouched. The host is where the
+# _pong reply gets delayed, so --net-sim-latency=/--net-sim-loss= belong on
+# the HOST invocation; the client just observes NetworkManager.rtt_ms.
+#
+# Usage:
+#   godot --headless --path Game res://tests/net_sim_smoke.tscn -- --role=host --net-sim-latency=80
+#   godot --headless --path Game res://tests/net_sim_smoke.tscn -- --role=client --min-rtt=70
+#
+# For the loss scenario:
+#   godot --headless --path Game res://tests/net_sim_smoke.tscn -- --role=host --net-sim-loss=1.0
+#   godot --headless --path Game res://tests/net_sim_smoke.tscn -- --role=client-loss
+
+const DEFAULT_PORT := 7810
+const TIMEOUT_SECONDS := 12.0
+const HOST_LIFETIME_SECONDS := 8.0
+# Loose ceiling, not a tight bound: real localhost jitter plus one full
+# PING_INTERVAL_SEC of scheduling slack is possible before the first sample
+# lands, so this only needs to catch a badly broken (e.g. no-op) NetSim.
+const MAX_RTT_SLACK_MS := 400.0
+
+var _role := ""
+var _port := DEFAULT_PORT
+var _min_rtt_ms := 0.0
+var _finished := false
+
+
+func _ready() -> void:
+	for arg: String in OS.get_cmdline_user_args():
+		if arg.begins_with("--role="):
+			_role = arg.substr("--role=".length())
+		elif arg.begins_with("--port="):
+			_port = int(arg.substr("--port=".length()))
+		elif arg.begins_with("--min-rtt="):
+			_min_rtt_ms = arg.substr("--min-rtt=".length()).to_float()
+
+	if _role == "host":
+		var err := NetworkManager.host(_port)
+		if err != OK:
+			_finish(false, "host() failed: %s" % error_string(err))
+			return
+		print("SMOKE: hosting on port %d (net-sim latency=%.1fms jitter=%.1fms loss=%.2f)" % [
+			_port, NetSim.latency_ms, NetSim.jitter_ms, NetSim.loss_fraction
+		])
+		get_tree().create_timer(HOST_LIFETIME_SECONDS).timeout.connect(func() -> void:
+			_finish(true, "host ran for %.1fs" % HOST_LIFETIME_SECONDS))
+	elif _role == "client":
+		NetworkManager.clock_updated.connect(_on_clock_updated)
+		var err := NetworkManager.join("127.0.0.1", _port)
+		if err != OK:
+			_finish(false, "join() failed: %s" % error_string(err))
+			return
+		print("SMOKE: joining 127.0.0.1:%d, expecting rtt >= %.1fms ..." % [_port, _min_rtt_ms])
+	elif _role == "client-loss":
+		NetworkManager.clock_updated.connect(_on_unexpected_clock_updated)
+		var err := NetworkManager.join("127.0.0.1", _port)
+		if err != OK:
+			_finish(false, "join() failed: %s" % error_string(err))
+			return
+		print("SMOKE: joining 127.0.0.1:%d, expecting NO rtt sample (100%% loss) ..." % _port)
+		get_tree().create_timer(HOST_LIFETIME_SECONDS - 1.0).timeout.connect(func() -> void:
+			_finish(NetworkManager.rtt_ms < 0.0, "rtt_ms=%.1f after %.1fs (expected -1, no pong ever arrived)" % [
+				NetworkManager.rtt_ms, HOST_LIFETIME_SECONDS - 1.0
+			]))
+	else:
+		_finish(false, "missing or unrecognised --role= (expected host|client|client-loss)")
+		return
+
+	get_tree().create_timer(TIMEOUT_SECONDS).timeout.connect(_on_timeout)
+
+
+func _process(_delta: float) -> void:
+	NetworkManager.poll()
+
+
+func _physics_process(_delta: float) -> void:
+	NetworkManager.poll()
+
+
+func _on_clock_updated(rtt_ms: float, _offset_ms: float) -> void:
+	var ceiling := _min_rtt_ms * 4.0 + MAX_RTT_SLACK_MS
+	var ok := rtt_ms >= _min_rtt_ms and rtt_ms <= ceiling
+	print("SMOKE INFO: observed rtt_ms=%.2f (want >= %.1f, <= %.1f)" % [rtt_ms, _min_rtt_ms, ceiling])
+	_finish(ok, "client observed rtt_ms=%.2f against min=%.1f" % [rtt_ms, _min_rtt_ms])
+
+
+func _on_unexpected_clock_updated(rtt_ms: float, _offset_ms: float) -> void:
+	_finish(false, "client received a pong (rtt_ms=%.2f) despite --net-sim-loss=1.0 on the host" % rtt_ms)
+
+
+func _on_timeout() -> void:
+	if not _finished:
+		_finish(false, "timed out waiting for a clock sample")
+
+
+func _finish(success: bool, message: String) -> void:
+	if _finished:
+		return
+	_finished = true
+	print("SMOKE %s: %s" % ["PASS" if success else "FAIL", message])
+	await get_tree().create_timer(0.3).timeout
+	NetworkManager.shutdown()
+	get_tree().quit(0 if success else 1)

Game/tests/net_sim_smoke.tscn

@@ -0,0 +1,6 @@
+[gd_scene load_steps=2 format=3]
+
+[ext_resource type="Script" path="res://tests/net_sim_smoke.gd" id="1_nss"]
+
+[node name="NetSimSmoke" type="Node"]
+script = ExtResource("1_nss")