fix(socket): follow HTTP 3xx during WebSocket handshake (OPENHUMAN-TAURI-9X) (#1547)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/openhuman/socket/event_handlers.rs

@@ -244,6 +244,7 @@ mod tests {
             webhook_router: RwLock::new(None),
             status: RwLock::new(ConnectionStatus::Disconnected),
             socket_id: RwLock::new(None),
+            error: RwLock::new(None),
         })
     }
 

src/openhuman/socket/manager.rs

@@ -51,6 +51,10 @@ pub(super) struct SharedState {
     pub(super) status: RwLock<ConnectionStatus>,
     /// Socket ID assigned by the server.
     pub(super) socket_id: RwLock<Option<String>>,
+    /// Last user-visible connection warning surfaced through `SocketState.error`
+    /// (e.g. "backend redirected ws→wss; update BACKEND_URL"). Cleared on every
+    /// successful handshake and on disconnect.
+    pub(super) error: RwLock<Option<String>>,
 }
 
 // ---------------------------------------------------------------------------
@@ -82,6 +86,7 @@ impl SocketManager {
                 webhook_router: RwLock::new(None),
                 status: RwLock::new(ConnectionStatus::Disconnected),
                 socket_id: RwLock::new(None),
+                error: RwLock::new(None),
             }),
             emit_tx: tokio::sync::Mutex::new(None),
             shutdown_tx: tokio::sync::Mutex::new(None),
@@ -105,7 +110,7 @@ impl SocketManager {
         SocketState {
             status: *self.shared.status.read(),
             socket_id: self.shared.socket_id.read().clone(),
-            error: None,
+            error: self.shared.error.read().clone(),
         }
     }
 
