deno_core 0.401.0

// Copyright 2018-2026 the Deno authors. MIT license.

//! The documentation for the inspector API is sparse, but these are helpful:
//! <https://chromedevtools.github.io/devtools-protocol/>
//! <https://web.archive.org/web/20210918052901/https://hyperandroid.com/2020/02/12/v8-inspector-from-an-embedder-standpoint/>

use std::cell::Cell;
use std::cell::RefCell;
use std::collections::HashMap;
use std::ffi::c_void;
use std::mem::take;
use std::pin::Pin;
use std::ptr::NonNull;
use std::rc::Rc;
use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
use std::thread;

use parking_lot::Mutex;

use crate::futures::channel::mpsc;
use crate::futures::channel::mpsc::UnboundedReceiver;
use crate::futures::channel::mpsc::UnboundedSender;
use crate::futures::channel::oneshot;
use crate::futures::prelude::*;
use crate::futures::stream::FuturesUnordered;
use crate::futures::stream::StreamExt;
use crate::futures::task;
use crate::serde_json::json;

#[derive(Debug)]
pub enum InspectorMsgKind {
  Notification,
  Message(i32),
}

#[derive(Debug)]
pub struct InspectorMsg {
  pub kind: InspectorMsgKind,
  pub content: String,
}

impl InspectorMsg {
  /// Create a notification message from a JSON value
  pub fn notification(content: serde_json::Value) -> Self {
    Self {
      kind: InspectorMsgKind::Notification,
      content: content.to_string(),
    }
  }
}

/// Maximum number of in-flight + completed requests kept for
/// Network.getResponseBody/streamResourceContent. Older entries are evicted.
const NETWORK_BUFFER_MAX_REQUESTS: usize = 32;
/// Per-request body cap. Beyond this, additional chunks are silently dropped
/// (matching Node's behavior in `RequestEntry::push_*_data_blob`).
const NETWORK_BUFFER_MAX_BYTES_PER_REQUEST: usize = 10 * 1024 * 1024;

#[derive(Debug, Default)]
struct NetworkDataEntry {
  /// Top-level `charset` from `Network.requestWillBeSent`, if any.
  /// Used by `Network.getRequestPostData` to decide whether the captured
  /// request body can be returned as a utf-8 string.
  request_charset: Option<String>,
  /// `response.charset` from `Network.responseReceived`, if any.
  /// Used by `Network.getResponseBody` to decide between utf-8 string
  /// and base64 binary output.
  response_charset: Option<String>,
  /// Accumulated request body bytes. Populated from `request.postData` on
  /// `requestWillBeSent`, then any subsequent `Network.dataSent` chunks.
  post_data: Vec<u8>,
  /// Decoded response body bytes, concatenated across `Network.dataReceived`
  /// events.
  data: Vec<u8>,
  /// `true` once the request body is fully sent: either no `hasPostData` was
  /// declared, or `Network.dataSent` with `finished: true` arrived.
  is_request_finished: bool,
  /// `true` once `Network.loadingFinished` arrived for this request.
  is_response_finished: bool,
  /// `true` once `Network.streamResourceContent` was called for this request:
  /// further `Network.dataReceived` events bypass the buffer and flow over the
  /// wire instead, matching Node's `is_streaming` switch.
  is_streaming: bool,
}

/// Bounded per-process buffer of request/response bodies, keyed by
/// CDP `requestId`. Sized to keep peak memory usage predictable so that
/// long-running processes attached to a debugger don't grow unbounded.
#[derive(Debug, Default)]
struct NetworkDataBuffer {
  entries: HashMap<String, NetworkDataEntry>,
  /// FIFO of request IDs in insertion order; oldest evicted first.
  order: std::collections::VecDeque<String>,
}

impl NetworkDataBuffer {
  fn ensure_capacity(&mut self) {
    while self.order.len() >= NETWORK_BUFFER_MAX_REQUESTS {
      if let Some(oldest) = self.order.pop_front() {
        self.entries.remove(&oldest);
      } else {
        break;
      }
    }
  }

  fn insert(&mut self, request_id: &str, entry: NetworkDataEntry) {
    if self.entries.contains_key(request_id) {
      return;
    }
    self.ensure_capacity();
    self.order.push_back(request_id.to_string());
    self.entries.insert(request_id.to_string(), entry);
  }

  fn get(&self, request_id: &str) -> Option<&NetworkDataEntry> {
    self.entries.get(request_id)
  }

  fn get_mut(&mut self, request_id: &str) -> Option<&mut NetworkDataEntry> {
    self.entries.get_mut(request_id)
  }

  fn erase(&mut self, request_id: &str) {
    if self.entries.remove(request_id).is_some() {
      self.order.retain(|id| id != request_id);
    }
  }
}

fn extend_capped(target: &mut Vec<u8>, bytes: &[u8]) {
  let remaining =
    NETWORK_BUFFER_MAX_BYTES_PER_REQUEST.saturating_sub(target.len());
  let take = bytes.len().min(remaining);
  target.extend_from_slice(&bytes[..take]);
}

/// Charset string matching Node's `request_charset == "utf-8"` check
/// in `network_agent.cc`. Anything else is treated as binary.
fn is_utf8_charset(charset: Option<&str>) -> bool {
  charset.is_some_and(|c| c.eq_ignore_ascii_case("utf-8"))
}

fn encode_b64(bytes: &[u8]) -> String {
  base64::Engine::encode(&base64::engine::general_purpose::STANDARD, bytes)
}

