feat(bench): ka/close A/B matrix — plain_serve baseline + box orchestrator

The jar's ka-vs-close throughput inversion was observed on the box but its
run configuration died with the job workspaces, so it can't be re-derived —
this commit makes the comparison a committed, controlled experiment instead
of a lost one-off.

- examples/plain_serve: the load_profile request path with zero causal
  machinery (same route/handler/Config), serving until killed. The plain
  half of the A/B; throughput measured externally by wrk.
- scripts/ka-close-matrix.sh: {ka,close} x {plain,causal} on fresh ports,
  byte-identical wrk args per mode pair except the Connection: close
  header, disjoint-core pinning for server vs wrk, per-cell curl
  verification of the negotiated connection behavior, ss/env capture, and
  a parsed summary with a close/ka ratio verdict. The causal cells double
  as the forgiveness-fix box validation (v13 pending item, pull-forward
  agreed): key on forgiven staying ~ injected x parked-actors in
  close-mode churn, not process-lifetime phantoms.

examples/plain_serve.rs

@@ -0,0 +1,35 @@
+//! Plain benchmark server: the `load_profile` request path with zero causal
+//! machinery — the baseline half of the ka/close A/B matrix (the
+//! throughput-inversion chase).
+//!
+//! Identical route, handler, and `Config` construction to `load_profile`;
+//! differs only in having no `smarm-causal` feature, no sweep, and no
+//! shutdown path — it serves until killed. Throughput is measured
+//! externally (wrk).
+//!
+//! Env:
+//!   URUS_PORT   listen port (default 8080; binds 0.0.0.0)
+//!
+//!   cargo run --release --example plain_serve
+
+use std::net::SocketAddr;
+
+use urus::{serve_with, Config, Conn, Next, Pipeline, Router};
+
+/// Mirrors `load_profile`'s handler byte for byte: param parse + JSON render.
+fn json_id(conn: Conn, _next: Next) -> Conn {
+    let id: u64 = conn.params.get("id").and_then(|s| s.parse().ok()).unwrap_or(0);
+    conn.put_status(200)
+        .put_header("content-type", "application/json")
+        .put_body(format!("{{\"id\":{id}}}"))
+}
+
+fn main() {
+    let port: u16 = std::env::var("URUS_PORT")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(8080);
+    let addr: SocketAddr = format!("0.0.0.0:{port}").parse().expect("addr");
+    let pipe = Pipeline::new().plug(Router::new().get("/json/:id", json_id));
+    serve_with(Config::new(addr), pipe).expect("serve");
+}

scripts/ka-close-matrix.sh

