Repository: denoland/deno
Pull request: feat(core): implement NodeRuntime CDP domain and fix --inspect-brk blocking

## Summary

Implements the `NodeRuntime` CDP (Chrome DevTools Protocol) domain for the inspector, and fixes several long-standing inspector issues.

### NodeRuntime CDP Domain
- `NodeRuntime.enable` — emits `NodeRuntime.waitingForDebugger` notification when the runtime is paused on start
- `NodeRuntime.disable` — disables the domain
- `NodeWorker.enable` — emits `NodeWorker.attachedToWorker` for worker sessions

### --inspect-brk Blocking Fix
Previously, `--inspect-brk` only blocked until a WebSocket session connected, then called `schedule_pause_on_next_statement` which was ineffective without `Debugger.enable`. Now:
- Blocks the main thread until `Runtime.runIfWaitingForDebugger` is received, matching Node.js behavior
- Schedules a V8 debugger break after the wait loop exits (when `Debugger.enable` is already active)
- Properly handles the `paused_on_start` flag in both the pump message handler and V8's `run_if_waiting_for_debugger` callback

### Inspector Session Cleanup
- Removes sessions from the `local` map when the pump future completes (WS disconnects), so `sessions_state()` correctly reports no active sessions
- Fixes `pump_websocket_messages` to handle `read_frame()` errors (remote disconnect without close frame), so child processes exit cleanly instead of hanging at "Waiting for inspector to disconnect"

### Tests
- Enables `parallel/test-inspector-console.js` Node.js compat test

## Test plan
- [x] `./x test-compat test-inspector-console.js` passes
- [x] `./x test-unit inspector` passes (all inspector unit tests)
- [x] CI green


🤖 Generated with [Claude Code](https://claude.com/claude-code)

Changed files:
filename: libs/core/inspector.rs
patch: @@ -194,7 +194,9 @@ impl v8::inspector::V8InspectorClientImpl for JsRuntimeInspectorClient {
 
   fn run_if_waiting_for_debugger(&self, context_group_id: i32) {
     assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID);
-    self.0.flags.borrow_mut().waiting_for_session = false;
+    let mut flags = self.0.flags.borrow_mut();
+    flags.waiting_for_session = false;
+    flags.paused_on_start = false;
   }
 
   fn ensure_default_context_in_group(
@@ -256,23 +258,25 @@ impl JsRuntimeInspectorState {
       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()).boxed_local();
+            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.
+          // 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);
-          }
-          let id = sessions.next_local_id;
-          sessions.next_local_id += 1;
-          sessions.local.insert(id, session);
+            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);
+            // 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;
@@ -356,6 +360,7 @@ impl JsRuntimeInspectorState {
                 self.nodeworker_enabled.clone(),
                 self.auto_attach_enabled.clone(),
                 self.discover_targets_enabled.clone(),
+                self.flags.clone(),
               );
 
               let prev = sessions.handshake.replace(session);