/// CDP error returned by the custom-domain handlers. The error code mirrors
/// Node's `protocol::DispatchResponse` constants used in `network_agent.cc`:
/// `InvalidParams` for caller-state issues (request not found, not finished,
/// streaming), `ServerError` for the binary-request-body case that the
/// protocol doesn't model.
enum CdpError {
  InvalidParams(&'static str),
  ServerError(&'static str),
}

impl CdpError {
  fn code(&self) -> i32 {
    match self {
      // Per JSON-RPC spec and Node's DispatchResponse::InvalidParams.
      Self::InvalidParams(_) => -32602,
      // Per Node's DispatchResponse::ServerError.
      Self::ServerError(_) => -32000,
    }
  }

  fn message(&self) -> &'static str {
    match self {
      Self::InvalidParams(m) | Self::ServerError(m) => m,
    }
  }
}

/// Build the CDP response for `Network.getResponseBody`. Matches Node's
/// `NetworkAgent::getResponseBody`: rejects if the response hasn't finished,
/// returns utf-8 as string or otherwise as base64, then erases the entry.
fn build_get_response_body(
  buf: &mut NetworkDataBuffer,
  request_id: &str,
) -> Result<serde_json::Value, CdpError> {
  let entry = buf
    .get(request_id)
    .ok_or(CdpError::InvalidParams("Request not found"))?;
  if entry.is_streaming {
    return Err(CdpError::InvalidParams(
      "Response body of the request is been streamed",
    ));
  }
  if !entry.is_response_finished {
    return Err(CdpError::InvalidParams("Response data is not finished yet"));
  }
  let result = if is_utf8_charset(entry.response_charset.as_deref())
    && let Ok(s) = std::str::from_utf8(&entry.data)
  {
    json!({ "body": s, "base64Encoded": false })
  } else {
    json!({
      "body": encode_b64(&entry.data),
      "base64Encoded": true,
    })
  };
  buf.erase(request_id);
  Ok(result)
}

/// Build the CDP response for `Network.streamResourceContent`. Matches Node's
/// `NetworkAgent::streamResourceContent`: returns currently-buffered data,
/// flips the entry to streaming mode (so subsequent `Network.dataReceived`
/// events flow over the wire), drains the buffer, and erases the entry if the
/// response already finished.
fn build_stream_resource_content(
  buf: &mut NetworkDataBuffer,
  request_id: &str,
) -> Result<serde_json::Value, CdpError> {
  let entry = buf
    .get_mut(request_id)
    .ok_or(CdpError::InvalidParams("Request not found"))?;
  entry.is_streaming = true;
  let buffered = std::mem::take(&mut entry.data);
  let is_response_finished = entry.is_response_finished;
  if is_response_finished {
    buf.erase(request_id);
  }
  Ok(json!({ "bufferedData": encode_b64(&buffered) }))
}

/// Build the CDP response for `Network.getRequestPostData`. Matches Node's
/// `NetworkAgent::getRequestPostData`: rejects if not yet finished, rejects
/// binary bodies, otherwise returns the body as a utf-8 string.
fn build_get_request_post_data(
  buf: &NetworkDataBuffer,
  request_id: &str,
) -> Result<serde_json::Value, CdpError> {
  let entry = buf
    .get(request_id)
    .ok_or(CdpError::InvalidParams("Request not found"))?;
  if !entry.is_request_finished {
    return Err(CdpError::InvalidParams("Request data is not finished yet"));
  }
  if !is_utf8_charset(entry.request_charset.as_deref()) {
    // Node returns ServerError here, not InvalidParams: the protocol simply
    // doesn't model binary post bodies, so this is a server-side limitation
    // rather than a caller mistake.
    return Err(CdpError::ServerError(
      "Unable to serialize binary request body",
    ));
  }
  let s = std::str::from_utf8(&entry.post_data)
    .map_err(|_| CdpError::ServerError("Request body is not valid UTF-8"))?;
  Ok(json!({ "postData": s }))
}

fn make_cdp_error_response(id: &serde_json::Value, error: &CdpError) -> String {
  json!({
    "id": id,
    "error": {
      "code": error.code(),
      "message": error.message(),
    }
  })
  .to_string()
}

fn make_cdp_ok_response(
  id: &serde_json::Value,
  result: serde_json::Value,
) -> String {
  json!({
    "id": id,
    "result": result,
  })
  .to_string()
}

enum CustomDomainOutcome {
  Empty,
  Result(serde_json::Value),
  Error(CdpError),
}

/// Dispatch a CDP command to one of the custom domains we own
/// (`Network.*`, `DOMStorage.enable/disable`). Returns `None` if the method
/// isn't ours and should be forwarded to V8.
fn dispatch_custom_domain(
  method: &str,
  parsed: &serde_json::Value,
  session: &InspectorSession,
) -> Option<CustomDomainOutcome> {
  match method {
    "Network.enable" => {
      session.state.network_enabled.set(true);
      Some(CustomDomainOutcome::Empty)
    }
    "Network.disable" => {
      session.state.network_enabled.set(false);
      Some(CustomDomainOutcome::Empty)
    }
    "DOMStorage.enable" => {
      session.state.dom_storage_enabled.set(true);
      Some(CustomDomainOutcome::Empty)
    }
    "DOMStorage.disable" => {
      session.state.dom_storage_enabled.set(false);
      Some(CustomDomainOutcome::Empty)
    }
    "Network.getResponseBody"
    | "Network.streamResourceContent"
    | "Network.getRequestPostData" => {
      let request_id = parsed
        .pointer("/params/requestId")
        .and_then(|v| v.as_str())
        .unwrap_or("");
      let mut buf = session.state.network_data.borrow_mut();
      let result = match method {
        "Network.getResponseBody" => {
          build_get_response_body(&mut buf, request_id)
        }
        "Network.streamResourceContent" => {
          build_stream_resource_content(&mut buf, request_id)
        }
        _ => build_get_request_post_data(&buf, request_id),
      };
      Some(match result {
        Ok(v) => CustomDomainOutcome::Result(v),
        Err(err) => CustomDomainOutcome::Error(err),
      })
    }
    _ => None,
  }
}

// TODO(bartlomieju): remove this
pub type SessionProxySender = UnboundedSender<InspectorMsg>;
// TODO(bartlomieju): remove this
pub type SessionProxyReceiver = UnboundedReceiver<String>;

/// Channels for an inspector session
pub enum InspectorSessionChannels {
  /// Regular inspector session with bidirectional communication
  Regular {
    tx: SessionProxySender,
    rx: SessionProxyReceiver,
  },
  /// Worker inspector session with separate channels for main<->worker communication
  Worker {
    /// Main thread sends commands TO worker
    main_to_worker_tx: UnboundedSender<String>,
    /// Worker sends responses/events TO main thread
    worker_to_main_rx: UnboundedReceiver<InspectorMsg>,
    /// Worker URL for identification
    worker_url: String,
  },
}

/// Encapsulates channels and metadata for creating an inspector session
pub struct InspectorSessionProxy {
  pub channels: InspectorSessionChannels,
  pub kind: InspectorSessionKind,
}

/// Creates a pair of inspector session proxies for worker-main communication.
pub fn create_worker_inspector_session_pair(
  worker_url: String,
) -> (InspectorSessionProxy, InspectorSessionProxy) {
  let (worker_to_main_tx, worker_to_main_rx) =
    mpsc::unbounded::<InspectorMsg>();
  let (main_to_worker_tx, main_to_worker_rx) = mpsc::unbounded::<String>();

  let main_side = InspectorSessionProxy {
    channels: InspectorSessionChannels::Worker {
      main_to_worker_tx,
      worker_to_main_rx,
      worker_url,
    },
    kind: InspectorSessionKind::NonBlocking {
      wait_for_disconnect: false,
    },
  };

  let worker_side = InspectorSessionProxy {
    channels: InspectorSessionChannels::Regular {
      tx: worker_to_main_tx,
      rx: main_to_worker_rx,
    },
    kind: InspectorSessionKind::NonBlocking {
      wait_for_disconnect: false,
    },
  };

  (main_side, worker_side)
}

pub type InspectorSessionSend = Box<dyn Fn(InspectorMsg)>;

#[derive(Clone, Copy, Debug)]
enum PollState {
  Idle,
  Woken,
  Polling,
  Parked,
  Dropped,
}

/// This structure is used responsible for providing inspector interface
/// to the `JsRuntime`.
///
/// It stores an instance of `v8::inspector::V8Inspector` and additionally
/// implements `v8::inspector::V8InspectorClientImpl`.
///
/// After creating this structure it's possible to connect multiple sessions
/// to the inspector, in case of Deno it's either: a "websocket session" that
/// provides integration with Chrome Devtools, or an "in-memory session" that
/// is used for REPL or coverage collection.
pub struct JsRuntimeInspector {
  v8_inspector: Rc<v8::inspector::V8Inspector>,
  new_session_tx: UnboundedSender<InspectorSessionProxy>,
  deregister_tx: RefCell<Option<oneshot::Sender<()>>>,
  state: Rc<JsRuntimeInspectorState>,
}

impl Drop for JsRuntimeInspector {
  fn drop(&mut self) {
    // Since the waker is cloneable, it might outlive the inspector itself.
    // Set the poll state to 'dropped' so it doesn't attempt to request an
    // interrupt from the isolate.
    self
      .state
      .waker
      .update(|w| w.poll_state = PollState::Dropped);

    // V8 automatically deletes all sessions when an `V8Inspector` instance is
    // deleted, however InspectorSession also has a drop handler that cleans
    // up after itself. To avoid a double free, make sure the inspector is
    // dropped last.
    self.state.sessions.borrow_mut().drop_sessions();

    // Notify counterparty that this instance is being destroyed. Ignoring
    // result because counterparty waiting for the signal might have already
    // dropped the other end of channel.
    if let Some(deregister_tx) = self.deregister_tx.borrow_mut().take() {
      let _ = deregister_tx.send(());
    }
  }
}

#[derive(Clone)]
struct JsRuntimeInspectorState {
  isolate_ptr: v8::UnsafeRawIsolatePtr,
  context: v8::Global<v8::Context>,
  flags: Rc<RefCell<InspectorFlags>>,
  waker: Arc<InspectorWaker>,
  sessions: Rc<RefCell<SessionContainer>>,
  is_dispatching_message: Rc<RefCell<bool>>,
  pending_worker_messages: Arc<Mutex<Vec<(String, String)>>>,
  nodeworker_enabled: Rc<Cell<bool>>,
  auto_attach_enabled: Rc<Cell<bool>>,
  discover_targets_enabled: Rc<Cell<bool>>,
  /// Shared buffer of captured Network.* request/response bodies, populated
  /// from `op_inspector_emit_protocol_event` and read by the dispatcher
  /// when handling `Network.getResponseBody`/`getRequestPostData`/
  /// `streamResourceContent`.
  network_data: Rc<RefCell<NetworkDataBuffer>>,
}

struct JsRuntimeInspectorClient(Rc<JsRuntimeInspectorState>);

impl v8::inspector::V8InspectorClientImpl for JsRuntimeInspectorClient {
  fn run_message_loop_on_pause(&self, context_group_id: i32) {
    assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID);
    self.0.flags.borrow_mut().on_pause = true;
    let _ = self.0.poll_sessions(None);
  }

  fn quit_message_loop_on_pause(&self) {
    self.0.flags.borrow_mut().on_pause = false;
  }

