feat(ci): IPC round-trip timing benchmark scaffold (#342)

Benchmark test in ipc_bench.rs: warmup + 100 rounds, p50/p95/p99 latency
reporting, 5ms threshold. Handshake step stubbed pending #555/#556.
Invoked via tests/run-ipc-benchmark.

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

server/tests/ipc_bench.rs

@@ -0,0 +1,195 @@
+//! IPC round-trip latency benchmark (#342, D-020)
+//!
+//! Measures end-to-end latency from client send (write_framed) to client receive
+//! (read_framed) over the real subprocess IPC channel. Reports p50/p95/p99.
+//!
+//! Latency budget: p99 must be <= 5ms (D-020: "~1-5ms serialization latency per tick").
+//!
+//! Run with:
+//!   cargo test --release --test ipc_bench -- --ignored --nocapture
+//!
+//! 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)
+
+use settled_reach_server::bridge::framing::{read_framed, write_framed};
+use settled_reach_server::bridge::types::*;
+use std::io::{BufRead, BufReader, BufWriter};
+use std::net::TcpStream;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+/// Number of warmup round-trips before timing begins.
+const WARMUP_ROUNDS: usize = 10;
+
+/// Number of timed round-trips (N in the spec).
+const MEASURE_ROUNDS: usize = 100;
+
+/// Latency threshold (p99 must be below this). D-020: "~1-5ms per tick".
+const THRESHOLD_MS: f64 = 5.0;
+
+/// Timeout for the server to emit LISTENING:{port} on stdout.
+const LISTEN_TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Timeout per round-trip read.
+const ROUND_TRIP_TIMEOUT: Duration = Duration::from_secs(5);
+
+fn percentile(sorted: &[f64], p: f64) -> f64 {
+    if sorted.is_empty() {
+        return 0.0;
+    }
+    let idx = ((sorted.len() - 1) as f64 * p).floor() as usize;
+    sorted[idx.min(sorted.len() - 1)]
+}
+
+#[test]
+#[ignore]
+fn ipc_round_trip_latency() {
+    // 1. Spawn server binary with --test-mode --port 0
+    let server_bin = env!("CARGO_BIN_EXE_settled-reach-server");
+    let mut child = Command::new(server_bin)
+        .args(["--test-mode", "--port", "0"])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("failed to spawn server binary");
+
+    let stdout = child.stdout.take().expect("stdout not captured");
+    let mut stdout_reader = BufReader::new(stdout);
+
+    // 2. Parse LISTENING:{port} from stdout
+    let port = {
+        let deadline = Instant::now() + LISTEN_TIMEOUT;
+        let mut line = String::new();
+        loop {
+            line.clear();
+            match stdout_reader.read_line(&mut line) {
+                Ok(0) => panic!("server stdout closed before LISTENING signal"),
+                Ok(_) => {
+                    let trimmed = line.trim();
+                    if let Some(port_str) = trimmed.strip_prefix("LISTENING:") {
+                        break port_str
+                            .parse::<u16>()
+                            .unwrap_or_else(|e| panic!("invalid port '{}': {}", port_str, e));
+                    }
+                }
+                Err(e) => panic!("failed to read server stdout: {}", e),
+            }
+            assert!(
+                Instant::now() < deadline,
+                "timed out waiting for LISTENING signal"
+            );
+        }
+    };
+
+    // 3. Connect via TCP
+    let addr = format!("127.0.0.1:{}", port);
+    let stream = TcpStream::connect(&addr)
+        .unwrap_or_else(|e| panic!("failed to connect to {}: {}", addr, e));
+    stream
+        .set_read_timeout(Some(ROUND_TRIP_TIMEOUT))
+        .expect("set read timeout");
+
+    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.
+
+    let make_input = |tick: u64| PlayerInput {
+        tick,
+        action: PlayerAction::MoveNorth,
+    };
+
+    let mut round_trip_ms: Vec<f64> = Vec::with_capacity(WARMUP_ROUNDS + MEASURE_ROUNDS);
+
+    // 4. 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");
+        write_framed(&mut writer, &payload).expect("send warmup input");
+        let _ = read_framed(&mut reader)
+            .expect("read warmup snapshot")
+            .expect("server closed during warmup");
+    }
+
+    // 5. 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");
+
+        let t_send = Instant::now();
+        write_framed(&mut writer, &payload).expect("send timed input");
+        let response = read_framed(&mut reader)
+            .expect("read timed snapshot")
+            .expect("server closed during measurement");
+        let elapsed_ms = t_send.elapsed().as_secs_f64() * 1000.0;
+
+        // Verify we received a valid snapshot (not just noise)
+        let _snapshot: ObserverSnapshot =
+            rmp_serde::from_slice(&response).expect("deserialize ObserverSnapshot");
+
+        round_trip_ms.push(elapsed_ms);
+    }
+
+    // 6. Clean up
+    drop(reader);
+    drop(writer);
+    let exit_deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        match child.try_wait() {
+            Ok(Some(_)) => break,
+            Ok(None) => {
+                if Instant::now() > exit_deadline {
+                    child.kill().ok();
+                    child.wait().ok();
+                    break;
+                }
+                std::thread::sleep(Duration::from_millis(50));
+            }
+            Err(_) => {
+                child.kill().ok();
+                break;
+            }
+        }
+    }
+
+    // 7. Compute percentiles
+    let mut sorted = round_trip_ms.clone();
+    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
+
+    let p50 = percentile(&sorted, 0.50);
+    let p95 = percentile(&sorted, 0.95);
+    let p99 = percentile(&sorted, 0.99);
+    let passed = p99 <= THRESHOLD_MS;
+
+    let result = serde_json::json!({
+        "p50_ms": (p50 * 100.0).round() / 100.0,
+        "p95_ms": (p95 * 100.0).round() / 100.0,
+        "p99_ms": (p99 * 100.0).round() / 100.0,
+        "threshold_ms": THRESHOLD_MS,
+        "passed": passed,
+        "rounds": MEASURE_ROUNDS,
+    });
+
+    println!(
+        "IPC_BENCH_RESULT:{}",
+        serde_json::to_string(&result).unwrap()
+    );
+
+    // Fail the test if we exceed the latency budget
+    assert!(
+        passed,
+        "IPC latency budget exceeded: p99={:.2}ms > threshold={}ms",
+        p99, THRESHOLD_MS
+    );
+}