fix(observability): bind OTLP receiver on capability confirmation

The supervisor bound the OTLP receiver on 127.0.0.1:4318 before the
session handshake, regardless of whether the gateway confirmed the
otel_export capability. When the capability was declined the port stayed
reserved and the receiver answered 200 OK for spans that were then dropped
on the way to a session that never drained them. Shutdown could also hang:
each HTTP connection task held a buffer sender and the forwarder only
exited once the last sender dropped, so a single keep-alive connection
blocked sandbox teardown.

Move the relay into openshell-supervisor-process and make the supervisor
session own it. The session binds the receiver on the first SessionAccepted
that confirms otel_export, keeps it across reconnects, and drains the
buffer straight into the session stream. This removes the forwarder task
and the intermediate session channel, leaving one bounded buffer between
receiver and gateway.

The receiver now tracks connections in a JoinSet under hyper-util's
GracefulShutdown with a header read timeout, so shutdown disables
keep-alive, waits a bounded 2s for in-flight requests, and aborts
stragglers. run_process asks the session to stop the receiver and flush
buffered telemetry after the entrypoint exits and before the exit is
reported, bounded at 3s, so final spans still reach the gateway. The
previous lib.rs drain ran after the session task had been aborted and
flushed into a channel nobody read.

Also drop the never-read RelayConfig.enabled field, factor the netns bind
into bind_tcp_in_netns_fd, and update the architecture doc.

Signed-off-by: Roland Huß <rhuss@redhat.com>

Author: rhuss

AGENTS.md

@@ -64,7 +64,7 @@ These pipelines connect skills into end-to-end workflows. Individual skill files
 | `crates/openshell-supervisor-middleware/` | Middleware runtime | Generic middleware registry, remote service integration, and chain execution |
 | `crates/openshell-supervisor-middleware-builtins/` | Built-in middleware | First-party in-process middleware implementations |
 | `crates/openshell-supervisor-network/` | Network supervisor | Proxying, L7 enforcement, policy evaluation, and provider credential injection |
-| `crates/openshell-supervisor-process/` | Process supervisor | Process lifecycle, namespace, and bypass monitoring |
+| `crates/openshell-supervisor-process/` | Process supervisor | Process lifecycle, namespace, bypass monitoring, and the OTLP telemetry relay |
 | `crates/openshell-vfio/` | VFIO support | PCI and GPU passthrough preparation and lifecycle |
 | `python/openshell/` | Python SDK | Python bindings and CLI packaging |
 | `sdk/typescript/` | TypeScript SDK | Native Connect client, curated sandbox API, and generated protobuf types |

Cargo.lock

@@ -469,17 +469,32 @@ collector without requiring direct egress from the sandbox.
 Agent process --> OTLP HTTP (127.0.0.1:4318) --> Supervisor receiver
   --> Enrichment (sandbox resource attributes)
   --> Bounded buffer (4096 slots, shared traces + OCSF)