  fn run_if_waiting_for_debugger(&self, context_group_id: i32) {
    assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID);
    let mut flags = self.0.flags.borrow_mut();
    flags.waiting_for_session = false;
    flags.paused_on_start = false;
  }

  fn ensure_default_context_in_group(
    &self,
    context_group_id: i32,
  ) -> Option<v8::Local<'_, v8::Context>> {
    assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID);
    let context = self.0.context.clone();
    let mut isolate =
      unsafe { v8::Isolate::from_raw_isolate_ptr(self.0.isolate_ptr) };
    let isolate = &mut isolate;
    v8::callback_scope!(unsafe scope, isolate);
    let local = v8::Local::new(scope, context);
    Some(unsafe { local.extend_lifetime_unchecked() })
  }

  fn resource_name_to_url(
    &self,
    resource_name: &v8::inspector::StringView,
  ) -> Option<v8::UniquePtr<v8::inspector::StringBuffer>> {
    let resource_name = resource_name.to_string();
    #[allow(
      clippy::disallowed_methods,
      reason = "Url::from_file_path is more efficient when not using the error and this code doesn't need Wasm support"
    )]
    let url = url::Url::from_file_path(resource_name).ok()?;
    let src_view = v8::inspector::StringView::from(url.as_str().as_bytes());
    Some(v8::inspector::StringBuffer::create(src_view))
  }
}

impl JsRuntimeInspectorState {
  #[allow(clippy::result_unit_err, reason = "error details not needed")]
  pub fn poll_sessions(
    &self,
    mut invoker_cx: Option<&mut Context>,
  ) -> Result<Poll<()>, ()> {
    // The futures this function uses do not have re-entrant poll() functions.
    // However it is can happen that poll_sessions() gets re-entered, e.g.
    // when an interrupt request is honored while the inspector future is polled
    // by the task executor. We let the caller know by returning some error.
    let Ok(mut sessions) = self.sessions.try_borrow_mut() else {
      return Err(());
    };

    self.waker.update(|w| {
      match w.poll_state {
        PollState::Idle | PollState::Woken => w.poll_state = PollState::Polling,
        _ => unreachable!(),
      };
    });

    // Create a new Context object that will make downstream futures
    // use the InspectorWaker when they are ready to be polled again.
    let waker_ref = task::waker_ref(&self.waker);
    let cx = &mut Context::from_waker(&waker_ref);

    loop {
      loop {
        // Do one "handshake" with a newly connected session at a time.
        if let Some(session) = sessions.handshake.take() {
          let id = sessions.next_local_id;
          sessions.next_local_id += 1;
          let mut fut =
            pump_inspector_session_messages(session.clone(), id).boxed_local();
          // Only add to established if the future is still pending.
          // If the channel is already closed (e.g., worker terminated quickly),
          // the future may complete immediately on first poll. Pushing a
          // completed future to FuturesUnordered would cause a panic when
          // polled again. Also skip inserting into local — the session is
          // already dead and would never be cleaned up (no established future
          // to yield back its ID).
          if fut.poll_unpin(cx).is_pending() {
            sessions.established.push(fut);
            sessions.local.insert(id, session);

            // Track the first session as the main session for Target events
            if sessions.main_session_id.is_none() {
              sessions.main_session_id = Some(id);
            }
          }

          continue;
        }

        // Accept new connections.
        if let Poll::Ready(Some(session_proxy)) =
          sessions.session_rx.poll_next_unpin(cx)
        {
          match session_proxy.channels {
            InspectorSessionChannels::Worker {
              main_to_worker_tx,
              worker_to_main_rx,
              worker_url,
            } => {
              // Get the next local ID for this worker
              let worker_id = sessions.next_local_id;
              sessions.next_local_id += 1;

              sessions.register_worker_session(worker_id, worker_url.clone());

              // Register the worker channels
              sessions.register_worker_channels(
                worker_id,
                main_to_worker_tx,
                worker_to_main_rx,
              );

              // Notify the main session about worker target creation/attachment.
              //
              // discover_targets_enabled: Set to true when the debugger calls
              // Target.setDiscoverTargets. When enabled, the inspector sends
              // Target.targetCreated events for new workers.
              //
              // auto_attach_enabled: Set to true when the debugger calls
              // Target.setAutoAttach. When enabled, the inspector automatically
              // attaches to new workers and sends Target.attachedToTarget events.
              if let Some(main_id) = sessions.main_session_id
                && let Some(main_session) = sessions.local.get(&main_id)
                && let Some(ts) =
                  sessions.target_sessions.get(&format!("{}", worker_id))
              {
                if self.discover_targets_enabled.get() {
                  (main_session.state.send)(InspectorMsg::notification(
                    json!({
                      "method": "Target.targetCreated",
                      "params": { "targetInfo": ts.target_info(false) }
                    }),
                  ));
                }

                if self.auto_attach_enabled.get() {
                  ts.attached.set(true);
                  (main_session.state.send)(InspectorMsg::notification(
                    json!({
                      "method": "Target.attachedToTarget",
                      "params": {
                        "sessionId": ts.session_id,
                        "targetInfo": ts.target_info(true),
                        "waitingForDebugger": false
                      }
                    }),
                  ));
                }
              }

              continue;
            }
            InspectorSessionChannels::Regular { tx, rx } => {
              // Normal session (not a worker)
              let session = InspectorSession::new(
                sessions.v8_inspector.as_ref().unwrap().clone(),
                self.is_dispatching_message.clone(),
                Box::new(move |msg| {
                  let _ = tx.unbounded_send(msg);
                }),
                Some(rx),
                session_proxy.kind,
                self.sessions.clone(),
                self.pending_worker_messages.clone(),
                self.nodeworker_enabled.clone(),
                self.auto_attach_enabled.clone(),
                self.discover_targets_enabled.clone(),
                self.flags.clone(),
                self.network_data.clone(),
              );

              let prev = sessions.handshake.replace(session);
              assert!(prev.is_none());
              continue;
            }
          }
        }

        // Poll worker message channels - forward messages from workers to main session
        if let Some(main_id) = sessions.main_session_id {
          // Get main session send function before mutably iterating over target_sessions
          let main_session_send =
            sessions.local.get(&main_id).map(|s| s.state.send.clone());

          if let Some(send) = main_session_send {
            let mut has_worker_message = false;
            let mut terminated_workers = Vec::new();

            for target_session in sessions.target_sessions.values() {
              // Skip sessions that haven't had their channels registered
              if !target_session.has_channels() {
                continue;
              }
              match target_session.poll_from_worker(cx) {
                Poll::Ready(Some(msg)) => {
                  // CDP Flattened Session Mode: Add sessionId at top level
                  // Used by Chrome DevTools (when NodeWorker.enable has NOT been called)
                  // VSCode uses NodeWorker.receivedMessageFromWorker instead
                  if !self.nodeworker_enabled.get() {
                    if let Ok(mut parsed) =
                      serde_json::from_str::<serde_json::Value>(&msg.content)
                    {
                      if let Some(obj) = parsed.as_object_mut() {
                        obj.insert(
                          "sessionId".to_string(),
                          json!(target_session.session_id),
                        );
                        let flattened_msg = parsed.to_string();

                        // Send the flattened response with sessionId at top level
                        send(InspectorMsg {
                          kind: msg.kind,
                          content: flattened_msg,
                        });
                      }
                    } else {
                      // Fallback: send as-is if not valid JSON
                      send(msg);
                    }
                  } else {
                    let wrapped_nodeworker = json!({
                      "method": "NodeWorker.receivedMessageFromWorker",
                      "params": {
                        "sessionId": target_session.session_id,
                        "message": msg.content,
                        "workerId": target_session.target_id
                      }
                    });

                    send(InspectorMsg {
                      kind: InspectorMsgKind::Notification,
                      content: wrapped_nodeworker.to_string(),
                    });
                  }

                  has_worker_message = true;
                }
                Poll::Ready(None) => {
                  // Worker channel closed - worker has terminated
                  // Notify debugger based on which protocol is in use
                  if self.nodeworker_enabled.get() {
                    // VSCode / Node.js style
                    send(InspectorMsg::notification(json!({
                      "method": "NodeWorker.detachedFromWorker",
                      "params": {
                        "sessionId": target_session.session_id
                      }
                    })));
                  } else if self.auto_attach_enabled.get()
                    || self.discover_targets_enabled.get()
                  {
                    // Chrome DevTools style
                    send(InspectorMsg::notification(json!({
                      "method": "Target.targetDestroyed",
                      "params": {
                        "targetId": target_session.target_id
                      }
                    })));
                  }
                  terminated_workers.push(target_session.session_id.clone());
                }
                Poll::Pending => {}
              }
            }

            // Clean up terminated worker sessions
            for session_id in terminated_workers {
              sessions.target_sessions.remove(&session_id);
              has_worker_message = true; // Trigger re-poll
            }

            if has_worker_message {
              continue;
            }
          }
        }

        // Poll established sessions.
        match sessions.established.poll_next_unpin(cx) {
          Poll::Ready(Some(completed_id)) => {
            // A session's pump future completed (WS disconnected).
            // Remove it from local so sessions_state() no longer
            // reports it as active.
            sessions.local.remove(&completed_id);
            // If this was the main session, promote another session
            // or clear, so new connections can become main and
            // Target/worker notifications continue to work.
            if sessions.main_session_id == Some(completed_id) {
              // Promote an arbitrary remaining session. HashMap iteration
              // order is non-deterministic, but in practice there is
              // typically only one debugger client connected at a time.
              sessions.main_session_id = sessions.local.keys().next().copied();
            }
            continue;
          }
          Poll::Ready(None) => {
            break;
          }
          Poll::Pending => {
            break;
          }
        };
      }

      let should_block = {
        let flags = self.flags.borrow();
        flags.on_pause || flags.waiting_for_session
      };

      // Process any queued NodeWorker messages after polling completes
      // Drain from the thread-safe Mutex queue (doesn't require borrowing sessions)
      let pending_messages: Vec<(String, String)> = {
        let mut queue = self.pending_worker_messages.lock();
        queue.drain(..).collect()
      };

      for (session_id, message) in pending_messages {
        if let Some(target_session) = sessions.target_sessions.get(&session_id)
        {
          target_session.send_to_worker(message);
        }
      }

      let new_state = self.waker.update(|w| {
        match w.poll_state {
          PollState::Woken => {
            // The inspector was woken while the session handler was being
            // polled, so we poll it another time.
            w.poll_state = PollState::Polling;
          }
          PollState::Polling if !should_block => {
            // The session handler doesn't need to be polled any longer, and
            // there's no reason to block (execution is not paused), so this
            // function is about to return.
            w.poll_state = PollState::Idle;
            // Register the task waker that can be used to wake the parent
            // task that will poll the inspector future.
            if let Some(cx) = invoker_cx.take() {
              w.task_waker.replace(cx.waker().clone());
            }
            // Register the address of the inspector, which allows the waker
            // to request an interrupt from the isolate.
            w.inspector_state_ptr = NonNull::new(self as *const _ as *mut Self);
          }
          PollState::Polling if should_block => {
            // Isolate execution has been paused but there are no more
            // events to process, so this thread will be parked. Therefore,
            // store the current thread handle in the waker so it knows
            // which thread to unpark when new events arrive.
            w.poll_state = PollState::Parked;
            w.parked_thread.replace(thread::current());
          }
          _ => unreachable!(),
        };
        w.poll_state
      });
      match new_state {
        PollState::Idle => break,            // Yield to task.
        PollState::Polling => continue,      // Poll the session handler again.
        PollState::Parked => thread::park(), // Park the thread.
        _ => unreachable!(),
      };
    }

    Ok(Poll::Pending)
  }
}

