feat(ci): protocol version handshake client + IPC benchmark (#556, #342)

#556: HANDSHAKING state in sim_bridge.gd — decodes first framed message
as HandshakeMessage, validates vs Protocol.PROTOCOL_VERSION, 5s timeout,
handshake_complete/handshake_failed signals.

#342: IPC benchmark now reads and validates HandshakeMessage before
starting the timing loop.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

client/scripts/autoloads/sim_bridge.gd

@@ -1,7 +1,7 @@
 extends Node
 
 # Connection states
-enum ConnectionState { DISCONNECTED, CONNECTING, CONNECTED, ERROR }
+enum ConnectionState { DISCONNECTED, CONNECTING, HANDSHAKING, CONNECTED, ERROR }
 
 var state: ConnectionState = ConnectionState.DISCONNECTED
 var test_mode: bool = OS.get_environment("SR_LIVE") != "1"  # SR_LIVE=1 connects to real server
@@ -27,9 +27,15 @@ const CONNECT_RETRY_INTERVAL: float = 0.1  # Seconds between retry attempts
 var _connect_retries: int = 0
 var _retry_timer: float = 0.0
 
+# Handshake state (#556)
+const HANDSHAKE_TIMEOUT_USEC: int = 5_000_000  # 5 seconds
+var _handshake_start_usec: int = 0
+
 # Signals
 signal connection_state_changed(old_state: ConnectionState, new_state: ConnectionState)
 signal snapshot_received(snapshot: Dictionary)
+signal handshake_complete(protocol_version: int)
+signal handshake_failed(reason: String)
 
 func _ready() -> void:
 	if test_mode:
@@ -121,7 +127,8 @@ func _process(delta: float) -> void:
 		_bridge.poll()
 		match _bridge.get_status():
 			StreamPeerTCP.STATUS_CONNECTED:
-				_set_state(ConnectionState.CONNECTED)
+				_handshake_start_usec = Time.get_ticks_usec()
+				_set_state(ConnectionState.HANDSHAKING)
 			StreamPeerTCP.STATUS_CONNECTING:
 				pass  # Still connecting, wait
 			StreamPeerTCP.STATUS_ERROR:
@@ -134,6 +141,62 @@ func _process(delta: float) -> void:
 				_bridge = null  # Reset and retry
 		return
 
+	# HANDSHAKING state: read first framed message, validate HandshakeMessage (#556)
+	if state == ConnectionState.HANDSHAKING:
+		if _bridge == null:
+			_set_state(ConnectionState.ERROR)
+			return
+		_bridge.poll()
+
+		# Check connection dropped during handshake
+		var bridge_status := _bridge.get_status()
+		if bridge_status == StreamPeerTCP.STATUS_ERROR or bridge_status == StreamPeerTCP.STATUS_NONE:
+			var reason := "Connection dropped during handshake"
+			push_error("SimBridge: %s" % reason)
+			handshake_failed.emit(reason)
+			_bridge = null
+			_set_state(ConnectionState.ERROR)
+			return
+
+		# Check timeout
+		if Time.get_ticks_usec() - _handshake_start_usec > HANDSHAKE_TIMEOUT_USEC:
+			var reason := "Handshake timeout: no message received within 5 seconds"
+			push_error("SimBridge: %s" % reason)
+			handshake_failed.emit(reason)
+			_bridge.disconnect_from_server()
+			_set_state(ConnectionState.ERROR)
+			return
+
+		# Try to read first message
+		var msg := _bridge.poll_message()
+		if msg.is_empty():
+			return  # Not ready yet, continue polling
+
+		# Decode HandshakeMessage: { "protocol_version": N }
+		var decoded: Variant = Messagepack.decode(msg)
+		if decoded.status != null or not (decoded.value is Dictionary) \
+				or not decoded.value.has("protocol_version"):
+			var reason := "Handshake decode failed: malformed HandshakeMessage"
+			push_error("SimBridge: %s" % reason)
+			handshake_failed.emit(reason)
+			_bridge.disconnect_from_server()
+			_set_state(ConnectionState.ERROR)
+			return
+
+		var server_version: int = decoded.value["protocol_version"]
+		if server_version != Protocol.PROTOCOL_VERSION:
+			var reason := "Protocol version mismatch: server=%d, client=%d" % [
+				server_version, Protocol.PROTOCOL_VERSION]
+			push_error("SimBridge: %s" % reason)
+			handshake_failed.emit(reason)
+			_bridge.disconnect_from_server()
+			_set_state(ConnectionState.ERROR)
+			return
+
+		handshake_complete.emit(server_version)
+		_set_state(ConnectionState.CONNECTED)
+		return
+
 	if _bridge == null:
 		return
 

server/tests/ipc_bench.rs

@@ -10,9 +10,6 @@
 //!
 //! Output: IPC_BENCH_RESULT:{json} on a single line for tooling to parse.
 //!
-//! BLOCKED (#342): Handshake step is stubbed pending #555 (server) + #556 (client).
-//! The test currently skips the HandshakeMessage exchange and starts timing
-//! immediately after TCP connection is established.
 //!
 //! Spec references: D-020 (subprocess IPC, 5ms budget), D-030 (Layer 3)
 
@@ -97,13 +94,20 @@ fn ipc_round_trip_latency() {
     let mut reader = BufReader::new(stream.try_clone().expect("clone stream"));
     let mut writer = BufWriter::new(stream);
 
-    // TODO (#342, #555/#556): Wait for HandshakeMessage here before starting timing.
-    // When the server sends HandshakeMessage { protocol_version: 14 } as the first
-    // framed message, read and validate it. If version != PROTOCOL_VERSION, abort.
-    // The timing loop below starts after a successful handshake.
-    //
-    // For now, connect and proceed directly — the timing loop handles whatever
-    // the server sends as its first message.
+    // 4. Handshake: read and validate HandshakeMessage before timing (#555/#556).
+    // Server sends HandshakeMessage { protocol_version } as the very first framed message.
+    let handshake_bytes = read_framed(&mut reader)
+        .expect("read handshake")
+        .expect("server closed before sending HandshakeMessage");
+    let handshake: HandshakeMessage =
+        rmp_serde::from_slice(&handshake_bytes).expect("deserialize HandshakeMessage");
+    assert_eq!(
+        handshake.protocol_version,
+        PROTOCOL_VERSION,
+        "handshake version mismatch: server={}, client={}",
+        handshake.protocol_version,
+        PROTOCOL_VERSION
+    );
 
     let make_input = |tick: u64| PlayerInput {
         tick,
@@ -112,7 +116,7 @@ fn ipc_round_trip_latency() {
 
     let mut round_trip_ms: Vec<f64> = Vec::with_capacity(WARMUP_ROUNDS + MEASURE_ROUNDS);
 
-    // 4. Warmup rounds (not timed)
+    // 5. Warmup rounds (not timed)
     for tick in 0..WARMUP_ROUNDS as u64 {
         let payload =
             rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput");
@@ -122,7 +126,7 @@ fn ipc_round_trip_latency() {
             .expect("server closed during warmup");
     }
 
-    // 5. Timed measurement rounds
+    // 6. Timed measurement rounds
     for tick in WARMUP_ROUNDS as u64..(WARMUP_ROUNDS + MEASURE_ROUNDS) as u64 {
         let payload =
             rmp_serde::to_vec_named(&vec![make_input(tick)]).expect("serialize PlayerInput");
@@ -141,7 +145,7 @@ fn ipc_round_trip_latency() {
         round_trip_ms.push(elapsed_ms);
     }
 
-    // 6. Clean up
+    // 7. Clean up
     drop(reader);
     drop(writer);
     let exit_deadline = Instant::now() + Duration::from_secs(5);
@@ -163,7 +167,7 @@ fn ipc_round_trip_latency() {
         }
     }
 
-    // 7. Compute percentiles
+    // 8. Compute percentiles
     let mut sorted = round_trip_ms.clone();
     sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());