@@ -466,7 +471,20 @@ impl JsRuntimeInspectorState {
 
         // Poll established sessions.
         match sessions.established.poll_next_unpin(cx) {
-          Poll::Ready(Some(())) => {
+          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) => {
@@ -691,16 +709,46 @@ impl JsRuntimeInspector {
   /// 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 let Some(session) =
-        self.state.sessions.borrow_mut().local.values().next()
-      {
-        break session.break_on_next_statement();
-      } else {
-        self.state.flags.borrow_mut().waiting_for_session = true;
-        let _ = self.state.poll_sessions(None).unwrap();
+      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.
@@ -742,6 +790,7 @@ impl JsRuntimeInspector {
         inspector.state.nodeworker_enabled.clone(),
         inspector.state.auto_attach_enabled.clone(),
         inspector.state.discover_targets_enabled.clone(),
+        inspector.state.flags.clone(),
       );
 
       let session_id = {
@@ -764,6 +813,13 @@ impl JsRuntimeInspector {
 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)]
@@ -1074,6 +1130,11 @@ struct InspectorSessionState {
   auto_attach_enabled: Rc<Cell<bool>>,
   // Track whether Target.setDiscoverTargets has been called (enables target discovery)
   discover_targets_enabled: Rc<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>,
+  // Inspector flags (shared with JsRuntimeInspectorState) for checking waiting_for_session
+  flags: Rc<RefCell<InspectorFlags>>,
 }
 
 /// An inspector session that proxies messages to concrete "transport layer",
@@ -1098,6 +1159,7 @@ impl InspectorSession {
     nodeworker_enabled: Rc<Cell<bool>>,
     auto_attach_enabled: Rc<Cell<bool>>,
     discover_targets_enabled: Rc<Cell<bool>>,
+    flags: Rc<RefCell<InspectorFlags>>,
   ) -> Rc<Self> {
     let state = InspectorSessionState {
       is_dispatching_message,
@@ -1109,6 +1171,8 @@ impl InspectorSession {
       nodeworker_enabled,
       auto_attach_enabled,
       discover_targets_enabled,
+      noderuntime_enabled: Cell::new(false),
+      flags,
     };
 
     let v8_session = v8_inspector.connect(
@@ -1129,14 +1193,6 @@ impl InspectorSession {
     *self.state.is_dispatching_message.borrow_mut() = false;
   }
 
-  pub fn break_on_next_statement(&self) {
-    let reason = v8::inspector::StringView::from(&b"debugCommand"[..]);
-    let detail = v8::inspector::StringView::empty();
-    self
-      .v8_session
-      .schedule_pause_on_next_statement(reason, detail);
-  }
-
   /// Queue a message to be sent to a worker
   fn queue_worker_message(&self, session_id: &str, message: String) {
     self
@@ -1194,7 +1250,7 @@ impl v8::inspector::ChannelImpl for InspectorSessionState {
 
   fn flush_protocol_notifications(&self) {}
 }
-type InspectorSessionPumpMessages = Pin<Box<dyn Future<Output = ()>>>;
+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
@@ -1238,7 +1294,10 @@ impl TargetSession {
   }
 }
 
-async fn pump_inspector_session_messages(session: Rc<InspectorSession>) {
+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 {
@@ -1266,6 +1325,38 @@ async fn pump_inspector_session_messages(session: Rc<InspectorSession>) {
     let msg_id = parsed.get("id").cloned();
 
     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);
+      }
+      "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| {
@@ -1341,6 +1432,7 @@ async fn pump_inspector_session_messages(session: Rc<InspectorSession>) {
       });
     }
   }
+  session_id
 }
 
 /// A local inspector session that can be used to send and receive protocol messages directly on

filename: libs/inspector_server/lib.rs
patch: @@ -671,22 +671,29 @@ async fn pump_websocket_messages(
             let msg = Frame::text(msg.content.into_bytes().into());
             let _ = websocket.write_frame(msg).await;
         }
-        Ok(msg) = websocket.read_frame() => {
-            match msg.opcode {
-                OpCode::Text => {
-                    if let Ok(s) = String::from_utf8(msg.payload.to_vec()) {
-                      let _ = inbound_tx.unbounded_send(s);
+        result = websocket.read_frame() => {
+            match result {
+                Ok(msg) => match msg.opcode {
+                    OpCode::Text => {
+                        if let Ok(s) = String::from_utf8(msg.payload.to_vec()) {
+                          let _ = inbound_tx.unbounded_send(s);
+                        }
                     }
-                }
-                OpCode::Close => {
-                    // Users don't care if there was an error coming from debugger,
-                    // just about the fact that debugger did disconnect.
-                    log::info!("Debugger session ended");
+                    OpCode::Close => {
+                        // Users don't care if there was an error coming from debugger,
+                        // just about the fact that debugger did disconnect.
+                        log::info!("Debugger session ended");
+                        break 'pump;
+                    }
+                    _ => {
+                        // Ignore other messages.
+                    }
+                },
+                Err(_) => {
+                    // WebSocket read error (remote disconnected without close frame).
+                    log::info!("Debugger session ended (connection lost)");
                     break 'pump;
                 }
-                _ => {
-                    // Ignore other messages.
-                }
             }
         }
         else => {

filename: tests/unit/inspector_test.ts
patch: @@ -896,6 +896,51 @@ Deno.test("inspector_node_worker_enable", async () => {
   }
 });
 
+Deno.test("inspector_noderuntime_waiting_for_debugger", async () => {
+  // Verifies that NodeRuntime.enable emits NodeRuntime.waitingForDebugger
+  // when --inspect-brk is used, and that Runtime.runIfWaitingForDebugger
+  // unblocks execution and triggers Debugger.paused.
+  const script = `${testdataPath}/inspector2.js`;
+  const tester = await InspectorTester.create(
+    ["run", "-A", "--inspect-brk=0", script],
+    { notificationFilter: ignoreScriptParsed },
+  );
+
+  try {
+    await tester.assertStderrForInspectBrk();
+
+    // Enable NodeRuntime first - should emit waitingForDebugger
+    tester.send({ id: 1, method: "NodeRuntime.enable" });
+    await tester.expectNotification("NodeRuntime.waitingForDebugger");
+
+    // Now enable Runtime and Debugger
+    tester.sendMany([
+      { id: 2, method: "Runtime.enable" },
+      { id: 3, method: "Debugger.enable" },
+    ]);
+
+    await tester.expectResponse(2);
+    await tester.expectResponse(3);
+
+    // Resume - should unblock and pause at first statement
+    tester.send({ id: 4, method: "Runtime.runIfWaitingForDebugger" });
+    await tester.expectResponse(4);
+    await tester.expectNotification("Debugger.paused");
+
+    // Resume execution
+    tester.send({ id: 5, method: "Debugger.resume" });
+    await tester.expectResponse(5);
+
+    // Script should run to completion
+    const scriptOutput = await tester.nextStdoutLine();
+    assertEquals(scriptOutput, "hello from the script");
+  } finally {
+    await tester.close();
+    tester.kill();
+    await tester.waitForExit();
+  }
+});
+
 Deno.test("inspector_runtime_evaluate_does_not_crash", async () => {
   const tester = await InspectorTester.create(
     ["repl", "-A", "--inspect=0"],