Run Evaluate requests with timeout

crates/ark/src/console/console_repl.rs

@@ -2509,6 +2509,9 @@ impl Console {
             return;
         }
 
+        // Interrupt any in-flight evaluations that have outlived their timeout
+        crate::timeout::check_timeout();
+
         // Skip running tasks if we don't have 128KB of stack space available.
         // This is 1/8th of the typical Windows stack space (1MB, whereas macOS
         // and Linux have 8MB).

crates/ark/src/dap/dap_jupyter_handler.rs

@@ -32,6 +32,7 @@ use stdext::result::ResultExt;
 use crate::dap::dap_notebook;
 use crate::dap::dap_server::DapConsoleEvent;
 use crate::dap::dap_server::DapHandler;
+use crate::dap::dap_server::EvaluateOutcome;
 use crate::dap::dap_state::Breakpoint;
 use crate::dap::dap_state::BreakpointState;
 use crate::dap::dap_state::Dap;
@@ -138,24 +139,37 @@ impl DapJupyterHandler {
         let expression = args.expression;
         let frame_id = args.frame_id;
 
-        // This only returns `Some()` if R is idle. Even though we stopped at
-        // some point, causing the Evaluate request, R could be busy evaluating
-        // the next expression already.
-        let result = r_task::try_idle_task(move |capture| {
+        // This runs only if R is idle. Even though we stopped at some point,
+        // causing the Evaluate request, R could be busy evaluating the next
+        // expression already, in which case `try_idle_task()` returns `None`.
+        // `DapHandler::evaluate()` runs the eval under a timeout so a runaway
+        // expression is interrupted and doesn't freeze the kernel.
+        let outcome = r_task::try_idle_task(move |capture| {
             DapHandler::evaluate(&state, &expression, frame_id, capture)
         });
 
-        match result {
-            None => Ok(self.error_response(seq, "evaluate", "R is busy")),
-            // The task raised an R error, caught by the sandbox on the R thread.
-            Some(Err(err)) => Ok(self.error_response(seq, "evaluate", &format!("{err}"))),
-            Some(Ok(Ok(body))) => {
-                let ResponseBody::Evaluate(eval) = body else {
-                    return Err(anyhow::anyhow!("Unexpected response body from evaluate"));
-                };
+        let Some(outcome) = outcome else {
+            return Ok(self.error_response(seq, "evaluate", "R is busy"));
+        };
+
+        let outcome = match outcome {
+            Ok(outcome) => outcome,
+            // A longjump that escaped `DapHandler::evaluate()`'s own `try_catch`
+            // (e.g. while building the response), caught by the try-idle sandbox.
+            Err(err) => return Ok(self.error_response(seq, "evaluate", &err.to_string())),
+        };
+
+        match outcome {
+            EvaluateOutcome::Ok(ResponseBody::Evaluate(eval)) => {
                 Ok(self.success_response(seq, "evaluate", serde_json::to_value(eval)?))
             },
-            Some(Ok(Err(err))) => Ok(self.error_response(seq, "evaluate", &format!("{err}"))),
+            EvaluateOutcome::Ok(_) => {
+                Err(anyhow::anyhow!("Unexpected response body from evaluate"))
+            },
+            EvaluateOutcome::TimedOut => {
+                Ok(self.error_response(seq, "evaluate", "Evaluation timed out"))
+            },
+            EvaluateOutcome::Error(err) => Ok(self.error_response(seq, "evaluate", &err)),
         }
     }
 

crates/ark/src/dap/dap_server.rs

@@ -53,6 +53,8 @@ use crate::r_task::RTask;
 use crate::request::debug_request_command;
 use crate::request::DebugRequest;
 use crate::request::RRequest;
+use crate::timeout::with_timeout;
+use crate::timeout::EVAL_TIMEOUT;
 
 /// Sentinel expression sent by the frontend to notify the kernel that the user
 /// selected a different stack frame in the debugger UI. Subsequent console
@@ -115,6 +117,14 @@ pub struct DapHandler {
     pub(crate) r_request_tx: Sender<RRequest>,
 }
 
+/// Outcome of a debugger `evaluate` request, produced on the R thread by
+/// [`DapHandler::evaluate()`] and turned into a DAP response by each transport.
+pub(crate) enum EvaluateOutcome {
+    Ok(ResponseBody),
+    TimedOut,
+    Error(String),
+}
+
 impl DapHandler {
     pub fn new(state: Arc<Mutex<Dap>>, r_request_tx: Sender<RRequest>) -> Self {
         Self {
@@ -125,10 +135,11 @@ impl DapHandler {
 
     /// Dispatch a parsed DAP request to the appropriate handler.
     ///
-    /// `Evaluate` is not handled here because `dispatch()` returns
-    /// synchronously, while Evaluate requires an async round-trip through the
-    /// R thread. Each transport calls [`DapHandler::evaluate()`] directly
-    /// inside its own async mechanism.
+    /// `Evaluate` is not handled here because it runs user code, which must
+    /// happen on the R thread. Each transport calls [`DapHandler::evaluate()`]
+    /// through its own mechanism (an idle task on the console, a try-idle one
+    /// on the notebook). The other commands run directly here without touching
+    /// the R thread.
     pub fn dispatch(&self, req: Request) -> DapOutput {
         let cmd = req.command.clone();
 
@@ -680,11 +691,11 @@ impl DapHandler {
         expression: &str,
         frame_id: Option<i64>,
         capture: &mut ConsoleOutputCapture,
-    ) -> anyhow::Result<ResponseBody> {
+    ) -> EvaluateOutcome {
         if expression == SELECTED_FRAME_EXPRESSION {
             log::trace!("DAP: Received frame selection sentinel, frame_id: {frame_id:?}");
             Console::get_mut().set_debug_selected_frame_id(frame_id);
-            return Ok(ResponseBody::Evaluate(EvaluateResponse {
+            return EvaluateOutcome::Ok(ResponseBody::Evaluate(EvaluateResponse {
                 result: String::new(),
                 type_field: None,
                 presentation_hint: None,
@@ -706,18 +717,31 @@ impl DapHandler {
         // would leave the mutex locked forever. The env stays valid without the
         // lock because its owning `RObject` remains in the state's map, which
         // only the R thread mutates.
-        let env = state
-            .lock()
-            .unwrap()
-            .frame_env(frame_id)
-            .map_err(|err| anyhow::anyhow!("{err}"))?;
+        let env = match state.lock().unwrap().frame_env(frame_id) {
+            Ok(env) => env,
+            Err(err) => return EvaluateOutcome::Error(err.to_string()),
+        };
 
         let capture = if print { Some(capture) } else { None };
-        let variable = Dap::evaluate(expr, env, capture)?;
+
+        // Run under `with_timeout()` so a long-running or inflooping Watch Pane
+        // expression is interrupted instead of freezing the kernel.
+        let (res, timed_out) = with_timeout(EVAL_TIMEOUT, || Dap::evaluate(expr, env, capture));
+
+        if timed_out {
+            return EvaluateOutcome::TimedOut;
+        }
+
+        let variable = match res {
+            Ok(Ok(variable)) => variable,
+            Ok(Err(err)) => return EvaluateOutcome::Error(format!("{err}")),
+            // An R longjump that escaped `Dap::evaluate()`'s own `try_catch`.
+            Err(err) => return EvaluateOutcome::Error(err.to_string()),
+        };
         log::trace!("DAP: Evaluate completed");
 
         let response = state.lock().unwrap().into_evaluate_response(variable);
-        Ok(ResponseBody::Evaluate(response))
+        EvaluateOutcome::Ok(ResponseBody::Evaluate(response))
     }
 }
 
@@ -918,8 +942,9 @@ impl<R: Read, W: Write> DapServer<R, W> {
         };
         log::trace!("DAP: Got request: {:#?}", req);
 
-        // Evaluate is async: the response is sent later via `responses_tx`.
-        // It is the only command that needs transport-specific handling.
+        // Evaluate runs user code on the R thread, so it's handled off on an
+        // idle task to keep `serve()` responsive to other requests. Its
+        // response is delivered asynchronously via `responses_tx`.
         if let Command::Evaluate(args) = &req.command {
             let args = args.clone();
             if let Err(err) = self.handle_evaluate(req, args) {
@@ -995,8 +1020,12 @@ impl<R: Read, W: Write> DapServer<R, W> {
         self.server.send_event(event)
     }
 
-    // Tied to the TCP transport for now. For Jupyter we need to figure out how
-    // to do async responses with Jupyter's Control channel.
+    /// Evaluate an expression on the R thread, with a timeout.
+    ///
+    /// Spawns an idle task to handle the evaluation request asynchronously
+    /// without blocking the DAP service while the evaluation is in flight on
+    /// the R thread. The eval runs under `timeout::with_timeout()`, which
+    /// guards against a runaway expression (e.g. an infloop in the watch pane).
     fn handle_evaluate(
         &mut self,
         req: Request,
@@ -1007,13 +1036,11 @@ impl<R: Read, W: Write> DapServer<R, W> {
         let state = self.handler.state.clone();
         let responses_tx = self.responses_tx.clone();
 
-        log::trace!("DAP: Spawning idle task for evaluate");
         r_task::spawn(RTask::send_idle_any_prompt(async move |mut capture| {
-            log::trace!("DAP: Idle task started for evaluate");
-
             let rsp = match DapHandler::evaluate(&state, &expression, frame_id, &mut capture) {
-                Ok(body) => req.success(body),
-                Err(err) => req.error(&format!("Error: {err}")),
+                EvaluateOutcome::Ok(body) => req.success(body),
+                EvaluateOutcome::TimedOut => req.error("Evaluation timed out"),
+                EvaluateOutcome::Error(err) => req.error(&format!("Error: {err}")),
             };
 
             responses_tx.send(rsp).log_err();

crates/ark/src/lib.rs

@@ -44,6 +44,7 @@ pub mod startup;
 pub mod strings;
 pub mod sys;
 pub mod thread;
+pub mod timeout;
 pub mod traps;
 pub mod treesitter;
 pub mod ui;

crates/ark/src/r_task.rs

@@ -20,7 +20,6 @@ use crossbeam::channel::Sender;
 use harp::exec::r_sandbox;
 #[cfg(debug_assertions)]
 use libr::SEXP;
-use stdext::result::ResultExt;
 use uuid::Uuid;
 
 use crate::console::Console;
@@ -113,12 +112,9 @@ const TRY_IDLE_TIMEOUT: Duration = Duration::from_millis(200);
 
 /// Attempt to run `f` on the R thread if it is currently idle at a prompt.
 ///
-/// Returns `None` if R was busy. Otherwise returns `Some` with the outcome:
+/// Returns `None` if R was busy, i.e. the Console event loop did not accept the
+/// task within `TRY_IDLE_TIMEOUT`. Otherwise returns `Some` with the outcome:
 /// `Ok` if `f` ran cleanly, `Err` if it raised an R error.
-///
-/// On the bounded(0) channel, `send_timeout` only hands off the task when the
-/// event loop's `Select` is parked receiving, so the task runs iff R is idle.
-/// Waiting up to `TRY_IDLE_TIMEOUT`.
 pub(crate) fn try_idle_task<F, T>(f: F) -> Option<harp::Result<T>>
 where
     F: FnOnce(&mut ConsoleOutputCapture) -> T + Send + 'static,
@@ -152,7 +148,10 @@ where
         },
     }
 
-    done_rx.recv().log_err()?;
+    if done_rx.recv().is_err() {
+        return None;
+    }
+
     let out = result.lock().unwrap().take();
     out
 }

crates/ark/src/timeout.rs

@@ -0,0 +1,132 @@
+//
+// timeout.rs
+//
+// Copyright (C) 2026 Posit Software, PBC. All rights reserved.
+//
+//
+
+//! A timeout that breaks runaway R-thread work with an interrupt.
+//!
+//! R runs on a single thread, so work handed to it (a debugger `evaluate`, say)
+//! can loop forever and freeze the kernel. There's no way to cancel R code from
+//! the outside other than an interrupt, which R polls for at loop iterations and
+//! other check points.
+//!
+//! Instead of watching the clock from another thread, we piggyback on R's own
+//! polled-events handler and call [`check_timeout()`] there. The interrupt
+//! unwinds R back to the nearest `try_catch`, which surfaces it as an error.
+//!
+//! FIXME: This is unix-only: on Windows R's process-events callback is a no-op
+//! (see `sys::windows::console`), so the handler never fires mid-computation
+//! and the deadline is never checked. Fixed by https://github.com/posit-dev/ark/pull/1222.
+
+use std::cell::Cell;
+use std::time::Duration;
+use std::time::Instant;
+
+use harp::raii::RLocalInterruptsSuspended;
+
+/// How long an evaluation in `with_timeout()` may run before we interrupt it.
+pub(crate) const EVAL_TIMEOUT: Duration = Duration::from_secs(1);
+
+// These variables are thread-local to provide safe lock-free interior
+// mutability on the R thread
+thread_local! {
+    /// When the in-flight evaluation should be interrupted, if it's still
+    /// running. Set by `with_timeout()`, read by `check_timeout()`.
+    static DEADLINE: Cell<Option<Instant>> = const { Cell::new(None) };
+
+    /// Set by `check_timeout()` when it trips the interrupt, read by
+    /// `with_timeout()` which reports the timeout.
+    static TIMED_OUT: Cell<bool> = const { Cell::new(false) };
+}
+
+/// Run `f` on the R thread, interrupting it if it outlasts `timeout`.
+///
+/// `f` runs in `try_catch()` with interrupts and polled events enabled so
+/// `check_timeout()` can fire at interrupt time and the resulting interrupt can
+/// longjump.
+///
+/// Note that the longjump is either caught by our `try_catch()`, or a
+/// `try_catch()` inside `f`. An alternative approach would be to try and
+/// propagate the interrupt with a Rust panic, but that needs to be carefully
+/// engineered. Be aware that because of the approach taken here, it's possible
+/// for code inside `f` to try and recover from Rust errors
+/// (`harp::Error::TopLevelExecError`) caused by the cancellation interrupt.
+///
+/// Returns `f`'s value wrapped in the `try_catch()` result and a boolean
+/// indicating whether the timeout fired.
+///
+/// Must run on the R thread.
+pub(crate) fn with_timeout<F, T>(timeout: Duration, f: F) -> (harp::Result<T>, bool)
+where
+    F: FnOnce() -> T,
+{
+    // Save and restore so a nested evaluation doesn't clobber an outer deadline.
+    let old_deadline = DEADLINE.replace(Some(Instant::now() + timeout));
+    let old_timed_out = TIMED_OUT.replace(false);
+
+    let res = try_catch_with_timeout(f);
+
+    let timed_out = TIMED_OUT.get();
+
+    // If `check_timeout()` tripped the interrupt but `f` finished before R
+    // acted on it, a stray interrupt is left pending. Clear it here so it can't
+    // fire on a later evaluation.
+    if timed_out {
+        crate::signals::set_interrupts_pending(false);
+    }
+
+    DEADLINE.set(old_deadline);
+    TIMED_OUT.set(old_timed_out);
+
+    (res, timed_out)
+}
+
+/// Called from the polled-events handler at R's interrupt check points. If the
+/// in-flight evaluation has outlived its deadline, ask R to interrupt itself.
+/// Runs on the R thread.
+pub(crate) fn check_timeout() {
+    if TIMED_OUT.get() {
+        // Already tripped, don't ask twice
+        return;
+    }
+    let Some(deadline) = DEADLINE.get() else {
+        // No evaluation under a timeout
+        return;
+    };
+    if Instant::now() >= deadline {
+        TIMED_OUT.set(true);
+        crate::signals::set_interrupts_pending(true);
+    }
+}
+
+/// Run `f` in a `try_catch` with interrupts and polled events live.
+///
+/// An outer sandbox (ReadConsole, or the task executor) suspends both; we
+/// re-enable them so the watchdog runs during `f` and the interrupt it trips can
+/// longjump. On Windows there's no polled-events handler to re-enable, so `f`
+/// runs interruptible but unwatched.
+#[cfg(unix)]
+fn try_catch_with_timeout<F, T>(f: F) -> harp::Result<T>
+where
+    F: FnOnce() -> T,
+{
+    let _interrupts = RLocalInterruptsSuspended::new(false);
+    // Re-enable Ark's polled-events handler so `check_timeout()` runs during `f`.
+    let _polled = harp::raii::RLocal::new(
+        Some(crate::console::r_polled_events as unsafe extern "C-unwind" fn()),
+        unsafe { libr::R_PolledEvents },
+    );
+    harp::try_catch(f)
+}
+
+#[cfg(not(unix))]
+fn try_catch_with_timeout<F, T>(f: F) -> harp::Result<T>
+where
+    F: FnOnce() -> T,
+{
+    // TODO: Update after https://github.com/posit-dev/ark/pull/1222 is merged
+    let _interrupts = RLocalInterruptsSuspended::new(false);
+    harp::try_catch(f)
+}