cli/tools/desktop_devtools.rs - main - githubmirror/deno

cli/tools/desktop_devtools.rs
2 441 строка82 KB
Nathan Whitaker
fix(inspector): validate request host headers (#36348)
05 авг 2026, 02:44
05 авг 2026, 02:4485a2842
// Copyright 2018-2026 the Deno authors. MIT license.

//! CDP multiplexer for `deno desktop --inspect`.
//!
//! Architecture:
//!
//! ```text
//!                 ┌──────────────────────────────────────────┐
//!                 │   CDP Multiplexer (this file)            │
//!   DevTools ◄──► │   HTTP: /json/{version,list,protocol}    │
//!   (single ws)   │         /debugger-attached (gate)        │
//!                 │   WS:   /unified  (primary entry)        │
//!                 │         /deno, /cef (direct bypass)      │
//!                 │   HTTP: /devtools/* (frontend proxy)     │
//!                 └──────┬─────────────────────────┬─────────┘
//!                        │                         │
//!              Deno inspector                CEF renderer
//!              (internal port)               (internal port)
//! ```
//!
//! ## Unified session
//!
//! `/unified` is the primary session. CEF is the default (un-sessioned)
//! target; the Deno runtime appears as an attached child via a
//! synthetic `Target.attachedToTarget` event with a stable `sessionId`
//! the mux invents. Frame routing:
//!
//! - Client → mux frames with `sessionId == deno_session_id` have the
//!   field stripped and are forwarded to Deno.
//! - Deno → client frames get the `sessionId` re-injected.
//! - `Target.attachToTarget`/`detachFromTarget` for the Deno child are
//!   answered locally — CEF never learns of the synthetic session.
//! - `Runtime.executionContextCreated` gets its `context.name`
//!   rewritten ("Renderer" on the CEF leg, "Deno" on the Deno leg) so
//!   the Console dropdown shows meaningful labels instead of V8's
//!   defaults.
//! - The synthetic `Target.attachedToTarget` event is emitted lazily,
//!   on the first `Target.setAutoAttach`/`setDiscoverTargets` from the
//!   client — firing earlier loses the event to the frontend's
//!   not-yet-ready auto-attach manager.
//!
//! DevTools sees both isolates in one window: the Console dropdown
//! shows "Renderer" / "Deno", and the Sources panel Threads sidebar
//! lists both.
//!
//! ## `--inspect-brk` / `--inspect-wait`
//!
//! The child process blocks on `GET /debugger-attached` before
//! navigating CEF. The mux flips that endpoint to 200 only after the
//! DevTools client has connected AND — under `--inspect-brk` — the
//! `Debugger.enable` + `Debugger.pause` injection against CEF has
//! acked both responses. This guarantees the renderer pauses on the
//! first JS statement of the loaded page instead of racing past it.
//! Deno's own `--inspect-brk` handles the Deno isolate separately.
//!
//! ## Direct / frontend endpoints
//!
//! `/deno` and `/cef` are direct passthrough WebSockets for debugging
//! each isolate in isolation when the unified session misbehaves.
//! `/devtools/*` proxies CEF's bundled DevTools frontend assets
//! through the mux port so `openDevtools()` can pop a CEF window
//! without triggering CEF's own remote-debugging-port frontend
//! interception.

use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use deno_core::anyhow::Context;
use deno_core::anyhow::anyhow;
use deno_core::anyhow::bail;
use deno_core::error::AnyError;
use deno_core::serde_json;
use deno_core::serde_json::Value;
use deno_core::serde_json::json;
use deno_core::url::Host;
use deno_core::url::Url;
use deno_runtime::deno_inspector_server::ValidatedHost;
use deno_runtime::deno_inspector_server::validated_host_header;
use fastwebsockets::Frame;
use fastwebsockets::OpCode;
use fastwebsockets::WebSocket;
use fastwebsockets::WebSocketError;
use fastwebsockets::handshake;
use http_body_util::BodyExt;
use http_body_util::Empty;
use http_body_util::Full;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use uuid::Uuid;

/// Configuration for the CDP multiplexer.
#[derive(Clone, Debug)]
pub struct MuxConfig {
  /// User-visible listen address (from `--inspect`).
  pub listen: SocketAddr,
  /// Internal listen address for Deno's native inspector.
  pub deno_internal: SocketAddr,
  /// Internal listen address for the CEF renderer debug port.
  pub cef_internal: SocketAddr,
  /// `true` when `--inspect-brk` was passed — the mux will inject
  /// `Debugger.enable` + `Debugger.pause` into the CEF session so the
  /// renderer breaks on the first JS statement after navigation.
  pub inspect_brk: bool,
  /// `true` when `--inspect-wait` or `--inspect-brk` was passed.
  /// Not read by the mux itself — the child process uses env vars to
  /// decide whether to poll `/debugger-attached` before navigating.
  #[allow(
    dead_code,
    reason = "read by the child process via env vars, not by the mux itself"
  )]
  pub wait_for_debugger: bool,
}

/// A running multiplexer. Dropping the handle shuts the server down.
pub struct MuxHandle {
  pub listen: SocketAddr,
  _shutdown_tx: oneshot::Sender<()>,
}

pub use deno_lib::util::net::allocate_random_port;

/// Spawn the mux on a background task. Returns once the listener is
/// bound — connection handling continues in the spawned task. The
/// upstream servers (Deno inspector, CEF) do not need to be up yet;
/// the mux polls them on demand when DevTools makes a request.
pub async fn spawn_mux(config: MuxConfig) -> Result<MuxHandle, AnyError> {
  let listener = TcpListener::bind(config.listen).await.with_context(|| {
    format!("failed to bind CDP multiplexer to {}", config.listen)
  })?;
  let listen = listener.local_addr()?;
  let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

  let state = Arc::new(MuxState::new(config.clone(), listen));

  tokio::spawn(async move {
    let mut shutdown_rx = shutdown_rx;
    let mut consecutive_accept_errors: u32 = 0;
    loop {
      tokio::select! {
        _ = &mut shutdown_rx => {
          log::debug!("[devtools-mux] shutdown requested");
          break;
        }
        accept = listener.accept() => {
          match accept {
            Ok((stream, _)) => {
              consecutive_accept_errors = 0;
              let state = state.clone();
              tokio::spawn(async move {
                if let Err(err) = serve_connection(stream, state).await {
                  log::debug!("[devtools-mux] connection error: {err:?}");
                }
              });
            }
            Err(err) => {
              // A persistent failure (EMFILE, etc.) used to spin forever
              // logging at 5Hz. Throttle the log to every Nth attempt
              // and back off harder.
              consecutive_accept_errors =
                consecutive_accept_errors.saturating_add(1);
              if consecutive_accept_errors == 1
                || consecutive_accept_errors.is_power_of_two()
              {
                log::error!(
                  "[devtools-mux] accept failed (attempt {}): {err:?}",
                  consecutive_accept_errors,
                );
              }
              tokio::time::sleep(Duration::from_millis(200)).await;
            }
          }
        }
      }
    }
  });

  Ok(MuxHandle {
    listen,
    _shutdown_tx: shutdown_tx,
  })
}

/// Per-target identification. Kept small and stable so DevTools'
/// `webSocketDebuggerUrl` doesn't change across polls.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum TargetKind {
  /// Single CDP session fronting both isolates. CEF is the primary;
  /// Deno appears as an attached child target via `Target.*`.
  Unified,
  /// Direct passthrough to the Deno inspector.
  Deno,
  /// Direct passthrough to the CEF renderer's debug port.
  Cef,
}

impl TargetKind {
  fn path(self) -> &'static str {
    match self {
      TargetKind::Unified => "/unified",
      TargetKind::Deno => "/deno",
      TargetKind::Cef => "/cef",
    }
  }

  fn title(self) -> &'static str {
    match self {
      TargetKind::Unified => "Deno Desktop (unified)",
      TargetKind::Deno => "Deno Runtime",
      TargetKind::Cef => "CEF Renderer",
    }
  }
}

/// Synthetic CDP target id for the Deno isolate inside the unified
/// session. Stable per process; DevTools uses it in `Target.attachToTarget`.
const DENO_CHILD_TARGET_ID: &str = "deno-runtime-isolate";

struct MuxState {
  config: MuxConfig,
  listen: SocketAddr,
  // Stable UUIDs per target so repeated /json/list calls return the
  // same IDs. DevTools caches these.
  unified_id: Uuid,
  deno_id: Uuid,
  cef_id: Uuid,
  // Stable session id we hand DevTools when it attaches to the Deno
  // child target inside the unified session.
  deno_session_id: String,
  // Set to `true` when a DevTools client has connected to any session.
  // The child process polls `/debugger-attached` to gate navigation
  // when `--inspect-wait` or `--inspect-brk` is active.
  debugger_attached: Arc<std::sync::atomic::AtomicBool>,
}

impl MuxState {
  fn new(config: MuxConfig, listen: SocketAddr) -> Self {
    Self {
      config,
      listen,
      unified_id: Uuid::new_v4(),
      deno_id: Uuid::new_v4(),
      cef_id: Uuid::new_v4(),
      deno_session_id: Uuid::new_v4().to_string(),
      debugger_attached: Arc::new(std::sync::atomic::AtomicBool::new(false)),
    }
  }

  fn target_for_path(&self, path: &str) -> Option<TargetKind> {
    if path == TargetKind::Unified.path() {
      Some(TargetKind::Unified)
    } else if path == TargetKind::Deno.path() {
      Some(TargetKind::Deno)
    } else if path == TargetKind::Cef.path() {
      Some(TargetKind::Cef)
    } else {
      None
    }
  }
}

async fn serve_connection(
  stream: TcpStream,
  state: Arc<MuxState>,
) -> Result<(), AnyError> {
  let io = TokioIo::new(stream);
  let service = hyper::service::service_fn(move |req| {
    let state = state.clone();
    async move { Ok::<_, Infallible>(handle_request(req, state).await) }
  });

  hyper::server::conn::http1::Builder::new()
    .serve_connection(io, service)
    .with_upgrades()
    .await
    .map_err(|e| anyhow!("hyper serve error: {e}"))?;
  Ok(())
}

