refactor(tunnel): rename aether-proxy to aether-tunnel

apps/aether-tunnel/src/tunnel/client.rs

@@ -0,0 +1,394 @@
+//! WebSocket tunnel client: connect, authenticate, and run the tunnel.
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use tokio::net::TcpStream;
+use tokio::sync::watch;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tokio_tungstenite::tungstenite::http;
+use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
+use tracing::{debug, info, warn};
+
+use crate::egress_proxy::{connect_target_via_proxy, ProxyConnectOptions, UpstreamProxyConfig};
+use crate::state::{AppState, ServerContext};
+use aether_contracts::tunnel::{CURRENT_TUNNEL_PROTOCOL_VERSION, TUNNEL_PROTOCOL_VERSION_HEADER};
+
+use super::{dispatcher, heartbeat, writer};
+
+/// Outcome of a tunnel session.
+pub enum TunnelOutcome {
+    /// Graceful shutdown requested by the local process.
+    Shutdown,
+    /// Remote side disconnected or connection lost — should reconnect.
+    Disconnected,
+}
+
+/// Connect to Aether's WebSocket tunnel endpoint and run until disconnected.
+///
+/// `conn_idx` identifies which connection in the pool this is (0-based).
+/// Only connection 0 sends heartbeats to avoid resetting shared metrics.
+pub async fn connect_and_run(
+    state: &Arc<AppState>,
+    server: &Arc<ServerContext>,
+    conn_idx: usize,
+    shutdown: &mut watch::Receiver<bool>,
+    drain: watch::Receiver<bool>,
+) -> Result<TunnelOutcome, anyhow::Error> {
+    let ws_url = build_tunnel_url(server);
+    debug!(url = %ws_url, conn = conn_idx, "connecting tunnel");
+
+    // Build WebSocket request with auth headers
+    let mut request = ws_url.clone().into_client_request()?;
+    let headers = request.headers_mut();
+    headers.insert(
+        "Authorization",
+        http::HeaderValue::from_str(&format!("Bearer {}", server.management_token))?,
+    );
+    headers.insert(
+        TUNNEL_PROTOCOL_VERSION_HEADER,
+        http::HeaderValue::from_str(&CURRENT_TUNNEL_PROTOCOL_VERSION.to_string())?,
+    );
+    let node_id = server.node_id.read().unwrap().clone();
+    headers.insert("X-Node-Id", http::HeaderValue::from_str(&node_id)?);
+    // Use dynamic node_name (may be updated by remote config) instead of
+    // the static server.node_name, so that remote name changes take effect
+    // on the next reconnect.
+    let dynamic_node_name = server.dynamic.load().node_name.clone();
+    headers.insert(
+        "X-Node-Name",
+        http::HeaderValue::from_str(&dynamic_node_name)?,
+    );
+    // Advertise per-connection max concurrent streams so the backend can
+    // respect the proxy's capacity limit.
+    let max_streams = state.config.tunnel_max_streams.unwrap_or(128);
+    headers.insert("X-Tunnel-Max-Streams", http::HeaderValue::from(max_streams));
+
+    // Parse host:port from URL
+    let uri: http::Uri = ws_url.parse()?;
+    let host = uri
+        .host()
+        .ok_or_else(|| anyhow::anyhow!("missing host in tunnel URL"))?;
+    let is_tls = uri.scheme_str() == Some("wss");
+    let port = uri.port_u16().unwrap_or(if is_tls { 443 } else { 80 });
+
+    // TCP connect with timeout
+    let connect_timeout = state
+        .config
+        .tunnel_connect_timeout()
+        .expect("validated config should resolve tunnel connect timeout");
+    let tcp_stream = connect_tunnel_tcp(state, host, port, connect_timeout).await?;
+
+    // Configure TCP parameters via socket2
+    configure_tcp_socket(&tcp_stream, state);
+
+    // WebSocket upgrade (with TLS if wss://)
+    let connector = if is_tls {
+        Some(tokio_tungstenite::Connector::Rustls(Arc::clone(
+            &state.tunnel_tls_config,
+        )))
+    } else {
+        None
+    };
+    // Match Python-side _MAX_FRAME_SIZE (64 MiB) to prevent tungstenite's
+    // default 16 MiB limit from rejecting large AI API payloads (multi-image
+    // base64 requests can exceed 16 MiB).
+    let ws_config = WebSocketConfig {
+        max_frame_size: Some(64 << 20),
+        max_message_size: Some(64 << 20),
+        ..Default::default()
+    };
+    let handshake_timeout = connect_timeout;
+    let (ws_stream, _response) = tokio::time::timeout(
+        handshake_timeout,
+        tokio_tungstenite::client_async_tls_with_config(
+            request,
+            tcp_stream,
+            Some(ws_config),
+            connector,
+        ),
+    )
+    .await
+    .map_err(|_| {
+        anyhow::anyhow!(
+            "tunnel WebSocket handshake timeout ({}ms)",
+            handshake_timeout.as_millis()
+        )
+    })??;
+    let stale_timeout = state
+        .config
+        .tunnel_stale_timeout()
+        .expect("validated config should resolve tunnel stale timeout");
+    let ping_interval = state
+        .config
+        .tunnel_ping_interval()
+        .expect("validated config should resolve tunnel ping interval");
+    debug!(
+        conn = conn_idx,
+        tcp_keepalive_secs = state.config.tunnel_tcp_keepalive_secs,
+        tcp_nodelay = state.config.tunnel_tcp_nodelay,
+        connect_timeout_ms = connect_timeout.as_millis(),
+        stale_timeout_ms = stale_timeout.as_millis(),
+        ping_interval_ms = ping_interval.as_millis(),
+        "tunnel connected"
+    );
+    server.tunnel_metrics.record_connect_success();
+    let connected_at = Instant::now();
+
+    // NOTE: reconnect_attempts reset is handled by the caller (mod.rs)
+    // based on how long the connection stayed alive.
+
+    // Split into read/write halves
+    let (ws_sink, ws_read) = futures_util::StreamExt::split(ws_stream);
+
+    // Spawn writer task (with WebSocket ping keepalive)
+    let (frame_tx, mut writer_handle) = writer::spawn_writer_with_metrics(
+        ws_sink,
+        ping_interval,
+        Some(Arc::clone(&server.tunnel_metrics)),
+    );
+    let drain_signal = spawn_drain_signal(conn_idx, frame_tx.clone(), drain.clone());
+
+    // Spawn heartbeat task (only for primary connection to avoid
+    // resetting shared atomic metrics via swap(0))
+    let hb_handle = if conn_idx == 0 {
+        heartbeat::spawn(
+            Arc::clone(state),
+            Arc::clone(server),
+            frame_tx.clone(),
+            shutdown.clone(),
+        )
+    } else {
+        heartbeat::spawn_noop()
+    };
+
+    // Run dispatcher (blocks until disconnect or shutdown).
+    // Also watch for writer exit — if the write half dies (e.g. the peer
+    // closed the connection) but the read half stays open, dispatcher would
+    // block forever on `ws_stream.next()`.  Monitoring `writer_handle`
+    // ensures we detect this and trigger a reconnect promptly.
+    let state_clone = Arc::clone(state);
+    let server_clone = Arc::clone(server);
+    let outcome = tokio::select! {
+        result = dispatcher::run(
+            state_clone,
+            server_clone,
+            ws_read,
+            frame_tx.clone(),
+            hb_handle,
+            drain.clone(),
+        ) => {
+            match result {
+                Ok(()) => Ok(TunnelOutcome::Disconnected),
+                Err(e) => {
+                    server
+                        .tunnel_metrics
+                        .record_error("dispatcher_error", &e.to_string());
+                    Err(e)
+                }
+            }
+        }
+        writer_result = &mut writer_handle => {
+            match writer_result {
+                Ok(()) => warn!("writer task exited normally, triggering reconnect"),
+                Err(e) => {
+                    if e.is_panic() {
+                        tracing::error!(error = %e, "writer task panicked, triggering reconnect");
+                        server
+                            .tunnel_metrics
+                            .record_error("writer_task_panic", &e.to_string());
+                    } else {
+                        warn!(error = %e, "writer task cancelled, triggering reconnect");
+                        server
+                            .tunnel_metrics
+                            .record_error("writer_task_cancelled", &e.to_string());
+                    }
+                }
+            }
+            Ok(TunnelOutcome::Disconnected)
+        }
+        _ = shutdown.changed() => {
+            debug!("shutdown during tunnel dispatch");
+            Ok(TunnelOutcome::Shutdown)
+        }
+    };
+
+    // Drop our sender; the writer will exit once all stream handler clones
+    // are also dropped (i.e. after they finish their in-flight work).
+    drop(frame_tx);
+    if !drain_signal.is_finished() {
+        drain_signal.abort();
+        let _ = drain_signal.await;
+    }
+
+    // Wait for the writer task to finish with a generous timeout — the
+    // dispatcher already waits up to 30s for stream handlers, so 35s here
+    // covers that plus a small margin.
+    // Skip if the writer already exited (the select branch that fired).
+    if !writer_handle.is_finished() {
+        let _ = tokio::time::timeout(Duration::from_secs(35), writer_handle).await;
+    }
+
+    let connected_for = connected_at.elapsed();
+    match &outcome {
+        Ok(TunnelOutcome::Shutdown) => info!(
+            conn = conn_idx,
+            connected_duration_ms = connected_for.as_millis() as u64,
+            close_reason = "shutdown",
+            "tunnel session ending"
+        ),
+        Ok(TunnelOutcome::Disconnected) => info!(
+            conn = conn_idx,
+            connected_duration_ms = connected_for.as_millis() as u64,
+            close_reason = "disconnected",
+            "tunnel session ending"
+        ),
+        Err(error) => warn!(
+            conn = conn_idx,
+            connected_duration_ms = connected_for.as_millis() as u64,
+            close_reason = "error",
+            error = %error,
+            "tunnel session ending"
+        ),
+    }
+
+    server.tunnel_metrics.record_disconnect(connected_for);
+
+    debug!("tunnel disconnected");
+    outcome
+}
+
+fn spawn_drain_signal(
+    conn_idx: usize,
+    frame_tx: writer::FrameSender,
+    mut drain: watch::Receiver<bool>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        if !*drain.borrow() {
+            loop {
+                if drain.changed().await.is_err() {
+                    return;
+                }
+                if *drain.borrow() {
+                    break;
+                }
+            }
+        }
+
+        debug!(conn = conn_idx, "sending GOAWAY for tunnel drain");
+        match tokio::time::timeout(
+            Duration::from_millis(250),
+            frame_tx.send(super::protocol::Frame::control(
+                super::protocol::MsgType::GoAway,
+                bytes::Bytes::new(),
+            )),
+        )
+        .await
+        {
+            Ok(Ok(())) => info!(conn = conn_idx, "sent GOAWAY for tunnel drain"),
+            Ok(Err(error)) => warn!(
+                conn = conn_idx,
+                error = ?error,
+                "failed to queue GOAWAY for tunnel drain"
+            ),
+            Err(_) => warn!(
+                conn = conn_idx,
+                "timed out queueing GOAWAY for tunnel drain"
+            ),
+        }
+    })
+}
+
+async fn connect_tunnel_tcp(
+    state: &Arc<AppState>,
+    host: &str,
+    port: u16,
+    connect_timeout: Duration,
+) -> Result<TcpStream, anyhow::Error> {
+    if let Some(proxy_url) = state.config.effective_aether_outbound_proxy_url() {
+        let proxy = UpstreamProxyConfig::parse(proxy_url)
+            .map_err(|err| anyhow::anyhow!("Aether outbound proxy URL invalid: {err}"))?;
+        debug!(
+            proxy_url = %proxy.redacted_url(),
+            host = %host,
+            port = port,
+            "connecting tunnel via Aether egress proxy"
+        );
+        return tokio::time::timeout(
+            connect_timeout,
+            connect_target_via_proxy(
+                &proxy,
+                host,
+                port,
+                ProxyConnectOptions {
+                    connect_timeout,
+                    tcp_nodelay: state.config.tunnel_tcp_nodelay,
+                    tcp_keepalive: (state.config.tunnel_tcp_keepalive_secs > 0)
+                        .then(|| Duration::from_secs(state.config.tunnel_tcp_keepalive_secs)),
+                },
+            ),
+        )
+        .await
+        .map_err(|_| {
+            anyhow::anyhow!(
+                "tunnel outbound proxy TCP connect timeout ({}ms)",
+                connect_timeout.as_millis()
+            )
+        })?
+        .map_err(anyhow::Error::from);
+    }
+
+    tokio::time::timeout(connect_timeout, TcpStream::connect((host, port)))
+        .await
+        .map_err(|_| {
+            anyhow::anyhow!(
+                "tunnel TCP connect timeout ({}ms)",
+                connect_timeout.as_millis()
+            )
+        })?
+        .map_err(anyhow::Error::from)
+}
+
+/// Configure TCP keepalive and NODELAY on an established socket.
+fn configure_tcp_socket(stream: &TcpStream, state: &Arc<AppState>) {
+    let sock_ref = socket2::SockRef::from(stream);
+
+    if state.config.tunnel_tcp_keepalive_secs > 0 {
+        let keepalive = socket2::TcpKeepalive::new()
+            .with_time(Duration::from_secs(state.config.tunnel_tcp_keepalive_secs))
+            .with_interval(Duration::from_secs(5));
+        #[cfg(not(target_os = "windows"))]
+        let keepalive = keepalive.with_retries(3);
+        if let Err(e) = sock_ref.set_tcp_keepalive(&keepalive) {
+            warn!(error = %e, "failed to set TCP keepalive on tunnel socket");
+        }
+    }
+
+    if state.config.tunnel_tcp_nodelay {
+        if let Err(e) = sock_ref.set_nodelay(true) {
+            warn!(error = %e, "failed to set TCP_NODELAY on tunnel socket");
+        }
+    }
+}
+
+/// Build rustls ClientConfig with system root certificates.
+pub fn build_tls_config() -> rustls::ClientConfig {
+    let _ = rustls::crypto::ring::default_provider().install_default();
+    let root_store =
+        rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+    rustls::ClientConfig::builder()
+        .with_root_certificates(root_store)
+        .with_no_client_auth()
+}
+
+fn build_tunnel_url(server: &ServerContext) -> String {
+    let base = server.aether_url.trim_end_matches('/');
+    let ws_base = if base.starts_with("https://") {
+        base.replacen("https://", "wss://", 1)
+    } else if base.starts_with("http://") {
+        base.replacen("http://", "ws://", 1)
+    } else {
+        format!("wss://{}", base)
+    };
+    format!("{}/api/internal/proxy-tunnel", ws_base)
+}