impl JsRuntimeInspector {
  /// Currently Deno supports only a single context in `JsRuntime`
  /// and thus it's id is provided as an associated constant.
  const CONTEXT_GROUP_ID: i32 = 1;

  pub fn new(
    isolate_ptr: v8::UnsafeRawIsolatePtr,
    scope: &mut v8::PinScope,
    context: v8::Local<v8::Context>,
    is_main_runtime: bool,
    worker_id: Option<u32>,
  ) -> Rc<Self> {
    let (new_session_tx, new_session_rx) =
      mpsc::unbounded::<InspectorSessionProxy>();

    let waker = InspectorWaker::new(scope.thread_safe_handle());
    let state = Rc::new(JsRuntimeInspectorState {
      waker,
      flags: Default::default(),
      isolate_ptr,
      context: v8::Global::new(scope, context),
      sessions: Rc::new(
        RefCell::new(SessionContainer::temporary_placeholder()),
      ),
      is_dispatching_message: Default::default(),
      pending_worker_messages: Arc::new(Mutex::new(Vec::new())),
      nodeworker_enabled: Rc::new(Cell::new(false)),
      auto_attach_enabled: Rc::new(Cell::new(false)),
      discover_targets_enabled: Rc::new(Cell::new(false)),
      network_data: Rc::new(RefCell::new(NetworkDataBuffer::default())),
    });
    let client = Box::new(JsRuntimeInspectorClient(state.clone()));
    let v8_inspector_client = v8::inspector::V8InspectorClient::new(client);
    let v8_inspector = Rc::new(v8::inspector::V8Inspector::create(
      scope,
      v8_inspector_client,
    ));

    *state.sessions.borrow_mut() =
      SessionContainer::new(v8_inspector.clone(), new_session_rx);

    // Tell the inspector about the main realm.
    let context_name_bytes = if is_main_runtime {
      &b"main realm"[..]
    } else {
      &format!("worker [{}]", worker_id.unwrap_or(1)).into_bytes()
    };

    let context_name = v8::inspector::StringView::from(context_name_bytes);
    // NOTE(bartlomieju): this is what Node.js does and it turns out some
    // debuggers (like VSCode) rely on this information to disconnect after
    // program completes.
    // The auxData structure should match {isDefault: boolean, type: 'default'|'isolated'|'worker'}
    // For Chrome DevTools to properly show workers in the execution context dropdown.
    let aux_data = if is_main_runtime {
      r#"{"isDefault": true, "type": "default"}"#
    } else {
      r#"{"isDefault": false, "type": "worker"}"#
    };
    let aux_data_view = v8::inspector::StringView::from(aux_data.as_bytes());
    v8_inspector.context_created(
      context,
      Self::CONTEXT_GROUP_ID,
      context_name,
      aux_data_view,
    );

    // Poll the session handler so we will get notified whenever there is
    // new incoming debugger activity.
    let _ = state.poll_sessions(None).unwrap();

    Rc::new(Self {
      v8_inspector,
      state,
      new_session_tx,
      deregister_tx: RefCell::new(None),
    })
  }

  pub fn is_dispatching_message(&self) -> bool {
    *self.state.is_dispatching_message.borrow()
  }

  pub fn context_destroyed(
    &self,
    scope: &mut v8::PinScope<'_, '_>,
    context: v8::Global<v8::Context>,
  ) {
    let context = v8::Local::new(scope, context);
    self.v8_inspector.context_destroyed(context);
  }

  pub fn exception_thrown(
    &self,
    scope: &mut v8::PinScope<'_, '_>,
    exception: v8::Local<'_, v8::Value>,
    in_promise: bool,
  ) {
    let context = scope.get_current_context();
    let message = v8::Exception::create_message(scope, exception);
    let stack_trace = message.get_stack_trace(scope);
    let stack_trace = self.v8_inspector.create_stack_trace(stack_trace);
    self.v8_inspector.exception_thrown(
      context,
      if in_promise {
        v8::inspector::StringView::from("Uncaught (in promise)".as_bytes())
      } else {
        v8::inspector::StringView::from("Uncaught".as_bytes())
      },
      exception,
      v8::inspector::StringView::from("".as_bytes()),
      v8::inspector::StringView::from("".as_bytes()),
      0,
      0,
      stack_trace,
      0,
    );
  }

  /// Broadcast `Runtime.executionContextDestroyed` to all connected sessions
  /// without requiring a V8 scope. This is used during `process.exit()` to
  /// notify debuggers that the execution context is being torn down before
  /// calling `std::process::exit()`, which would skip the normal event-loop
  /// shutdown path where V8's `context_destroyed` is called.
  ///
  /// Sessions that opted into `NodeRuntime.notifyWhenWaitingForDisconnect`
  /// are skipped here — they receive `NodeRuntime.waitingForDisconnect`
  /// instead, via `broadcast_waiting_for_disconnect`.
  pub fn broadcast_context_destroyed(&self) {
    let sessions = self.state.sessions.borrow();
    for session in sessions.local.values() {
      if session.state.notify_waiting_for_disconnect.get() {
        continue;
      }
      (session.state.send)(InspectorMsg::notification(json!({
        "method": "Runtime.executionContextDestroyed",
        "params": { "executionContextId": Self::CONTEXT_GROUP_ID }
      })));
    }
  }

  /// Emit `NodeRuntime.waitingForDisconnect` to every session that opted in
  /// via `NodeRuntime.notifyWhenWaitingForDisconnect(enabled: true)`. Called
  /// during the process-exit handler before blocking for disconnect, so
  /// debugger clients can close cleanly instead of waiting for a TCP RST.
  pub fn broadcast_waiting_for_disconnect(&self) {
    let sessions = self.state.sessions.borrow();
    for session in sessions.local.values() {
      if !session.state.notify_waiting_for_disconnect.get() {
        continue;
      }
      (session.state.send)(InspectorMsg::notification(json!({
        "method": "NodeRuntime.waitingForDisconnect"
      })));
    }
  }