async fn handle_request(
  req: hyper::Request<Incoming>,
  state: Arc<MuxState>,
) -> hyper::Response<Full<Bytes>> {
  let host = match validated_host_header(&req) {
    Ok(host) => host,
    Err(_) => {
      return simple_response(
        http::StatusCode::BAD_REQUEST,
        "Invalid Host header",
      );
    }
  };
  if req.method() != http::Method::GET {
    return simple_response(
      http::StatusCode::METHOD_NOT_ALLOWED,
      "Not Allowed",
    );
  }
  let path = req.uri().path().to_string();
  match path.as_str() {
    "/json/version" => json_version(&state, host.as_ref()),
    "/json" | "/json/list" => json_list(&state, host.as_ref()),
    "/json/protocol" => json_protocol(),
    "/debugger-attached" => {
      if state
        .debugger_attached
        .load(std::sync::atomic::Ordering::SeqCst)
      {
        simple_response(http::StatusCode::OK, "attached")
      } else {
        simple_response(http::StatusCode::SERVICE_UNAVAILABLE, "waiting")
      }
    }
    other => {
      if let Some(kind) = state.target_for_path(other) {
        if !valid_ws_origin(&req, host.as_ref()) {
          return simple_response(
            http::StatusCode::FORBIDDEN,
            "Origin does not match request authority",
          );
        }
        match handle_upgrade(req, kind, state.clone()) {
          Ok(resp) => resp,
          Err(err) => {
            log::error!("[devtools-mux] upgrade failed for {other}: {err:?}");
            simple_response(http::StatusCode::BAD_REQUEST, "upgrade failed")
          }
        }
      } else if other.starts_with("/devtools/") {
        // Proxy DevTools frontend assets from CEF's bundled HTTP server.
        // Serving them through our port means the CEF window navigating
        // to the DevTools URL sees only `127.0.0.1:<mux_port>`, so it
        // doesn't special-case the remote-debugging port and steal the
        // frontend for the new window's own renderer.
        match proxy_devtools_asset(&req, state.config.cef_internal).await {
          Ok(resp) => resp,
          Err(err) => {
            log::error!("[devtools-mux] devtools asset proxy failed: {err:?}");
            simple_response(http::StatusCode::BAD_GATEWAY, "proxy failed")
          }
        }
      } else {
        simple_response(http::StatusCode::NOT_FOUND, "Not Found")
      }
    }
  }
}

fn valid_ws_origin<T>(
  req: &hyper::Request<T>,
  host: Option<&ValidatedHost>,
) -> bool {
  let mut origins = req.headers().get_all(http::header::ORIGIN).iter();
  let Some(origin) = origins.next() else {
    return true;
  };
  if origins.next().is_some() {
    return false;
  }
  let Ok(origin) = origin.to_str() else {
    return false;
  };
  let Ok(url) = Url::parse(origin) else {
    return false;
  };

  if matches!(url.scheme(), "devtools" | "chrome" | "chrome-devtools") {
    return true;
  }
  if !matches!(url.scheme(), "http" | "https")
    || !url.username().is_empty()
    || url.password().is_some()
    || url.path() != "/"
    || url.query().is_some()
    || url.fragment().is_some()
  {
    return false;
  }

  let Some(host) = host else {
    return false;
  };
  let origin_hostname = match url.host() {
    Some(Host::Domain(hostname))
      if hostname.eq_ignore_ascii_case("localhost") =>
    {
      "localhost".to_string()
    }
    Some(Host::Ipv4(ip)) => ip.to_string(),
    Some(Host::Ipv6(ip)) => ip.to_string(),
    _ => return false,
  };
  origin_hostname == host.hostname()
    && url.port_or_known_default() == Some(host.port().unwrap_or(80))
}

/// GET `http://<cef_internal><path>` and return the response verbatim.
/// Used to proxy the bundled DevTools frontend (inspector.html + its
/// JS/CSS assets) through the mux's own port, bypassing CEF's
/// remote-debugging-port special-case that would otherwise wire the
/// frontend to the requesting window's renderer.
async fn proxy_devtools_asset(
  req: &hyper::Request<Incoming>,
  cef_internal: SocketAddr,
) -> Result<hyper::Response<Full<Bytes>>, AnyError> {
  let path_and_query = req
    .uri()
    .path_and_query()
    .map(|p| p.as_str())
    .unwrap_or("/");

  let stream = TcpStream::connect(cef_internal).await?;
  let io = TokioIo::new(stream);
  let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
    .await
    .map_err(|e| anyhow!("devtools asset handshake failed: {e}"))?;
  tokio::spawn(async move {
    if let Err(err) = conn.await {
      log::trace!("[devtools-mux] devtools asset conn closed: {err:?}");
    }
  });

  let upstream_req = hyper::Request::builder()
    .method(http::Method::GET)
    .uri(path_and_query)
    .header(http::header::HOST, cef_internal.to_string())
    .body(Empty::<Bytes>::new())?;
  let resp = sender.send_request(upstream_req).await?;
  let (parts, body) = resp.into_parts();
  let bytes = body.collect().await?.to_bytes();

  let mut builder = hyper::Response::builder().status(parts.status);
  // Drop hop-by-hop headers that don't apply to our re-packaged body.
  for (name, value) in parts.headers.iter() {
    let skip = matches!(
      name.as_str().to_ascii_lowercase().as_str(),
      "transfer-encoding"
        | "content-length"
        | "connection"
        | "keep-alive"
        | "proxy-authenticate"
        | "proxy-authorization"
        | "te"
        | "trailer"
        | "upgrade"
    );
    if !skip {
      builder = builder.header(name, value);
    }
  }
  Ok(
    builder
      .header(http::header::CONTENT_LENGTH, bytes.len())
      .body(Full::new(bytes))
      .unwrap(),
  )
}

fn json_version(
  state: &MuxState,
  host: Option<&ValidatedHost>,
) -> hyper::Response<Full<Bytes>> {
  let authority = host
    .map(ValidatedHost::authority)
    .map(String::from)
    .unwrap_or_else(|| state.listen.to_string());
  let body = json!({
    "Browser": format!("deno-desktop/{}", env!("CARGO_PKG_VERSION")),
    "Protocol-Version": "1.3",
    "V8-Version": deno_core::v8::VERSION_STRING,
    // Advertise one of the two upstream WS URLs as the "browser" URL.
    // DevTools' `chrome://inspect` uses this to drive auto-attach; the
    // CEF upstream exposes the richer Target.* domain.
    "webSocketDebuggerUrl": format!(
      "ws://{}{}",
      authority,
      TargetKind::Cef.path(),
    ),
  });
  json_response(body)
}

fn json_list(
  state: &MuxState,
  host: Option<&ValidatedHost>,
) -> hyper::Response<Full<Bytes>> {
  let authority = host
    .map(ValidatedHost::authority)
    .map(String::from)
    .unwrap_or_else(|| state.listen.to_string());
  let unified_url = format!("ws://{authority}{}", TargetKind::Unified.path());
  let deno_url = format!("ws://{authority}{}", TargetKind::Deno.path());
  let cef_url = format!("ws://{authority}{}", TargetKind::Cef.path());

  // Primary entry: the unified session. DevTools opens one window
  // (inspector.html — full browser DevTools) and gets both isolates as
  // attached targets in the Sources panel.
  let unified_entry = json!({
    "id": state.unified_id.to_string(),
    "type": "page",
    "title": TargetKind::Unified.title(),
    "description": "Unified DevTools (CEF page + Deno runtime)",
    "url": "deno-desktop://unified",
    "faviconUrl": "https://deno.land/favicon.ico",
    "devtoolsFrontendUrl": format!(
      "devtools://devtools/bundled/inspector.html?ws={}",
      strip_scheme(&unified_url),
    ),
    "webSocketDebuggerUrl": unified_url,
  });
  // Fallback entries: direct passthrough to each isolate. Useful when
  // the unified session misbehaves and you want to debug each side in
  // isolation.
  let deno_entry = json!({
    "id": state.deno_id.to_string(),
    "type": "node",
    "title": TargetKind::Deno.title(),
    "description": "Deno runtime V8 isolate (direct)",
    "url": format!("deno://{}", state.config.deno_internal),
    "faviconUrl": "https://deno.land/favicon.ico",
    "devtoolsFrontendUrl": format!(
      "devtools://devtools/bundled/js_app.html?ws={}&experiments=true&v8only=true",
      strip_scheme(&deno_url),
    ),
    "webSocketDebuggerUrl": deno_url,
  });
  let cef_entry = json!({
    "id": state.cef_id.to_string(),
    "type": "page",
    "title": TargetKind::Cef.title(),
    "description": "CEF renderer V8 isolate (direct)",
    "url": format!("cef://{}", state.config.cef_internal),
    "faviconUrl": "https://deno.land/favicon.ico",
    "devtoolsFrontendUrl": format!(
      "devtools://devtools/bundled/inspector.html?ws={}",
      strip_scheme(&cef_url),
    ),
    "webSocketDebuggerUrl": cef_url,
  });

  json_response(Value::Array(vec![unified_entry, deno_entry, cef_entry]))
}

/// Return an empty protocol descriptor. DevTools tolerates this and
/// falls back to the built-in protocol.
fn json_protocol() -> hyper::Response<Full<Bytes>> {
  json_response(json!({
    "version": { "major": "1", "minor": "3" },
    "domains": [],
  }))
}

fn json_response(value: Value) -> hyper::Response<Full<Bytes>> {
  let body = Full::new(Bytes::from(serde_json::to_vec(&value).unwrap()));
  hyper::Response::builder()
    .status(http::StatusCode::OK)
    .header(http::header::CONTENT_TYPE, "application/json")
    .body(body)
    .unwrap()
}

fn simple_response(
  status: http::StatusCode,
  msg: &'static str,
) -> hyper::Response<Full<Bytes>> {
  hyper::Response::builder()
    .status(status)
    .body(Full::new(Bytes::from(msg)))
    .unwrap()
}

fn strip_scheme(ws_url: &str) -> String {
  ws_url
    .strip_prefix("ws://")
    .or_else(|| ws_url.strip_prefix("wss://"))
    .unwrap_or(ws_url)
    .to_string()
}