-  --> Forwarder --> Session channel (OtelExportData message)
+  --> Supervisor session (OtelExportData message on the session stream)
   --> Gateway --> Dedicated SpanExporter --> External OTLP collector
 ```
 
+The supervisor session owns the relay (`openshell-supervisor-process::otlp`).
+There is no separate forwarder task: the session drains the buffer directly
+into its outbound stream, so the buffer is the only queue between the
+receiver and the gateway.
+
 ### Receiver Binding
 
 The OTLP HTTP receiver binds to `127.0.0.1:4318` only when the relay is
 active (the gateway has `[openshell.gateway.otlp]` configured and confirms
 the `otel_export` capability). When OTLP is not configured, no port is
 bound and no receiver runs.
 
+The bind is lazy: the session binds the receiver on the first
+`SessionAccepted` that confirms the capability and keeps it bound across
+gateway reconnects. The agent process starts before the session handshake
+completes, so an exporter that flushes in that window sees a refused
+connection rather than a silent drop; OTel SDK exporters retry with
+backoff, so those spans land once the port is up. If a later reconnect
+declines the capability, the receiver stays bound and forwarding pauses
+until a session confirms again. A bind failure is logged and disables the
+relay for the rest of the sandbox lifetime.
+
 The bind address depends on the supervisor topology. In all current
 topologies, the process supervisor runs co-located with the agent workload,
 so `127.0.0.1` is reachable from the agent. For Docker/Podman drivers
@@ -515,10 +530,22 @@ relay is active, the supervisor sets `OTEL_EXPORTER_OTLP_ENDPOINT` and
 
 ### Non-Interference
 
-The buffer uses `try_send` (non-blocking) on the session channel. When the
-buffer reaches capacity, the oldest entries are dropped and a counter records
-each drop. A queue depth gauge tracks buffer pressure. This ensures telemetry
-cannot block or degrade sandbox control operations.
+Both hops are non-blocking. The receiver uses `try_send` into the bounded
+buffer: when the buffer is full, the newest item is dropped and a counter
+records each drop. The session uses `try_send` into its outbound stream and
+drops the message when that stream is backed up. A queue depth gauge tracks
+buffer pressure. This ensures telemetry cannot block or degrade sandbox
+control operations.
+
+### Shutdown
+
+After the entrypoint exits and before the exit is reported to the gateway,
+the supervisor asks the session to stop the receiver and flush the buffer.
+The receiver stops accepting, disables keep-alive on open connections (idle
+ones close immediately), waits up to 2 seconds for in-flight requests, and
+aborts stragglers. The session then pushes whatever is still buffered onto
+the session stream. The whole flush is bounded at 3 seconds so an
+unreachable gateway cannot delay the exit report.
 
 ### OCSF Event Relay
 

architecture/sandbox.md

@@ -850,9 +850,6 @@ pub async fn run_sandbox(
     };
     tokio::pin!(proxy_exited);
 
-    #[cfg_attr(not(target_os = "linux"), allow(unused_mut))]
-    let mut otel_relay_handle: Option<openshell_supervisor_network::otlp::RelayHandle> = None;
-
     let exit_code = if process_enabled {
         let ca_file_paths = networking
             .as_ref()
@@ -953,22 +950,21 @@ pub async fn run_sandbox(
             None
         };
 
-        // OTEL relay: bind OTLP receiver for all Linux topologies.
-        // All current topologies keep the process supervisor co-located with
-        // the agent, so 127.0.0.1 is reachable from agent processes. Future
-        // topologies that move the supervisor out of the workload pod would
-        // need to derive the bind address from the topology (e.g., pod IP
-        // via downward API) and update OTEL_EXPORTER_OTLP_ENDPOINT to match.
-        let otel_rx = {
+        // OTEL relay: describe the OTLP receiver for the supervisor session,
+        // which binds it only once the gateway confirms the otel_export
+        // capability. All current topologies keep the process supervisor
+        // co-located with the agent, so 127.0.0.1 is reachable from agent
+        // processes. Future topologies that move the supervisor out of the
+        // workload pod would need to derive the bind address from the
+        // topology and update OTEL_EXPORTER_OTLP_ENDPOINT to match.
+        let otel_relay = {
             #[cfg(target_os = "linux")]
             {
-                let otlp_addr = openshell_core::sandbox_env::OTLP_RECEIVER_ADDR;
-
-                let (otel_session_tx, otel_session_rx) =
-                    tokio::sync::mpsc::channel::<openshell_core::proto::SupervisorMessage>(64);
-
-                let relay_config = openshell_supervisor_network::otlp::RelayConfig::default();
-                let metadata = openshell_supervisor_network::otlp::SandboxMetadata {
+                let bind_addr: std::net::SocketAddr =
+                    openshell_core::sandbox_env::OTLP_RECEIVER_ADDR
+                        .parse()
+                        .expect("OTLP_RECEIVER_ADDR is a valid socket address");
+                let metadata = openshell_supervisor_process::otlp::SandboxMetadata {
                     sandbox_id: sandbox_id.clone().unwrap_or_default(),
                     workspace_id: workspace_rx.borrow().clone(),
                     policy: sandbox_name_for_agg.clone().unwrap_or_default(),
@@ -979,37 +975,11 @@ pub async fn run_sandbox(
                     driver: std::env::var(openshell_core::sandbox_env::SUPERVISOR_TOPOLOGY)
                         .unwrap_or_else(|_| "container".to_string()),
                 };
-                let relay = openshell_supervisor_network::otlp::OtelRelay::new(
-                    relay_config,
+                Some(openshell_supervisor_process::otlp::RelaySetup {
+                    config: openshell_supervisor_process::otlp::RelayConfig::default(),
                     metadata,
-                    otel_session_tx,
-                );
-
-                let bind_addr: std::net::SocketAddr = otlp_addr.parse().unwrap();
-
-                if let Some(ns) = netns.as_ref() {
-                    match ns.bind_tcp_in_netns(otlp_addr).await {
-                        Ok(listener) => {
-                            let handle = relay.start_with_listener(listener);
-                            tracing::info!(bind = %bind_addr, "OTEL relay started (netns)");
-                            otel_relay_handle = Some(handle);
-                        }
-                        Err(e) => {
-                            tracing::warn!(error = %e, "OTEL relay failed to bind in netns; continuing without relay");
-                        }
-                    }
-                } else {
-                    match relay.start(bind_addr).await {
-                        Ok(handle) => {
-                            tracing::info!(bind = %bind_addr, "OTEL relay started");
-                            otel_relay_handle = Some(handle);
-                        }
-                        Err(e) => {
-                            tracing::warn!(error = %e, "OTEL relay failed to start; continuing without relay");
-                        }
-                    }
-                }
-                Some(otel_session_rx)
+                    bind_addr,
+                })
             }
             #[cfg(not(target_os = "linux"))]
             {
@@ -1046,7 +1016,7 @@ pub async fn run_sandbox(
             bypass_denial_tx,
             #[cfg(target_os = "linux")]
             bypass_activity_tx,
-            otel_rx,
+            otel_relay,
         );
 
         if let Some(control_closed) = process_control_closed.as_mut() {
@@ -1206,12 +1176,6 @@ pub async fn run_sandbox(
         }
     };
 
-    // Drain OTEL relay before tearing down networking so short-lived
-    // agents don't lose their final spans.
-    if let Some(handle) = otel_relay_handle {
-        handle.shutdown().await;
-    }
-
     // Drop networking explicitly so the proxy + bypass monitor RAII
     // handles tear down before we return.
     drop(networking);

crates/openshell-sandbox/src/lib.rs

@@ -25,6 +25,7 @@ mod grpc;
 mod http;
 mod middleware;
 mod multiplex;
+pub(crate) mod otel_relay;
 mod otel_tracing;
 mod persistence;
 pub(crate) mod policy_store;
@@ -37,7 +38,6 @@ mod service_routing;
 mod ssh_sessions;
 pub mod supervisor_session;
 mod telemetry;
-pub(crate) mod otel_relay;
 #[cfg(any(test, feature = "test-support"))]
 pub mod test_support;
 mod tls;
@@ -666,8 +666,7 @@ pub(crate) async fn run_server(
     state.middleware_registry = middleware_registry;
     state.gateway_interceptors = gateway_interceptors;
     state.provider_profile_sources = provider_profile_sources;
-    state.otel_relay_exporter =
-        otel_relay::try_create_exporter(config_file.as_ref()).await;
+    state.otel_relay_exporter = otel_relay::try_create_exporter(config_file.as_ref()).await;
     state.sandbox_jwt_issuer = sandbox_jwt_issuer.clone();
     state.sandbox_jwt_authenticator = sandbox_jwt_authenticator;
     if let Some(issuer) = sandbox_jwt_issuer {

crates/openshell-server/src/lib.rs

@@ -1059,6 +1059,9 @@ fn confirm_capabilities(advertised: &[String], state: &Arc<ServerState>) -> Vec<
     confirmed
 }
 
+/// Upper bound on a single relayed OCSF event; larger events are skipped.
+const MAX_OCSF_EVENT_SIZE: usize = 256 * 1024;
+
 /// Handle incoming OTEL export data from the supervisor: forward trace data to
 /// the configured OTLP collector and dispatch OCSF events to the log sink.
 fn handle_otel_export(
@@ -1069,39 +1072,33 @@ fn handle_otel_export(
 ) {
     if let Some(openshell_core::proto::otel_export_data::Signal::TraceData(trace_data)) =
         otel_data.signal
+        && !trace_data.is_empty()
+        && let Some(relay_exporter) = state.otel_relay_exporter.as_ref()
     {
-        if !trace_data.is_empty() {
-            if let Some(relay_exporter) = state.otel_relay_exporter.as_ref() {
-                let exporter = relay_exporter.clone();
-                let sandbox_id = sandbox_id.to_string();
-                tokio::spawn(async move {
-                    match tokio::time::timeout(
-                        Duration::from_secs(10),
-                        exporter.export_raw(trace_data),
-                    )
-                    .await
-                    {
-                        Ok(Err(e)) => {
-                            debug!(
-                                sandbox_id = %sandbox_id,
-                                error = %e,
-                                "OTEL relay: failed to export trace data"
-                            );
-                        }
-                        Err(_) => {
-                            debug!(
-                                sandbox_id = %sandbox_id,
-                                "OTEL relay: export timed out"
-                            );
-                        }
-                        Ok(Ok(())) => {}
-                    }
-                });
+        let exporter = relay_exporter.clone();
+        let sandbox_id = sandbox_id.to_string();
+        tokio::spawn(async move {
+            match tokio::time::timeout(Duration::from_secs(10), exporter.export_raw(trace_data))
+                .await
+            {
+                Ok(Err(e)) => {
+                    debug!(
+                        sandbox_id = %sandbox_id,
+                        error = %e,
+                        "OTEL relay: failed to export trace data"
+                    );
+                }
+                Err(_) => {
+                    debug!(
+                        sandbox_id = %sandbox_id,
+                        "OTEL relay: export timed out"
+                    );
+                }
+                Ok(Ok(())) => {}
             }
-        }
+        });
     }
 
-    const MAX_OCSF_EVENT_SIZE: usize = 256 * 1024;
     for ocsf_event in &otel_data.ocsf_events {
         if ocsf_event.len() > MAX_OCSF_EVENT_SIZE {
             debug!(

crates/openshell-server/src/supervisor_session.rs

@@ -17,11 +17,6 @@ openshell-otel = { path = "../openshell-otel" }
 openshell-policy = { path = "../openshell-policy" }
 openshell-supervisor-middleware = { path = "../openshell-supervisor-middleware" }
 
-hyper = { workspace = true }
-hyper-util = { version = "0.1", features = ["tokio", "http1"] }
-http-body-util = "0.1"
-opentelemetry-proto = { workspace = true }
-prost = { workspace = true }
 apollo-parser = { workspace = true }
 aws-sigv4 = { version = "1", features = ["sign-http", "http1"] }
 aws-credential-types = { version = "1", features = ["hardcoded-credentials"] }

crates/openshell-supervisor-network/Cargo.toml

@@ -11,7 +11,6 @@
 pub mod identity;
 pub mod l7;
 pub mod opa;
-pub mod otlp;
 pub(crate) mod policy_dns;
 pub mod policy_local;
 pub mod procfs;