  /// Block the current thread until all inspector sessions have disconnected.
  /// Used after `broadcast_context_destroyed` to give debuggers a chance to
  /// process the notification before the process exits.
  ///
  /// Note: this intentionally does not set a blocking flag on `poll_sessions`.
  /// Unlike `wait_for_session` (which blocks waiting for a *new* WS
  /// connection handled by a separate server thread), disconnect detection
  /// requires the async I/O reactor to process WebSocket close frames.
  /// Parking the thread would prevent those events from being delivered.
  /// The loop spins with `poll_sessions(None)` which returns quickly when
  /// idle, yielding a brief busy-wait that is acceptable given this only
  /// runs during process exit.
  pub fn wait_for_sessions_disconnect(&self) {
    loop {
      {
        let sessions = self.state.sessions.borrow();
        if sessions.local.is_empty() && sessions.established.is_empty() {
          break;
        }
      }
      let _ = self.state.poll_sessions(None);
    }
  }

  pub fn sessions_state(&self) -> SessionsState {
    self.state.sessions.borrow().sessions_state()
  }

  pub fn poll_sessions_from_event_loop(&self, cx: &mut Context) {
    let _ = self.state.poll_sessions(Some(cx)).unwrap();
  }

  /// This function blocks the thread until at least one inspector client has
  /// established a websocket connection.
  pub fn wait_for_session(&self) {
    loop {
      if let Some(_session) =
        self.state.sessions.borrow_mut().local.values().next()
      {
        self.state.flags.borrow_mut().waiting_for_session = false;
        break;
      } else {
        self.state.flags.borrow_mut().waiting_for_session = true;
        let _ = self.state.poll_sessions(None).unwrap();
      }
    }
  }

  /// This function blocks the thread until at least one inspector client has
  /// established a websocket connection.
  ///
  /// After that, it instructs V8 to pause at the next statement.
  /// Frontend must send "Runtime.runIfWaitingForDebugger" message to resume
  /// execution.
  pub fn wait_for_session_and_break_on_next_statement(&self) {
    self.state.flags.borrow_mut().paused_on_start = true;
    // Block the main thread until Runtime.runIfWaitingForDebugger is
    // received (which clears paused_on_start). This matches Node.js
    // --inspect-brk behavior where execution is blocked until the
    // frontend explicitly resumes.
    //
    // poll_sessions will block when waiting_for_session is true and
    // process incoming messages (including from WS clients). The pump
    // handler intercepts Runtime.runIfWaitingForDebugger and clears
    // paused_on_start. Since poll_sessions processes all pending
    // messages before returning, paused_on_start will be false when
    // poll_sessions returns (if the message was received).
    loop {
      if !self.state.flags.borrow().paused_on_start {
        break;
      }
      self.state.flags.borrow_mut().waiting_for_session = true;
      let _ = self.state.poll_sessions(None).unwrap();
    }
    // Schedule a V8 debugger break on the next statement. By this point
    // the frontend has sent Debugger.enable (enabling the debugger agent)
    // and Runtime.runIfWaitingForDebugger (which unblocked us above).
    // The break causes a Debugger.paused notification, matching the
    // --inspect-brk behavior where execution pauses at the first statement.
    // Use the session that sent runIfWaitingForDebugger (it has Debugger
    // enabled), falling back to any available session.
    let sessions = self.state.sessions.borrow();
    let resumed_by = self.state.flags.borrow().resumed_by_session_id;
    let session = resumed_by
      .and_then(|id| sessions.local.get(&id))
      .or_else(|| sessions.local.values().next());
    if let Some(session) = session {
      let reason = v8::inspector::StringView::from(&b"debugCommand"[..]);
      let detail = v8::inspector::StringView::empty();
      session
        .v8_session
        .schedule_pause_on_next_statement(reason, detail);
    }
    // Clear so it doesn't stale-reference a session in future cycles.
    self.state.flags.borrow_mut().resumed_by_session_id = None;
  }

  /// Obtain a sender for proxy channels.
  pub fn get_session_sender(&self) -> UnboundedSender<InspectorSessionProxy> {
    self.new_session_tx.clone()
  }

  /// Create a channel that notifies the frontend when inspector is dropped.
  ///
  /// NOTE: Only a single handler is currently available.
  pub fn add_deregister_handler(&self) -> oneshot::Receiver<()> {
    let maybe_deregister_tx = self.deregister_tx.borrow_mut().take();
    if let Some(deregister_tx) = maybe_deregister_tx
      && !deregister_tx.is_canceled()
    {
      panic!("Inspector deregister handler already exists and is alive.");
    }
    let (tx, rx) = oneshot::channel::<()>();
    self.deregister_tx.borrow_mut().replace(tx);
    rx
  }

  /// Capture an incoming `Network.*` event payload into the shared body buffer
  /// before it is broadcast to sessions. Called from
  /// `op_inspector_emit_protocol_event` so subsequent
  /// `Network.getResponseBody`/`streamResourceContent`/`getRequestPostData`
  /// CDP commands have something to return.
  ///
  /// Returns `true` if the event should still be broadcast to sessions, or
  /// `false` if the event was fully consumed by the buffer. Matches Node's
  /// `NetworkAgent`: `dataSent` is purely a capture event, and `dataReceived`
  /// is only forwarded once `streamResourceContent` has flipped the entry
  /// into streaming mode.
  pub fn capture_network_event(
    &self,
    event_name: &str,
    params: &serde_json::Value,
  ) -> bool {
    let Some(request_id) = params.get("requestId").and_then(|v| v.as_str())
    else {
      return true;
    };

    let mut buf = self.state.network_data.borrow_mut();
    match event_name {
      "Network.requestWillBeSent" => {
        let charset = params
          .get("charset")
          .and_then(|v| v.as_str())
          .map(|s| s.to_string());
        let has_post_data = params
          .pointer("/request/hasPostData")
          .and_then(|v| v.as_bool())
          .unwrap_or(false);
        let mut entry = NetworkDataEntry {
          request_charset: charset,
          is_request_finished: !has_post_data,
          ..NetworkDataEntry::default()
        };
        if let Some(post_data) =
          params.pointer("/request/postData").and_then(|v| v.as_str())
        {
          extend_capped(&mut entry.post_data, post_data.as_bytes());
        }
        buf.insert(request_id, entry);
        true
      }
      "Network.responseReceived" => {
        if let Some(entry) = buf.get_mut(request_id) {
          entry.response_charset = params
            .pointer("/response/charset")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        }
        true
      }
      "Network.loadingFinished" => {
        if let Some(entry) = buf.get_mut(request_id) {
          entry.is_response_finished = true;
          if entry.is_streaming {
            buf.erase(request_id);
          }
        }
        true
      }
      "Network.loadingFailed" => {
        buf.erase(request_id);
        true
      }
      "Network.dataReceived" => {
        let Some(entry) = buf.get_mut(request_id) else {
          // No tracked request — pass through to the wire.
          return true;
        };
        if entry.is_streaming {
          // Streaming mode: bypass the buffer and forward to sessions.
          return true;
        }
        if let Some(data_b64) = params.get("data").and_then(|v| v.as_str())
          && let Ok(bytes) = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            data_b64,
          )
        {
          extend_capped(&mut entry.data, &bytes);
        }
        false
      }
      "Network.dataSent" => {
        let Some(entry) = buf.get_mut(request_id) else {
          return false;
        };
        if params
          .get("finished")
          .and_then(|v| v.as_bool())
          .unwrap_or(false)
        {
          entry.is_request_finished = true;
          return false;
        }
        if let Some(data_b64) = params.get("data").and_then(|v| v.as_str())
          && let Ok(bytes) = base64::Engine::decode(
            &base64::engine::general_purpose::STANDARD,
            data_b64,
          )
        {
          extend_capped(&mut entry.post_data, &bytes);
        }
        false
      }
      _ => true,
    }
  }

  /// Broadcast a protocol event notification to all sessions that have the
  /// given domain enabled. Used for custom domains like Network.
  pub fn broadcast_to_sessions(&self, event_name: &str, params_json: &str) {
    let domain = event_name.split('.').next().unwrap_or("");
    let notification = json!({
      "method": event_name,
      "params": serde_json::from_str::<serde_json::Value>(params_json)
        .unwrap_or_else(|_| json!({}))
    })
    .to_string();

    let sessions = self.state.sessions.borrow();

    // `sessions.local` contains all active sessions — both programmatic
    // (created via create_local_session) and WebSocket-based (added during
    // handshake). The `established` field only holds their pump futures.
    for session in sessions.local.values() {
      let enabled = match domain {
        "Network" => session.state.network_enabled.get(),
        "DOMStorage" => session.state.dom_storage_enabled.get(),
        _ => false,
      };
      if enabled {
        (session.state.send)(InspectorMsg {
          kind: InspectorMsgKind::Notification,
          content: notification.clone(),
        });
      }
    }
  }

  pub fn create_local_session(
    inspector: Rc<JsRuntimeInspector>,
    callback: InspectorSessionSend,
    kind: InspectorSessionKind,
  ) -> LocalInspectorSession {
    let (session_id, sessions) = {
      let sessions = inspector.state.sessions.clone();

      let inspector_session = InspectorSession::new(
        inspector.v8_inspector.clone(),
        inspector.state.is_dispatching_message.clone(),
        callback,
        None,
        kind,
        sessions.clone(),
        inspector.state.pending_worker_messages.clone(),
        inspector.state.nodeworker_enabled.clone(),
        inspector.state.auto_attach_enabled.clone(),
        inspector.state.discover_targets_enabled.clone(),
        inspector.state.flags.clone(),
        inspector.state.network_data.clone(),
      );

      let session_id = {
        let mut s = sessions.borrow_mut();
        let id = s.next_local_id;
        s.next_local_id += 1;
        assert!(s.local.insert(id, inspector_session).is_none());
        id
      };

      take(&mut inspector.state.flags.borrow_mut().waiting_for_session);
      (session_id, sessions)
    };

    LocalInspectorSession::new(session_id, sessions)
  }
}