/// Upgrade an incoming HTTP request to a WebSocket, open a matching
/// WebSocket to the upstream target, and shuttle frames in both
/// directions.
fn handle_upgrade(
  mut req: hyper::Request<Incoming>,
  kind: TargetKind,
  state: Arc<MuxState>,
) -> Result<hyper::Response<Full<Bytes>>, AnyError> {
  let (resp, upgrade_fut) = fastwebsockets::upgrade::upgrade(&mut req)
    .map_err(|e| anyhow!("not a valid websocket upgrade: {e}"))?;

  tokio::spawn(async move {
    let client = match upgrade_fut.await {
      Ok(ws) => ws,
      Err(err) => {
        log::error!("[devtools-mux] client upgrade failed: {err:?}");
        return;
      }
    };

    // `debugger_attached` is what releases the child process from its
    // `/debugger-attached` poll so it can navigate CEF. Under
    // `--inspect-brk` we MUST inject `Debugger.enable` + `Debugger.pause`
    // into the CEF isolate BEFORE the child navigates, otherwise the
    // page's JS executes before the pause request arrives. So each
    // target-specific handler signals attachment at its own right moment.

    match kind {
      TargetKind::Unified => {
        if let Err(err) = run_unified_session(client, state).await {
          log::debug!("[devtools-mux] unified session ended: {err:?}");
        }
      }
      TargetKind::Deno => {
        // Deno has its own `--inspect-brk` mechanism that blocks the
        // isolate until a client attaches — nothing to inject here.
        mark_debugger_attached(&state);
        match connect_upstream(&state, kind).await {
          Ok(upstream) => {
            if let Err(err) = proxy_frames(client, upstream).await {
              log::debug!("[devtools-mux] proxy ended: {err:?}");
            }
          }
          Err(err) => {
            log::error!(
              "[devtools-mux] failed to connect upstream for Deno: {err:?}"
            );
          }
        }
      }
      TargetKind::Cef => match connect_upstream(&state, kind).await {
        Ok(mut upstream) => {
          if state.config.inspect_brk
            && let Err(err) = inject_cef_pause(&mut upstream).await
          {
            log::error!(
              "[devtools-mux] failed to inject pause into CEF: {err:?}"
            );
          }
          mark_debugger_attached(&state);
          if let Err(err) = proxy_frames(client, upstream).await {
            log::debug!("[devtools-mux] proxy ended: {err:?}");
          }
        }
        Err(err) => {
          log::error!(
            "[devtools-mux] failed to connect upstream for CEF: {err:?}"
          );
        }
      },
    }
  });

  let (parts, _) = resp.into_parts();
  Ok(hyper::Response::from_parts(parts, Full::new(Bytes::new())))
}

/// Open a WebSocket to the right upstream target, performing target
/// discovery as needed. For CEF we hit `/json/list` to find the live
/// `webSocketDebuggerUrl`; for Deno we call the inspector server's
/// same endpoint. The upstream is retried briefly since the renderer
/// may not have finished booting.
async fn connect_upstream(
  state: &MuxState,
  kind: TargetKind,
) -> Result<WebSocket<TokioIo<hyper::upgrade::Upgraded>>, AnyError> {
  let upstream_host = match kind {
    TargetKind::Deno => state.config.deno_internal,
    TargetKind::Cef => state.config.cef_internal,
    TargetKind::Unified => {
      bail!("connect_upstream cannot be called with the Unified target")
    }
  };

  // Poll until the upstream has a live WebSocket debugger URL.
  let mut last_err: Option<AnyError> = None;
  let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
  while tokio::time::Instant::now() < deadline {
    match fetch_upstream_ws_url(upstream_host).await {
      Ok(ws_url) => match connect_ws(&ws_url).await {
        Ok(ws) => return Ok(ws),
        Err(err) => {
          last_err = Some(err);
        }
      },
      Err(err) => {
        last_err = Some(err);
      }
    }
    tokio::time::sleep(Duration::from_millis(250)).await;
  }
  Err(last_err.unwrap_or_else(|| anyhow!("upstream connect timed out")))
}

/// GET `http://<host>/json/list` and pick the first entry's
/// `webSocketDebuggerUrl`. Both Deno's inspector server and CEF's
/// remote-debugging endpoint implement this.
async fn fetch_upstream_ws_url(host: SocketAddr) -> Result<String, AnyError> {
  let stream = TcpStream::connect(host).await?;
  let io = TokioIo::new(stream);
  let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
    .await
    .map_err(|e| anyhow!("http handshake to {host} failed: {e}"))?;
  tokio::spawn(async move {
    if let Err(err) = conn.await {
      log::trace!("[devtools-mux] upstream conn closed: {err:?}");
    }
  });

  let req = hyper::Request::builder()
    .method(http::Method::GET)
    .uri("/json/list")
    .header(http::header::HOST, host.to_string())
    .body(Empty::<Bytes>::new())?;
  let resp = sender.send_request(req).await?;
  if !resp.status().is_success() {
    bail!("upstream /json/list at {host} returned {}", resp.status());
  }
  let body = resp.collect().await?.to_bytes();
  let value: Value = serde_json::from_slice(&body)
    .with_context(|| format!("upstream /json/list at {host} not JSON"))?;

  let ws_url = value
    .as_array()
    .and_then(|arr| {
      arr.iter().find_map(|v| {
        // Skip targets that are our own DevTools frontend window — when
        // openDevtools() creates a CEF window pointed at inspector.html,
        // CEF registers it as a debuggable target. Connecting to it
        // instead of the real app window would show "DevTools for
        // DevTools".
        let url = v.get("url").and_then(|u| u.as_str()).unwrap_or("");
        if url.contains("/devtools/") || url.contains("devtools://") {
          return None;
        }
        v.get("webSocketDebuggerUrl")
      })
    })
    .and_then(|v| v.as_str())
    .ok_or_else(|| {
      anyhow!("no webSocketDebuggerUrl in /json/list at {host}")
    })?;

  // Upstream responds with its own listen host; some backends return
  // `0.0.0.0` or `localhost`. Force to the target host so we connect
  // to the right address.
  let rewritten = rewrite_ws_host(ws_url, host);
  Ok(rewritten)
}

fn rewrite_ws_host(ws_url: &str, host: SocketAddr) -> String {
  let rest = ws_url
    .strip_prefix("ws://")
    .or_else(|| ws_url.strip_prefix("wss://"))
    .unwrap_or(ws_url);
  let path = rest.find('/').map(|i| &rest[i..]).unwrap_or("/");
  format!("ws://{host}{path}")
}

async fn connect_ws(
  ws_url: &str,
) -> Result<WebSocket<TokioIo<hyper::upgrade::Upgraded>>, AnyError> {
  let url: http::Uri = ws_url.parse()?;
  let host = url
    .host()
    .ok_or_else(|| anyhow!("ws url missing host: {ws_url}"))?;
  // Don't fall back to port 80: a malformed `/json/list` response
  // missing the port would otherwise route the WS connect to whatever
  // is listening on port 80 of the upstream host.
  let port = url
    .port_u16()
    .ok_or_else(|| anyhow!("ws url missing port: {ws_url}"))?;
  let authority = format!("{host}:{port}");

  let stream = TcpStream::connect(&authority).await?;
  let req = hyper::Request::builder()
    .method(http::Method::GET)
    .uri(url.path_and_query().map(|p| p.as_str()).unwrap_or("/"))
    .header(http::header::HOST, &authority)
    .header(http::header::UPGRADE, "websocket")
    .header(http::header::CONNECTION, "upgrade")
    .header("Sec-WebSocket-Key", handshake::generate_key())
    .header("Sec-WebSocket-Version", "13")
    .body(Empty::<Bytes>::new())?;

  let (ws, _) = handshake::client(&TokioExec, req, stream).await?;
  Ok(ws)
}

struct TokioExec;
impl<F> hyper::rt::Executor<F> for TokioExec
where
  F: std::future::Future + Send + 'static,
  F::Output: Send + 'static,
{
  fn execute(&self, fut: F) {
    tokio::spawn(fut);
  }
}

/// Bidirectionally forward frames between the DevTools client and the
/// upstream inspector. The loop ends when either side closes or errors.
async fn proxy_frames(
  mut client: WebSocket<TokioIo<hyper::upgrade::Upgraded>>,
  mut upstream: WebSocket<TokioIo<hyper::upgrade::Upgraded>>,
) -> Result<(), AnyError> {
  // We forward control frames (ping/pong/close) verbatim between the
  // two peers, so disable fastwebsockets' built-in handling.
  client.set_auto_close(false);
  client.set_auto_pong(false);
  upstream.set_auto_close(false);
  upstream.set_auto_pong(false);

  // Split both sides so the two pump directions can run concurrently
  // without holding a single mutex across `.await` points.
  let (mut client_rx, mut client_tx) = client.split(tokio::io::split);
  let (mut up_rx, mut up_tx) = upstream.split(tokio::io::split);

  let client_to_up = async {
    // The send_fn is used by fastwebsockets to auto-respond to control
    // frames; with auto_close/auto_pong disabled it is never called.
    let mut noop = |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
    loop {
      let frame = match client_rx.read_frame(&mut noop).await {
        Ok(f) => f,
        Err(err) => {
          log::debug!("[devtools-mux] client read: {err:?}");
          return;
        }
      };
      let is_close = frame.opcode == OpCode::Close;
      if let Err(err) = up_tx.write_frame(frame).await {
        log::debug!("[devtools-mux] upstream write: {err:?}");
        return;
      }
      if is_close {
        return;
      }
    }
  };

  let up_to_client = async {
    let mut noop = |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
    loop {
      let frame = match up_rx.read_frame(&mut noop).await {
        Ok(f) => f,
        Err(err) => {
          log::debug!("[devtools-mux] upstream read: {err:?}");
          return;
        }
      };
      let is_close = frame.opcode == OpCode::Close;
      if let Err(err) = client_tx.write_frame(frame).await {
        log::debug!("[devtools-mux] client write: {err:?}");
        return;
      }
      if is_close {
        return;
      }
    }
  };

  tokio::join!(client_to_up, up_to_client);
  Ok(())
}

// ─── Unified session (v2) ──────────────────────────────────────────
//
// One DevTools window, two isolates. CEF is the primary session;
// Deno appears as an "attached" child target via the standard
// `Target.attachedToTarget` event with a synthetic `sessionId`.
// Frames flow through CDP-aware routers that strip/inject `sessionId`
// on the Deno leg and pass everything else through to CEF.