@@ -133,6 +138,7 @@ impl SocketManager {
         log::info!("[socket] Connecting to {}", url);
 
         *self.shared.status.write() = ConnectionStatus::Connecting;
+        *self.shared.error.write() = None;
         emit_state_change(&self.shared);
 
         let (emit_tx, emit_rx) = mpsc::unbounded_channel::<String>();
@@ -165,6 +171,7 @@ impl SocketManager {
         }
         *self.shared.status.write() = ConnectionStatus::Disconnected;
         *self.shared.socket_id.write() = None;
+        *self.shared.error.write() = None;
         emit_state_change(&self.shared);
         log::debug!("[socket] Disconnected");
         Ok(())
@@ -247,6 +254,18 @@ mod tests {
         assert_eq!(state.socket_id.as_deref(), Some("sid-abc"));
     }
 
+    #[test]
+    fn get_state_surfaces_stored_error_to_callers() {
+        let mgr = SocketManager::new();
+        *mgr.shared.error.write() =
+            Some("backend redirected ws→wss; update BACKEND_URL".to_string());
+        let state = mgr.get_state();
+        assert_eq!(
+            state.error.as_deref(),
+            Some("backend redirected ws→wss; update BACKEND_URL")
+        );
+    }
+
     #[tokio::test]
     async fn emit_without_connection_errors_without_panic() {
         let mgr = SocketManager::new();
@@ -269,6 +288,7 @@ mod tests {
             webhook_router: RwLock::new(None),
             status: RwLock::new(ConnectionStatus::Connecting),
             socket_id: RwLock::new(None),
+            error: RwLock::new(None),
         };
         // Must not panic even with all default state.
         emit_state_change(&shared);
@@ -280,6 +300,7 @@ mod tests {
             webhook_router: RwLock::new(None),
             status: RwLock::new(ConnectionStatus::Connected),
             socket_id: RwLock::new(Some("x".into())),
+            error: RwLock::new(None),
         };
         // Pure logging — must not touch state or panic.
         emit_server_event(&shared, "any.event", json!({}));

src/openhuman/socket/ws_loop.rs

@@ -6,7 +6,10 @@ use futures_util::{SinkExt, StreamExt};
 use serde_json::json;
 use tokio::sync::{mpsc, watch};
 use tokio::time::{Duration, Instant};
-use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage};
+use tokio_tungstenite::{
+    connect_async,
+    tungstenite::{http::StatusCode, Error as WsError, Message as WsMessage},
+};
 
 use crate::api::models::socket::ConnectionStatus;
 
@@ -14,6 +17,14 @@ use super::event_handlers::{handle_sio_event, parse_sio_event};
 use super::manager::{emit_state_change, SharedState};
 use super::types::{ConnectionOutcome, WsStream};
 
+/// Maximum HTTP redirect hops to follow during a single WebSocket connect attempt.
+///
+/// Cloudflare and similar edges return a single 301 (e.g. when the configured
+/// `BACKEND_URL` is `http://...` and the server only serves the upgrade over TLS)
+/// before the upgrade succeeds. Three hops is enough headroom for chained
+/// redirects while still bounding pathological loops.
+const MAX_REDIRECT_HOPS: u8 = 3;
+
 // ---------------------------------------------------------------------------
 // Background loop
 // ---------------------------------------------------------------------------
@@ -30,6 +41,13 @@ pub(super) async fn ws_loop(
     let mut backoff = Duration::from_millis(1000);
     let max_backoff = Duration::from_secs(30);
 
+    // `ws_url` is the *resolved* socket URL we're currently connecting to.
+    // If the backend responds with an HTTP 3xx during the upgrade (typical when
+    // BACKEND_URL is configured as `http://` and the edge forces TLS), we
+    // follow the Location header and pin the resolved URL here so subsequent
+    // reconnects skip the redirect round-trip entirely.
+    let mut ws_url = crate::api::socket::websocket_url(&url);
+
     loop {
         if *shutdown_rx.borrow() {
             break;
@@ -40,7 +58,7 @@ pub(super) async fn ws_loop(
         emit_state_change(&shared);
 
         let outcome = run_connection(
-            &url,
+            &mut ws_url,
             &token,
             &shared,
             &mut emit_rx,
@@ -93,21 +111,27 @@ pub(super) async fn ws_loop(
 // ---------------------------------------------------------------------------
 
 /// Run a single WebSocket connection through handshake and event loop.
+///
+/// `ws_url` is taken by mutable reference so that any HTTP redirect we follow
+/// during the upgrade (see `connect_with_redirects`) is pinned for the next
+/// reconnect attempt — we don't want to re-hit the redirect every time the
+/// loop backs off and retries.
 async fn run_connection(
-    url: &str,
+    ws_url: &mut String,
     token: &str,
     shared: &Arc<SharedState>,
     emit_rx: &mut mpsc::UnboundedReceiver<String>,
     shutdown_rx: &mut watch::Receiver<bool>,
     internal_tx: &mpsc::UnboundedSender<String>,
 ) -> ConnectionOutcome {
-    // 1. Build WebSocket URL (appends /socket.io/?EIO=4&transport=websocket)
-    let ws_url = crate::api::socket::websocket_url(url);
     log::info!("[socket] WS URL: {}", ws_url);
 
-    // 2. Connect via WebSocket (uses rustls TLS for wss://)
-    let (ws_stream, _response) = match connect_async(&ws_url).await {
-        Ok(r) => r,
+    // 2. Connect via WebSocket (uses rustls TLS for wss://). Follow HTTP 3xx
+    //    redirects up to MAX_REDIRECT_HOPS so a `http://` config behind a
+    //    Cloudflare-style edge that 301s to `https://` connects cleanly
+    //    instead of looping at error-level forever.
+    let ws_stream = match connect_with_redirects(ws_url, shared).await {
+        Ok(stream) => stream,
         Err(e) => return ConnectionOutcome::Failed(format!("WebSocket connect: {e}")),
     };
 
@@ -404,6 +428,154 @@ fn handle_sio_packet(
     }
 }
 
+// ---------------------------------------------------------------------------
+// Redirect-following connect
+// ---------------------------------------------------------------------------
+
+/// Connect to `ws_url`, following HTTP 3xx redirects up to `MAX_REDIRECT_HOPS`.
+///
+/// Plain `connect_async` returns an error on any non-`101 Switching Protocols`
+/// response, so a Cloudflare-style `http://… → https://…` 301 (which happens
+/// whenever `BACKEND_URL` is configured without TLS) used to be fatal — the
+/// reconnect loop would hammer the same dead URL forever at error level.
+///
+/// On each redirect we:
+///   1. resolve the `Location` header against the current URL (handles relative
+///      Location values),
+///   2. upgrade the scheme so the next attempt is still a WebSocket
+///      (`http` → `ws`, `https` → `wss`; `ws`/`wss` pass through),
+///   3. mutate `ws_url` in place so the redirect target is pinned for
+///      subsequent reconnects (no need to re-hit the redirect every retry),
+///   4. record a one-shot warning in `SharedState.error` the first time we
+///      follow a redirect so the UI can surface "your `BACKEND_URL` is stale".
+///
+/// On non-redirect failures the original error is returned and the caller
+/// counts it toward the exponential backoff like before.
+async fn connect_with_redirects(
+    ws_url: &mut String,
+    shared: &Arc<SharedState>,
+) -> Result<WsStream, WsError> {
+    let original = ws_url.clone();
+    for hop in 0..=MAX_REDIRECT_HOPS {
+        match connect_async(ws_url.as_str()).await {
+            Ok((stream, _response)) => return Ok(stream),
+            Err(WsError::Http(response)) if is_redirect_status(response.status()) => {
+                if hop == MAX_REDIRECT_HOPS {
+                    log::error!(
+                        "[socket] Exceeded {MAX_REDIRECT_HOPS} redirect hops starting from {original}; giving up"
+                    );
+                    return Err(WsError::Http(response));
+                }
+                let location = match extract_location_header(&response) {
+                    Some(loc) => loc,
+                    None => {
+                        log::error!(
+                            "[socket] Redirect {} from {ws_url} missing Location header",
+                            response.status()
+                        );
+                        return Err(WsError::Http(response));
+                    }
+                };
+                let next_url = match resolve_redirect_target(ws_url, &location) {
+                    Ok(url) => url,
+                    Err(e) => {
+                        log::error!(
+                            "[socket] Cannot follow redirect to {location} from {ws_url}: {e}"
+                        );
+                        return Err(WsError::Http(response));
+                    }
+                };
+                log::warn!(
+                    "[socket] Server redirected ({}) {} → {}",
+                    response.status(),
+                    ws_url,
+                    next_url
+                );
+                // Only persist a stale-BACKEND_URL warning for permanent
+                // redirects (301 / 308). Temporary redirects (302 / 307) say
+                // "this time, go elsewhere" — the configured BACKEND_URL is
+                // still correct, and surfacing a "please update config" hint
+                // for a transient hop would be misleading. Per CodeRabbit
+                // review on PR #1547.
+                if matches!(
+                    response.status(),
+                    StatusCode::MOVED_PERMANENTLY | StatusCode::PERMANENT_REDIRECT
+                ) {
+                    record_redirect_warning(shared, &original, &next_url);
+                }
+                *ws_url = next_url;
+            }
+            Err(e) => return Err(e),
+        }
+    }
+    // Unreachable: the loop either returns Ok, returns the redirect error after
+    // exhausting hops, or returns a non-redirect Err.
+    unreachable!("connect_with_redirects exited loop without returning")
+}
+
+/// Statuses we treat as "follow the Location and retry".
+///
+/// 308 (Permanent Redirect) and 307 (Temporary Redirect) explicitly preserve
+/// the method; 301/302 historically do too for upgrade requests in practice.
+/// Anything else (300, 304, ...) stays an error.
+fn is_redirect_status(status: StatusCode) -> bool {
+    matches!(
+        status,
+        StatusCode::MOVED_PERMANENTLY
+            | StatusCode::FOUND
+            | StatusCode::TEMPORARY_REDIRECT
+            | StatusCode::PERMANENT_REDIRECT
+    )
+}
+
+fn extract_location_header(
+    response: &tokio_tungstenite::tungstenite::http::Response<Option<Vec<u8>>>,
+) -> Option<String> {
+    response
+        .headers()
+        .get(tokio_tungstenite::tungstenite::http::header::LOCATION)
+        .and_then(|v| v.to_str().ok())
+        .map(|s| s.to_string())
+}
+
+/// Resolve `location` against `current_ws_url` and rewrite the scheme so the
+/// result is still a valid WebSocket URL.
+///
+/// `location` may be absolute (`https://host/path?q=1`) or relative
+/// (`/socket.io/?EIO=4`). We use the `url` crate's relative-URL parser to do
+/// the join the same way browsers do, then map `http`→`ws` / `https`→`wss`.
+fn resolve_redirect_target(current_ws_url: &str, location: &str) -> Result<String, String> {
+    let base = url::Url::parse(current_ws_url).map_err(|e| format!("invalid current URL: {e}"))?;
+    let resolved = base
+        .join(location)
+        .map_err(|e| format!("invalid Location {location:?}: {e}"))?;
+
+    let upgraded_scheme = match resolved.scheme() {
+        "http" => "ws",
+        "https" => "wss",
+        "ws" | "wss" => resolved.scheme(),
+        other => return Err(format!("unsupported scheme in Location: {other}")),
+    };
+
+    let mut next = resolved.clone();
+    next.set_scheme(upgraded_scheme)
+        .map_err(|_| format!("failed to set scheme {upgraded_scheme} on {resolved}"))?;
+    Ok(next.to_string())
+}
+
+/// Persist a one-shot, user-visible warning that the backend redirected the
+/// configured socket URL. Subsequent redirects in the same connect attempt
+/// don't overwrite — the first hop carries the actionable signal.
+fn record_redirect_warning(shared: &Arc<SharedState>, original: &str, resolved: &str) {
+    let mut slot = shared.error.write();
+    if slot.is_some() {
+        return;
+    }
+    *slot = Some(format!(
+        "Backend redirected {original} → {resolved}. Update BACKEND_URL to the resolved URL to avoid the extra hop."
+    ));
+}
+
 #[cfg(test)]
 #[path = "ws_loop_tests.rs"]
 mod tests;

src/openhuman/socket/ws_loop_tests.rs

@@ -1,14 +1,125 @@
 use super::*;
 use parking_lot::RwLock;
+use tokio_tungstenite::tungstenite::http::{header::LOCATION, Response, StatusCode};
 
 fn make_shared() -> Arc<SharedState> {
     Arc::new(SharedState {
         webhook_router: RwLock::new(None),
         status: RwLock::new(ConnectionStatus::Connected),
         socket_id: RwLock::new(None),
+        error: RwLock::new(None),
     })
 }
 
+// ── Redirect resolution (the real fix for OPENHUMAN-TAURI-9X) ──
+
+#[test]
+fn resolve_redirect_upgrades_http_to_ws_for_absolute_location() {
+    // Cloudflare's exact behaviour: ws://host/path → 301 Location: https://host:443/path.
+    // We must rewrite https→wss so connect_async sees a WebSocket URL.
+    let next = resolve_redirect_target(
+        "ws://api.tinyhumans.ai/socket.io/?EIO=4&transport=websocket",
+        "https://api.tinyhumans.ai:443/socket.io/?EIO=4&transport=websocket",
+    )
+    .expect("scheme upgrade");
+    assert!(
+        next.starts_with("wss://api.tinyhumans.ai"),
+        "expected wss:// after upgrade, got {next}"
+    );
+    assert!(next.contains("/socket.io/?EIO=4&transport=websocket"));
+}
+
+#[test]
+fn resolve_redirect_handles_relative_location_against_current_url() {
+    // RFC 7230 allows a relative Location — must be resolved against the
+    // request URL, not treated as an error.
+    let next = resolve_redirect_target(
+        "ws://api.example.com/socket.io/?EIO=4&transport=websocket",
+        "/v2/socket.io/?EIO=4&transport=websocket",
+    )
+    .expect("relative resolve");
+    assert_eq!(
+        next,
+        "ws://api.example.com/v2/socket.io/?EIO=4&transport=websocket"
+    );
+}
+
+#[test]
+fn resolve_redirect_preserves_ws_and_wss_schemes_verbatim() {
+    let next = resolve_redirect_target("wss://a.example/socket.io/", "wss://b.example/socket.io/")
+        .unwrap();
+    assert!(next.starts_with("wss://b.example"));
+}
+
+#[test]
+fn resolve_redirect_rejects_unsupported_scheme() {
+    let err = resolve_redirect_target("wss://a.example/socket.io/", "ftp://elsewhere/socket.io/")
+        .unwrap_err();
+    assert!(err.contains("ftp"), "{err}");
+}
+
+#[test]
+fn redirect_status_matches_only_followable_codes() {
+    assert!(is_redirect_status(StatusCode::MOVED_PERMANENTLY));
+    assert!(is_redirect_status(StatusCode::FOUND));
+    assert!(is_redirect_status(StatusCode::TEMPORARY_REDIRECT));
+    assert!(is_redirect_status(StatusCode::PERMANENT_REDIRECT));
+    // 304 / 300 / 4xx / 5xx all stay errors that the backoff loop handles.
+    assert!(!is_redirect_status(StatusCode::NOT_MODIFIED));
+    assert!(!is_redirect_status(StatusCode::MULTIPLE_CHOICES));
+    assert!(!is_redirect_status(StatusCode::BAD_REQUEST));
+    assert!(!is_redirect_status(StatusCode::BAD_GATEWAY));
+}
+
+#[test]
+fn extract_location_header_returns_value_when_present() {
+    let resp = Response::builder()
+        .status(StatusCode::MOVED_PERMANENTLY)
+        .header(LOCATION, "https://api.example.com/socket.io/")
+        .body(None)
+        .unwrap();
+    assert_eq!(
+        extract_location_header(&resp).as_deref(),
+        Some("https://api.example.com/socket.io/")
+    );
+}
+
+#[test]
+fn extract_location_header_returns_none_when_missing() {
+    let resp = Response::builder()
+        .status(StatusCode::MOVED_PERMANENTLY)
+        .body(None)
+        .unwrap();
+    assert!(extract_location_header(&resp).is_none());
+}
+
+#[test]
+fn redirect_warning_is_recorded_once_and_pinned_to_first_hop() {
+    // First call records original→resolved. Second call (a second hop in the
+    // same attempt) must NOT overwrite — the first warning carries the
+    // user-actionable signal (your configured BACKEND_URL is stale).
+    let shared = make_shared();
+    record_redirect_warning(
+        &shared,
+        "ws://api.example.com/socket.io/",
+        "wss://api.example.com/socket.io/",
+    );
+    let first = shared.error.read().clone().unwrap();
+    assert!(first.contains("ws://api.example.com"));
+    assert!(first.contains("wss://api.example.com"));
+
+    record_redirect_warning(
+        &shared,
+        "wss://api.example.com/socket.io/",
+        "wss://api.example.com/v2/socket.io/",
+    );
+    let after_second = shared.error.read().clone().unwrap();
+    assert_eq!(
+        after_second, first,
+        "second redirect must not overwrite the first warning"
+    );
+}
+
 // ── handle_eio_message ─────────────────────────────────────────
 
 #[test]
@@ -383,3 +494,138 @@ fn connect_behavior_variants_are_distinct() {
         ConnectBehavior::GarbageOpenPacket => {}
     }
 }
+
+// ── End-to-end redirect-follow (the real fix for the 301 noise) ──
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+/// Spawn a one-shot HTTP/1.1 server that replies with a 301 redirect to
+/// `location` and closes — used to prove that `connect_with_redirects`
+/// follows the redirect end-to-end through `connect_async` instead of
+/// surfacing the 301 as a recurring error.
+async fn spawn_mock_301_redirect(location: String) -> std::net::SocketAddr {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let addr = listener.local_addr().expect("addr");
+    tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.expect("accept");
+        // Drain the incoming upgrade request so the client doesn't see RST.
+        let mut buf = [0u8; 4096];
+        let _ = stream.read(&mut buf).await;
+        let response = format!(
+            "HTTP/1.1 301 Moved Permanently\r\n\
+             Location: {location}\r\n\
+             Content-Length: 0\r\n\
+             Connection: close\r\n\r\n"
+        );
+        let _ = stream.write_all(response.as_bytes()).await;
+        let _ = stream.shutdown().await;
+    });
+    addr
+}
+
+/// Driver-level proof: when the configured URL responds with a 301 pointing
+/// at a working Engine.IO server, `ws_loop` follows the redirect, completes
+/// the handshake, and records a one-shot warning in `SharedState.error` so
+/// the UI can surface the stale-config signal.
+#[tokio::test]
+async fn ws_loop_follows_301_to_working_backend() {
+    // 1. Real EIO server on `ws://127.0.0.1:PORT`.
+    let (fwd_tx, mut fwd_rx) = mpsc::unbounded_channel::<String>();
+    let real_addr = spawn_mock_eio_server(ConnectBehavior::Ack, fwd_tx).await;
+    let real_ws_url = format!("ws://{real_addr}/socket.io/?EIO=4&transport=websocket");
+
+    // 2. Redirect server that 301s every request to the real EIO server.
+    let redirect_addr = spawn_mock_301_redirect(real_ws_url.clone()).await;
+
+    // 3. Drive `ws_loop` with the redirect address as the base URL. This is
+    //    exactly the production failure mode: BACKEND_URL points at a host
+    //    that 301s the WebSocket upgrade.
+    let shared = make_shared();
+    *shared.status.write() = ConnectionStatus::Disconnected;
+    let (emit_tx, emit_rx) = mpsc::unbounded_channel::<String>();
+    let (shutdown_tx, shutdown_rx) = watch::channel(false);
+    let internal_tx = emit_tx.clone();
+    drop(emit_tx);
+
+    let loop_shared = Arc::clone(&shared);
+    let handle = tokio::spawn(async move {
+        ws_loop(
+            format!("http://{redirect_addr}"),
+            "redirect-test-token".into(),
+            loop_shared,
+            emit_rx,
+            shutdown_rx,
+            internal_tx,
+        )
+        .await;
+    });
+
+    // The SIO CONNECT frame arriving on the *real* server proves the redirect
+    // was followed and the WebSocket handshake completed against the redirect
+    // target — not the redirect host.
+    let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(5);
+    let mut saw_connect = false;
+    while tokio::time::Instant::now() < deadline {
+        if let Ok(Some(frame)) =
+            tokio::time::timeout(tokio::time::Duration::from_millis(200), fwd_rx.recv()).await
+        {
+            if frame.starts_with("40") && frame.contains("redirect-test-token") {
+                saw_connect = true;
+                break;
+            }
+        }
+    }
+    assert!(
+        saw_connect,
+        "redirect was not followed — SIO CONNECT never reached the real EIO server"
+    );
+
+    for _ in 0..50 {
+        if *shared.status.read() == ConnectionStatus::Connected {
+            break;
+        }
+        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+    }
+    assert_eq!(*shared.status.read(), ConnectionStatus::Connected);
+
+    let warning = shared.error.read().clone();
+    assert!(
+        warning
+            .as_deref()
+            .map(|w| w.contains("redirected") && w.contains("BACKEND_URL"))
+            .unwrap_or(false),
+        "expected redirect warning in SharedState.error, got {warning:?}"
+    );
+
+    let _ = shutdown_tx.send(true);
+    let _ = tokio::time::timeout(tokio::time::Duration::from_secs(5), handle).await;
+}
+
+/// 301 without a Location header is unrecoverable — must surface as a real
+/// error and not loop forever attempting to follow nothing.
+#[tokio::test]
+async fn connect_with_redirects_fails_when_location_missing() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let addr = listener.local_addr().expect("addr");
+    tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.expect("accept");
+        let mut buf = [0u8; 4096];
+        let _ = stream.read(&mut buf).await;
+        // 301 but no Location header.
+        let _ = stream
+            .write_all(
+                b"HTTP/1.1 301 Moved Permanently\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
+            )
+            .await;
+        let _ = stream.shutdown().await;
+    });
+
+    let shared = make_shared();
+    let mut url = format!("ws://{addr}/socket.io/?EIO=4&transport=websocket");
+    let err = connect_with_redirects(&mut url, &shared)
+        .await
+        .expect_err("must surface failure when Location is absent");
+    assert!(matches!(err, WsError::Http(_)));
+    // No warning recorded because the redirect was never actually followed.
+    assert!(shared.error.read().is_none());
+}