@@ -0,0 +1,179 @@
+#!/usr/bin/env bash
+# ka-close-matrix.sh — controlled ka/close A/B for the throughput-inversion
+# chase (handoff jar item), with the forgiveness-fix box validation folded
+# into the causal cells (handoff v13 PENDING, pull-forward agreed 2026-07-20).
+#
+# Cells, run in order on fresh ports:
+#   plain-ka, plain-close     examples/plain_serve (no causal feature).
+#                             Metric: wrk Requests/sec, REPS reps after a
+#                             discarded warmup.
+#   causal-ka, causal-close   examples/load_profile (smarm-causal). Metric:
+#                             the sweep summary + ledger audit (forgiveness
+#                             column, books balance); wrk is backdrop load
+#                             whose own numbers are injection-contaminated.
+#
+# Controls: byte-identical wrk invocation per mode pair except the
+# `Connection: close` header; server and wrk pinned to disjoint core sets
+# (SMT siblings left idle by default on the 5900X); the negotiated
+# connection behavior is verified via curl per cell and logged, so a
+# loadgen-config asymmetry can never silently explain a result again.
+#
+# Knobs (env, defaults for the 5900X box):
+#   DUR=30 REPS=2 CONNS=64 THREADS=4 PIN=1
+#   SERVER_CPUS=0-7 WRK_CPUS=8-11
+#   CAUSAL_WRK_DUR=300 BASE_PORT=8080
+#   OUT=/workspace/results
+#   PLAIN_BIN, CAUSAL_BIN     binary paths (default: target/release/examples/*)
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+DUR="${DUR:-30}"
+REPS="${REPS:-2}"
+CONNS="${CONNS:-64}"
+THREADS="${THREADS:-4}"
+PIN="${PIN:-1}"
+SERVER_CPUS="${SERVER_CPUS:-0-7}"
+WRK_CPUS="${WRK_CPUS:-8-11}"
+CAUSAL_WRK_DUR="${CAUSAL_WRK_DUR:-300}"
+BASE_PORT="${BASE_PORT:-8080}"
+OUT="${OUT:-/workspace/results}"
+PLAIN_BIN="${PLAIN_BIN:-target/release/examples/plain_serve}"
+CAUSAL_BIN="${CAUSAL_BIN:-target/release/examples/load_profile}"
+
+mkdir -p "$OUT"
+say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$OUT/matrix.log"; }
+
+pin_server=(); pin_wrk=()
+if [ "$PIN" = 1 ]; then
+  pin_server=(taskset -c "$SERVER_CPUS")
+  pin_wrk=(taskset -c "$WRK_CPUS")
+fi
+
+# ---- environment record --------------------------------------------------
+{
+  date -u
+  uname -a
+  echo "nproc: $(nproc)"
+  lscpu -e 2>/dev/null || true
+  echo "port range: $(cat /proc/sys/net/ipv4/ip_local_port_range 2>/dev/null)"
+  echo "somaxconn:  $(cat /proc/sys/net/core/somaxconn 2>/dev/null)"
+  echo "wrk: $(wrk --version 2>&1 | head -1 || true)"
+  echo "PIN=$PIN SERVER_CPUS=$SERVER_CPUS WRK_CPUS=$WRK_CPUS"
+  echo "DUR=$DUR REPS=$REPS CONNS=$CONNS THREADS=$THREADS CAUSAL_WRK_DUR=$CAUSAL_WRK_DUR"
+} > "$OUT/env.txt"
+say "env recorded -> env.txt"
+
+# ---- helpers -------------------------------------------------------------
+run_wrk() { # $1=mode $2=duration_s $3=port
+  if [ "$1" = close ]; then
+    "${pin_wrk[@]}" wrk -t "$THREADS" -c "$CONNS" -d "${2}s" --latency \
+      -H "Connection: close" "http://127.0.0.1:$3/json/7"
+  else
+    "${pin_wrk[@]}" wrk -t "$THREADS" -c "$CONNS" -d "${2}s" --latency \
+      "http://127.0.0.1:$3/json/7"
+  fi
+}
+
+wait_port() { # $1=port
+  for _ in $(seq 1 150); do
+    curl -s -o /dev/null "http://127.0.0.1:$1/json/1" && return 0
+    sleep 0.2
+  done
+  return 1
+}
+
+verify_mode() { # $1=mode $2=port — record what actually goes over the wire
+  echo "--- single request, mode=$1 ---"
+  if [ "$1" = close ]; then
+    curl -sv --http1.1 -H "Connection: close" -o /dev/null \
+      "http://127.0.0.1:$2/json/7" 2>&1 | grep -iE "^(> |< )(GET|HTTP|connection)" || true
+  else
+    curl -sv --http1.1 -o /dev/null "http://127.0.0.1:$2/json/7" 2>&1 \
+      | grep -iE "^(> |< )(GET|HTTP|connection)" || true
+  fi
+  echo "--- reuse probe (two requests, one curl) ---"
+  curl -sv --http1.1 -o /dev/null -o /dev/null \
+    "http://127.0.0.1:$2/json/1" "http://127.0.0.1:$2/json/2" 2>&1 \
+    | grep -icE "re-us(ed|ing)" || true
+}
+
+# ---- plain cells ---------------------------------------------------------
+run_plain() { # $1=mode $2=port
+  local mode="$1" port="$2" d="$OUT/plain-$1"
+  mkdir -p "$d"
+  say "=== plain / $mode (port $port) ==="
+  URUS_PORT="$port" "${pin_server[@]}" "$PLAIN_BIN" > "$d/server.log" 2>&1 &
+  local spid=$!
+  wait_port "$port" || { say "FATAL: plain server never came up"; cat "$d/server.log"; exit 1; }
+  verify_mode "$mode" "$port" > "$d/mode-verify.txt" 2>&1
+  ss -s > "$d/ss-before.txt" 2>/dev/null || true
+  run_wrk "$mode" 5 "$port" > "$d/warmup.txt" 2>&1
+  for r in $(seq 1 "$REPS"); do
+    run_wrk "$mode" "$DUR" "$port" > "$d/rep$r.txt" 2>&1
+    grep -E "Requests/sec|Latency |requests in|Socket errors|Non-2xx" "$d/rep$r.txt" \
+      | sed "s/^/  [plain-$mode r$r] /" | tee -a "$OUT/matrix.log" || true
+  done
+  ss -s > "$d/ss-after.txt" 2>/dev/null || true
+  awk '{printf "server cpu jiffies (utime+stime): %d\n", $14+$15}' \
+    "/proc/$spid/stat" > "$d/server-cpu.txt" 2>/dev/null || true
+  kill "$spid" 2>/dev/null || true
+  wait "$spid" 2>/dev/null || true
+}
+
+# ---- causal cells --------------------------------------------------------
+run_causal() { # $1=mode $2=port
+  local mode="$1" port="$2" d="$OUT/causal-$1"
+  mkdir -p "$d"
+  say "=== causal / $mode (port $port) ==="
+  URUS_PORT="$port" COZ_OUT="$d/profile.coz" WARMUP_MS=2000 \
+    "${pin_server[@]}" "$CAUSAL_BIN" > "$d/sweep.log" 2>&1 &
+  local spid=$!
+  wait_port "$port" || { say "FATAL: causal server never came up"; cat "$d/sweep.log"; exit 1; }
+  verify_mode "$mode" "$port" > "$d/mode-verify.txt" 2>&1
+  run_wrk "$mode" "$CAUSAL_WRK_DUR" "$port" > "$d/wrk-backdrop.txt" 2>&1 &
+  local wpid=$!
+  local rc=0
+  wait "$spid" || rc=$?
+  kill "$wpid" 2>/dev/null || true
+  wait "$wpid" 2>/dev/null || true
+  say "causal/$mode server exit=$rc (sweep + audit in sweep.log)"
+  if [ "$rc" -ne 0 ]; then say "WARNING: causal/$mode exited nonzero"; fi
+}
+
+# ---- matrix --------------------------------------------------------------
+run_plain  ka    "$BASE_PORT"
+run_plain  close "$((BASE_PORT + 1))"
+run_causal ka    "$((BASE_PORT + 2))"
+run_causal close "$((BASE_PORT + 3))"
+
+# ---- summary -------------------------------------------------------------
+say "=== SUMMARY ==="
+python3 - "$OUT" <<'PY' | tee -a "$OUT/matrix.log"
+import re, sys, pathlib, statistics
+out = pathlib.Path(sys.argv[1])
+means = {}
+for mode in ("ka", "close"):
+    vals = []
+    for rep in sorted((out / f"plain-{mode}").glob("rep*.txt")):
+        m = re.search(r"Requests/sec:\s+([\d.]+)", rep.read_text())
+        if m:
+            vals.append(float(m.group(1)))
+    if vals:
+        means[mode] = statistics.mean(vals)
+        print(f"plain-{mode}: reps={[f'{v:.0f}' for v in vals]} mean={means[mode]:.0f} req/s")
+if len(means) == 2:
+    r = means["close"] / means["ka"]
+    print(f"close/ka ratio: {r:.3f}", "-> INVERSION PRESENT (close beats ka)" if r > 1.05
+          else "-> no inversion (ka >= close)" if r < 0.95 else "-> parity")
+for mode in ("ka", "close"):
+    log = out / f"causal-{mode}" / "sweep.log"
+    if log.exists():
+        t = log.read_text()
+        keep = [l for l in t.splitlines()
+                if re.search(r"forgiv|books|audit|balance|total responses", l, re.I)]
+        print(f"-- causal-{mode} audit lines --")
+        for l in keep[:14]:
+            print("  " + l)
+PY
+say "=== MATRIX DONE ==="