apps/aether-tunnel/src/tunnel/dispatcher.rs

@@ -0,0 +1,440 @@
+//! Frame dispatcher: reads incoming WebSocket frames and routes them.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures_util::StreamExt;
+use tokio::sync::mpsc;
+use tokio::sync::watch;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::tungstenite::Message;
+use tracing::{debug, error, info, warn};
+
+use crate::state::{AppState, ServerContext};
+
+use super::heartbeat::HeartbeatHandle;
+use super::protocol::{decompress_if_gzip, Frame, MsgType, RequestMeta};
+use super::stream_handler;
+use super::writer::FrameSender;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum StreamDispatchStatus {
+    Delivered,
+    Closed,
+    TimedOut,
+}
+
+/// Run the dispatcher loop, reading from the WebSocket stream.
+pub async fn run<S>(
+    state: Arc<AppState>,
+    server: Arc<ServerContext>,
+    mut ws_stream: S,
+    frame_tx: FrameSender,
+    heartbeat: HeartbeatHandle,
+    mut drain: watch::Receiver<bool>,
+) -> Result<(), anyhow::Error>
+where
+    S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
+        + Unpin
+        + Send
+        + 'static,
+{
+    // Active streams: stream_id -> body sender
+    let mut streams: HashMap<u32, mpsc::Sender<Frame>> = HashMap::new();
+    // Track spawned stream handlers so we can wait for them on shutdown
+    let mut handler_handles: Vec<JoinHandle<()>> = Vec::new();
+    let max_streams = state.config.tunnel_max_streams.unwrap_or(128) as usize;
+    let mut frames_since_cleanup: u32 = 0;
+    let stale_timeout = state
+        .config
+        .tunnel_stale_timeout()
+        .expect("validated config should resolve tunnel stale timeout");
+
+    // Track last time we received any data to detect stale connections
+    let mut last_data_at = tokio::time::Instant::now();
+    let mut draining = *drain.borrow();
+
+    let read_err = loop {
+        if draining && streams.is_empty() {
+            info!("tunnel drained after in-flight streams completed");
+            break None;
+        }
+
+        let msg_result = tokio::select! {
+            msg = ws_stream.next() => {
+                match msg {
+                    Some(r) => r,
+                    None => break None,
+                }
+            }
+            changed = drain.changed() => {
+                if changed.is_err() {
+                    continue;
+                }
+                if *drain.borrow() {
+                    info!("tunnel drain requested, waiting for in-flight streams");
+                    draining = true;
+                }
+                continue;
+            }
+            _ = tokio::time::sleep_until(last_data_at + stale_timeout) => {
+                warn!(
+                    stale_ms = stale_timeout.as_millis(),
+                    "tunnel connection stale, no data received"
+                );
+                server.tunnel_metrics.record_error(
+                    "stale_timeout",
+                    &format!("no tunnel frame received for {}ms", stale_timeout.as_millis()),
+                );
+                break None;
+            }
+        };
+
+        let msg = match msg_result {
+            Ok(m) => m,
+            Err(e) => {
+                error!(error = %e, "WebSocket read error");
+                server
+                    .tunnel_metrics
+                    .record_error("ws_read_error", &e.to_string());
+                break Some(e);
+            }
+        };
+
+        // Any successfully received message proves the connection is alive
+        last_data_at = tokio::time::Instant::now();
+
+        let data = match msg {
+            Message::Binary(data) => {
+                server.tunnel_metrics.record_ws_incoming_frame(data.len());
+                Bytes::from(data)
+            }
+            Message::Ping(_) => continue,
+            Message::Pong(_) => continue,
+            Message::Close(_) => {
+                debug!("received WebSocket close");
+                break None;
+            }
+            _ => continue,
+        };
+
+        let frame = match Frame::decode(data) {
+            Ok(f) => f,
+            Err(e) => {
+                warn!(error = %e, "failed to decode frame");
+                server
+                    .tunnel_metrics
+                    .record_error("frame_decode_error", &e.to_string());
+                continue;
+            }
+        };
+
+        match frame.msg_type {
+            MsgType::RequestHeaders => {
+                if draining {
+                    if frame_tx
+                        .try_send(Frame::new(
+                            frame.stream_id,
+                            MsgType::StreamError,
+                            0,
+                            Bytes::from("tunnel draining"),
+                        ))
+                        .is_err()
+                    {
+                        warn!(
+                            stream_id = frame.stream_id,
+                            "writer channel full, StreamError dropped during drain"
+                        );
+                    }
+                    continue;
+                }
+
+                // Decompress if the frame is gzip-compressed, then parse metadata
+                let payload = match decompress_if_gzip(&frame) {
+                    Ok(p) => p,
+                    Err(e) => {
+                        warn!(stream_id = frame.stream_id, error = %e, "frame decompress failed");
+                        continue;
+                    }
+                };
+                let meta: RequestMeta = match serde_json::from_slice(&payload) {
+                    Ok(m) => m,
+                    Err(e) => {
+                        warn!(stream_id = frame.stream_id, error = %e, "invalid request metadata");
+                        // Use try_send to avoid blocking the read loop
+                        if frame_tx
+                            .try_send(Frame::new(
+                                frame.stream_id,
+                                MsgType::StreamError,
+                                0,
+                                Bytes::from(format!("invalid request metadata: {e}")),
+                            ))
+                            .is_err()
+                        {
+                            warn!(
+                                stream_id = frame.stream_id,
+                                "writer channel full, StreamError dropped"
+                            );
+                        }
+                        continue;
+                    }
+                };
+
+                if streams.len() >= max_streams {
+                    warn!(
+                        stream_id = frame.stream_id,
+                        "max concurrent streams reached"
+                    );
+                    if frame_tx
+                        .try_send(Frame::new(
+                            frame.stream_id,
+                            MsgType::StreamError,
+                            0,
+                            Bytes::from("max concurrent streams reached"),
+                        ))
+                        .is_err()
+                    {
+                        warn!(
+                            stream_id = frame.stream_id,
+                            "writer channel full, StreamError dropped"
+                        );
+                    }
+                    continue;
+                }
+
+                // Create body channel and spawn handler
+                let (body_tx, body_rx) = mpsc::channel::<Frame>(64);
+                streams.insert(frame.stream_id, body_tx);
+
+                let state_clone = Arc::clone(&state);
+                let server_clone = Arc::clone(&server);
+                let tx_clone = frame_tx.clone();
+                let sid = frame.stream_id;
+                let handle = tokio::spawn(async move {
+                    stream_handler::handle_stream(
+                        state_clone,
+                        server_clone,
+                        sid,
+                        meta,
+                        body_rx,
+                        tx_clone,
+                    )
+                    .await;
+                });
+                handler_handles.push(handle);
+
+                debug!(stream_id = frame.stream_id, "new stream started");
+            }
+
+            MsgType::RequestBody => {
+                if let Some(tx) = streams.get(&frame.stream_id).cloned() {
+                    let is_end = frame.is_end_stream();
+                    let sid = frame.stream_id;
+                    let dispatch = dispatch_stream_frame(&tx, frame).await;
+                    if is_end || dispatch != StreamDispatchStatus::Delivered {
+                        streams.remove(&sid);
+                        if dispatch == StreamDispatchStatus::TimedOut {
+                            server.tunnel_metrics.record_error(
+                                "stream_dispatch_timeout",
+                                &format!("request body dispatch timed out for stream {}", sid),
+                            );
+                            try_send_stream_error(
+                                &frame_tx,
+                                sid,
+                                "tunnel request body dispatch stalled",
+                            );
+                        }
+                        if draining && streams.is_empty() {
+                            info!("tunnel drained after request body completion");
+                            break None;
+                        }
+                    }
+                }
+            }
+
+            MsgType::StreamEnd | MsgType::StreamError => {
+                // Client-side cancellation or end
+                if let Some(tx) = streams.remove(&frame.stream_id) {
+                    let _ = dispatch_stream_frame(&tx, frame).await;
+                    if draining && streams.is_empty() {
+                        info!("tunnel drained after stream termination");
+                        break None;
+                    }
+                }
+            }
+
+            MsgType::Ping => {
+                // Use try_send to avoid blocking the read loop when writer is congested
+                if frame_tx
+                    .try_send(Frame::control(MsgType::Pong, frame.payload))
+                    .is_err()
+                {
+                    warn!("writer channel full, Pong dropped");
+                }
+            }
+
+            MsgType::HeartbeatAck => {
+                heartbeat.on_ack(frame.payload).await;
+            }
+
+            MsgType::GoAway => {
+                info!("received GOAWAY");
+                break None;
+            }
+
+            _ => {
+                debug!(msg_type = ?frame.msg_type, "ignoring unexpected frame type");
+            }
+        }
+
+        // Periodically clean up finished handles to avoid unbounded growth.
+        // Trigger every 64 frames OR when the count exceeds max_streams.
+        frames_since_cleanup += 1;
+        if frames_since_cleanup >= 64 || handler_handles.len() > max_streams {
+            handler_handles.retain(|h| !h.is_finished());
+            frames_since_cleanup = 0;
+        }
+    };
+
+    // Drop body senders so stream handlers waiting on body_rx will unblock
+    streams.clear();
+
+    // Wait for active stream handlers to finish so their frame_tx clones
+    // are dropped before the writer closes the sink.
+    drain_handlers(handler_handles).await;
+
+    match read_err {
+        Some(e) => Err(e.into()),
+        None => Ok(()),
+    }
+}
+
+async fn dispatch_stream_frame(tx: &mpsc::Sender<Frame>, frame: Frame) -> StreamDispatchStatus {
+    let stream_id = frame.stream_id;
+    match tokio::time::timeout(stream_frame_dispatch_timeout(), tx.send(frame)).await {
+        Ok(Ok(())) => StreamDispatchStatus::Delivered,
+        Ok(Err(_)) => {
+            warn!(
+                stream_id,
+                "stream handler channel closed while dispatching tunnel frame"
+            );
+            StreamDispatchStatus::Closed
+        }
+        Err(_) => {
+            warn!(
+                stream_id,
+                timeout_ms = stream_frame_dispatch_timeout().as_millis(),
+                "stream handler channel blocked while dispatching tunnel frame"
+            );
+            StreamDispatchStatus::TimedOut
+        }
+    }
+}
+
+/// Bound how long a single stream handler is allowed to block the shared
+/// WebSocket read loop while receiving request-body frames.
+fn stream_frame_dispatch_timeout() -> Duration {
+    #[cfg(test)]
+    {
+        Duration::from_millis(25)
+    }
+
+    #[cfg(not(test))]
+    {
+        Duration::from_millis(500)
+    }
+}
+
+fn try_send_stream_error(frame_tx: &FrameSender, stream_id: u32, message: &'static str) {
+    if frame_tx
+        .try_send(Frame::new(
+            stream_id,
+            MsgType::StreamError,
+            0,
+            Bytes::from(message),
+        ))
+        .is_err()
+    {
+        warn!(
+            stream_id,
+            "writer channel full, StreamError dropped while aborting stalled stream"
+        );
+    }
+}
+
+/// Wait for all active stream handlers to finish (with a timeout).
+async fn drain_handlers(handles: Vec<JoinHandle<()>>) {
+    if handles.is_empty() {
+        return;
+    }
+    let count = handles.len();
+    debug!(count, "waiting for active stream handlers to finish");
+    let _ = tokio::time::timeout(Duration::from_secs(30), async {
+        for h in handles {
+            let _ = h.await;
+        }
+    })
+    .await;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aether_runtime::bounded_queue;
+
+    #[tokio::test]
+    async fn dispatch_stream_frame_times_out_when_handler_stops_draining() {
+        let (tx, mut rx) = mpsc::channel::<Frame>(1);
+        tx.send(Frame::new(
+            7,
+            MsgType::RequestBody,
+            0,
+            Bytes::from_static(b"first"),
+        ))
+        .await
+        .expect("first frame should enqueue");
+
+        let stalled_send = tokio::spawn({
+            let tx = tx.clone();
+            async move {
+                dispatch_stream_frame(
+                    &tx,
+                    Frame::new(7, MsgType::RequestBody, 0, Bytes::from_static(b"second")),
+                )
+                .await
+            }
+        });
+
+        assert_eq!(
+            stalled_send.await.expect("dispatch task should join"),
+            StreamDispatchStatus::TimedOut
+        );
+
+        let retained = rx
+            .recv()
+            .await
+            .expect("queued frame should still be present");
+        assert_eq!(retained.payload, Bytes::from_static(b"first"));
+    }
+
+    #[tokio::test]
+    async fn try_send_stream_error_emits_stream_error_frame() {
+        let (high_tx, mut high_rx) = bounded_queue::<Frame>(4);
+        let (normal_tx, _normal_rx) = bounded_queue::<Frame>(4);
+        let frame_tx = FrameSender::from_test_queues(high_tx, normal_tx);
+        try_send_stream_error(&frame_tx, 9, "tunnel request body dispatch stalled");
+
+        let frame = high_rx
+            .recv()
+            .await
+            .expect("stream error frame should enqueue");
+        assert_eq!(frame.stream_id, 9);
+        assert_eq!(frame.msg_type, MsgType::StreamError);
+        assert_eq!(
+            frame.payload,
+            Bytes::from_static(b"tunnel request body dispatch stalled")
+        );
+    }
+}