#[derive(Default)]
struct InspectorFlags {
  waiting_for_session: bool,
  on_pause: bool,
  /// Set when --inspect-brk is used. Remains true until
  /// Runtime.runIfWaitingForDebugger is received, allowing
  /// NodeRuntime.waitingForDebugger to be emitted.
  paused_on_start: bool,
  /// Tracks which session sent Runtime.runIfWaitingForDebugger,
  /// so schedule_pause_on_next_statement targets the correct session.
  resumed_by_session_id: Option<i32>,
}

#[derive(Debug)]
pub struct SessionsState {
  pub has_active: bool,
  pub has_blocking: bool,
  pub has_nonblocking: bool,
  pub has_nonblocking_wait_for_disconnect: bool,
}

/// A helper structure that helps coordinate sessions during different
/// parts of their lifecycle.
pub struct SessionContainer {
  v8_inspector: Option<Rc<v8::inspector::V8Inspector>>,
  session_rx: UnboundedReceiver<InspectorSessionProxy>,
  handshake: Option<Rc<InspectorSession>>,
  established: FuturesUnordered<InspectorSessionPumpMessages>,
  next_local_id: i32,
  local: HashMap<i32, Rc<InspectorSession>>,

  target_sessions: HashMap<String, Rc<TargetSession>>, // sessionId -> TargetSession
  main_session_id: Option<i32>, // The first session that should receive Target events
  next_worker_id: u32, // Sequential ID for worker display naming (1, 2, 3, ...)
}

struct MainWorkerChannels {
  main_to_worker_tx: UnboundedSender<String>,
  worker_to_main_rx: UnboundedReceiver<InspectorMsg>,
}

/// Represents a CDP Target session (e.g., a worker)
struct TargetSession {
  target_id: String,
  session_id: String,
  local_session_id: i32,
  /// Sequential worker ID for display (1, 2, 3, ...) - independent from session IDs
  worker_id: u32,
  main_worker_channels: RefCell<Option<MainWorkerChannels>>,
  url: String,
  /// Track if we've already sent attachedToTarget for this session
  attached: Cell<bool>,
}

impl TargetSession {
  /// Get a display title for the worker using Node.js style naming
  /// e.g., "worker [1]", "worker [2]"
  fn title(&self) -> String {
    format!("worker [{}]", self.worker_id)
  }

  /// Send a message to the worker (main → worker direction)
  fn send_to_worker(&self, message: String) {
    if let Some(channels) = self.main_worker_channels.borrow().as_ref() {
      let _ = channels.main_to_worker_tx.unbounded_send(message);
    }
  }

  /// Returns true if worker channels have been registered
  fn has_channels(&self) -> bool {
    self.main_worker_channels.borrow().is_some()
  }

  /// Poll for messages from the worker (worker → main direction).
  /// Panics if channels have not been registered yet - caller should
  /// check has_channels() first.
  fn poll_from_worker(&self, cx: &mut Context) -> Poll<Option<InspectorMsg>> {
    self
      .main_worker_channels
      .borrow_mut()
      .as_mut()
      .expect("poll_from_worker called before channels were registered")
      .worker_to_main_rx
      .poll_next_unpin(cx)
  }
}

impl SessionContainer {
  fn new(
    v8_inspector: Rc<v8::inspector::V8Inspector>,
    new_session_rx: UnboundedReceiver<InspectorSessionProxy>,
  ) -> Self {
    Self {
      v8_inspector: Some(v8_inspector),
      session_rx: new_session_rx,
      handshake: None,
      established: FuturesUnordered::new(),
      next_local_id: 1,
      local: HashMap::new(),

      target_sessions: HashMap::new(),
      main_session_id: None,
      next_worker_id: 1, // Workers are numbered starting from 1
    }
  }

  /// V8 automatically deletes all sessions when an `V8Inspector` instance is
  /// deleted, however InspectorSession also has a drop handler that cleans
  /// up after itself. To avoid a double free, we need to manually drop
  /// all sessions before dropping the inspector instance.
  fn drop_sessions(&mut self) {
    self.v8_inspector = Default::default();
    self.handshake.take();
    self.established.clear();
    self.local.clear();
  }

  fn sessions_state(&self) -> SessionsState {
    SessionsState {
      has_active: !self.established.is_empty()
        || self.handshake.is_some()
        || !self.local.is_empty(),
      has_blocking: self
        .local
        .values()
        .any(|s| matches!(s.state.kind, InspectorSessionKind::Blocking)),
      has_nonblocking: self.local.values().any(|s| {
        matches!(s.state.kind, InspectorSessionKind::NonBlocking { .. })
      }),
      has_nonblocking_wait_for_disconnect: self.local.values().any(|s| {
        matches!(
          s.state.kind,
          InspectorSessionKind::NonBlocking {
            wait_for_disconnect: true
          }
        )
      }),
    }
  }

  /// A temporary placeholder that should be used before actual
  /// instance of V8Inspector is created. It's used in favor
  /// of `Default` implementation to signal that it's not meant
  /// for actual use.
  fn temporary_placeholder() -> Self {
    let (_tx, rx) = mpsc::unbounded::<InspectorSessionProxy>();
    Self {
      v8_inspector: Default::default(),
      session_rx: rx,
      handshake: None,
      established: FuturesUnordered::new(),
      next_local_id: 1,
      local: HashMap::new(),

      target_sessions: HashMap::new(),
      main_session_id: None,
      next_worker_id: 1,
    }
  }

  pub fn dispatch_message_from_frontend(
    &mut self,
    session_id: i32,
    message: String,
  ) {
    let session = self.local.get(&session_id).unwrap();

    // Try custom-domain dispatch before sending to V8. For methods we own
    // (Network.*, DOMStorage.enable/disable), generate the response here
    // and skip V8 entirely.
    if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&message)
      && let Some(method) = parsed.get("method").and_then(|m| m.as_str())
      && let Some(outcome) = dispatch_custom_domain(method, &parsed, session)
    {
      if let Some(id) = parsed.get("id") {
        let call_id = id.as_i64().unwrap_or(0) as i32;
        let content = match outcome {
          CustomDomainOutcome::Empty => make_cdp_ok_response(id, json!({})),
          CustomDomainOutcome::Result(v) => make_cdp_ok_response(id, v),
          CustomDomainOutcome::Error(err) => make_cdp_error_response(id, &err),
        };
        (session.state.send)(InspectorMsg {
          kind: InspectorMsgKind::Message(call_id),
          content,
        });
      }
      return;
    }

