Add Watch Pane support

crates/ark/src/console.rs

@@ -17,8 +17,6 @@ use std::collections::HashMap;
 use std::ffi::*;
 use std::os::raw::c_uchar;
 use std::result::Result::Ok;
-use std::sync::atomic::AtomicBool;
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::task::Poll;
@@ -138,7 +136,6 @@ use crate::ui::UiCommMessage;
 use crate::ui::UiCommSender;
 use crate::url::ExtUrl;
 
-pub static CAPTURE_CONSOLE_OUTPUT: AtomicBool = AtomicBool::new(false);
 static RE_DEBUG_PROMPT: Lazy<Regex> = Lazy::new(|| Regex::new(r"Browse\[\d+\]").unwrap());
 
 /// All debug commands as documented in `?browser`
@@ -247,6 +244,7 @@ pub struct Console {
     /// Channel to send and receive tasks from `RTask`s
     tasks_interrupt_rx: Receiver<RTask>,
     tasks_idle_rx: Receiver<RTask>,
+    tasks_idle_any_rx: Receiver<RTask>,
     pending_futures: HashMap<Uuid, (BoxFuture<'static, ()>, RTaskStartInfo)>,
 
     /// Channel to communicate requests and events to the frontend
@@ -281,9 +279,9 @@ pub struct Console {
     /// Stored in `Console` to avoid memory leakage when `Rf_error()` jumps.
     r_error_buffer: Option<CString>,
 
-    /// `WriteConsole` output diverted from IOPub is stored here. This is only used
-    /// to return R output to the debugger.
-    pub(crate) captured_output: String,
+    /// When `Some`, console output is captured here instead of being sent to IOPub.
+    /// Interact with this via `ConsoleOutputCapture` from `start_capture()`.
+    pub(crate) captured_output: Option<String>,
 
     /// Whether we should preserve focus when stopping in a debug session. We
     /// should only preserve focus if we're explicitly stepping through code as
@@ -311,8 +309,9 @@ pub struct Console {
     /// valid for a single session.
     pub(crate) debug_session_index: u32,
 
-    /// The current frame `id`. Unique across all frames within a single debug session.
-    /// Reset after `debug_stop()`, not between debug steps.
+    /// The current frame `id`. Monotonically increasing, unique across all
+    /// frames and debug sessions. It's important that each frame gets a unique
+    /// ID across the process lifetime so that we can invalidate stale requests.
     pub(crate) debug_current_frame_id: i64,
 
     /// Reason for entering the debugger. Used to determine which DAP event to send.
@@ -541,6 +540,64 @@ pub(crate) enum ConsoleResult {
     Error(String),
 }
 
+/// Guard for capturing console output during idle tasks.
+///
+/// Created via `Console::start_capture()`, which sets `Console::captured_output`
+/// to `Some` so that all `write_console` output goes there instead of IOPub.
+/// When dropped, restores the previous state and logs any remaining output.
+///
+/// Use `take()` to retrieve captured output. Can be called multiple times to
+/// get output accumulated since the last take.
+pub struct ConsoleOutputCapture {
+    previous_output: Option<String>,
+    connected: bool,
+}
+
+impl ConsoleOutputCapture {
+    /// Create a dummy capture that doesn't interact with Console.
+    /// Used in test contexts where Console is not initialized.
+    pub(crate) fn dummy() -> Self {
+        Self {
+            previous_output: None,
+            connected: false,
+        }
+    }
+
+    /// Take the captured output so far, clearing the buffer.
+    /// Can be called multiple times; each call returns output accumulated since the last take.
+    pub fn take(&mut self) -> String {
+        if !self.connected {
+            return String::new();
+        }
+
+        if let Some(captured) = Console::get_mut().captured_output.as_mut() {
+            return std::mem::take(captured);
+        }
+
+        String::new()
+    }
+}
+
+impl Drop for ConsoleOutputCapture {
+    fn drop(&mut self) {
+        if !self.connected {
+            return;
+        }
+
+        let console = Console::get_mut();
+
+        // Log any remaining output that wasn't taken
+        if let Some(output) = console.captured_output.take() {
+            if !output.trim().is_empty() {
+                log::info!("[Captured idle output]\n{}", output.trim_end());
+            }
+        }
+
+        // Restore previous capture state
+        console.captured_output = self.previous_output.take();
+    }
+}
+
 impl Console {
     /// Sets up the main R thread, initializes the `CONSOLE` singleton,
     /// and starts R. Does not return!
@@ -571,11 +628,12 @@ impl Console {
             };
         }
 
-        let (tasks_interrupt_rx, tasks_idle_rx) = r_task::take_receivers();
+        let (tasks_interrupt_rx, tasks_idle_rx, tasks_idle_any_rx) = r_task::take_receivers();
 
         CONSOLE.set(UnsafeCell::new(Console::new(
             tasks_interrupt_rx,
             tasks_idle_rx,
+            tasks_idle_any_rx,
             comm_event_tx,
             r_request_rx,
             stdin_request_tx,
@@ -810,6 +868,7 @@ impl Console {
     pub fn new(
         tasks_interrupt_rx: Receiver<RTask>,
         tasks_idle_rx: Receiver<RTask>,
+        tasks_idle_any_rx: Receiver<RTask>,
         comm_event_tx: Sender<CommEvent>,
         r_request_rx: Receiver<RRequest>,
         stdin_request_tx: Sender<StdInRequest>,
@@ -840,12 +899,13 @@ impl Console {
             debug_stopped_reason: None,
             tasks_interrupt_rx,
             tasks_idle_rx,
+            tasks_idle_any_rx,
             pending_futures: HashMap::new(),
             session_mode,
             positron_ns: None,
             banner: None,
             r_error_buffer: None,
-            captured_output: String::new(),
+            captured_output: None,
             debug_call_text: DebugCallText::None,
             debug_last_line: None,
             debug_preserve_focus: false,
@@ -919,6 +979,17 @@ impl Console {
         &self.iopub_tx
     }
 
+    /// Start capturing console output.
+    /// Returns a guard that saves and restores the previous capture state on drop.
+    pub(crate) fn start_capture(&mut self) -> ConsoleOutputCapture {
+        let previous_output = self.captured_output.replace(String::new());
+
+        ConsoleOutputCapture {
+            previous_output,
+            connected: true,
+        }
+    }
+
     /// Get the current execution context if an active request exists.
     /// Returns (execution_id, code) tuple where execution_id is the Jupyter message ID.
     pub fn get_execution_context(&self) -> Option<(String, String)> {
@@ -1117,6 +1188,7 @@ impl Console {
         let kernel_request_rx = self.kernel_request_rx.clone();
         let tasks_interrupt_rx = self.tasks_interrupt_rx.clone();
         let tasks_idle_rx = self.tasks_idle_rx.clone();
+        let tasks_idle_any_rx = self.tasks_idle_any_rx.clone();
 
         // Process R's polled events regularly while waiting for console input.
         // We used to poll every 200ms but that lead to visible delays for the
@@ -1138,16 +1210,20 @@ impl Console {
 
         // Only process idle at top level. We currently don't want idle tasks
         // (e.g. for srcref generation) to run when the call stack is not empty.
-        // We could make this configurable though if needed, i.e. some idle
-        // tasks would be able to run in the browser. Those should be sent to a
-        // dedicated channel that would always be included in the set of recv
-        // channels.
         let tasks_idle_index = if matches!(info.kind, PromptKind::TopLevel) {
             Some(select.recv(&tasks_idle_rx))
         } else {
             None
         };
 
+        // "Idle any" tasks run at both top-level and browser prompts
+        let tasks_idle_any_index =
+            if matches!(info.kind, PromptKind::TopLevel | PromptKind::Browser) {
+                Some(select.recv(&tasks_idle_any_rx))
+            } else {
+                None
+            };
+
         loop {
             // If an interrupt was signaled and we are in a user
             // request prompt, e.g. `readline()`, we need to propagate
@@ -1216,6 +1292,12 @@ impl Console {
                     self.handle_task(task);
                 },
 
+                // An "idle any" task woke us up
+                i if Some(i) == tasks_idle_any_index => {
+                    let task = oper.recv(&tasks_idle_any_rx).unwrap();
+                    self.handle_task(task);
+                },
+
                 // It's time to run R's polled events
                 i if i == polled_events_index => {
                     let _ = oper.recv(&polled_events_rx).unwrap();
@@ -2237,10 +2319,10 @@ impl Console {
 
     /// Invoked by R to write output to the console.
     fn write_console(buf: *const c_char, _buflen: i32, otype: i32) {
-        if CAPTURE_CONSOLE_OUTPUT.load(Ordering::SeqCst) {
-            Console::get_mut()
-                .captured_output
-                .push_str(&console_to_utf8(buf).unwrap());
+        let console = Console::get_mut();
+
+        if let Some(captured) = &mut console.captured_output {
+            captured.push_str(&console_to_utf8(buf).unwrap());
             return;
         }
 
@@ -2249,8 +2331,6 @@ impl Console {
             Err(err) => panic!("Failed to read from R buffer: {err:?}"),
         };
 
-        let console = Console::get_mut();
-
         if !Console::is_initialized() {
             // During init, consider all output to be part of the startup banner
             match console.banner.as_mut() {
@@ -2929,7 +3009,7 @@ unsafe extern "C-unwind" fn ps_onload_hook(pkg: SEXP, _path: SEXP) -> anyhow::Re
 
     // Populate fake source refs if needed
     if do_resource_namespaces() {
-        r_task::spawn_idle(|| async move {
+        r_task::spawn_idle(|_| async move {
             if let Err(err) = ns_populate_srcref(pkg.clone()).await {
                 log::error!("Can't populate srcref for `{pkg}`: {err:?}");
             }

crates/ark/src/console_debug.rs

@@ -125,7 +125,6 @@ impl Console {
         self.debug_stopped_reason = None;
         self.debug_last_stack = vec![];
         self.clear_fallback_sources();
-        self.debug_reset_frame_id();
         self.debug_session_index += 1;
 
         let mut dap = self.debug_dap.lock().unwrap();
@@ -295,10 +294,6 @@ impl Console {
         out
     }
 
-    pub(crate) fn debug_reset_frame_id(&mut self) {
-        self.debug_current_frame_id = 0;
-    }
-
     pub(crate) fn ark_debug_uri(
         debug_session_index: u32,
         source_name: &str,