apps/aether-tunnel/src/tunnel/heartbeat.rs

@@ -0,0 +1,537 @@
+//! Tunnel heartbeat: sends metrics over the tunnel, processes ACKs.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
+
+use bytes::Bytes;
+use tokio::sync::watch;
+use tokio::time::Instant;
+use tracing::{debug, info, warn};
+
+use crate::registration::client::RemoteConfig;
+use crate::runtime;
+use crate::state::{AppState, ServerContext, TunnelRequestMetricsSnapshot};
+
+use super::protocol::{Frame, MsgType};
+use super::writer::FrameSender;
+
+const CURRENT_VERSION: &str = env!("CARGO_PKG_VERSION");
+static UPGRADE_IN_PROGRESS: AtomicBool = AtomicBool::new(false);
+static NON_ROOT_UPGRADE_WARNED: AtomicBool = AtomicBool::new(false);
+
+enum AckDecision {
+    Accept {
+        heartbeat_id: u64,
+        upgrade_to: Option<String>,
+    },
+    Ignore,
+}
+
+/// Handle for the dispatcher to forward HeartbeatAck frames.
+#[derive(Clone)]
+pub struct HeartbeatHandle {
+    ack_tx: tokio::sync::mpsc::Sender<Bytes>,
+}
+
+impl HeartbeatHandle {
+    pub async fn on_ack(&self, payload: Bytes) {
+        let _ = self.ack_tx.send(payload).await;
+    }
+}
+
+/// Create a no-op heartbeat handle that silently discards ACKs.
+/// Used for non-primary tunnel connections (conn_idx > 0) to avoid
+/// duplicating heartbeat ACK processing.
+pub fn spawn_noop() -> HeartbeatHandle {
+    let (ack_tx, _) = tokio::sync::mpsc::channel::<Bytes>(1);
+    // receiver is immediately dropped; on_ack() calls will silently fail
+    HeartbeatHandle { ack_tx }
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+struct HeartbeatSnapshot {
+    cumulative: TunnelRequestMetricsSnapshot,
+    window: TunnelRequestMetricsSnapshot,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct PendingHeartbeat {
+    heartbeat_id: u64,
+    snapshot: HeartbeatSnapshot,
+    cumulative: TunnelRequestMetricsSnapshot,
+    sent_at: Option<Instant>,
+}
+
+/// Spawn the heartbeat task. Returns a handle for forwarding ACKs.
+pub fn spawn(
+    state: Arc<AppState>,
+    server: Arc<ServerContext>,
+    frame_tx: FrameSender,
+    mut shutdown: watch::Receiver<bool>,
+) -> HeartbeatHandle {
+    let (ack_tx, mut ack_rx) = tokio::sync::mpsc::channel::<Bytes>(4);
+
+    tokio::spawn(async move {
+        // Read initial interval from dynamic config (may be updated by remote config).
+        let initial_interval = Duration::from_secs(server.dynamic.load().heartbeat_interval);
+        let mut current_interval = initial_interval;
+        // At most one in-flight heartbeat snapshot is tracked at a time.
+        // We keep the last ACKed cumulative snapshot so each payload can
+        // report both monotonic totals and the delta since the previous ACK.
+        let mut pending: Option<PendingHeartbeat> = None;
+        let mut last_acked_snapshot = TunnelRequestMetricsSnapshot::default();
+        let mut next_heartbeat_id: u64 = 1;
+        let heartbeat_session_id = format!(
+            "{}-{}",
+            std::process::id(),
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_nanos()
+        );
+
+        // Skip first immediate tick by sleeping first.
+        tokio::time::sleep(current_interval).await;
+
+        loop {
+            tokio::select! {
+                _ = tokio::time::sleep(current_interval) => {
+                    let pending_entry = if let Some(entry) = pending {
+                        entry
+                    } else {
+                        let cumulative = server.metrics.snapshot();
+                        let id = next_heartbeat_id;
+                        next_heartbeat_id = next_heartbeat_id.wrapping_add(1);
+                        if next_heartbeat_id == 0 {
+                            next_heartbeat_id = 1;
+                        }
+                        let window = cumulative.delta_since(last_acked_snapshot);
+                        let entry = PendingHeartbeat {
+                            heartbeat_id: id,
+                            snapshot: HeartbeatSnapshot { cumulative, window },
+                            cumulative,
+                            sent_at: None,
+                        };
+                        pending = Some(entry);
+                        entry
+                    };
+
+                    let payload = build_heartbeat_payload(
+                        &state,
+                        &server,
+                        &heartbeat_session_id,
+                        pending_entry.heartbeat_id,
+                        pending_entry.snapshot
+                    ).await;
+                    let frame = Frame::control(MsgType::HeartbeatData, payload);
+                    if frame_tx.send(frame).await.is_err() {
+                        break; // Writer closed
+                    }
+                    server.tunnel_metrics.record_heartbeat_sent();
+                    if let Some(mut entry) = pending {
+                        entry.sent_at = Some(Instant::now());
+                        pending = Some(entry);
+                    }
+                    debug!("sent heartbeat data");
+
+                    // Re-read interval from dynamic config (remote config may have
+                    // updated it since the last heartbeat).
+                    let new_interval = Duration::from_secs(
+                        server.dynamic.load().heartbeat_interval
+                    );
+                    if new_interval != current_interval {
+                        debug!(
+                            old_secs = current_interval.as_secs(),
+                            new_secs = new_interval.as_secs(),
+                            "heartbeat interval updated from dynamic config"
+                        );
+                        current_interval = new_interval;
+                    }
+                }
+                Some(ack_payload) = ack_rx.recv() => {
+                    match handle_ack(&server, &ack_payload) {
+                        AckDecision::Accept {
+                            heartbeat_id: ack_id,
+                            upgrade_to,
+                        } => {
+                            if let Some(entry) = pending {
+                                if ack_id == entry.heartbeat_id {
+                                    if let Some(sent_at) = entry.sent_at {
+                                        server.tunnel_metrics.record_heartbeat_ack(sent_at.elapsed());
+                                    }
+                                    last_acked_snapshot = entry.cumulative;
+                                    pending = None;
+                                }
+                            }
+                            maybe_trigger_upgrade(upgrade_to);
+                        }
+                        AckDecision::Ignore => {}
+                    }
+                }
+                _ = shutdown.changed() => {
+                    debug!("heartbeat task shutting down");
+                    break;
+                }
+            }
+        }
+    });
+
+    HeartbeatHandle { ack_tx }
+}
+
+async fn build_heartbeat_payload(
+    state: &AppState,
+    server: &ServerContext,
+    heartbeat_session_id: &str,
+    heartbeat_id: u64,
+    snapshot: HeartbeatSnapshot,
+) -> Bytes {
+    let node_id = server.node_id.read().unwrap().clone();
+    let tunnel_snapshot = server.tunnel_metrics.snapshot();
+    let recent_errors = server.tunnel_metrics.recent_errors(8);
+    let resource_usage = state.resource_monitor.snapshot();
+
+    let cumulative = snapshot.cumulative;
+    let window = snapshot.window;
+    let cumulative_metrics = serde_json::json!({
+        "total_requests": cumulative.total_requests,
+        "total_latency_ns": cumulative.total_latency_ns,
+        "avg_latency_ms": cumulative.average_latency_ms(),
+        "failed_requests": cumulative.failed_requests,
+        "dns_failures": cumulative.dns_failures,
+        "stream_errors": cumulative.stream_errors,
+        "slow_requests": cumulative.slow_requests,
+    });
+    let window_metrics = serde_json::json!({
+        "total_requests": window.total_requests,
+        "total_latency_ns": window.total_latency_ns,
+        "avg_latency_ms": window.average_latency_ms(),
+        "failed_requests": window.failed_requests,
+        "dns_failures": window.dns_failures,
+        "stream_errors": window.stream_errors,
+        "slow_requests": window.slow_requests,
+    });
+    let local_admission = state.stream_concurrency_snapshot().map(|snapshot| {
+        serde_json::json!({
+            "limit": snapshot.limit,
+            "in_flight": snapshot.in_flight,
+            "available_permits": snapshot.available_permits,
+            "high_watermark": snapshot.high_watermark,
+            "rejected_total": snapshot.rejected,
+        })
+    });
+    let distributed_admission = match state.distributed_stream_concurrency_snapshot().await {
+        Ok(Some(snapshot)) => Some(serde_json::json!({
+            "limit": snapshot.limit,
+            "in_flight": snapshot.in_flight,
+            "available_permits": snapshot.available_permits,
+            "high_watermark": snapshot.high_watermark,
+            "rejected_total": snapshot.rejected,
+        })),
+        Ok(None) => None,
+        Err(err) => Some(serde_json::json!({
+            "error": err.to_string(),
+        })),
+    };
+    let admission = match (local_admission, distributed_admission) {
+        (None, None) => None,
+        (local, distributed) => Some(serde_json::json!({
+            "local_streams": local,
+            "distributed_streams": distributed,
+        })),
+    };
+
+    let payload = serde_json::json!({
+        "node_id": node_id,
+        "heartbeat_session_id": heartbeat_session_id,
+        "heartbeat_id": heartbeat_id,
+        "heartbeat_interval": server.dynamic.load().heartbeat_interval,
+        "active_connections": server.active_connections.load(Ordering::Acquire),
+        "total_requests": cumulative.total_requests,
+        "avg_latency_ms": cumulative.average_latency_ms(),
+        "failed_requests": cumulative.failed_requests,
+        "dns_failures": cumulative.dns_failures,
+        "stream_errors": cumulative.stream_errors,
+        "slow_requests": cumulative.slow_requests,
+        "window_total_requests": window.total_requests,
+        "window_total_latency_ns": window.total_latency_ns,
+        "window_avg_latency_ms": window.average_latency_ms(),
+        "window_failed_requests": window.failed_requests,
+        "window_dns_failures": window.dns_failures,
+        "window_stream_errors": window.stream_errors,
+        "window_slow_requests": window.slow_requests,
+        "proxy_metrics": {
+            "cumulative": cumulative_metrics,
+            "window": window_metrics,
+        },
+        "proxy_metadata": {
+            "version": CURRENT_VERSION,
+            "admission": admission,
+            "resource_usage": resource_usage,
+            "tunnel_metrics": {
+                "connect_attempts": tunnel_snapshot.connect_attempts,
+                "connect_successes": tunnel_snapshot.connect_successes,
+                "connect_errors": tunnel_snapshot.connect_errors,
+                "disconnects": tunnel_snapshot.disconnects,
+                "last_connected_at_unix_secs": tunnel_snapshot.last_connected_at_unix_secs,
+                "last_disconnected_at_unix_secs": tunnel_snapshot.last_disconnected_at_unix_secs,
+                "last_connected_duration_ms": tunnel_snapshot.last_connected_duration_ms,
+                "connected_duration_total_ms": tunnel_snapshot.connected_duration_total_ms,
+                "heartbeat_sent": tunnel_snapshot.heartbeat_sent,
+                "heartbeat_ack": tunnel_snapshot.heartbeat_ack,
+                "heartbeat_rtt_last_ms": tunnel_snapshot.heartbeat_rtt_last_ms,
+                "heartbeat_rtt_avg_ms": tunnel_snapshot.heartbeat_rtt_avg_ms(),
+                "ws_in_frames": tunnel_snapshot.ws_in_frames,
+                "ws_in_bytes": tunnel_snapshot.ws_in_bytes,
+                "ws_out_frames": tunnel_snapshot.ws_out_frames,
+                "ws_out_bytes": tunnel_snapshot.ws_out_bytes,
+                "error_events_total": tunnel_snapshot.error_events_total,
+            },
+            "recent_tunnel_errors": recent_errors,
+        },
+    });
+
+    Bytes::from(serde_json::to_vec(&payload).unwrap_or_default())
+}
+
+fn handle_ack(server: &ServerContext, payload: &[u8]) -> AckDecision {
+    if payload.is_empty() {
+        warn!("received empty heartbeat ACK");
+        server
+            .tunnel_metrics
+            .record_error("heartbeat_ack_empty", "received empty heartbeat ACK");
+        return AckDecision::Ignore;
+    }
+
+    #[derive(serde::Deserialize)]
+    struct AckPayload {
+        #[serde(default)]
+        remote_config: Option<RemoteConfig>,
+        #[serde(default)]
+        config_version: u64,
+        heartbeat_id: u64,
+        #[serde(default)]
+        upgrade_to: Option<String>,
+    }
+
+    match serde_json::from_slice::<AckPayload>(payload) {
+        Ok(ack) => {
+            if let Some(ref rc) = ack.remote_config {
+                runtime::apply_remote_config(&server.dynamic, rc, ack.config_version);
+            }
+            AckDecision::Accept {
+                heartbeat_id: ack.heartbeat_id,
+                upgrade_to: ack.upgrade_to.and_then(normalize_upgrade_target),
+            }
+        }
+        Err(e) => {
+            warn!(error = %e, "failed to parse heartbeat ACK");
+            server
+                .tunnel_metrics
+                .record_error("heartbeat_ack_parse", &e.to_string());
+            AckDecision::Ignore
+        }
+    }
+}
+
+fn normalize_upgrade_target(raw: String) -> Option<String> {
+    let trimmed = raw.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+    let normalized = trimmed
+        .strip_prefix("tunnel-v")
+        .or_else(|| trimmed.strip_prefix("proxy-v"))
+        .unwrap_or(trimmed);
+    if normalized == CURRENT_VERSION {
+        return None;
+    }
+    Some(normalized.to_string())
+}
+
+fn maybe_trigger_upgrade(version: Option<String>) {
+    let Some(target_version) = version else {
+        return;
+    };
+    if !crate::setup::service::is_root() {
+        if NON_ROOT_UPGRADE_WARNED
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+        {
+            warn!(
+                target_version = %target_version,
+                "remote upgrade skipped: root privileges are required"
+            );
+        }
+        return;
+    }
+    if UPGRADE_IN_PROGRESS
+        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+        .is_err()
+    {
+        debug!(target_version = %target_version, "upgrade already in progress, ignoring");
+        return;
+    }
+
+    tokio::spawn(async move {
+        info!(target_version = %target_version, "received remote upgrade instruction");
+        match crate::setup::upgrade::perform_upgrade(&target_version).await {
+            Ok(()) => {
+                info!(target_version = %target_version, "remote upgrade finished");
+            }
+            Err(e) => {
+                warn!(
+                    target_version = %target_version,
+                    error = %e,
+                    "remote upgrade failed"
+                );
+                UPGRADE_IN_PROGRESS.store(false, Ordering::Release);
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::AtomicU64;
+    use std::sync::{Arc, RwLock};
+
+    use arc_swap::ArcSwap;
+    use clap::Parser;
+
+    use super::{build_heartbeat_payload, handle_ack, AckDecision, HeartbeatSnapshot};
+    use crate::registration::client::AetherClient;
+    use crate::runtime::DynamicConfig;
+    use crate::state::{AppState, ServerContext, TunnelMetrics, TunnelRequestMetrics};
+
+    fn sample_config() -> Arc<crate::config::Config> {
+        Arc::new(crate::config::Config::parse_from([
+            "aether-tunnel",
+            "--aether-url",
+            "https://example.com",
+            "--management-token",
+            "ae_test",
+            "--node-name",
+            "tunnel-test",
+        ]))
+    }
+
+    fn sample_server() -> Arc<ServerContext> {
+        let config = sample_config();
+        Arc::new(ServerContext {
+            server_label: "heartbeat-test".to_string(),
+            aether_url: config.aether_url.clone(),
+            management_token: config.management_token.clone(),
+            node_name: config.node_name.clone(),
+            node_id: Arc::new(RwLock::new("node-123".to_string())),
+            aether_client: Arc::new(AetherClient::new(
+                &config,
+                &config.aether_url,
+                &config.management_token,
+            )),
+            dynamic: Arc::new(ArcSwap::from_pointee(DynamicConfig::from_config(&config))),
+            active_connections: Arc::new(AtomicU64::new(0)),
+            metrics: Arc::new(TunnelRequestMetrics::new()),
+            tunnel_metrics: Arc::new(TunnelMetrics::new()),
+        })
+    }
+
+    fn sample_state(config: Arc<crate::config::Config>) -> AppState {
+        let dns_cache = Arc::new(crate::target_filter::DnsCache::new(
+            std::time::Duration::from_secs(config.dns_cache_ttl_secs),
+            config.dns_cache_capacity,
+        ));
+        AppState {
+            config: Arc::clone(&config),
+            dns_cache: Arc::clone(&dns_cache),
+            upstream_client_pool: crate::upstream_client::UpstreamClientPool::new(
+                config, dns_cache,
+            ),
+            tunnel_tls_config: Arc::new(crate::tunnel::client::build_tls_config()),
+            resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
+            stream_gate: None,
+            distributed_stream_gate: None,
+        }
+    }
+
+    #[test]
+    fn heartbeat_ack_requires_heartbeat_id() {
+        let server = sample_server();
+        let decision = handle_ack(
+            &server,
+            br#"{"config_version":1,"remote_config":{"heartbeat_interval":9}}"#,
+        );
+
+        assert!(matches!(decision, AckDecision::Ignore));
+        assert_eq!(server.dynamic.load().heartbeat_interval, 5);
+    }
+
+    #[test]
+    fn heartbeat_ack_applies_remote_config_with_heartbeat_id() {
+        let server = sample_server();
+        let decision = handle_ack(
+            &server,
+            br#"{"heartbeat_id":7,"config_version":1,"remote_config":{"heartbeat_interval":9}}"#,
+        );
+
+        assert!(matches!(
+            decision,
+            AckDecision::Accept {
+                heartbeat_id: 7,
+                upgrade_to: None
+            }
+        ));
+        assert_eq!(server.dynamic.load().heartbeat_interval, 9);
+    }
+
+    #[tokio::test]
+    async fn heartbeat_payload_reports_resource_usage_and_tunnel_error_diagnostics() {
+        let config = sample_config();
+        let server = sample_server();
+        server
+            .tunnel_metrics
+            .record_error("ws_write_error", "IO error: Connection reset by peer");
+        let state = sample_state(config);
+
+        let payload = build_heartbeat_payload(
+            &state,
+            &server,
+            "session-1",
+            42,
+            HeartbeatSnapshot::default(),
+        )
+        .await;
+        let payload: serde_json::Value =
+            serde_json::from_slice(&payload).expect("heartbeat payload should be JSON");
+        let resource_usage = payload
+            .pointer("/proxy_metadata/resource_usage")
+            .and_then(serde_json::Value::as_object)
+            .expect("resource usage should be reported");
+        assert!(resource_usage.contains_key("system_cpu_usage_percent"));
+        assert!(resource_usage.contains_key("process_memory_bytes"));
+
+        let recent_error = payload
+            .pointer("/proxy_metadata/recent_tunnel_errors/0")
+            .and_then(serde_json::Value::as_object)
+            .expect("recent tunnel error should be reported");
+        assert!(recent_error
+            .get("timestamp_unix_ms")
+            .and_then(serde_json::Value::as_u64)
+            .is_some());
+        assert_eq!(
+            recent_error
+                .get("component")
+                .and_then(serde_json::Value::as_str),
+            Some("tunnel_write")
+        );
+        assert_eq!(
+            recent_error
+                .get("severity")
+                .and_then(serde_json::Value::as_str),
+            Some("error")
+        );
+    }
+}

apps/aether-tunnel/src/tunnel/mod.rs

@@ -0,0 +1,562 @@
+pub mod client;
+pub mod dispatcher;
+pub mod heartbeat;
+pub mod protocol;
+pub mod stream_handler;
+pub mod writer;
+
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use tokio::sync::watch;
+use tracing::{debug, error, info};
+
+use crate::state::{AppState, ServerContext};
+
+/// If a tunnel stays connected at least this long, treat the next disconnect
+/// as a non-failure and reset reconnect backoff.
+const STABLE_SESSION_RESET_AFTER: Duration = Duration::from_secs(30);
+/// Startup staggering step per secondary connection, used to avoid
+/// simultaneous bursts when a pool of tunnels starts together.
+const STARTUP_STAGGER_STEP_MS: u64 = 150;
+/// Upper bound for startup staggering.
+const MAX_STARTUP_STAGGER_MS: u64 = 1_500;
+/// Keep a tiny floor for repeated reconnects; first retry is still immediate.
+const MIN_RECONNECT_DELAY_MS: u64 = 50;
+/// Even under sustained failures, keep probing frequently so recovery is fast
+/// once cross-border network quality improves.
+const RECONNECT_PROBE_MAX_DELAY_MS: u64 = 3_000;
+
+/// Run the tunnel mode main loop (connect, dispatch, reconnect).
+///
+/// `conn_idx` identifies which connection in the pool this is (0-based).
+/// Only connection 0 sends heartbeats to avoid resetting shared metrics.
+pub async fn run(
+    state: &Arc<AppState>,
+    server: &Arc<ServerContext>,
+    conn_idx: usize,
+    mut shutdown: watch::Receiver<bool>,
+    mut drain: watch::Receiver<bool>,
+) {
+    info!(server = %server.server_label, conn = conn_idx, "starting tunnel");
+    let reconnect_salt = compute_connection_salt(server, conn_idx);
+
+    if *drain.borrow() {
+        info!(server = %server.server_label, conn = conn_idx, "tunnel drain requested before startup");
+        return;
+    }
+
+    let startup_delay = compute_startup_stagger(conn_idx, reconnect_salt);
+    if !startup_delay.is_zero() {
+        info!(
+            server = %server.server_label,
+            conn = conn_idx,
+            delay_ms = startup_delay.as_millis(),
+            "startup stagger before first connect"
+        );
+        tokio::select! {
+            _ = tokio::time::sleep(startup_delay) => {}
+            _ = shutdown.changed() => {
+                info!(server = %server.server_label, conn = conn_idx, "shutdown requested during startup stagger");
+                return;
+            }
+            _ = drain.changed() => {
+                if *drain.borrow() {
+                    info!(server = %server.server_label, conn = conn_idx, "tunnel drain requested during startup stagger");
+                    return;
+                }
+            }
+        }
+    }
+
+    let mut consecutive_failures: u32 = 0;
+
+    loop {
+        if *drain.borrow() {
+            info!(server = %server.server_label, conn = conn_idx, "tunnel drained, exiting slot");
+            return;
+        }
+        server.tunnel_metrics.record_connect_attempt();
+        let started_at = Instant::now();
+        match client::connect_and_run(state, server, conn_idx, &mut shutdown, drain.clone()).await {
+            Ok(client::TunnelOutcome::Shutdown) => {
+                info!(server = %server.server_label, conn = conn_idx, "tunnel shut down gracefully");
+                return;
+            }
+            Ok(client::TunnelOutcome::Disconnected) => {
+                debug!(server = %server.server_label, conn = conn_idx, "tunnel disconnected, reconnecting");
+            }
+            Err(e) => {
+                server.tunnel_metrics.record_connect_error();
+                server
+                    .tunnel_metrics
+                    .record_error("tunnel_connect_error", &e.to_string());
+                error!(server = %server.server_label, conn = conn_idx, error = %e, "tunnel connection error, reconnecting");
+            }
+        }
+
+        if *shutdown.borrow() {
+            info!(server = %server.server_label, conn = conn_idx, "shutdown requested, not reconnecting");
+            return;
+        }
+        if *drain.borrow() {
+            info!(server = %server.server_label, conn = conn_idx, "tunnel drained after disconnect");
+            return;
+        }
+
+        // Reset backoff after a stable session to keep recovery snappy when
+        // failures are only occasional.
+        let connected_for = started_at.elapsed();
+        if connected_for >= STABLE_SESSION_RESET_AFTER {
+            consecutive_failures = 0;
+        } else {
+            consecutive_failures = consecutive_failures.saturating_add(1);
+        }
+
+        let reconnect_delay = compute_reconnect_delay(
+            state.config.tunnel_reconnect_base_ms,
+            state.config.tunnel_reconnect_max_ms,
+            consecutive_failures,
+            reconnect_salt,
+        );
+        if reconnect_delay.is_zero() && consecutive_failures <= 1 {
+            debug!(
+                server = %server.server_label,
+                conn = conn_idx,
+                failures = consecutive_failures,
+                delay_ms = reconnect_delay.as_millis(),
+                "waiting before reconnect"
+            );
+        } else {
+            info!(
+                server = %server.server_label,
+                conn = conn_idx,
+                failures = consecutive_failures,
+                delay_ms = reconnect_delay.as_millis(),
+                "waiting before reconnect"
+            );
+        }
+
+        tokio::select! {
+            _ = tokio::time::sleep(reconnect_delay) => {}
+            _ = shutdown.changed() => {
+                info!(server = %server.server_label, conn = conn_idx, "shutdown requested during reconnect wait");
+                return;
+            }
+            _ = drain.changed() => {
+                if *drain.borrow() {
+                    info!(server = %server.server_label, conn = conn_idx, "tunnel drain requested during reconnect wait");
+                    return;
+                }
+            }
+        }
+    }
+}
+
+fn compute_connection_salt(server: &ServerContext, conn_idx: usize) -> u64 {
+    // FNV-1a style hash over server label + connection index.
+    let mut h: u64 = 0xcbf29ce484222325;
+    for &b in server.server_label.as_bytes() {
+        h ^= b as u64;
+        h = h.wrapping_mul(0x100000001b3);
+    }
+    h ^= conn_idx as u64;
+    mix_u64(h)
+}
+
+fn compute_startup_stagger(conn_idx: usize, salt: u64) -> Duration {
+    if conn_idx == 0 {
+        return Duration::ZERO;
+    }
+    let base = (conn_idx as u64).saturating_mul(STARTUP_STAGGER_STEP_MS);
+    let jitter = mix_u64(salt) % 301; // 0..=300ms
+    Duration::from_millis((base + jitter).min(MAX_STARTUP_STAGGER_MS))
+}
+
+fn compute_reconnect_delay(
+    base_ms: u64,
+    max_ms: u64,
+    consecutive_failures: u32,
+    salt: u64,
+) -> Duration {
+    // First retry should be immediate to maximize recovery speed on transient
+    // blips (the user's primary expectation in poor networks).
+    if consecutive_failures <= 1 {
+        return Duration::ZERO;
+    }
+
+    // Keep a sane minimum for repeated failures.
+    let base_ms = base_ms.max(MIN_RECONNECT_DELAY_MS);
+    let max_ms = max_ms.max(base_ms);
+    let cap_ms = compute_reconnect_cap_ms(base_ms, max_ms, consecutive_failures)
+        .min(RECONNECT_PROBE_MAX_DELAY_MS.max(base_ms));
+
+    // Equal-jitter: randomize in [cap/2, cap], preventing synchronized reconnect
+    // storms while keeping reconnect latency bounded.
+    if cap_ms <= 1 {
+        return Duration::from_millis(cap_ms);
+    }
+
+    let half = cap_ms / 2;
+    let span = cap_ms - half;
+    let now_nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.subsec_nanos() as u64)
+        .unwrap_or(0);
+    let mixed = mix_u64(now_nanos ^ salt);
+    let jitter = if span == 0 { 0 } else { mixed % (span + 1) };
+    Duration::from_millis(half + jitter)
+}
+
+fn compute_reconnect_cap_ms(base_ms: u64, max_ms: u64, consecutive_failures: u32) -> u64 {
+    if consecutive_failures <= 1 {
+        return base_ms.min(max_ms);
+    }
+
+    let shift = (consecutive_failures - 1).min(31);
+    let factor = 1u64 << shift;
+    base_ms.saturating_mul(factor).min(max_ms)
+}
+
+fn mix_u64(mut x: u64) -> u64 {
+    // SplitMix64 finalizer - cheap bit mixing for pseudo-random jitter.
+    x ^= x >> 30;
+    x = x.wrapping_mul(0xbf58476d1ce4e5b9);
+    x ^= x >> 27;
+    x = x.wrapping_mul(0x94d049bb133111eb);
+    x ^ (x >> 31)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::AtomicU64;
+    use std::sync::{Arc, Once};
+    use std::time::Duration;
+
+    use aether_gateway::{build_router_with_state, AppState as GatewayAppState};
+    use arc_swap::ArcSwap;
+    use axum::Router;
+    use reqwest::StatusCode;
+    use tokio::sync::watch;
+
+    use crate::config::Config;
+    use crate::registration::client::AetherClient;
+    use crate::runtime::DynamicConfig;
+    use crate::state::{
+        AppState as TunnelAppState, ServerContext, TunnelMetrics, TunnelRequestMetrics,
+    };
+    use crate::target_filter::DnsCache;
+    use crate::tunnel::protocol;
+    use crate::upstream_client;
+
+    use super::{
+        compute_reconnect_cap_ms, compute_reconnect_delay, compute_startup_stagger, run,
+        MAX_STARTUP_STAGGER_MS, RECONNECT_PROBE_MAX_DELAY_MS, STARTUP_STAGGER_STEP_MS,
+    };
+
+    #[test]
+    fn reconnect_cap_grows_exponentially_and_caps() {
+        let base = 500;
+        let max = 30_000;
+        assert_eq!(compute_reconnect_cap_ms(base, max, 0), 500);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 1), 500);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 2), 1_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 3), 2_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 4), 4_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 5), 8_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 6), 16_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 7), 30_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 20), 30_000);
+    }
+
+    #[test]
+    fn startup_stagger_is_zero_for_primary_and_bounded_for_secondary() {
+        assert_eq!(compute_startup_stagger(0, 42), Duration::ZERO);
+
+        let d1 = compute_startup_stagger(1, 42);
+        let d2 = compute_startup_stagger(2, 42);
+
+        assert!(d1 >= Duration::from_millis(STARTUP_STAGGER_STEP_MS));
+        assert!(d1 <= Duration::from_millis(MAX_STARTUP_STAGGER_MS));
+        assert!(d2 >= Duration::from_millis(STARTUP_STAGGER_STEP_MS * 2));
+        assert!(d2 <= Duration::from_millis(MAX_STARTUP_STAGGER_MS));
+    }
+
+    #[test]
+    fn reconnect_delay_is_immediate_on_first_failure() {
+        assert_eq!(compute_reconnect_delay(700, 45_000, 1, 123), Duration::ZERO);
+    }
+
+    #[test]
+    fn reconnect_delay_stays_within_probe_ceiling_after_many_failures() {
+        let d = compute_reconnect_delay(500, 45_000, 100, 12345);
+        assert!(d <= Duration::from_millis(RECONNECT_PROBE_MAX_DELAY_MS));
+    }
+
+    #[tokio::test]
+    async fn tunnel_reconnects_after_gateway_restart() {
+        ensure_rustls_provider();
+
+        let gateway_port = reserve_local_port().expect("gateway port should reserve");
+        let gateway_base_url = format!("http://127.0.0.1:{gateway_port}");
+        let (gateway_state, mut gateway_handle) = start_gateway_on_port(gateway_port)
+            .await
+            .expect("gateway should start");
+
+        let state = sample_state(sample_config(&gateway_base_url));
+        let server = sample_server(&state, "node-recovery");
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
+        let tunnel_task = tokio::spawn({
+            let state = Arc::clone(&state);
+            let server = Arc::clone(&server);
+            let (_drain_tx, drain_rx) = watch::channel(false);
+            async move {
+                run(&state, &server, 0, shutdown_rx, drain_rx).await;
+            }
+        });
+
+        wait_until_relay_status(
+            &gateway_base_url,
+            "node-recovery",
+            StatusCode::GATEWAY_TIMEOUT,
+        )
+        .await;
+
+        assert_eq!(gateway_state.force_close_all_tunnel_proxies(), 1);
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        gateway_handle.abort();
+
+        let (_restarted_gateway_state, restarted_gateway_handle) =
+            start_gateway_on_port_retry(gateway_port)
+                .await
+                .expect("gateway should restart on fixed port");
+        gateway_handle = restarted_gateway_handle;
+
+        wait_until_relay_status(
+            &gateway_base_url,
+            "node-recovery",
+            StatusCode::GATEWAY_TIMEOUT,
+        )
+        .await;
+
+        let _ = shutdown_tx.send(true);
+        tokio::time::timeout(Duration::from_secs(5), tunnel_task)
+            .await
+            .expect("tunnel task should stop")
+            .expect("tunnel task should join");
+        gateway_handle.abort();
+    }
+
+    async fn wait_until_relay_status(gateway_base_url: &str, node_id: &str, expected: StatusCode) {
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+        let mut last_observed = None::<String>;
+        loop {
+            if let Some((status, body)) = probe_relay_status(gateway_base_url, node_id).await {
+                last_observed = Some(format!("{status} body={body}"));
+                if status == expected {
+                    return;
+                }
+            }
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "relay status did not become {expected} within timeout; last={:?}",
+                last_observed
+            );
+            tokio::time::sleep(Duration::from_millis(25)).await;
+        }
+    }
+
+    async fn probe_relay_status(
+        gateway_base_url: &str,
+        node_id: &str,
+    ) -> Option<(StatusCode, String)> {
+        let response = reqwest::Client::new()
+            .post(format!(
+                "{gateway_base_url}/api/internal/tunnel/relay/{node_id}"
+            ))
+            .header("content-type", "application/octet-stream")
+            .body(relay_probe_envelope())
+            .send()
+            .await
+            .ok()?;
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        Some((status, body))
+    }
+
+    fn relay_probe_envelope() -> Vec<u8> {
+        let meta = protocol::RequestMeta {
+            provider_id: None,
+            endpoint_id: None,
+            key_id: None,
+            method: "GET".to_string(),
+            url: "http://127.0.0.1:80/blocked".to_string(),
+            headers: std::collections::HashMap::new(),
+            timeout: 5,
+            follow_redirects: None,
+            http1_only: false,
+            transport_profile: None,
+        };
+        let meta_json =
+            serde_json::to_vec(&meta).expect("tunnel relay probe metadata should serialize");
+        let mut envelope = Vec::with_capacity(4 + meta_json.len());
+        envelope.extend_from_slice(&(meta_json.len() as u32).to_be_bytes());
+        envelope.extend_from_slice(&meta_json);
+        envelope
+    }
+
+    async fn start_gateway_on_port(
+        port: u16,
+    ) -> Result<(GatewayAppState, tokio::task::JoinHandle<()>), std::io::Error> {
+        let state = GatewayAppState::new().expect("gateway test state should build");
+        let router = build_router_with_state(state.clone());
+        let handle = spawn_router_on_port(port, router).await?;
+        Ok((state, handle))
+    }
+
+    async fn start_gateway_on_port_retry(
+        port: u16,
+    ) -> Result<(GatewayAppState, tokio::task::JoinHandle<()>), std::io::Error> {
+        let mut attempts = 0usize;
+        loop {
+            match start_gateway_on_port(port).await {
+                Ok(server) => return Ok(server),
+                Err(err) => {
+                    attempts += 1;
+                    if attempts >= 20 {
+                        return Err(err);
+                    }
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                }
+            }
+        }
+    }
+
+    async fn spawn_router_on_port(
+        port: u16,
+        app: Router,
+    ) -> Result<tokio::task::JoinHandle<()>, std::io::Error> {
+        let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await?;
+        Ok(tokio::spawn(async move {
+            axum::serve(
+                listener,
+                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
+            )
+            .await
+            .expect("gateway test server should run");
+        }))
+    }
+
+    fn reserve_local_port() -> Result<u16, std::io::Error> {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
+        let port = listener.local_addr()?.port();
+        drop(listener);
+        Ok(port)
+    }
+
+    fn sample_state(config: Config) -> Arc<TunnelAppState> {
+        let config = Arc::new(config);
+        let dns_cache = Arc::new(DnsCache::new(Duration::from_secs(60), 128));
+        let upstream_client_pool =
+            upstream_client::UpstreamClientPool::new(Arc::clone(&config), Arc::clone(&dns_cache));
+        Arc::new(TunnelAppState {
+            config,
+            dns_cache,
+            upstream_client_pool,
+            tunnel_tls_config: Arc::new(crate::tunnel::client::build_tls_config()),
+            resource_monitor: Arc::new(crate::hardware::RuntimeResourceMonitor::new()),
+            stream_gate: None,
+            distributed_stream_gate: None,
+        })
+    }
+
+    fn sample_server(state: &Arc<TunnelAppState>, node_id: &str) -> Arc<ServerContext> {
+        let config = Arc::clone(&state.config);
+        Arc::new(ServerContext {
+            server_label: "gateway-owned-tunnel".to_string(),
+            aether_url: config.aether_url.clone(),
+            management_token: config.management_token.clone(),
+            node_name: config.node_name.clone(),
+            node_id: Arc::new(std::sync::RwLock::new(node_id.to_string())),
+            aether_client: Arc::new(AetherClient::new(
+                &config,
+                &config.aether_url,
+                &config.management_token,
+            )),
+            dynamic: Arc::new(ArcSwap::from_pointee(DynamicConfig::from_config(&config))),
+            active_connections: Arc::new(AtomicU64::new(0)),
+            metrics: Arc::new(TunnelRequestMetrics::new()),
+            tunnel_metrics: Arc::new(TunnelMetrics::new()),
+        })
+    }
+
+    fn sample_config(aether_url: &str) -> Config {
+        Config {
+            aether_url: aether_url.to_string(),
+            management_token: "token".to_string(),
+            public_ip: None,
+            node_name: "tunnel-test".to_string(),
+            node_region: None,
+            heartbeat_interval: 1,
+            allowed_ports: vec![80, 443],
+            allow_private_targets: false,
+            aether_request_timeout_secs: 10,
+            aether_connect_timeout_secs: 2,
+            aether_pool_max_idle_per_host: 8,
+            aether_pool_idle_timeout_secs: 90,
+            aether_tcp_keepalive_secs: 60,
+            aether_tcp_nodelay: true,
+            aether_http2: true,
+            aether_outbound_proxy_url: None,
+            aether_retry_max_attempts: 1,
+            aether_retry_base_delay_ms: 50,
+            aether_retry_max_delay_ms: 100,
+            diagnostics_bind: None,
+            max_concurrent_connections: None,
+            max_in_flight_streams: None,
+            distributed_stream_limit: None,
+            distributed_stream_redis_url: None,
+            distributed_stream_redis_key_prefix: None,
+            distributed_stream_lease_ttl_ms: 30_000,
+            distributed_stream_renew_interval_ms: 10_000,
+            distributed_stream_command_timeout_ms: 1_000,
+            dns_cache_ttl_secs: 60,
+            dns_cache_capacity: 128,
+            upstream_connect_timeout_secs: 30,
+            upstream_pool_max_idle_per_host: 4,
+            upstream_pool_idle_timeout_secs: 60,
+            upstream_tcp_keepalive_secs: 60,
+            upstream_tcp_nodelay: true,
+            upstream_proxy_url: None,
+            redirect_replay_budget_bytes: crate::config::DEFAULT_REDIRECT_REPLAY_BUDGET_BYTES,
+            emit_proxy_timing_header: true,
+            log_level: "info".to_string(),
+            log_destination: crate::config::TunnelLogDestinationArg::Stdout,
+            log_dir: None,
+            log_rotation: crate::config::TunnelLogRotationArg::Daily,
+            log_retention_days: 7,
+            log_max_files: 30,
+            tunnel_reconnect_base_ms: 50,
+            tunnel_reconnect_max_ms: 250,
+            tunnel_ping_interval_ms: 1_000,
+            tunnel_max_streams: Some(8),
+            tunnel_connect_timeout_ms: 2_000,
+            tunnel_tcp_keepalive_secs: 30,
+            tunnel_tcp_nodelay: true,
+            tunnel_stale_timeout_ms: 5_000,
+            tunnel_connections: Some(1),
+            tunnel_connections_max: Some(1),
+            tunnel_scale_check_interval_ms: 1_000,
+            tunnel_scale_up_threshold_percent: 70,
+            tunnel_scale_down_threshold_percent: 35,
+            tunnel_scale_down_grace_secs: 15,
+        }
+    }
+
+    fn ensure_rustls_provider() {
+        static INIT: Once = Once::new();
+        INIT.call_once(|| {
+            let _ = rustls::crypto::ring::default_provider().install_default();
+        });
+    }
+}