    session.dispatch_message(message);
  }

  /// Register a worker session and return the assigned worker ID
  fn register_worker_session(
    &mut self,
    local_session_id: i32,
    worker_url: String,
  ) -> u32 {
    // Assign a sequential worker ID for display purposes
    let worker_id = self.next_worker_id;
    self.next_worker_id += 1;

    // Use the local_session_id for internal session routing
    let target_id = format!("{}", local_session_id);
    let session_id = format!("{}", local_session_id);

    let target_session = Rc::new(TargetSession {
      target_id: target_id.clone(),
      session_id: session_id.clone(),
      local_session_id,
      worker_id,
      main_worker_channels: RefCell::new(None),
      url: worker_url.clone(),
      attached: Cell::new(false),
    });
    self
      .target_sessions
      .insert(session_id.clone(), target_session.clone());

    worker_id
  }

  /// Register the communication channels for a worker's V8 inspector
  /// This is called from the worker side to establish bidirectional communication
  pub fn register_worker_channels(
    &mut self,
    local_session_id: i32,
    main_to_worker_tx: UnboundedSender<String>,
    worker_to_main_rx: UnboundedReceiver<InspectorMsg>,
  ) -> bool {
    // Find the target session for this local session ID
    for target_session in self.target_sessions.values() {
      if target_session.local_session_id == local_session_id {
        *target_session.main_worker_channels.borrow_mut() =
          Some(MainWorkerChannels {
            main_to_worker_tx,
            worker_to_main_rx,
          });
        return true;
      }
    }
    false
  }
}

struct InspectorWakerInner {
  poll_state: PollState,
  task_waker: Option<task::Waker>,
  parked_thread: Option<thread::Thread>,
  inspector_state_ptr: Option<NonNull<JsRuntimeInspectorState>>,
  isolate_handle: v8::IsolateHandle,
}

// SAFETY: unsafe trait must have unsafe implementation
unsafe impl Send for InspectorWakerInner {}

struct InspectorWaker(Mutex<InspectorWakerInner>);

impl InspectorWaker {
  fn new(isolate_handle: v8::IsolateHandle) -> Arc<Self> {
    let inner = InspectorWakerInner {
      poll_state: PollState::Idle,
      task_waker: None,
      parked_thread: None,
      inspector_state_ptr: None,
      isolate_handle,
    };
    Arc::new(Self(Mutex::new(inner)))
  }

  fn update<F, R>(&self, update_fn: F) -> R
  where
    F: FnOnce(&mut InspectorWakerInner) -> R,
  {
    let mut g = self.0.lock();
    update_fn(&mut g)
  }
}

impl task::ArcWake for InspectorWaker {
  fn wake_by_ref(arc_self: &Arc<Self>) {
    arc_self.update(|w| {
      match w.poll_state {
        PollState::Idle => {
          // Wake the task, if any, that has polled the Inspector future last.
          if let Some(waker) = w.task_waker.take() {
            waker.wake()
          }
          // Request an interrupt from the isolate if it's running and there's
          // not unhandled interrupt request in flight.
          if let Some(arg) = w
            .inspector_state_ptr
            .take()
            .map(|ptr| ptr.as_ptr() as *mut c_void)
          {
            w.isolate_handle.request_interrupt(handle_interrupt, arg);
          }
          unsafe extern "C" fn handle_interrupt(
            _isolate: v8::UnsafeRawIsolatePtr,
            arg: *mut c_void,
          ) {
            // SAFETY: `InspectorWaker` is owned by `JsRuntimeInspector`, so the
            // pointer to the latter is valid as long as waker is alive.
            let inspector_state =
              unsafe { &*(arg as *mut JsRuntimeInspectorState) };
            let _ = inspector_state.poll_sessions(None);
          }
        }
        PollState::Parked => {
          // Unpark the isolate thread.
          let parked_thread = w.parked_thread.take().unwrap();
          assert_ne!(parked_thread.id(), thread::current().id());
          parked_thread.unpark();
        }
        _ => {}
      };
      w.poll_state = PollState::Woken;
    });
  }
}

#[derive(Clone, Copy, Debug)]
pub enum InspectorSessionKind {
  Blocking,
  NonBlocking { wait_for_disconnect: bool },
}

#[derive(Clone)]
struct InspectorSessionState {
  is_dispatching_message: Rc<RefCell<bool>>,
  send: Rc<InspectorSessionSend>,
  rx: Rc<RefCell<Option<SessionProxyReceiver>>>,
  // Describes if session should keep event loop alive, eg. a local REPL
  // session should keep event loop alive, but a Websocket session shouldn't.
  kind: InspectorSessionKind,
  sessions: Rc<RefCell<SessionContainer>>,
  // Thread-safe queue for NodeWorker messages that need to be sent to workers
  pending_worker_messages: Arc<Mutex<Vec<(String, String)>>>,
  // Track whether NodeWorker.enable has been called (enables VSCode-style worker debugging)
  nodeworker_enabled: Rc<Cell<bool>>,
  // Track whether Target.setAutoAttach has been called (enables worker auto-attach)
  auto_attach_enabled: Rc<Cell<bool>>,
  // Track whether Target.setDiscoverTargets has been called (enables target discovery)
  discover_targets_enabled: Rc<Cell<bool>>,
  // Track whether Network.enable has been called (per-session, not shared)
  network_enabled: Cell<bool>,
  // Track whether DOMStorage.enable has been called (per-session, not shared)
  dom_storage_enabled: Cell<bool>,
  // Track whether NodeRuntime.enable has been called (per-session, not shared,
  // because one client disabling it should not affect another client's state)
  noderuntime_enabled: Cell<bool>,
  // Track whether NodeRuntime.notifyWhenWaitingForDisconnect has been called
  // with `enabled: true`. When set, the runtime emits
  // NodeRuntime.waitingForDisconnect (instead of
  // Runtime.executionContextDestroyed) to this session at exit, and waits for
  // the session to disconnect before terminating the process.
  notify_waiting_for_disconnect: Cell<bool>,
  // Inspector flags (shared with JsRuntimeInspectorState) for checking waiting_for_session
  flags: Rc<RefCell<InspectorFlags>>,
  // Shared body buffer used by the Network.* command handlers.
  network_data: Rc<RefCell<NetworkDataBuffer>>,
}

/// An inspector session that proxies messages to concrete "transport layer",
/// eg. Websocket or another set of channels.
struct InspectorSession {
  v8_session: v8::inspector::V8InspectorSession,
  state: InspectorSessionState,
}

impl InspectorSession {
  const CONTEXT_GROUP_ID: i32 = 1;

  #[allow(clippy::too_many_arguments, reason = "construction")]
  pub fn new(
    v8_inspector: Rc<v8::inspector::V8Inspector>,
    is_dispatching_message: Rc<RefCell<bool>>,
    send: InspectorSessionSend,
    rx: Option<SessionProxyReceiver>,
    kind: InspectorSessionKind,
    sessions: Rc<RefCell<SessionContainer>>,
    pending_worker_messages: Arc<Mutex<Vec<(String, String)>>>,
    nodeworker_enabled: Rc<Cell<bool>>,
    auto_attach_enabled: Rc<Cell<bool>>,
    discover_targets_enabled: Rc<Cell<bool>>,
    flags: Rc<RefCell<InspectorFlags>>,
    network_data: Rc<RefCell<NetworkDataBuffer>>,
  ) -> Rc<Self> {
    let state = InspectorSessionState {
      is_dispatching_message,
      send: Rc::new(send),
      rx: Rc::new(RefCell::new(rx)),
      kind,
      sessions,
      pending_worker_messages,
      nodeworker_enabled,
      auto_attach_enabled,
      discover_targets_enabled,
      network_enabled: Cell::new(false),
      dom_storage_enabled: Cell::new(false),
      noderuntime_enabled: Cell::new(false),
      notify_waiting_for_disconnect: Cell::new(false),
      flags,
      network_data,
    };

    let v8_session = v8_inspector.connect(
      Self::CONTEXT_GROUP_ID,
      v8::inspector::Channel::new(Box::new(state.clone())),
      v8::inspector::StringView::empty(),
      v8::inspector::V8InspectorClientTrustLevel::FullyTrusted,
    );

    Rc::new(Self { v8_session, state })
  }

  // Dispatch message to V8 session
  fn dispatch_message(&self, msg: String) {
    *self.state.is_dispatching_message.borrow_mut() = true;
    let msg = v8::inspector::StringView::from(msg.as_bytes());
    self.v8_session.dispatch_protocol_message(msg);
    *self.state.is_dispatching_message.borrow_mut() = false;
  }

  /// Queue a message to be sent to a worker
  fn queue_worker_message(&self, session_id: &str, message: String) {
    self
      .state
      .pending_worker_messages
      .lock()
      .push((session_id.to_string(), message));
  }