/// Owned representation of a WebSocket frame, suitable for sending
/// through an mpsc channel. `fastwebsockets::Frame` borrows from the
/// reader's internal buffer and so can't cross task boundaries.
struct OwnedFrame {
  opcode: OpCode,
  payload: Vec<u8>,
}

impl OwnedFrame {
  fn text(payload: Vec<u8>) -> Self {
    Self {
      opcode: OpCode::Text,
      payload,
    }
  }

  fn into_frame(self) -> Frame<'static> {
    Frame::new(
      true,
      self.opcode,
      None,
      fastwebsockets::Payload::Owned(self.payload),
    )
  }
}

/// Set `debugger_attached` so the child process's `/debugger-attached`
/// poll returns 200 and it can proceed with navigation. Only call this
/// once the CEF pause injection (if any) has completed.
fn mark_debugger_attached(state: &MuxState) {
  state
    .debugger_attached
    .store(true, std::sync::atomic::Ordering::SeqCst);
}

/// Send `Debugger.enable` + `Debugger.pause` to the CEF upstream,
/// swallowing their responses. Called before the child is released to
/// navigate, so the renderer stops on the first JS statement of the
/// loaded page.
///
/// Any CEF → client frames that arrive during the injection window are
/// discarded. This is acceptable because the renderer has not yet
/// navigated to a user page (the child is blocked on
/// `/debugger-attached`), so the only traffic is CEF's own protocol
/// setup (e.g. unsolicited `Target.targetCreated` for about:blank),
/// which DevTools will re-observe via `Target.setDiscoverTargets` once
/// the session opens.
async fn inject_cef_pause<S>(cef: &mut WebSocket<S>) -> Result<(), AnyError>
where
  S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
  let enable = json!({"id": -1, "method": "Debugger.enable"});
  cef
    .write_frame(Frame::new(
      true,
      OpCode::Text,
      None,
      fastwebsockets::Payload::Owned(serde_json::to_vec(&enable).unwrap()),
    ))
    .await?;

  let pause = json!({"id": -2, "method": "Debugger.pause"});
  cef
    .write_frame(Frame::new(
      true,
      OpCode::Text,
      None,
      fastwebsockets::Payload::Owned(serde_json::to_vec(&pause).unwrap()),
    ))
    .await?;

  // Read frames until we've observed both responses. CEF may interleave
  // unsolicited events, which we drop.
  let mut saw_enable = false;
  let mut saw_pause = false;
  while !(saw_enable && saw_pause) {
    let frame = cef.read_frame().await?;
    if frame.opcode != OpCode::Text {
      continue;
    }
    let value: Value = match serde_json::from_slice(&frame.payload) {
      Ok(v) => v,
      Err(_) => continue,
    };
    match value.get("id").and_then(|v| v.as_i64()) {
      Some(-1) => saw_enable = true,
      Some(-2) => saw_pause = true,
      _ => {}
    }
  }

  log::debug!(
    "[devtools-mux] injected Debugger.enable + Debugger.pause into CEF"
  );
  Ok(())
}

/// Run the unified DevTools session: one client WebSocket fronting two
/// upstreams (CEF as the default session, Deno attached via a
/// synthetic `sessionId`).
async fn run_unified_session(
  client: WebSocket<TokioIo<hyper::upgrade::Upgraded>>,
  state: Arc<MuxState>,
) -> Result<(), AnyError> {
  let mut client = client;
  client.set_auto_close(false);
  client.set_auto_pong(false);

  let (cef, deno) = tokio::try_join!(
    connect_upstream(&state, TargetKind::Cef),
    connect_upstream(&state, TargetKind::Deno),
  )?;
  let mut cef = cef;
  let mut deno = deno;
  cef.set_auto_close(false);
  cef.set_auto_pong(false);
  deno.set_auto_close(false);
  deno.set_auto_pong(false);

  let session_id = state.deno_session_id.clone();

  // When --inspect-brk is active, inject Debugger.enable + Debugger.pause
  // into the CEF session BEFORE the child is allowed to navigate. The
  // child polls `/debugger-attached` and will only navigate once we
  // signal attachment, so doing the injection first — and releasing
  // attachment afterwards — guarantees the renderer pauses on the very
  // first JS statement of the loaded page.
  if state.config.inspect_brk {
    inject_cef_pause(&mut cef).await?;
  }
  mark_debugger_attached(&state);

  let (mut client_rx, mut client_tx) = client.split(tokio::io::split);
  let (mut cef_rx, mut cef_tx) = cef.split(tokio::io::split);
  let (mut deno_rx, mut deno_tx) = deno.split(tokio::io::split);

  let (client_send, mut client_recv) = mpsc::unbounded_channel::<OwnedFrame>();
  let (cef_send, mut cef_recv) = mpsc::unbounded_channel::<OwnedFrame>();
  let (deno_send, mut deno_recv) = mpsc::unbounded_channel::<OwnedFrame>();

  // Deno is announced lazily, the first time DevTools asks about
  // targets — firing too early causes the event to be dropped before
  // the frontend's auto-attach manager has subscribed. See
  // `route_client_text` for the trigger.
  let deno_announced = Arc::new(std::sync::atomic::AtomicBool::new(false));

  // Writer tasks: pump owned frames from a channel into the WS half.
  let client_writer = tokio::spawn(async move {
    while let Some(owned) = client_recv.recv().await {
      let close = owned.opcode == OpCode::Close;
      if let Err(err) = client_tx.write_frame(owned.into_frame()).await {
        log::debug!("[devtools-mux] unified client write: {err:?}");
        return;
      }
      if close {
        return;
      }
    }
  });
  let cef_writer = tokio::spawn(async move {
    while let Some(owned) = cef_recv.recv().await {
      let close = owned.opcode == OpCode::Close;
      if let Err(err) = cef_tx.write_frame(owned.into_frame()).await {
        log::debug!("[devtools-mux] unified cef write: {err:?}");
        return;
      }
      if close {
        return;
      }
    }
  });
  let deno_writer = tokio::spawn(async move {
    while let Some(owned) = deno_recv.recv().await {
      let close = owned.opcode == OpCode::Close;
      if let Err(err) = deno_tx.write_frame(owned.into_frame()).await {
        log::debug!("[devtools-mux] unified deno write: {err:?}");
        return;
      }
      if close {
        return;
      }
    }
  });

  // Reader: client → CEF/Deno (with CDP-aware routing).
  let mut client_reader = {
    let client_send = client_send.clone();
    let cef_send = cef_send.clone();
    let deno_send = deno_send.clone();
    let session_id = session_id.clone();
    let deno_announced = deno_announced.clone();
    tokio::spawn(async move {
      let mut noop = |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
      loop {
        let frame = match client_rx.read_frame(&mut noop).await {
          Ok(f) => f,
          Err(err) => {
            log::debug!("[devtools-mux] unified client read: {err:?}");
            return;
          }
        };
        let opcode = frame.opcode;
        let payload = frame.payload.to_vec();
        match opcode {
          OpCode::Text => {
            route_client_text(
              &payload,
              &session_id,
              &client_send,
              &cef_send,
              &deno_send,
              &deno_announced,
            );
          }
          OpCode::Close => {
            let _ = cef_send.send(OwnedFrame {
              opcode,
              payload: payload.clone(),
            });
            let _ = deno_send.send(OwnedFrame { opcode, payload });
            return;
          }
          _ => {
            // Binary, Ping, Pong, Continuation: forward to CEF (the
            // primary session). Ping/pong on the client connection is
            // for keep-alive; CEF will reply.
            let _ = cef_send.send(OwnedFrame { opcode, payload });
          }
        }
      }
    })
  };

  // Reader: CEF → client. No sessionId injection, but we do rewrite
  // the execution-context name so the Console dropdown reads
  // "Renderer" instead of V8's default "top".
  let mut cef_reader = {
    let client_send = client_send.clone();
    tokio::spawn(async move {
      let mut noop = |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
      loop {
        let frame = match cef_rx.read_frame(&mut noop).await {
          Ok(f) => f,
          Err(err) => {
            log::debug!("[devtools-mux] unified cef read: {err:?}");
            return;
          }
        };
        let opcode = frame.opcode;
        let payload = frame.payload.to_vec();
        let owned = if opcode == OpCode::Text {
          OwnedFrame::text(rewrite_text_from_upstream(
            &payload, None, "Renderer",
          ))
        } else {
          OwnedFrame { opcode, payload }
        };
        if client_send.send(owned).is_err() {
          return;
        }
      }
    })
  };

  // Reader: Deno → client. Inject sessionId so DevTools routes frames
  // to the synthetic child target, and relabel the execution context
  // to "Deno" instead of V8's "main realm".
  let mut deno_reader = {
    let client_send = client_send.clone();
    let session_id = session_id.clone();
    tokio::spawn(async move {
      let mut noop = |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
      loop {
        let frame = match deno_rx.read_frame(&mut noop).await {
          Ok(f) => f,
          Err(err) => {
            log::debug!("[devtools-mux] unified deno read: {err:?}");
            return;
          }
        };
        let opcode = frame.opcode;
        let payload = frame.payload.to_vec();
        let owned = if opcode == OpCode::Text {
          OwnedFrame::text(rewrite_text_from_upstream(
            &payload,
            Some(&session_id),
            "Deno",
          ))
        } else {
          OwnedFrame { opcode, payload }
        };
        if client_send.send(owned).is_err() {
          return;
        }
      }
    })
  };

  // Drop the originals so writers exit once readers finish.
  drop(client_send);
  drop(cef_send);
  drop(deno_send);

  // Wait for any reader to exit, then tear down everything else.
  tokio::select! {
    _ = &mut client_reader => {},
    _ = &mut cef_reader => {},
    _ = &mut deno_reader => {},
  }

  client_reader.abort();
  cef_reader.abort();
  deno_reader.abort();
  client_writer.abort();
  cef_writer.abort();
  deno_writer.abort();

  Ok(())
}