apps/aether-tunnel/src/tunnel/protocol.rs

@@ -0,0 +1 @@
+pub use aether_contracts::tunnel::*;

apps/aether-tunnel/src/tunnel/writer.rs

@@ -0,0 +1,308 @@
+//! Dedicated WebSocket writer task.
+//!
+//! All frame writes go through an mpsc channel to a single writer task,
+//! avoiding contention on the WebSocket sink.  The writer also sends
+//! periodic WebSocket Ping frames to keep the connection alive through
+//! intermediary proxies (Nginx, Cloudflare, etc.).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use aether_contracts::tunnel::{MsgType, HEADER_SIZE};
+#[cfg(test)]
+use aether_runtime::QueueSnapshot;
+use aether_runtime::{bounded_queue, BoundedQueueSender, QueueSendError};
+use futures_util::SinkExt;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::tungstenite::Message;
+use tracing::{debug, error, trace};
+
+use crate::state::TunnelMetrics;
+
+use super::protocol::Frame;
+
+const HIGH_PRIORITY_QUEUE_CAPACITY: usize = 64;
+const NORMAL_PRIORITY_QUEUE_CAPACITY: usize = 256;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FramePriority {
+    High,
+    Normal,
+}
+
+#[cfg(test)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FrameQueueSnapshots {
+    pub high: QueueSnapshot,
+    pub normal: QueueSnapshot,
+}
+
+/// Sender half — cloned by stream handlers and heartbeat.
+#[derive(Debug, Clone)]
+pub struct FrameSender {
+    high_tx: BoundedQueueSender<Frame>,
+    normal_tx: BoundedQueueSender<Frame>,
+}
+
+impl FrameSender {
+    pub async fn send(&self, frame: Frame) -> Result<(), QueueSendError<Frame>> {
+        match classify_frame_priority(&frame) {
+            FramePriority::High => self.high_tx.send(frame).await,
+            FramePriority::Normal => self.normal_tx.send(frame).await,
+        }
+    }
+
+    pub fn try_send(&self, frame: Frame) -> Result<(), QueueSendError<Frame>> {
+        match classify_frame_priority(&frame) {
+            FramePriority::High => self.high_tx.try_send(frame),
+            FramePriority::Normal => self.normal_tx.try_send(frame),
+        }
+    }
+
+    #[cfg(test)]
+    pub fn snapshots(&self) -> FrameQueueSnapshots {
+        FrameQueueSnapshots {
+            high: self.high_tx.snapshot(),
+            normal: self.normal_tx.snapshot(),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn from_test_queues(
+        high_tx: BoundedQueueSender<Frame>,
+        normal_tx: BoundedQueueSender<Frame>,
+    ) -> Self {
+        Self { high_tx, normal_tx }
+    }
+}
+
+/// Spawn the writer task. Returns the sender and a JoinHandle for cleanup.
+///
+/// `ping_interval` controls WebSocket-level Ping frequency (typically 15s).
+/// This keeps the connection alive through intermediary proxies/load-balancers.
+#[cfg(test)]
+pub fn spawn_writer<S>(sink: S, ping_interval: Duration) -> (FrameSender, JoinHandle<()>)
+where
+    S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin + Send + 'static,
+{
+    spawn_writer_with_metrics(sink, ping_interval, None)
+}
+
+/// Spawn the writer task with optional tunnel metrics instrumentation.
+pub fn spawn_writer_with_metrics<S>(
+    mut sink: S,
+    ping_interval: Duration,
+    tunnel_metrics: Option<Arc<TunnelMetrics>>,
+) -> (FrameSender, JoinHandle<()>)
+where
+    S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin + Send + 'static,
+{
+    let (high_tx, mut high_rx) = bounded_queue::<Frame>(HIGH_PRIORITY_QUEUE_CAPACITY);
+    let (normal_tx, mut normal_rx) = bounded_queue::<Frame>(NORMAL_PRIORITY_QUEUE_CAPACITY);
+    let tx = FrameSender { high_tx, normal_tx };
+
+    let handle = tokio::spawn(async move {
+        let mut ping_ticker = tokio::time::interval(ping_interval);
+        let mut high_open = true;
+        let mut normal_open = true;
+        ping_ticker.tick().await; // skip first immediate tick
+
+        loop {
+            if let Ok(frame) = high_rx.try_recv() {
+                if !write_frame(&mut sink, frame, tunnel_metrics.as_deref()).await {
+                    break;
+                }
+                continue;
+            }
+            if !high_open && !normal_open {
+                break;
+            }
+
+            tokio::select! {
+                biased;
+                frame = high_rx.recv(), if high_open => {
+                    match frame {
+                        Some(frame) => {
+                            if !write_frame(&mut sink, frame, tunnel_metrics.as_deref()).await {
+                                break;
+                            }
+                        }
+                        None => high_open = false,
+                    }
+                }
+                _ = ping_ticker.tick(), if high_open || normal_open => {
+                    if let Err(e) = sink.send(Message::Ping(vec![])).await {
+                        error!(error = %e, "failed to send WebSocket ping");
+                        if let Some(metrics) = tunnel_metrics.as_deref() {
+                            metrics.record_error("ws_ping_error", &e.to_string());
+                        }
+                        break;
+                    }
+                    trace!("sent WebSocket ping");
+                }
+                frame = normal_rx.recv(), if normal_open => {
+                    match frame {
+                        Some(frame) => {
+                            if !write_frame(&mut sink, frame, tunnel_metrics.as_deref()).await {
+                                break;
+                            }
+                        }
+                        None => normal_open = false,
+                    }
+                }
+            }
+        }
+        debug!("writer task exiting");
+        let _ = sink.close().await;
+    });
+
+    (tx, handle)
+}
+
+fn classify_frame_priority(frame: &Frame) -> FramePriority {
+    match frame.msg_type {
+        MsgType::ResponseHeaders
+        | MsgType::StreamError
+        | MsgType::Ping
+        | MsgType::Pong
+        | MsgType::GoAway
+        | MsgType::HeartbeatData
+        | MsgType::HeartbeatAck => FramePriority::High,
+        MsgType::RequestHeaders
+        | MsgType::RequestBody
+        | MsgType::ResponseBody
+        | MsgType::StreamEnd => FramePriority::Normal,
+    }
+}
+
+async fn write_frame<S>(sink: &mut S, frame: Frame, tunnel_metrics: Option<&TunnelMetrics>) -> bool
+where
+    S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin + Send + 'static,
+{
+    let stream_id = frame.stream_id;
+    let msg_type = frame.msg_type;
+    let flags = frame.flags;
+    let data = frame.encode();
+    let wire_len = data.len().max(HEADER_SIZE);
+    if let Err(e) = sink.send(Message::Binary(data.into())).await {
+        error!(
+            stream_id = stream_id,
+            msg_type = ?msg_type,
+            flags = flags,
+            wire_len = wire_len,
+            error = %e,
+            "failed to write frame to WebSocket"
+        );
+        if let Some(metrics) = tunnel_metrics {
+            metrics.record_error("ws_write_error", &e.to_string());
+        }
+        return false;
+    }
+    if let Some(metrics) = tunnel_metrics {
+        metrics.record_ws_outgoing_frame(wire_len);
+    }
+    true
+}
+
+#[cfg(test)]
+mod tests {
+    use std::pin::Pin;
+    use std::sync::{Arc, Mutex};
+    use std::task::{Context, Poll};
+    use std::time::Duration;
+
+    use futures_util::Sink;
+    use tokio_tungstenite::tungstenite::{Error, Message};
+
+    use super::spawn_writer;
+    use crate::tunnel::protocol::Frame;
+    use aether_contracts::tunnel::MsgType;
+
+    #[derive(Clone, Default)]
+    struct VecSink {
+        sent: Arc<Mutex<Vec<Message>>>,
+    }
+
+    impl Sink<Message> for VecSink {
+        type Error = Error;
+
+        fn poll_ready(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn start_send(self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
+            self.sent.lock().expect("sink lock").push(item);
+            Ok(())
+        }
+
+        fn poll_flush(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_close(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[tokio::test]
+    async fn prioritizes_control_frames_ahead_of_buffered_body_frames() {
+        let sink = VecSink::default();
+        let sent = Arc::clone(&sink.sent);
+        let (sender, handle) = spawn_writer(sink, Duration::from_secs(60));
+
+        for idx in 0..8u8 {
+            sender
+                .try_send(Frame::new(
+                    7,
+                    MsgType::ResponseBody,
+                    0,
+                    bytes::Bytes::from(vec![idx; 32]),
+                ))
+                .expect("frame send should succeed");
+        }
+        sender
+            .try_send(Frame::new(
+                7,
+                MsgType::StreamError,
+                0,
+                bytes::Bytes::from_static(b"boom"),
+            ))
+            .expect("frame send should succeed");
+        let snapshots = sender.snapshots();
+        assert!(snapshots.high.enqueued_total >= 1);
+        assert!(snapshots.normal.enqueued_total >= 8);
+
+        tokio::time::sleep(Duration::from_millis(30)).await;
+        drop(sender);
+        handle.await.expect("writer should exit cleanly");
+
+        let sent = sent.lock().expect("sink lock");
+        assert!(
+            sent.len() >= 2,
+            "writer should flush both body and control frames"
+        );
+        let first = match &sent[0] {
+            Message::Binary(data) => {
+                Frame::decode(data.clone().into()).expect("frame should decode")
+            }
+            other => panic!("unexpected first message: {other:?}"),
+        };
+        let second = match &sent[1] {
+            Message::Binary(data) => {
+                Frame::decode(data.clone().into()).expect("frame should decode")
+            }
+            other => panic!("unexpected second message: {other:?}"),
+        };
+        assert_eq!(first.msg_type, MsgType::StreamError);
+        assert_eq!(second.msg_type, MsgType::ResponseBody);
+    }
+}