fix: eliminate parallel Rust test flakes

architecture/sandbox.md

@@ -53,7 +53,10 @@ paths, such as proxy support files or GPU device paths when a GPU is present.
 
 All ordinary agent egress is routed through the sandbox proxy. The proxy
 identifies the calling binary, checks trust-on-first-use binary identity, rejects
-unsafe internal destinations, and evaluates the active policy.
+unsafe internal destinations, and evaluates the active policy. On Linux, it
+maps an accepted proxy connection back to the workload socket by matching the
+complete local-to-remote TCP tuple before resolving every process that owns the
+socket inode.
 For inspected HTTP traffic, the proxy can enforce REST method/path rules,
 WebSocket upgrade and text-message rules, GraphQL operation rules, and
 MCP method, tool, and supported params rules or generic JSON-RPC method rules

crates/openshell-cli/src/ssh.rs

@@ -18,6 +18,7 @@ use openshell_core::proto::{
 };
 use owo_colors::OwoColorize;
 use std::fs;
+use std::future::Future;
 use std::io::{IsTerminal, Write};
 #[cfg(unix)]
 use std::os::unix::process::CommandExt;
@@ -447,9 +448,24 @@ async fn wait_for_forward_start(child: &mut Child, spec: &ForwardSpec) -> Result
 /// last probe error is folded into the timeout diagnostic, so a failure reports
 /// why the listener never opened, not just that it timed out.
 async fn wait_for_forward_listener(spec: &ForwardSpec, wait_for: Duration) -> Result<()> {
+    wait_for_forward_listener_with_probe(spec, wait_for, |spec| async move {
+        probe_forward_listener(&spec).await
+    })
+    .await
+}
+
+async fn wait_for_forward_listener_with_probe<F, Fut>(
+    spec: &ForwardSpec,
+    wait_for: Duration,
+    mut probe: F,
+) -> Result<()>
+where
+    F: FnMut(ForwardSpec) -> Fut,
+    Fut: Future<Output = std::result::Result<(), String>>,
+{
     let deadline = tokio::time::Instant::now() + wait_for;
     loop {
-        let probe_error = match probe_forward_listener(spec).await {
+        let probe_error = match probe(spec.clone()).await {
             Ok(()) => return Ok(()),
             Err(err) => err,
         };
@@ -1816,17 +1832,17 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn wait_for_forward_listener_rejects_missing_listener() {
-        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
-        let port = listener.local_addr().unwrap().port();
-        drop(listener);
-        let spec = ForwardSpec::new(port);
+    async fn wait_for_forward_listener_reports_failed_probe() {
+        let spec = ForwardSpec::new(12345);
 
-        let err = wait_for_forward_listener(&spec, Duration::from_millis(20))
-            .await
-            .unwrap_err();
+        let err = wait_for_forward_listener_with_probe(&spec, Duration::ZERO, |_| {
+            std::future::ready(Err("connection refused".to_string()))
+        })
+        .await
+        .unwrap_err();
         let text = format!("{err:?}");
         assert!(text.contains("local forward listener did not open"));
+        assert!(text.contains("connection refused"));
     }
 
     #[test]

crates/openshell-gateway-interceptors/src/runtime.rs

@@ -911,7 +911,6 @@ mod tests {
             .with_ansi(false)
             .without_time();
         let subscriber = tracing_subscriber::registry().with(fmt_layer);
-        let dispatch = tracing::Dispatch::new(subscriber);
         let plan = BindingPlan {
             interceptor_name: "test".to_string(),
             binding_id: "binding".to_string(),
@@ -940,7 +939,11 @@ mod tests {
             ..InterceptorResult::default()
         };
 
-        tracing::dispatcher::with_default(&dispatch, || {
+        tracing::subscriber::with_default(subscriber, || {
+            // Other parallel tests may register this callsite while no subscriber
+            // is active. Refresh the process-wide cache after installing this
+            // thread-local subscriber so the event cannot remain disabled.
+            tracing::callsite::rebuild_interest_cache();
             emit_evaluation_log(&plan, &result, "allow", 2);
         });
 

crates/openshell-server/src/multiplex.rs

@@ -1724,16 +1724,21 @@ mod tests {
             .with_span_events(FmtSpan::CLOSE);
 
         let subscriber = tracing_subscriber::registry().with(fmt_layer);
-        let _guard = tracing::subscriber::set_default(subscriber);
-
-        let req = Request::builder()
-            .uri("/test-path")
-            .header("x-request-id", "trace-test-id-12345")
-            .body(Empty::<Bytes>::new())
-            .unwrap();
-        let span = make_request_span(&req);
-        drop(span.enter());
-        drop(span);
+        tracing::subscriber::with_default(subscriber, || {
+            // Other parallel tests may register this callsite while no subscriber
+            // is active. Refresh the process-wide cache after installing this
+            // thread-local subscriber so the span cannot remain disabled.
+            tracing::callsite::rebuild_interest_cache();
+
+            let req = Request::builder()
+                .uri("/test-path")
+                .header("x-request-id", "trace-test-id-12345")
+                .body(Empty::<Bytes>::new())
+                .unwrap();
+            let span = make_request_span(&req);
+            drop(span.enter());
+            drop(span);
+        });
 
         let output = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap();
         assert!(