  /// Notify all registered workers via a callback
  fn notify_workers<F>(&self, mut f: F)
  where
    F: FnMut(&TargetSession, &dyn Fn(InspectorMsg)) + 'static,
  {
    let sessions = self.state.sessions.clone();
    let send = self.state.send.clone();
    deno_core::unsync::spawn(async move {
      let sessions = sessions.borrow();
      for ts in sessions.target_sessions.values() {
        f(ts, &|msg| send(msg));
      }
    });
  }
}

impl InspectorSessionState {
  fn send_message(
    &self,
    msg_kind: InspectorMsgKind,
    msg: v8::UniquePtr<v8::inspector::StringBuffer>,
  ) {
    let msg = msg.unwrap().string().to_string();
    (self.send)(InspectorMsg {
      kind: msg_kind,
      content: msg,
    });
  }
}

impl v8::inspector::ChannelImpl for InspectorSessionState {
  fn send_response(
    &self,
    call_id: i32,
    message: v8::UniquePtr<v8::inspector::StringBuffer>,
  ) {
    self.send_message(InspectorMsgKind::Message(call_id), message);
  }

  fn send_notification(
    &self,
    message: v8::UniquePtr<v8::inspector::StringBuffer>,
  ) {
    self.send_message(InspectorMsgKind::Notification, message);
  }

  fn flush_protocol_notifications(&self) {}
}
type InspectorSessionPumpMessages = Pin<Box<dyn Future<Output = i32>>>;
/// Helper to extract a string param from CDP params
fn get_str_param(params: &Option<serde_json::Value>, key: &str) -> String {
  params
    .as_ref()
    .and_then(|p| p.get(key))
    .and_then(|v| v.as_str())
    .unwrap_or_default()
    .to_owned()
}

/// Helper to extract a bool param from CDP params
fn get_bool_param(params: &Option<serde_json::Value>, key: &str) -> bool {
  params
    .as_ref()
    .and_then(|p| p.get(key))
    .and_then(|v| v.as_bool())
    .unwrap_or(false)
}

impl TargetSession {
  /// Build target info JSON for CDP events
  fn target_info(&self, attached: bool) -> serde_json::Value {
    json!({
      "targetId": self.target_id,
      "type": "node_worker",
      "title": self.title(),
      "url": self.url,
      "attached": attached,
      "canAccessOpener": true
    })
  }

  /// Build worker info JSON for NodeWorker events
  fn worker_info(&self) -> serde_json::Value {
    json!({
      "workerId": self.target_id,
      "type": "node_worker",
      "title": self.title(),
      "url": self.url
    })
  }
}

async fn pump_inspector_session_messages(
  session: Rc<InspectorSession>,
  session_id: i32,
) -> i32 {
  let mut rx = session.state.rx.borrow_mut().take().unwrap();

  while let Some(msg) = rx.next().await {
    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&msg) else {
      session.dispatch_message(msg);
      continue;
    };

    // CDP Flattened Session Mode: route messages with top-level sessionId to workers
    if let Some(session_id) = parsed.get("sessionId").and_then(|s| s.as_str()) {
      let mut worker_msg = parsed.clone();
      if let Some(obj) = worker_msg.as_object_mut() {
        obj.remove("sessionId");
        session.queue_worker_message(session_id, worker_msg.to_string());
      }
      continue;
    }

    let Some(method) = parsed.get("method").and_then(|m| m.as_str()) else {
      session.dispatch_message(msg);
      continue;
    };

    let params = parsed.get("params").cloned();
    let msg_id = parsed.get("id").cloned();

    if let Some(outcome) = dispatch_custom_domain(method, &parsed, &session) {
      if let Some(id) = msg_id {
        let call_id = id.as_i64().unwrap_or(0) as i32;
        let content = match outcome {
          CustomDomainOutcome::Empty => make_cdp_ok_response(&id, json!({})),
          CustomDomainOutcome::Result(v) => make_cdp_ok_response(&id, v),
          CustomDomainOutcome::Error(err) => make_cdp_error_response(&id, &err),
        };
        (session.state.send)(InspectorMsg {
          kind: InspectorMsgKind::Message(call_id),
          content,
        });
      }
      continue;
    }

    match method {
      "NodeRuntime.enable" => {
        session.state.noderuntime_enabled.set(true);
        // If the runtime is paused on start (--inspect-brk), emit
        // NodeRuntime.waitingForDebugger. We check paused_on_start
        // (not waiting_for_session) because the session has already
        // connected by the time this message is received.
        let flags = session.state.flags.borrow();
        if flags.waiting_for_session || flags.paused_on_start {
          drop(flags);
          (session.state.send)(InspectorMsg::notification(json!({
            "method": "NodeRuntime.waitingForDebugger"
          })));
        }
      }
      "NodeRuntime.disable" => {
        session.state.noderuntime_enabled.set(false);
      }
      "NodeRuntime.notifyWhenWaitingForDisconnect" => {
        let enabled = get_bool_param(&params, "enabled");
        session.state.notify_waiting_for_disconnect.set(enabled);
      }
      "Runtime.runIfWaitingForDebugger" => {
        // Clear paused_on_start so wait_for_session_and_break_on_next_statement
        // unblocks. Also clear waiting_for_session so poll_sessions stops
        // blocking. Track which session resumed us so we schedule the pause
        // on the correct session (the one that has Debugger.enable active).
        {
          let mut flags = session.state.flags.borrow_mut();
          flags.paused_on_start = false;
          flags.waiting_for_session = false;
          flags.resumed_by_session_id = Some(session_id);
        }
        // Forward to V8 for actual processing
        session.dispatch_message(msg);
        continue;
      }
      "NodeWorker.enable" => {
        session.state.nodeworker_enabled.set(true);
        session.notify_workers(|ts, send| {
          send(InspectorMsg::notification(json!({
            "method": "NodeWorker.attachedToWorker",
            "params": {
              "sessionId": ts.session_id,
              "workerInfo": ts.worker_info(),
              "waitingForDebugger": false
            }
          })));
        });
      }
      "NodeWorker.sendMessageToWorker" | "Target.sendMessageToTarget" => {
        session.queue_worker_message(
          &get_str_param(&params, "sessionId"),
          get_str_param(&params, "message"),
        );
      }
      "Target.setDiscoverTargets" => {
        let discover = get_bool_param(&params, "discover");
        session.state.discover_targets_enabled.set(discover);

        if discover {
          session.notify_workers(|ts, send| {
            send(InspectorMsg::notification(json!({
              "method": "Target.targetCreated",
              "params": { "targetInfo": ts.target_info(false) }
            })));
          });
        }
      }
      "Target.setAutoAttach" => {
        let auto_attach = get_bool_param(&params, "autoAttach");
        let send = session.state.send.clone();
        let sessions = session.state.sessions.clone();
        session.state.auto_attach_enabled.set(auto_attach);
        if auto_attach {
          deno_core::unsync::spawn(async move {
            let sessions = sessions.borrow();
            for ts in sessions.target_sessions.values() {
              if ts.attached.replace(true) {
                continue; // Skip if already attached
              }
              send(InspectorMsg::notification(json!({
                "method": "Target.attachedToTarget",
                "params": {
                  "sessionId": ts.session_id,
                  "targetInfo": ts.target_info(true),
                  "waitingForDebugger": false
                }
              })));
            }
          });
        }
      }
      _ => {
        session.dispatch_message(msg);
        continue;
      }
    }

    // Send default `{result: {}}` ack for commands handled above
    // that don't produce their own response payload.
    if let Some(id) = msg_id {
      let call_id = id.as_i64().unwrap_or(0) as i32;
      (session.state.send)(InspectorMsg {
        kind: InspectorMsgKind::Message(call_id),
        content: json!({
          "id": id,
          "result": {}
        })
        .to_string(),
      });
    }
  }
  session_id
}

/// A local inspector session that can be used to send and receive protocol messages directly on
/// the same thread as an isolate.
///
/// Does not provide any abstraction over CDP messages.
pub struct LocalInspectorSession {
  sessions: Rc<RefCell<SessionContainer>>,
  session_id: i32,
}

impl LocalInspectorSession {
  pub fn new(session_id: i32, sessions: Rc<RefCell<SessionContainer>>) -> Self {
    Self {
      sessions,
      session_id,
    }
  }

  pub fn dispatch(&mut self, msg: String) {
    self
      .sessions
      .borrow_mut()
      .dispatch_message_from_frontend(self.session_id, msg);
  }

  pub fn post_message<T: serde::Serialize>(
    &mut self,
    id: i32,
    method: &str,
    params: Option<T>,
  ) {
    let message = json!({
        "id": id,
        "method": method,
        "params": params,
    });

    let stringified_msg = serde_json::to_string(&message).unwrap();
    self.dispatch(stringified_msg);
  }
}