/// CDP-aware routing for a text frame coming from the DevTools client.
///
/// - `sessionId == deno_session_id` → strip and send to Deno.
/// - `Target.setAutoAttach` / `Target.setDiscoverTargets(true)` →
///   forward to CEF AND lazily emit our synthetic
///   `Target.attachedToTarget` for Deno (once). We piggyback on these
///   calls because they are the frontend's signal that it's ready to
///   process target events; firing earlier causes the event to be
///   silently dropped.
/// - `Target.attachToTarget(deno-id)` → reply locally with the
///   synthetic sessionId; never reaches CEF (which doesn't know it).
/// - `Target.detachFromTarget(deno-session)` → reply locally and
///   synthesize the corresponding `Target.detachedFromTarget` event.
/// - everything else → forward to CEF.
fn route_client_text(
  payload: &[u8],
  session_id: &str,
  client_send: &mpsc::UnboundedSender<OwnedFrame>,
  cef_send: &mpsc::UnboundedSender<OwnedFrame>,
  deno_send: &mpsc::UnboundedSender<OwnedFrame>,
  deno_announced: &Arc<std::sync::atomic::AtomicBool>,
) {
  let mut value: Value = match serde_json::from_slice(payload) {
    Ok(v) => v,
    Err(_) => {
      let _ = cef_send.send(OwnedFrame::text(payload.to_vec()));
      return;
    }
  };

  let session = value.get("sessionId").and_then(|v| v.as_str());
  if session == Some(session_id) {
    if let Some(obj) = value.as_object_mut() {
      obj.remove("sessionId");
    }
    let bytes = serde_json::to_vec(&value).unwrap_or_else(|_| payload.to_vec());
    let _ = deno_send.send(OwnedFrame::text(bytes));
    return;
  }

  let id = value.get("id").and_then(|v| v.as_i64());
  let method = value
    .get("method")
    .and_then(|v| v.as_str())
    .map(str::to_owned);

  // The frontend is now listening for target events — announce Deno.
  if matches!(
    method.as_deref(),
    Some("Target.setAutoAttach") | Some("Target.setDiscoverTargets")
  ) && !deno_announced.swap(true, std::sync::atomic::Ordering::SeqCst)
  {
    let event = attached_to_target_event(session_id);
    let _ =
      client_send.send(OwnedFrame::text(serde_json::to_vec(&event).unwrap()));
  }

  if method.as_deref() == Some("Target.attachToTarget") {
    let target_id = value
      .get("params")
      .and_then(|p| p.get("targetId"))
      .and_then(|v| v.as_str());
    if target_id == Some(DENO_CHILD_TARGET_ID) {
      if let Some(rid) = id {
        let reply = json!({
          "id": rid,
          "result": { "sessionId": session_id },
        });
        let _ = client_send
          .send(OwnedFrame::text(serde_json::to_vec(&reply).unwrap()));
      }
      return;
    }
  }

  if method.as_deref() == Some("Target.detachFromTarget") {
    let detach_session = value
      .get("params")
      .and_then(|p| p.get("sessionId"))
      .and_then(|v| v.as_str());
    if detach_session == Some(session_id) {
      if let Some(rid) = id {
        let reply = json!({ "id": rid, "result": {} });
        let _ = client_send
          .send(OwnedFrame::text(serde_json::to_vec(&reply).unwrap()));
      }
      let event = json!({
        "method": "Target.detachedFromTarget",
        "params": {
          "sessionId": session_id,
          "targetId": DENO_CHILD_TARGET_ID,
        },
      });
      let _ =
        client_send.send(OwnedFrame::text(serde_json::to_vec(&event).unwrap()));
      return;
    }
  }

  let _ = cef_send.send(OwnedFrame::text(payload.to_vec()));
}

/// Rewrite a JSON CDP frame on its way from an upstream to the
/// DevTools client:
///
/// - Optionally inject `sessionId = session_id` so DevTools attributes
///   the frame to the synthetic Deno child target.
/// - Rename `Runtime.executionContextCreated.params.context.name` to
///   `context_name` so the Console "execution context" dropdown shows
///   a meaningful label instead of V8's defaults (`"top"`, `"main
///   realm"`).
///
/// If the payload isn't valid JSON, return it unchanged.
fn rewrite_text_from_upstream(
  payload: &[u8],
  inject_session_id: Option<&str>,
  context_name: &str,
) -> Vec<u8> {
  let mut value: Value = match serde_json::from_slice(payload) {
    Ok(v) => v,
    Err(_) => return payload.to_vec(),
  };
  let Some(obj) = value.as_object_mut() else {
    return payload.to_vec();
  };
  if let Some(sid) = inject_session_id {
    obj.insert("sessionId".to_string(), Value::String(sid.to_string()));
  }
  if obj.get("method").and_then(|v| v.as_str())
    == Some("Runtime.executionContextCreated")
    && let Some(ctx) = obj
      .get_mut("params")
      .and_then(|v| v.get_mut("context"))
      .and_then(|v| v.as_object_mut())
  {
    ctx.insert("name".to_string(), Value::String(context_name.to_string()));
  }
  serde_json::to_vec(&value).unwrap_or_else(|_| payload.to_vec())
}

/// Build a `Target.attachedToTarget` event advertising the Deno
/// runtime isolate as a child of the unified session.
fn attached_to_target_event(session_id: &str) -> Value {
  // `inspector.html` only renders attached child targets in the Sources
  // panel "Threads" sidebar when their type matches a known
  // worker-style kind (`worker`, `shared_worker`, `service_worker`).
  // We pick `worker` so the Deno runtime isolate shows up alongside
  // the CEF renderer's main thread.
  json!({
    "method": "Target.attachedToTarget",
    "params": {
      "sessionId": session_id,
      "targetInfo": {
        "targetId": DENO_CHILD_TARGET_ID,
        "type": "worker",
        "title": "Deno Runtime",
        "url": "deno://runtime",
        "attached": true,
        "canAccessOpener": false,
      },
      "waitingForDebugger": false,
    },
  })
}

#[cfg(test)]
mod tests {
  use super::*;

  #[test]
  fn rewrite_ws_host_forces_host() {
    let host: SocketAddr = "127.0.0.1:9230".parse().unwrap();
    assert_eq!(
      rewrite_ws_host("ws://0.0.0.0:9230/devtools/browser/abc", host),
      "ws://127.0.0.1:9230/devtools/browser/abc"
    );
    assert_eq!(
      rewrite_ws_host("ws://localhost/ws/deadbeef", host),
      "ws://127.0.0.1:9230/ws/deadbeef"
    );
  }

  fn test_config() -> MuxConfig {
    MuxConfig {
      listen: "127.0.0.1:9229".parse().unwrap(),
      deno_internal: "127.0.0.1:9230".parse().unwrap(),
      cef_internal: "127.0.0.1:9231".parse().unwrap(),
      inspect_brk: false,
      wait_for_debugger: false,
    }
  }

  #[test]
  fn target_path_round_trip() {
    let state = MuxState::new(test_config(), "127.0.0.1:9229".parse().unwrap());
    assert_eq!(state.target_for_path("/unified"), Some(TargetKind::Unified));
    assert_eq!(state.target_for_path("/deno"), Some(TargetKind::Deno));
    assert_eq!(state.target_for_path("/cef"), Some(TargetKind::Cef));
    assert_eq!(state.target_for_path("/bogus"), None);
  }

  fn request_with_host_and_origin(
    host: Option<&str>,
    origin: Option<&str>,
  ) -> http::Request<()> {
    let mut request = http::Request::builder().uri("/unified");
    if let Some(host) = host {
      request = request.header(http::header::HOST, host);
    }
    if let Some(origin) = origin {
      request = request.header(http::header::ORIGIN, origin);
    }
    request.body(()).unwrap()
  }

  #[test]
  fn websocket_origins_match_client_visible_authority() {
    for origin in [
      None,
      Some("devtools://devtools"),
      Some("chrome://inspect"),
      Some("chrome-devtools://devtools"),
      Some("http://localhost:43123"),
      Some("https://localhost:43123"),
    ] {
      let request =
        request_with_host_and_origin(Some("localhost:43123"), origin);
      let host = validated_host_header(&request).unwrap();
      assert!(valid_ws_origin(&request, host.as_ref()));
    }

    for origin in [
      "http://example.test:43123",
      "http://localhost:43124",
      "file://localhost",
      "null",
    ] {
      let request =
        request_with_host_and_origin(Some("localhost:43123"), Some(origin));
      let host = validated_host_header(&request).unwrap();
      assert!(!valid_ws_origin(&request, host.as_ref()));
    }

    let request =
      request_with_host_and_origin(None, Some("http://localhost:43123"));
    assert!(!valid_ws_origin(&request, None));
  }

  // ── sessionId dispatch ────────────────────────────────────────────

  #[test]
  fn route_client_text_strips_session_and_forwards_to_deno() {
    let (client_tx, _client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, _cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, mut deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(false));

    let session_id = "test-session-123";
    let msg = json!({
      "id": 1,
      "method": "Debugger.enable",
      "sessionId": session_id,
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload, session_id, &client_tx, &cef_tx, &deno_tx, &announced,
    );

    // Should arrive at Deno with sessionId stripped.
    let frame = deno_rx.try_recv().expect("expected frame on deno channel");
    let value: Value = serde_json::from_slice(&frame.payload).unwrap();
    assert_eq!(value.get("id").unwrap(), 1);
    assert_eq!(value.get("method").unwrap(), "Debugger.enable");
    assert!(
      value.get("sessionId").is_none(),
      "sessionId should be stripped"
    );
  }

  #[test]
  fn route_client_text_forwards_non_session_to_cef() {
    let (client_tx, _client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let msg = json!({"id": 5, "method": "DOM.getDocument"});
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload,
      "some-session",
      &client_tx,
      &cef_tx,
      &deno_tx,
      &announced,
    );

    let frame = cef_rx.try_recv().expect("expected frame on cef channel");
    let value: Value = serde_json::from_slice(&frame.payload).unwrap();
    assert_eq!(value.get("id").unwrap(), 5);
  }

  #[test]
  fn route_client_text_attach_to_deno_target_replies_locally() {
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let session_id = "deno-sess";
    let msg = json!({
      "id": 10,
      "method": "Target.attachToTarget",
      "params": { "targetId": DENO_CHILD_TARGET_ID },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload, session_id, &client_tx, &cef_tx, &deno_tx, &announced,
    );

    // Should reply to client with the synthetic sessionId.
    let frame = client_rx.try_recv().expect("expected reply on client");
    let value: Value = serde_json::from_slice(&frame.payload).unwrap();
    assert_eq!(value["id"], 10);
    assert_eq!(value["result"]["sessionId"], session_id);

    // Should NOT have forwarded to CEF.
    assert!(cef_rx.try_recv().is_err());
  }

  #[test]
  fn route_client_text_lazily_announces_deno() {
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, _cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(false));

    let msg = json!({
      "id": 1,
      "method": "Target.setAutoAttach",
      "params": { "autoAttach": true, "waitForDebuggerOnStart": false },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload, "sess", &client_tx, &cef_tx, &deno_tx, &announced,
    );

    // Should have emitted Target.attachedToTarget event.
    let frame = client_rx.try_recv().expect("expected announce event");
    let value: Value = serde_json::from_slice(&frame.payload).unwrap();
    assert_eq!(value["method"], "Target.attachedToTarget");
    assert_eq!(value["params"]["targetInfo"]["type"], "worker");
    assert!(announced.load(std::sync::atomic::Ordering::SeqCst));

    // Calling again should NOT emit a second event.
    route_client_text(
      &payload, "sess", &client_tx, &cef_tx, &deno_tx, &announced,
    );
    // Only the forwarded-to-cef frame, no second announce.
    assert!(client_rx.try_recv().is_err());
  }

  #[test]
  fn route_client_text_set_discover_targets_also_triggers_announce() {
    // The lazy-announce trigger is either setAutoAttach OR
    // setDiscoverTargets — DevTools sometimes uses one, sometimes the
    // other, depending on which panel initialised first.
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, _cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(false));

    let msg = json!({
      "id": 7,
      "method": "Target.setDiscoverTargets",
      "params": { "discover": true },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload, "sess", &client_tx, &cef_tx, &deno_tx, &announced,
    );

    let frame = client_rx
      .try_recv()
      .expect("setDiscoverTargets should also trigger announce");
    let value: Value = serde_json::from_slice(&frame.payload).unwrap();
    assert_eq!(value["method"], "Target.attachedToTarget");
    assert!(announced.load(std::sync::atomic::Ordering::SeqCst));
  }

  #[test]
  fn route_client_text_attach_to_cef_target_forwards_to_cef() {
    // `Target.attachToTarget` for any target ID that isn't ours (e.g.
    // a real CEF subframe or worker) must pass through — only the
    // synthetic Deno target is answered locally.
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let msg = json!({
      "id": 11,
      "method": "Target.attachToTarget",
      "params": { "targetId": "some-cef-subframe-target" },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload,
      "deno-sess",
      &client_tx,
      &cef_tx,
      &deno_tx,
      &announced,
    );

    let forwarded = cef_rx.try_recv().expect("expected frame on cef");
    let value: Value = serde_json::from_slice(&forwarded.payload).unwrap();
    assert_eq!(value["method"], "Target.attachToTarget");
    assert_eq!(value["params"]["targetId"], "some-cef-subframe-target");
    assert!(client_rx.try_recv().is_err());
  }

  // ── context name rewriting ────────────────────────────────────────

  #[test]
  fn rewrite_injects_session_id() {
    let input = json!({"id": 1, "result": {}});
    let payload = serde_json::to_vec(&input).unwrap();
    let out = rewrite_text_from_upstream(&payload, Some("my-sess"), "Deno");
    let value: Value = serde_json::from_slice(&out).unwrap();
    assert_eq!(value["sessionId"], "my-sess");
  }

  #[test]
  fn rewrite_renames_execution_context() {
    let input = json!({
      "method": "Runtime.executionContextCreated",
      "params": {
        "context": {
          "id": 1,
          "origin": "",
          "name": "top",
        },
      },
    });
    let payload = serde_json::to_vec(&input).unwrap();
    let out = rewrite_text_from_upstream(&payload, None, "Renderer");
    let value: Value = serde_json::from_slice(&out).unwrap();
    assert_eq!(value["params"]["context"]["name"], "Renderer");
  }

  #[test]
  fn rewrite_no_session_leaves_field_absent() {
    let input = json!({"method": "Console.messageAdded"});
    let payload = serde_json::to_vec(&input).unwrap();
    let out = rewrite_text_from_upstream(&payload, None, "Renderer");
    let value: Value = serde_json::from_slice(&out).unwrap();
    assert!(value.get("sessionId").is_none());
  }

  #[test]
  fn rewrite_leaves_non_execution_context_messages_unchanged() {
    // Any `name` field on other methods must not be touched — only
    // `Runtime.executionContextCreated.params.context.name` is rewritten.
    let input = json!({
      "method": "Target.targetInfoChanged",
      "params": { "targetInfo": { "name": "something" } },
    });
    let payload = serde_json::to_vec(&input).unwrap();
    let out = rewrite_text_from_upstream(&payload, None, "Renderer");
    let value: Value = serde_json::from_slice(&out).unwrap();
    assert_eq!(value["params"]["targetInfo"]["name"], "something");
  }

  #[test]
  fn rewrite_invalid_json_returned_verbatim() {
    let payload = b"totally not json".to_vec();
    let out = rewrite_text_from_upstream(&payload, Some("sid"), "Renderer");
    assert_eq!(out, payload);
  }

  #[test]
  fn rewrite_non_object_json_returned_verbatim() {
    // Arrays / scalars have no "sessionId" slot to inject into —
    // they must pass through untouched rather than being wrapped.
    let input = json!([1, 2, 3]);
    let payload = serde_json::to_vec(&input).unwrap();
    let out = rewrite_text_from_upstream(&payload, Some("sid"), "Renderer");
    assert_eq!(out, payload);
  }

  // ── /json/list shape ──────────────────────────────────────────────

  #[tokio::test]
  async fn json_list_returns_three_targets() {
    let state = MuxState::new(test_config(), "127.0.0.1:9229".parse().unwrap());
    let resp = json_list(&state, None);
    let body = resp.into_body();
    let bytes = body.collect().await.unwrap().to_bytes();
    let list: Vec<Value> = serde_json::from_slice(&bytes).unwrap();

    assert_eq!(list.len(), 3);

    // Unified target.
    assert_eq!(list[0]["type"], "page");
    assert_eq!(list[0]["title"], "Deno Desktop (unified)");
    assert!(
      list[0]["webSocketDebuggerUrl"]
        .as_str()
        .unwrap()
        .ends_with("/unified")
    );

    // Deno direct target.
    assert_eq!(list[1]["type"], "node");
    assert_eq!(list[1]["title"], "Deno Runtime");
    assert!(
      list[1]["webSocketDebuggerUrl"]
        .as_str()
        .unwrap()
        .ends_with("/deno")
    );

    // CEF direct target.
    assert_eq!(list[2]["type"], "page");
    assert_eq!(list[2]["title"], "CEF Renderer");
    assert!(
      list[2]["webSocketDebuggerUrl"]
        .as_str()
        .unwrap()
        .ends_with("/cef")
    );

    // Every entry must expose a devtoolsFrontendUrl that points at
    // its own ws:// endpoint — DevTools uses it to launch the right
    // frontend (inspector.html vs js_app.html) against the right mux
    // route. Every entry must also advertise a stable UUID id and
    // include description + faviconUrl, which DevTools renders in the
    // target picker.
    for entry in &list {
      let frontend = entry["devtoolsFrontendUrl"].as_str().unwrap();
      let ws = entry["webSocketDebuggerUrl"].as_str().unwrap();
      let ws_host_path = ws.strip_prefix("ws://").unwrap();
      assert!(
        frontend.contains(ws_host_path),
        "frontend {frontend} must embed ws host+path {ws_host_path}"
      );

      let id = entry["id"].as_str().unwrap();
      Uuid::parse_str(id).unwrap_or_else(|_| panic!("id {id} is not a UUID"));
      assert!(entry["description"].is_string());
      assert!(entry["faviconUrl"].is_string());
    }

    // Deno direct entry uses `js_app.html` (DevTools' node variant)
    // while CEF and unified use the full `inspector.html`.
    assert!(
      list[1]["devtoolsFrontendUrl"]
        .as_str()
        .unwrap()
        .contains("js_app.html"),
      "deno direct entry should launch js_app.html"
    );
    for page_entry in [&list[0], &list[2]] {
      assert!(
        page_entry["devtoolsFrontendUrl"]
          .as_str()
          .unwrap()
          .contains("inspector.html"),
      );
    }

    // IDs must be stable across calls — DevTools caches them.
    let resp2 = json_list(&state, None);
    let bytes2 = resp2.into_body().collect().await.unwrap().to_bytes();
    let list2: Vec<Value> = serde_json::from_slice(&bytes2).unwrap();
    for (a, b) in list.iter().zip(list2.iter()) {
      assert_eq!(
        a["id"], b["id"],
        "target id changed across /json/list calls"
      );
    }
  }

  #[tokio::test]
  async fn json_version_shape() {
    let state = MuxState::new(test_config(), "127.0.0.1:9229".parse().unwrap());
    let resp = json_version(&state, None);
    assert_eq!(resp.status(), http::StatusCode::OK);
    let ct = resp.headers().get(http::header::CONTENT_TYPE).unwrap();
    assert_eq!(ct, "application/json");

    let bytes = resp.into_body().collect().await.unwrap().to_bytes();
    let value: Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(value["Protocol-Version"], "1.3");
    assert!(
      value["Browser"]
        .as_str()
        .unwrap()
        .starts_with("deno-desktop/")
    );
    assert!(value["V8-Version"].is_string());
    // The browser-level webSocketDebuggerUrl must point at the CEF
    // target — DevTools' chrome://inspect auto-attach needs the
    // richer Target.* domain that CEF implements.
    let ws = value["webSocketDebuggerUrl"].as_str().unwrap();
    assert!(ws.ends_with("/cef"), "got {ws}");
  }

  // ── Target.detachFromTarget dispatch ──────────────────────────────

  #[test]
  fn route_client_text_detach_from_deno_session_synthesizes_reply_and_event() {
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let session_id = "deno-sess-xyz";
    let msg = json!({
      "id": 42,
      "method": "Target.detachFromTarget",
      "params": { "sessionId": session_id },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload, session_id, &client_tx, &cef_tx, &deno_tx, &announced,
    );

    // First frame: the id=42 reply.
    let reply = client_rx.try_recv().expect("expected detach reply");
    let reply_val: Value = serde_json::from_slice(&reply.payload).unwrap();
    assert_eq!(reply_val["id"], 42);
    assert!(reply_val["result"].is_object());

    // Second frame: the synthesized Target.detachedFromTarget event.
    let event = client_rx.try_recv().expect("expected detached event");
    let event_val: Value = serde_json::from_slice(&event.payload).unwrap();
    assert_eq!(event_val["method"], "Target.detachedFromTarget");
    assert_eq!(event_val["params"]["sessionId"], session_id);
    assert_eq!(event_val["params"]["targetId"], DENO_CHILD_TARGET_ID);

    // Must NOT be forwarded to CEF — CEF has no knowledge of our
    // synthetic Deno session.
    assert!(cef_rx.try_recv().is_err());
  }

  #[test]
  fn route_client_text_detach_from_unknown_session_forwards_to_cef() {
    // A detach for a session CEF owns (not our synthetic Deno one)
    // must flow through to CEF unchanged.
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let msg = json!({
      "id": 99,
      "method": "Target.detachFromTarget",
      "params": { "sessionId": "some-cef-session" },
    });
    let payload = serde_json::to_vec(&msg).unwrap();

    route_client_text(
      &payload,
      "deno-sess",
      &client_tx,
      &cef_tx,
      &deno_tx,
      &announced,
    );

    let forwarded = cef_rx.try_recv().expect("expected frame forwarded");
    let value: Value = serde_json::from_slice(&forwarded.payload).unwrap();
    assert_eq!(value["id"], 99);
    assert!(client_rx.try_recv().is_err());
  }

  // ── Malformed / edge-case dispatch ────────────────────────────────

  #[test]
  fn route_client_text_invalid_json_forwards_to_cef() {
    let (client_tx, mut client_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (cef_tx, mut cef_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (deno_tx, _deno_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let announced = Arc::new(std::sync::atomic::AtomicBool::new(true));

    let payload = b"not-json".to_vec();
    route_client_text(
      &payload, "sess", &client_tx, &cef_tx, &deno_tx, &announced,
    );

    let frame = cef_rx.try_recv().expect("expected frame on cef");
    assert_eq!(frame.payload, b"not-json");
    assert!(client_rx.try_recv().is_err());
  }

  // ── End-to-end integration ────────────────────────────────────────
  //
  // Spin up mock upstream HTTP+WS servers for CEF and Deno, start the
  // mux in front of them, then drive a real WebSocket client through
  // the `/unified` endpoint to exercise routing, context rewriting,
  // and the `--inspect-brk` pause injection.

  /// Simple mock upstream: serves `/json/list` pointing at its own
  /// `/ws` path, accepts a WebSocket upgrade there, and exposes
  /// unbounded channels so tests can pump frames in/out.
  struct MockUpstream {
    listen: SocketAddr,
    /// Frames received from the mux.
    from_mux: tokio::sync::Mutex<mpsc::UnboundedReceiver<OwnedFrame>>,
    /// Frames to send to the mux.
    to_mux: mpsc::UnboundedSender<OwnedFrame>,
  }

  async fn spawn_mock_upstream() -> Arc<MockUpstream> {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let listen = listener.local_addr().unwrap();
    let (in_tx, in_rx) = mpsc::unbounded_channel::<OwnedFrame>();
    let (out_tx, out_rx) = mpsc::unbounded_channel::<OwnedFrame>();

    let upstream = Arc::new(MockUpstream {
      listen,
      from_mux: tokio::sync::Mutex::new(in_rx),
      to_mux: out_tx,
    });

    // Shared across every connection the mux makes to this upstream —
    // the mux opens one TCP connection for `/json/list` and a separate
    // one for the WS upgrade, so the out_rx must only be consumed when
    // the upgrade actually happens.
    let shared_out_rx = Arc::new(std::sync::Mutex::new(Some(out_rx)));

    tokio::spawn(async move {
      loop {
        let (stream, _) = match listener.accept().await {
          Ok(v) => v,
          Err(_) => return,
        };
        let in_tx = in_tx.clone();
        let shared_out_rx = shared_out_rx.clone();
        let listen_str = listen.to_string();
        tokio::spawn(async move {
          let io = TokioIo::new(stream);
          let service = hyper::service::service_fn(move |mut req| {
            let in_tx = in_tx.clone();
            let shared_out_rx = shared_out_rx.clone();
            let listen_str = listen_str.clone();
            async move {
              let path = req.uri().path().to_string();
              if path == "/json/list" {
                let body = serde_json::to_vec(&json!([{
                  "id": "mock-target",
                  "type": "page",
                  "webSocketDebuggerUrl": format!("ws://{listen_str}/ws"),
                }]))
                .unwrap();
                return Ok::<_, Infallible>(
                  hyper::Response::builder()
                    .status(http::StatusCode::OK)
                    .header(http::header::CONTENT_TYPE, "application/json")
                    .body(Full::new(Bytes::from(body)))
                    .unwrap(),
                );
              }
              if path == "/ws" {
                let Ok((resp, upgrade_fut)) =
                  fastwebsockets::upgrade::upgrade(&mut req)
                else {
                  return Ok(simple_response(
                    http::StatusCode::BAD_REQUEST,
                    "bad upgrade",
                  ));
                };
                let in_tx = in_tx.clone();
                let out_rx = shared_out_rx.lock().unwrap().take();
                tokio::spawn(async move {
                  let mut ws = match upgrade_fut.await {
                    Ok(w) => w,
                    Err(_) => return,
                  };
                  ws.set_auto_close(false);
                  ws.set_auto_pong(false);
                  let (mut rx, mut tx) = ws.split(tokio::io::split);
                  let reader = async move {
                    let mut noop =
                      |_: Frame<'_>| async { Ok::<(), WebSocketError>(()) };
                    loop {
                      let frame = match rx.read_frame(&mut noop).await {
                        Ok(f) => f,
                        Err(_) => return,
                      };
                      let owned = OwnedFrame {
                        opcode: frame.opcode,
                        payload: frame.payload.to_vec(),
                      };
                      let is_close = owned.opcode == OpCode::Close;
                      if in_tx.send(owned).is_err() || is_close {
                        return;
                      }
                    }
                  };
                  let writer = async move {
                    let Some(mut out_rx) = out_rx else { return };
                    while let Some(owned) = out_rx.recv().await {
                      let close = owned.opcode == OpCode::Close;
                      if tx.write_frame(owned.into_frame()).await.is_err() {
                        return;
                      }
                      if close {
                        return;
                      }
                    }
                  };
                  tokio::join!(reader, writer);
                });
                let (parts, _) = resp.into_parts();
                return Ok(hyper::Response::from_parts(
                  parts,
                  Full::new(Bytes::new()),
                ));
              }
              Ok(simple_response(http::StatusCode::NOT_FOUND, "Not Found"))
            }
          });
          let _ = hyper::server::conn::http1::Builder::new()
            .serve_connection(io, service)
            .with_upgrades()
            .await;
        });
      }
    });

    upstream
  }

  /// Open a WebSocket from the test to `ws://<addr><path>`.
  async fn test_connect_ws(
    ws_url: &str,
  ) -> WebSocket<TokioIo<hyper::upgrade::Upgraded>> {
    connect_ws(ws_url).await.unwrap()
  }

  async fn read_text_value<S>(ws: &mut WebSocket<S>) -> Value
  where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
  {
    loop {
      let frame = ws.read_frame().await.unwrap();
      if frame.opcode != OpCode::Text {
        continue;
      }
      return serde_json::from_slice(&frame.payload).unwrap();
    }
  }

  async fn write_json<S>(ws: &mut WebSocket<S>, v: &Value)
  where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
  {
    let payload = serde_json::to_vec(v).unwrap();
    ws.write_frame(Frame::new(
      true,
      OpCode::Text,
      None,
      fastwebsockets::Payload::Owned(payload),
    ))
    .await
    .unwrap();
  }

  async fn spawn_test_mux(
    cef: &MockUpstream,
    deno: &MockUpstream,
    inspect_brk: bool,
  ) -> MuxHandle {
    let listen_port = allocate_random_port().unwrap();
    spawn_mux(MuxConfig {
      listen: format!("127.0.0.1:{listen_port}").parse().unwrap(),
      deno_internal: deno.listen,
      cef_internal: cef.listen,
      inspect_brk,
      wait_for_debugger: inspect_brk,
    })
    .await
    .unwrap()
  }

  /// GET a path from the mux over raw HTTP and return the body bytes.
  async fn http_get_with_host(
    addr: SocketAddr,
    path: &str,
    host: &str,
  ) -> (http::StatusCode, Bytes) {
    let stream = TcpStream::connect(addr).await.unwrap();
    let io = TokioIo::new(stream);
    let (mut sender, conn) =
      hyper::client::conn::http1::handshake(io).await.unwrap();
    tokio::spawn(async move {
      let _ = conn.await;
    });
    let req = hyper::Request::builder()
      .method(http::Method::GET)
      .uri(path)
      .header(http::header::HOST, host)
      .body(Empty::<Bytes>::new())
      .unwrap();
    let resp = sender.send_request(req).await.unwrap();
    let status = resp.status();
    let bytes = resp.collect().await.unwrap().to_bytes();
    (status, bytes)
  }

  async fn http_get(addr: SocketAddr, path: &str) -> (http::StatusCode, Bytes) {
    http_get_with_host(addr, path, &addr.to_string()).await
  }

  async fn upgrade_status(
    addr: SocketAddr,
    path: &str,
    host: &str,
    origin: Option<&str>,
  ) -> http::StatusCode {
    let stream = TcpStream::connect(addr).await.unwrap();
    let io = TokioIo::new(stream);
    let (mut sender, conn) =
      hyper::client::conn::http1::handshake(io).await.unwrap();
    tokio::spawn(async move {
      let _ = conn.with_upgrades().await;
    });
    let mut request = hyper::Request::builder()
      .method(http::Method::GET)
      .uri(path)
      .header(http::header::HOST, host)
      .header(http::header::UPGRADE, "websocket")
      .header(http::header::CONNECTION, "upgrade")
      .header("Sec-WebSocket-Key", handshake::generate_key())
      .header("Sec-WebSocket-Version", "13");
    if let Some(origin) = origin {
      request = request.header(http::header::ORIGIN, origin);
    }
    sender
      .send_request(request.body(Empty::<Bytes>::new()).unwrap())
      .await
      .unwrap()
      .status()
  }

  #[tokio::test]
  async fn integration_http_endpoints() {
    let cef = spawn_mock_upstream().await;
    let deno = spawn_mock_upstream().await;
    let mux = spawn_test_mux(&cef, &deno, false).await;

    // /json/list — the mux advertises three targets; the HTTP path
    // does not need upstream WS to be live.
    let (status, body) = http_get(mux.listen, "/json/list").await;
    assert_eq!(status, http::StatusCode::OK);
    let list: Vec<Value> = serde_json::from_slice(&body).unwrap();
    assert_eq!(list.len(), 3);

    // Discovery preserves a client-visible forwarded authority.
    let (status, body) =
      http_get_with_host(mux.listen, "/json/list", "localhost:43123").await;
    assert_eq!(status, http::StatusCode::OK);
    let list: Vec<Value> = serde_json::from_slice(&body).unwrap();
    assert_eq!(
      list[0]["webSocketDebuggerUrl"],
      "ws://localhost:43123/unified"
    );

    // Named authorities are rejected on discovery and WebSocket routes.
    let (status, _) =
      http_get_with_host(mux.listen, "/json/list", "example.test:43123").await;
    assert_eq!(status, http::StatusCode::BAD_REQUEST);
    assert_eq!(
      upgrade_status(
        mux.listen,
        "/unified",
        "example.test:43123",
        Some("http://example.test:43123"),
      )
      .await,
      http::StatusCode::BAD_REQUEST
    );

    // Browser WebSocket requests must come from their request authority.
    assert_eq!(
      upgrade_status(
        mux.listen,
        "/unified",
        &mux.listen.to_string(),
        Some("http://example.test:43123"),
      )
      .await,
      http::StatusCode::FORBIDDEN
    );

    // /json/version.
    let (status, body) = http_get(mux.listen, "/json/version").await;
    assert_eq!(status, http::StatusCode::OK);
    let v: Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(v["Protocol-Version"], "1.3");

    // /debugger-attached returns 503 before any client connects.
    let (status, _) = http_get(mux.listen, "/debugger-attached").await;
    assert_eq!(status, http::StatusCode::SERVICE_UNAVAILABLE);

    // Unknown path → 404.
    let (status, _) = http_get(mux.listen, "/nope").await;
    assert_eq!(status, http::StatusCode::NOT_FOUND);
  }

  #[tokio::test]
  async fn integration_forwarded_same_origin_websocket_is_allowed() {
    let cef = spawn_mock_upstream().await;
    let deno = spawn_mock_upstream().await;
    let mux = spawn_test_mux(&cef, &deno, false).await;

    let stream = TcpStream::connect(mux.listen).await.unwrap();
    let request = hyper::Request::builder()
      .method(http::Method::GET)
      .uri("/unified")
      .header(http::header::HOST, "localhost:43123")
      .header(http::header::ORIGIN, "http://localhost:43123")
      .header(http::header::UPGRADE, "websocket")
      .header(http::header::CONNECTION, "upgrade")
      .header("Sec-WebSocket-Key", handshake::generate_key())
      .header("Sec-WebSocket-Version", "13")
      .body(Empty::<Bytes>::new())
      .unwrap();
    let (_client, response) = handshake::client(&TokioExec, request, stream)
      .await
      .unwrap();
    assert_eq!(response.status(), http::StatusCode::SWITCHING_PROTOCOLS);
  }

  #[tokio::test]
  async fn integration_unified_session_routes_and_rewrites() {
    let cef = spawn_mock_upstream().await;
    let deno = spawn_mock_upstream().await;
    let mux = spawn_test_mux(&cef, &deno, false).await;

    let ws_url = format!("ws://{}/unified", mux.listen);
    let mut client = test_connect_ws(&ws_url).await;
    client.set_auto_close(false);
    client.set_auto_pong(false);

    // Before any client activity, /debugger-attached should flip to 200.
    // Give the spawned upgrade task a moment to run.
    for _ in 0..50 {
      let (status, _) = http_get(mux.listen, "/debugger-attached").await;
      if status == http::StatusCode::OK {
        break;
      }
      tokio::time::sleep(Duration::from_millis(20)).await;
    }

    // 1) Client sends Target.setAutoAttach. Mux should forward it to
    //    CEF *and* synthesize Target.attachedToTarget to the client.
    write_json(
      &mut client,
      &json!({
        "id": 1,
        "method": "Target.setAutoAttach",
        "params": { "autoAttach": true, "waitForDebuggerOnStart": false },
      }),
    )
    .await;

    let attached = read_text_value(&mut client).await;
    assert_eq!(attached["method"], "Target.attachedToTarget");
    let session_id = attached["params"]["sessionId"]
      .as_str()
      .unwrap()
      .to_string();
    assert_eq!(attached["params"]["targetInfo"]["type"], "worker");

    // CEF upstream should have received the setAutoAttach verbatim.
    let received = {
      let mut rx = cef.from_mux.lock().await;
      tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .unwrap()
        .unwrap()
    };
    let received_val: Value =
      serde_json::from_slice(&received.payload).unwrap();
    assert_eq!(received_val["method"], "Target.setAutoAttach");

    // 2) Client sends a frame with sessionId=deno. Mux strips it and
    //    forwards to Deno upstream.
    write_json(
      &mut client,
      &json!({
        "id": 2,
        "method": "Debugger.enable",
        "sessionId": session_id,
      }),
    )
    .await;
    let deno_received = {
      let mut rx = deno.from_mux.lock().await;
      tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .unwrap()
        .unwrap()
    };
    let deno_val: Value =
      serde_json::from_slice(&deno_received.payload).unwrap();
    assert_eq!(deno_val["id"], 2);
    assert_eq!(deno_val["method"], "Debugger.enable");
    assert!(deno_val["sessionId"].is_null());

    // 3) Upstream CEF emits Runtime.executionContextCreated. The mux
    //    rewrites context.name to "Renderer" and forwards to client
    //    WITHOUT a sessionId.
    cef
      .to_mux
      .send(OwnedFrame::text(
        serde_json::to_vec(&json!({
          "method": "Runtime.executionContextCreated",
          "params": { "context": { "id": 1, "origin": "", "name": "top" } },
        }))
        .unwrap(),
      ))
      .unwrap();
    let from_cef = read_text_value(&mut client).await;
    assert_eq!(from_cef["method"], "Runtime.executionContextCreated");
    assert_eq!(from_cef["params"]["context"]["name"], "Renderer");
    assert!(from_cef["sessionId"].is_null());

    // 4) Upstream Deno emits the same. Mux rewrites to "Deno" AND
    //    injects the synthetic sessionId.
    deno
      .to_mux
      .send(OwnedFrame::text(
        serde_json::to_vec(&json!({
          "method": "Runtime.executionContextCreated",
          "params": { "context": { "id": 1, "origin": "", "name": "main realm" } },
        }))
        .unwrap(),
      ))
      .unwrap();
    let from_deno = read_text_value(&mut client).await;
    assert_eq!(from_deno["method"], "Runtime.executionContextCreated");
    assert_eq!(from_deno["params"]["context"]["name"], "Deno");
    assert_eq!(from_deno["sessionId"], session_id);

    drop(client);
    drop(mux);
  }

  #[tokio::test]
  async fn integration_inspect_brk_injects_before_marking_attached() {
    // The core contract: under --inspect-brk the mux must send
    // Debugger.enable + Debugger.pause to the CEF upstream BEFORE
    // /debugger-attached flips to 200. If that order is reversed the
    // child process navigates CEF before the pause is in flight and
    // the renderer races past the first JS statement.
    let cef = spawn_mock_upstream().await;
    let deno = spawn_mock_upstream().await;
    let mux = spawn_test_mux(&cef, &deno, true).await;

    let ws_url = format!("ws://{}/unified", mux.listen);
    let client_connect =
      tokio::spawn(async move { test_connect_ws(&ws_url).await });

    // Observe the first two frames CEF receives. With inspect_brk=true
    // they must be Debugger.enable then Debugger.pause.
    let enable_frame = {
      let mut rx = cef.from_mux.lock().await;
      tokio::time::timeout(Duration::from_secs(5), rx.recv())
        .await
        .expect("no frame received on CEF upstream")
        .unwrap()
    };
    let enable_val: Value =
      serde_json::from_slice(&enable_frame.payload).unwrap();
    assert_eq!(enable_val["method"], "Debugger.enable");

    // /debugger-attached MUST still be 503 until we ack enable+pause.
    // We haven't replied yet, so the mux is still blocked waiting.
    let (status, _) = http_get(mux.listen, "/debugger-attached").await;
    assert_eq!(
      status,
      http::StatusCode::SERVICE_UNAVAILABLE,
      "debugger_attached must not be signalled until pause injection completes"
    );

    let pause_frame = {
      let mut rx = cef.from_mux.lock().await;
      tokio::time::timeout(Duration::from_secs(5), rx.recv())
        .await
        .expect("no pause frame received")
        .unwrap()
    };
    let pause_val: Value =
      serde_json::from_slice(&pause_frame.payload).unwrap();
    assert_eq!(pause_val["method"], "Debugger.pause");

    // Now ack both from the upstream side, simulating CEF's responses.
    cef
      .to_mux
      .send(OwnedFrame::text(
        serde_json::to_vec(&json!({"id": -1, "result": {}})).unwrap(),
      ))
      .unwrap();
    cef
      .to_mux
      .send(OwnedFrame::text(
        serde_json::to_vec(&json!({"id": -2, "result": {}})).unwrap(),
      ))
      .unwrap();

    // With both injection acks in flight, the mux should mark
    // attached. Give it a moment to run.
    let mut saw_attached = false;
    for _ in 0..100 {
      let (status, _) = http_get(mux.listen, "/debugger-attached").await;
      if status == http::StatusCode::OK {
        saw_attached = true;
        break;
      }
      tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(
      saw_attached,
      "debugger_attached never flipped to 200 after injection completed"
    );

    let _client = client_connect.await.unwrap();
    drop(mux);
  }
}