diff --git a/crates/ark/src/console.rs b/crates/ark/src/console.rs
index 317dfe88b..8610f02a8 100644
--- a/crates/ark/src/console.rs
+++ b/crates/ark/src/console.rs
@@ -126,7 +126,6 @@ pub(crate) use console_debug::FrameSource;
 use console_error::stack_overflow_occurred;
 use console_filter::strip_step_lines;
 use console_filter::ConsoleFilter;
-pub use console_repl::catching_panics;
 pub(crate) use console_repl::console_inputs;
 pub(crate) use console_repl::r_busy;
 pub(crate) use console_repl::r_interrupt_events;
diff --git a/crates/ark/src/console/console_repl.rs b/crates/ark/src/console/console_repl.rs
index fc62b8cc4..6516c51a3 100644
--- a/crates/ark/src/console/console_repl.rs
+++ b/crates/ark/src/console/console_repl.rs
@@ -13,12 +13,13 @@
 use std::path::Path;
 use std::rc::Rc;
 
-use stdext::panic_message;
 use stdext::DebugRefCell;
 
 use super::*;
 use crate::dap::dap_notebook;
 use crate::data_explorer::r_data_explorer::POSITRON_DATA_EXPLORER_MIME;
+use crate::panic;
+use crate::panic::Recovery;
 use crate::r_task::QueuedRTask;
 use crate::r_task::RTask;
 use crate::r_task::TryIdleTask;
@@ -34,18 +35,6 @@ const DEBUG_COMMANDS: &[&str] = &["c", "cont", "f", "help", "n", "s", "where", "
 // These are not transient evals: they represent deliberate debugger navigation.
 const DEBUG_COMMANDS_CONTINUE: &[&str] = &["n", "f", "c", "cont", "Q"];
 
-thread_local! {
-    /// When `true`, the global panic hook should return early instead of
-    /// aborting, so that `catch_unwind` can catch the panic in `Console::with`.
-    static CATCHING_PANICS: Cell<bool> = const { Cell::new(false) };
-}
-
-/// Returns `true` when we are inside a `Console::with` catch boundary.
-/// Checked by the global panic hook to decide whether to abort.
-pub fn catching_panics() -> bool {
-    CATCHING_PANICS.get()
-}
-
 /// Used to wait for complete R startup in `Console::wait_initialized()` or
 /// check for it in `Console::is_initialized()`.
 ///
@@ -740,19 +729,11 @@ impl Console {
     /// caught and converted to `anyhow::Error`, which `harp::register`'s
     /// `r_unwrap()` then surfaces as a clean R error.
     pub fn with<T>(f: impl FnOnce(&Console) -> anyhow::Result<T>) -> anyhow::Result<T> {
-        if cfg!(debug_assertions) {
-            return f(Console::get());
-        }
-
-        CATCHING_PANICS.set(true);
-        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(Console::get())));
-        CATCHING_PANICS.set(false);
-
-        match result {
+        match panic::catch_unwind(Recovery::ReleaseOnly, || f(Console::get())) {
             Ok(result) => result,
-            Err(panic) => {
-                let msg = panic_message(panic.as_ref());
-                Err(anyhow!("Panic in Console callback: {msg}"))
+            Err(payload) => {
+                let message = panic::message(&payload);
+                Err(anyhow!("Panic in Console callback: {message}"))
             },
         }
     }
diff --git a/crates/ark/src/lib.rs b/crates/ark/src/lib.rs
index 02e367e42..f282d3b80 100644
--- a/crates/ark/src/lib.rs
+++ b/crates/ark/src/lib.rs
@@ -31,6 +31,7 @@ pub mod lsp;
 pub mod methods;
 pub mod modules;
 pub mod modules_utils;
+pub mod panic;
 pub mod plots;
 pub mod r_task;
 pub mod repos;
diff --git a/crates/ark/src/lsp/analysis.rs b/crates/ark/src/lsp/analysis.rs
index 6a372852e..7e94b945d 100644
--- a/crates/ark/src/lsp/analysis.rs
+++ b/crates/ark/src/lsp/analysis.rs
@@ -25,6 +25,6 @@ pub(crate) use snapshot::WorldStateSnapshot;
 pub(crate) use warmup::warm_workspace_index;
 
 /// Run `f`, swallowing a salsa cancellation as `None`. Any other panic propagates.
-fn catch_cancellation<T>(f: impl FnOnce() -> T) -> Option<T> {
+pub(crate) fn catch_cancellation<T>(f: impl FnOnce() -> T) -> Option<T> {
     salsa::Cancelled::catch(AssertUnwindSafe(f)).ok()
 }
diff --git a/crates/ark/src/lsp/analysis/pool.rs b/crates/ark/src/lsp/analysis/pool.rs
index 4181a3dec..da82da7be 100644
--- a/crates/ark/src/lsp/analysis/pool.rs
+++ b/crates/ark/src/lsp/analysis/pool.rs
@@ -6,19 +6,20 @@
 //
 
 use std::collections::VecDeque;
-use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
 use std::sync::Condvar;
 use std::sync::Mutex;
 use std::sync::MutexGuard;
 
 use aether_path::FilePath;
-use stdext::panic_message;
 use stdext::spawn;
 
 use super::catch_cancellation;
 use super::snapshot::WorldStateSnapshot;
 use crate::lsp;
+use crate::lsp::main_loop::LspServiceContext;
+use crate::panic;
+use crate::panic::Recovery;
 
 /// Enough threads that a handful of open files all get diagnosed in parallel,
 /// few enough that they don't crowd out the main loop or the R session we share
@@ -41,11 +42,11 @@ pub(crate) struct AnalysisPool {
 }
 
 impl AnalysisPool {
-    pub(crate) fn new() -> Self {
-        Self::with_threads(analysis_threads())
+    pub(crate) fn new(service_context: Arc<LspServiceContext>) -> Self {
+        Self::with_threads(analysis_threads(), service_context)
     }
 
-    fn with_threads(threads: usize) -> Self {
+    fn with_threads(threads: usize, service_context: Arc<LspServiceContext>) -> Self {
         let shared = Arc::new(Shared {
             queue: Mutex::new(Queue {
                 entries: VecDeque::new(),
@@ -56,7 +57,8 @@ impl AnalysisPool {
 
         for _ in 0..threads {
             let shared = Arc::clone(&shared);
-            spawn!("oak-analysis", move || work(shared));
+            let service_context = Arc::clone(&service_context);
+            spawn!("oak-analysis", move || work(shared, service_context));
         }
 
         Self { shared }
@@ -156,12 +158,12 @@ struct Entry {
     run: Box<dyn FnOnce(WorldStateSnapshot) + Send>,
 }
 
-fn work(shared: Arc<Shared>) {
+fn work(shared: Arc<Shared>, service_context: Arc<LspServiceContext>) {
     // `run_entry` takes the entry by value, so the snapshot has dropped by the
     // time we ask for the next one. A worker parked on `next_entry` doesn't
     // hold a db handle and can't block a writer.
     while let Some(entry) = shared.next_entry() {
-        run_entry(entry);
+        run_entry(entry, &service_context);
     }
 }
 
@@ -192,7 +194,7 @@ impl Shared {
     }
 }
 
-fn run_entry(entry: Entry) {
+fn run_entry(entry: Entry, service_context: &LspServiceContext) {
     let Entry { snapshot, run, .. } = entry;
 
     // A writer parked on this handle would only cancel the task at its first
@@ -202,12 +204,12 @@ fn run_entry(entry: Entry) {
         return;
     }
 
-    let task = AssertUnwindSafe(|| catch_cancellation(|| run(snapshot)));
-    if let Err(err) = std::panic::catch_unwind(task) {
-        lsp::log_error!(
-            "An analysis task panicked: {msg}",
-            msg = panic_message(err.as_ref())
-        );
+    if let Err(payload) =
+        panic::catch_unwind(Recovery::Always, || catch_cancellation(|| run(snapshot)))
+    {
+        let message = panic::message(&payload);
+        lsp::log_error!("An analysis task panicked: {message}");
+        service_context.report_background_panic();
     }
 }
 
@@ -218,6 +220,7 @@ mod tests {
     use std::sync::Arc;
 
     use super::AnalysisPool;
+    use crate::lsp::main_loop::LspServiceContext;
     use crate::lsp::state::WorldState;
 
     /// A queued task whose snapshot is already cancelled must be dropped without
@@ -229,7 +232,8 @@ mod tests {
     #[test]
     fn test_pool_drops_cancelled_task_without_running() {
         let state = WorldState::default();
-        let pool = AnalysisPool::with_threads(1);
+        let context = Arc::new(LspServiceContext::new());
+        let pool = AnalysisPool::with_threads(1, context);
 
         let cancelled = state.snapshot();
         cancelled.cancellation_token().cancel();
@@ -250,4 +254,28 @@ mod tests {
             .unwrap();
         assert!(!ran.load(Ordering::Acquire));
     }
+
+    /// Install the production hook so a missing `catch_unwind()` aborts the
+    /// process instead of silently losing the worker panic.
+    #[test]
+    fn test_pool_survives_panicking_task() {
+        crate::panic::install();
+
+        let state = WorldState::default();
+        let context = Arc::new(LspServiceContext::new());
+        let pool = AnalysisPool::with_threads(1, context);
+
+        pool.spawn(state.snapshot(), |_snapshot| {
+            panic!("Test panic in an analysis task")
+        });
+
+        let (barrier_tx, barrier_rx) = std::sync::mpsc::channel();
+        pool.spawn(state.snapshot(), move |_snapshot| {
+            barrier_tx.send(()).unwrap()
+        });
+
+        barrier_rx
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .unwrap();
+    }
 }
diff --git a/crates/ark/src/lsp/backend.rs b/crates/ark/src/lsp/backend.rs
index 3380112b5..5a53e2e58 100644
--- a/crates/ark/src/lsp/backend.rs
+++ b/crates/ark/src/lsp/backend.rs
@@ -8,9 +8,7 @@
 #![allow(deprecated)]
 
 use std::path::PathBuf;
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
-use std::time::Duration;
 
 use amalthea::comm::server_comm::ServerStartMessage;
 use amalthea::comm::server_comm::ServerStartedMessage;
@@ -34,7 +32,7 @@ use tower_lsp_server::LanguageServer;
 use tower_lsp_server::LspService;
 use tower_lsp_server::Server;
 
-use super::main_loop::LSP_HAS_CRASHED;
+use super::main_loop::LspServiceContext;
 use crate::console::Console;
 use crate::console::ConsoleNotification;
 use crate::lsp::handlers::VirtualDocumentParams;
@@ -55,10 +53,6 @@ use crate::lsp::statement_range::StatementRangeParams;
 use crate::lsp::statement_range::StatementRangeResponse;
 use crate::r_task;
 
-// This enum is useful for two things. First it allows us to distinguish a
-// normal request failure from a crash. In the latter case we send a
-// notification to the client so the user knows the LSP has crashed.
-//
 // Once the LSP has crashed all requests respond with an error. This prevents
 // any handler from running while we process the message to shut down the
 // server. The `Disabled` enum variant is an indicator of this state. We could
@@ -67,13 +61,12 @@ use crate::r_task;
 #[expect(clippy::large_enum_variant)]
 pub(crate) enum RequestResponse {
     Disabled,
-    Crashed(anyhow::Error),
     Result(LspResult<LspResponse>),
 }
 
 // Based on https://stackoverflow.com/a/69324393/1725177
 macro_rules! cast_response {
-    ($self:expr, $target:expr, $pat:path) => {{
+    ($target:expr, $pat:path) => {{
         match $target {
             RequestResponse::Result(Ok($pat(resp))) => Ok(resp),
             RequestResponse::Result(Ok(_)) => {
@@ -88,15 +81,6 @@ macro_rules! cast_response {
                 // is set, which then surfaces in the client's error popup.
                 LspError::Anyhow(err) => Err(new_jsonrpc_error(format!("{err}"))),
             },
-            RequestResponse::Crashed(err) => {
-                // Notify user that the LSP has crashed and is no longer active
-                report_crash($self.client()).await;
-
-                // The backtrace is reported via `err` and eventually shows up
-                // in the LSP logs on the client side
-                let _ = $self.shutdown_tx.send(()).await;
-                Err(new_jsonrpc_error(format!("{err:?}")))
-            },
             RequestResponse::Disabled => Err(new_jsonrpc_error(String::from(
                 "The LSP server has crashed and is now shut down!",
             ))),
@@ -104,36 +88,6 @@ macro_rules! cast_response {
     }};
 }
 
-/// Send via `request::ShowMessageRequest` not `notification::ShowMessage` so that we can
-/// ensure that the message has been received on the frontend side. We are about to shut
-/// the LSP down, and sending out a fire-and-forget notification often won't get sent out
-/// before shutdown occurs. The request returns control to us when the user acknowledges
-/// the message. It doesn't matter if that takes awhile because we shut down right after,
-/// and we've already flipped the `LSP_HAS_CRASHED` global flag, but we do bound it with
-/// a 5 second timeout just in case the user ignores the message entirely, so we can still
-/// shutdown.
-async fn report_crash(client: &Client) {
-    let user_message = concat!(
-        "The R language server has crashed and has been disabled. ",
-        "Smart features such as completions will no longer work in this session. ",
-        "Please report this crash to https://github.com/posit-dev/positron/issues ",
-        "with full logs (see https://positron.posit.co/troubleshooting.html#python-and-r-logs)."
-    );
-    let request = client.send_request::<request::ShowMessageRequest>(ShowMessageRequestParams {
-        typ: MessageType::ERROR,
-        message: String::from(user_message),
-        actions: None,
-    });
-    match tokio::time::timeout(Duration::from_secs(5), request).await {
-        Ok(result) => {
-            result.log_err();
-        },
-        Err(_) => {
-            log::warn!("Timed out waiting for frontend to acknowledge LSP crash notification");
-        },
-    }
-}
-
 #[derive(Debug)]
 #[expect(clippy::large_enum_variant)]
 pub(crate) enum LspMessage {
@@ -151,6 +105,12 @@ pub(crate) enum LspNotification {
     DidChangeTextDocument(DidChangeTextDocumentParams),
     DidSaveTextDocument(DidSaveTextDocumentParams),
     DidCloseTextDocument(DidCloseTextDocumentParams),
+    #[cfg(feature = "testing")]
+    TestPanic,
+    #[cfg(feature = "testing")]
+    TestCancelRTask,
+    #[cfg(feature = "testing")]
+    TestPanicRTask,
 }
 
 #[derive(Debug)]
@@ -177,6 +137,8 @@ pub(crate) enum LspRequest {
     CodeAction(CodeActionParams),
     VirtualDocument(VirtualDocumentParams),
     InputBoundaries(InputBoundariesParams),
+    #[cfg(feature = "testing")]
+    TestPanic,
 }
 
 #[derive(Debug)]
@@ -203,6 +165,8 @@ pub(crate) enum LspResponse {
     CodeAction(Option<CodeActionResponse>),
     VirtualDocument(VirtualDocumentResponse),
     InputBoundaries(InputBoundariesResponse),
+    #[cfg(feature = "testing")]
+    TestPanic(()),
 }
 
 pub(crate) type LspResult<T> = std::result::Result<T, LspError>;
@@ -240,15 +204,11 @@ impl From<anyhow::Error> for LspError {
 
 #[derive(Debug)]
 struct Backend {
-    /// Shutdown notifier used to unwind tower-lsp and disconnect from the
-    /// client when an LSP handler panics.
-    shutdown_tx: tokio::sync::mpsc::Sender<()>,
-
     /// Channel for communication with the main loop.
     events_tx: TokioUnboundedSender<Event>,
 
-    /// Copy of the Client, for reporting crash messages.
-    client: Client,
+    /// State shared with the main loop and its background pools.
+    service_context: Arc<LspServiceContext>,
 
     /// Handle to the LSP loops. Drop it to shut the loops down and drop all
     /// owned state.
@@ -257,37 +217,47 @@ struct Backend {
 
 impl Backend {
     async fn request(&self, request: LspRequest) -> RequestResponse {
-        if LSP_HAS_CRASHED.load(Ordering::Acquire) {
+        if self.service_context.has_crashed() {
             return RequestResponse::Disabled;
         }
 
         let (response_tx, mut response_rx) = tokio_unbounded_channel::<RequestResponse>();
 
         // Relay request to main loop
-        self.events_tx
+        if self
+            .events_tx
             .send(Event::Lsp(LspMessage::Request(request, response_tx)))
-            .unwrap();
+            .is_err()
+        {
+            return RequestResponse::Disabled;
+        }
 
         // Wait for response from main loop
-        response_rx.recv().await.unwrap()
+        match response_rx.recv().await {
+            Some(response) => response,
+            None => RequestResponse::Disabled,
+        }
     }
 
     fn notify(&self, notif: LspNotification) {
+        if self.service_context.has_crashed() {
+            return;
+        }
+
         // Relay notification to main loop
-        self.events_tx
+        if self
+            .events_tx
             .send(Event::Lsp(LspMessage::Notification(notif)))
-            .unwrap();
-    }
-
-    fn client(&self) -> &Client {
-        &self.client
+            .is_err()
+        {
+            log::error!("Can't relay notification, the main loop is gone");
+        }
     }
 }
 
 impl LanguageServer for Backend {
     async fn initialize(&self, params: InitializeParams) -> Result<InitializeResult> {
         cast_response!(
-            self,
             self.request(LspRequest::Initialize(params)).await,
             LspResponse::Initialize
         )
@@ -320,7 +290,6 @@ impl LanguageServer for Backend {
         params: WorkspaceSymbolParams,
     ) -> Result<Option<WorkspaceSymbolResponse>> {
         let info: Option<Vec<SymbolInformation>> = cast_response!(
-            self,
             self.request(LspRequest::WorkspaceSymbol(params)).await,
             LspResponse::WorkspaceSymbol
         )?;
@@ -332,7 +301,6 @@ impl LanguageServer for Backend {
         params: DocumentSymbolParams,
     ) -> Result<Option<DocumentSymbolResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::DocumentSymbol(params)).await,
             LspResponse::DocumentSymbol
         )
@@ -340,7 +308,6 @@ impl LanguageServer for Backend {
 
     async fn folding_range(&self, params: FoldingRangeParams) -> Result<Option<Vec<FoldingRange>>> {
         cast_response!(
-            self,
             self.request(LspRequest::FoldingRange(params)).await,
             LspResponse::FoldingRange
         )
@@ -351,7 +318,6 @@ impl LanguageServer for Backend {
         params: ExecuteCommandParams,
     ) -> jsonrpc::Result<Option<Value>> {
         cast_response!(
-            self,
             self.request(LspRequest::ExecuteCommand(params)).await,
             LspResponse::ExecuteCommand
         )
@@ -375,7 +341,6 @@ impl LanguageServer for Backend {
 
     async fn completion(&self, params: CompletionParams) -> Result<Option<CompletionResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::Completion(params)).await,
             LspResponse::Completion
         )
@@ -383,7 +348,6 @@ impl LanguageServer for Backend {
 
     async fn completion_resolve(&self, item: CompletionItem) -> Result<CompletionItem> {
         cast_response!(
-            self,
             self.request(LspRequest::CompletionResolve(item)).await,
             LspResponse::CompletionResolve
         )
@@ -391,7 +355,6 @@ impl LanguageServer for Backend {
 
     async fn hover(&self, params: HoverParams) -> Result<Option<Hover>> {
         cast_response!(
-            self,
             self.request(LspRequest::Hover(params)).await,
             LspResponse::Hover
         )
@@ -399,7 +362,6 @@ impl LanguageServer for Backend {
 
     async fn signature_help(&self, params: SignatureHelpParams) -> Result<Option<SignatureHelp>> {
         cast_response!(
-            self,
             self.request(LspRequest::SignatureHelp(params)).await,
             LspResponse::SignatureHelp
         )
@@ -410,7 +372,6 @@ impl LanguageServer for Backend {
         params: GotoDefinitionParams,
     ) -> Result<Option<GotoDefinitionResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::GotoDefinition(params)).await,
             LspResponse::GotoDefinition
         )
@@ -421,7 +382,6 @@ impl LanguageServer for Backend {
         params: GotoImplementationParams,
     ) -> Result<Option<GotoImplementationResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::GotoImplementation(params)).await,
             LspResponse::GotoImplementation
         )
@@ -432,7 +392,6 @@ impl LanguageServer for Backend {
         params: SelectionRangeParams,
     ) -> Result<Option<Vec<SelectionRange>>> {
         cast_response!(
-            self,
             self.request(LspRequest::SelectionRange(params)).await,
             LspResponse::SelectionRange
         )
@@ -440,7 +399,6 @@ impl LanguageServer for Backend {
 
     async fn references(&self, params: ReferenceParams) -> Result<Option<Vec<Location>>> {
         cast_response!(
-            self,
             self.request(LspRequest::References(params)).await,
             LspResponse::References
         )
@@ -451,7 +409,6 @@ impl LanguageServer for Backend {
         params: TextDocumentPositionParams,
     ) -> Result<Option<PrepareRenameResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::PrepareRename(params)).await,
             LspResponse::PrepareRename
         )
@@ -459,7 +416,6 @@ impl LanguageServer for Backend {
 
     async fn rename(&self, params: RenameParams) -> Result<Option<WorkspaceEdit>> {
         cast_response!(
-            self,
             self.request(LspRequest::Rename(params)).await,
             LspResponse::Rename
         )
@@ -470,7 +426,6 @@ impl LanguageServer for Backend {
         params: DocumentOnTypeFormattingParams,
     ) -> Result<Option<Vec<TextEdit>>> {
         cast_response!(
-            self,
             self.request(LspRequest::OnTypeFormatting(params)).await,
             LspResponse::OnTypeFormatting
         )
@@ -478,7 +433,6 @@ impl LanguageServer for Backend {
 
     async fn code_action(&self, params: CodeActionParams) -> Result<Option<CodeActionResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::CodeAction(params)).await,
             LspResponse::CodeAction
         )
@@ -506,7 +460,6 @@ impl Backend {
         params: StatementRangeParams,
     ) -> jsonrpc::Result<Option<StatementRangeResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::StatementRange(params)).await,
             LspResponse::StatementRange
         )
@@ -517,7 +470,6 @@ impl Backend {
         params: HelpTopicParams,
     ) -> jsonrpc::Result<Option<HelpTopicResponse>> {
         cast_response!(
-            self,
             self.request(LspRequest::HelpTopic(params)).await,
             LspResponse::HelpTopic
         )
@@ -528,7 +480,6 @@ impl Backend {
         params: VirtualDocumentParams,
     ) -> tower_lsp_server::jsonrpc::Result<VirtualDocumentResponse> {
         cast_response!(
-            self,
             self.request(LspRequest::VirtualDocument(params)).await,
             LspResponse::VirtualDocument
         )
@@ -539,7 +490,6 @@ impl Backend {
         params: InputBoundariesParams,
     ) -> tower_lsp_server::jsonrpc::Result<InputBoundariesResponse> {
         cast_response!(
-            self,
             self.request(LspRequest::InputBoundaries(params)).await,
             LspResponse::InputBoundaries
         )
@@ -548,8 +498,47 @@ impl Backend {
     async fn notification(&self, params: Option<Value>) {
         log::info!("Received Positron notification: {:?}", params);
     }
+
+    #[cfg(feature = "testing")]
+    async fn test_panic(&self, _params: Option<Value>) -> jsonrpc::Result<()> {
+        cast_response!(
+            self.request(LspRequest::TestPanic).await,
+            LspResponse::TestPanic
+        )
+    }
+
+    #[cfg(feature = "testing")]
+    async fn test_panic_notification(&self, _params: Option<Value>) {
+        self.notify(LspNotification::TestPanic);
+    }
+
+    #[cfg(feature = "testing")]
+    async fn test_cancel_r_task(&self, _params: Option<Value>) {
+        self.notify(LspNotification::TestCancelRTask);
+    }
+
+    #[cfg(feature = "testing")]
+    async fn test_panic_r_task(&self, _params: Option<Value>) {
+        self.notify(LspNotification::TestPanicRTask);
+    }
+
+    #[cfg(feature = "testing")]
+    async fn test_panic_main_loop(&self, _params: Option<Value>) {
+        let _ = self.events_tx.send(Event::TestPanicMainLoop);
+    }
 }
 
+#[cfg(feature = "testing")]
+pub(crate) static ARK_TEST_PANIC_REQUEST: &str = "ark/testPanic";
+#[cfg(feature = "testing")]
+pub(crate) static ARK_TEST_PANIC_NOTIFICATION: &str = "ark/testPanicNotification";
+#[cfg(feature = "testing")]
+pub(crate) static ARK_TEST_PANIC_MAIN_LOOP: &str = "ark/testPanicMainLoop";
+#[cfg(feature = "testing")]
+pub(crate) static ARK_TEST_CANCEL_R_TASK: &str = "ark/testCancelRTask";
+#[cfg(feature = "testing")]
+pub(crate) static ARK_TEST_PANIC_R_TASK: &str = "ark/testPanicRTask";
+
 pub(crate) fn start_lsp(
     r_home: PathBuf,
     runtime: Arc<Runtime>,
@@ -590,11 +579,12 @@ pub(crate) fn start_lsp(
         let (shutdown_tx, mut shutdown_rx) = tokio::sync::mpsc::channel::<()>(1);
 
         let init = |client: Client| {
-            let state = GlobalState::new(client.clone(), r_home, console_notification_tx);
+            let state = GlobalState::new(client, r_home, console_notification_tx);
             let events_tx = state.events_tx();
+            let service_context = Arc::clone(state.service_context());
 
             // Start main loop and hold onto the handle that keeps it alive
-            let main_loop = state.start();
+            let main_loop = state.start(shutdown_tx);
 
             // Forward event channel along to `Console`.
             // This also updates an outdated channel after a reconnect.
@@ -610,14 +600,13 @@ pub(crate) fn start_lsp(
             });
 
             Backend {
-                shutdown_tx,
                 events_tx,
-                client,
+                service_context,
                 _main_loop: main_loop,
             }
         };
 
-        let (service, socket) = LspService::build(init)
+        let builder = LspService::build(init)
             .custom_method(
                 statement_range::POSITRON_STATEMENT_RANGE_REQUEST,
                 Backend::statement_range,
@@ -629,8 +618,20 @@ pub(crate) fn start_lsp(
                 input_boundaries::POSITRON_INPUT_BOUNDARIES_REQUEST,
                 Backend::input_boundaries,
             )
-            .custom_method("positron/notification", Backend::notification)
-            .finish();
+            .custom_method("positron/notification", Backend::notification);
+
+        #[cfg(feature = "testing")]
+        let builder = builder
+            .custom_method(ARK_TEST_PANIC_REQUEST, Backend::test_panic)
+            .custom_method(
+                ARK_TEST_PANIC_NOTIFICATION,
+                Backend::test_panic_notification,
+            )
+            .custom_method(ARK_TEST_PANIC_MAIN_LOOP, Backend::test_panic_main_loop)
+            .custom_method(ARK_TEST_CANCEL_R_TASK, Backend::test_cancel_r_task)
+            .custom_method(ARK_TEST_PANIC_R_TASK, Backend::test_panic_r_task);
+
+        let (service, socket) = builder.finish();
 
         let server = Server::new(read, write, socket);
 
diff --git a/crates/ark/src/lsp/io_pool.rs b/crates/ark/src/lsp/io_pool.rs
index e2c579258..ef401a1db 100644
--- a/crates/ark/src/lsp/io_pool.rs
+++ b/crates/ark/src/lsp/io_pool.rs
@@ -5,13 +5,15 @@
 //
 //
 
-use std::panic::AssertUnwindSafe;
+use std::sync::Arc;
 
 use crossbeam::channel::Sender;
-use stdext::panic_message;
 use stdext::spawn_with_stack_size;
 
 use crate::lsp;
+use crate::lsp::main_loop::LspServiceContext;
+use crate::panic;
+use crate::panic::Recovery;
 
 type Job = Box<dyn FnOnce() + Send + 'static>;
 
@@ -31,14 +33,20 @@ impl IoPool {
     /// of stack. Each lane picks its own size from the deepest call tree its
     /// jobs can reach, so use [`stdext::DEFAULT_STACK_SIZE`] unless you've
     /// bounded that.
-    pub(crate) fn new(name: &'static str, threads: usize, stack_size: usize) -> Self {
+    pub(crate) fn new(
+        name: &'static str,
+        threads: usize,
+        stack_size: usize,
+        service_context: Arc<LspServiceContext>,
+    ) -> Self {
         let (jobs_tx, jobs_rx) = crossbeam::channel::unbounded::<Job>();
 
         for _ in 0..threads {
             let jobs_rx = jobs_rx.clone();
+            let service_context = Arc::clone(&service_context);
             spawn_with_stack_size!(name, stack_size, move || {
                 while let Ok(job) = jobs_rx.recv() {
-                    run_job(job);
+                    run_job(job, &service_context);
                 }
             });
         }
@@ -53,11 +61,31 @@ impl IoPool {
     }
 }
 
-fn run_job(job: Job) {
-    if let Err(err) = std::panic::catch_unwind(AssertUnwindSafe(job)) {
-        lsp::log_error!(
-            "An I/O job panicked: {msg}",
-            msg = panic_message(err.as_ref())
-        );
+fn run_job(job: Job, service_context: &LspServiceContext) {
+    if let Err(payload) = panic::catch_unwind(Recovery::Always, job) {
+        let message = panic::message(&payload);
+        lsp::log_error!("An I/O job panicked: {message}");
+        service_context.report_background_panic();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Install the production hook so a missing `catch_unwind()` aborts the
+    /// process instead of silently losing the worker panic.
+    #[test]
+    fn test_pool_survives_panicking_job() {
+        crate::panic::install();
+
+        let context = Arc::new(LspServiceContext::new());
+        let pool = IoPool::new("test-io-pool", 1, stdext::DEFAULT_STACK_SIZE, context);
+        pool.submit(|| panic!("Test panic in an I/O job"));
+
+        let (tx, rx) = std::sync::mpsc::channel();
+        pool.submit(move || tx.send(()).unwrap());
+
+        rx.recv_timeout(std::time::Duration::from_secs(10)).unwrap();
     }
 }
diff --git a/crates/ark/src/lsp/main_loop.rs b/crates/ark/src/lsp/main_loop.rs
index cf1f0658f..db1f00362 100644
--- a/crates/ark/src/lsp/main_loop.rs
+++ b/crates/ark/src/lsp/main_loop.rs
@@ -7,12 +7,15 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::mem::discriminant;
+use std::mem::Discriminant;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::sync::RwLock;
+use std::time::Duration;
 
 use aether_path::FilePath;
 use anyhow::anyhow;
@@ -22,16 +25,18 @@ use oak_scan::DbScan;
 use oak_scan::ScanCompleted;
 use oak_scan::ScanRequest;
 use oak_scan::ScanScheduler;
-use stdext::panic_message;
 use stdext::result::ResultExt;
 use stdext::spawn;
 use tokio::runtime::Handle;
 use tokio::sync::mpsc::unbounded_channel as tokio_unbounded_channel;
+use tokio::sync::mpsc::Sender;
 use tokio::sync::oneshot;
 use tower_lsp_server::jsonrpc;
 use tower_lsp_server::ls_types as lsp_types;
+use tower_lsp_server::ls_types::request;
 use tower_lsp_server::ls_types::Diagnostic;
 use tower_lsp_server::ls_types::MessageType;
+use tower_lsp_server::ls_types::ShowMessageRequestParams;
 use tower_lsp_server::ls_types::Uri;
 use tower_lsp_server::Client;
 
@@ -39,6 +44,7 @@ use super::backend::RequestResponse;
 use crate::console::ConsoleNotification;
 use crate::lsp;
 use crate::lsp::analysis;
+use crate::lsp::analysis::catch_cancellation;
 use crate::lsp::analysis::AnalysisPool;
 use crate::lsp::analysis::DiagnosticsReady;
 use crate::lsp::analysis::DiagnosticsState;
@@ -64,6 +70,8 @@ use crate::lsp::state_handlers;
 use crate::lsp::state_handlers::ConsoleInputs;
 use crate::lsp::traits::url::UriExt;
 use crate::lsp::watchdog::Watchdog;
+use crate::panic;
+use crate::panic::Recovery;
 
 pub(crate) type TokioUnboundedSender<T> = tokio::sync::mpsc::UnboundedSender<T>;
 pub(crate) type TokioUnboundedReceiver<T> = tokio::sync::mpsc::UnboundedReceiver<T>;
@@ -83,7 +91,37 @@ pub(crate) type TokioUnboundedReceiver<T> = tokio::sync::mpsc::UnboundedReceiver
 /// LSPs to send log messages and tasks to the newer LSPs.
 static AUXILIARY_EVENT_TX: RwLock<Option<TokioUnboundedSender<AuxiliaryEvent>>> = RwLock::new(None);
 
-pub static LSP_HAS_CRASHED: AtomicBool = AtomicBool::new(false);
+/// State shared across one LSP service.
+#[derive(Debug)]
+pub(crate) struct LspServiceContext {
+    crashed: AtomicBool,
+    background_panic_reported: AtomicBool,
+}
+
+impl LspServiceContext {
+    pub(crate) fn new() -> Self {
+        Self {
+            crashed: AtomicBool::new(false),
+            background_panic_reported: AtomicBool::new(false),
+        }
+    }
+
+    pub(crate) fn has_crashed(&self) -> bool {
+        self.crashed.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn set_crashed(&self) {
+        self.crashed.store(true, Ordering::Release);
+    }
+
+    pub(crate) fn report_background_panic(&self) {
+        if self.background_panic_reported.swap(true, Ordering::AcqRel) {
+            return;
+        }
+
+        report_background_panic();
+    }
+}
 
 #[derive(Debug)]
 #[expect(clippy::large_enum_variant)]
@@ -93,6 +131,8 @@ pub(crate) enum Event {
     OakScanCompleted(ScanCompleted),
     SourceCompleted(SourceCompleted),
     DiagnosticsReady(DiagnosticsReady),
+    #[cfg(feature = "testing")]
+    TestPanicMainLoop,
 }
 
 #[derive(Debug)]
@@ -123,6 +163,7 @@ pub(crate) struct DidCloseVirtualDocumentParams {
 pub(crate) enum AuxiliaryEvent {
     Log(lsp_types::MessageType, String),
     PublishDiagnostics(DiagnosticsPublication),
+    ReportBackgroundPanic,
     Shutdown,
 }
 
@@ -146,6 +187,9 @@ pub(crate) struct GlobalState {
     /// LSP client shared with tower-lsp and the log loop
     client: Client,
 
+    /// Request handlers whose panic has already been reported to the user.
+    reported_request_panics: HashSet<Discriminant<LspRequest>>,
+
     /// Event channels for the main loop. The tower-lsp methods forward
     /// notifications and requests here via `Event::Lsp`. We also receive
     /// messages from the kernel via `Event::Kernel`, and from ourselves via
@@ -212,6 +256,9 @@ pub(crate) struct LspState {
     /// behind a background task that can't drop its db snapshot. See
     /// [`crate::lsp::watchdog`].
     pub(crate) watchdog: Watchdog,
+
+    /// State shared with the backend, main loop, and background pools.
+    service_context: Arc<LspServiceContext>,
 }
 
 impl LspState {
@@ -219,25 +266,34 @@ impl LspState {
         console_notification_tx: TokioUnboundedSender<ConsoleNotification>,
         source_scheduler: SourceScheduler,
     ) -> Self {
+        let service_context = Arc::new(LspServiceContext::new());
+
         Self {
             capabilities: Capabilities::default(),
             console_notification_tx,
             oak_scheduler: ScanScheduler::new(),
             source_scheduler,
-            analysis_pool: AnalysisPool::new(),
+            analysis_pool: AnalysisPool::new(Arc::clone(&service_context)),
             diagnostics: DiagnosticsState::default(),
             // Stack size: `ScanRequest::run()` walks the filesystem with
             // `ignore::Walk` and `WalkDir`, both iterative, and parses
             // DESCRIPTION line by line.
-            scan_pool: IoPool::new("oak-scan", 1, stdext::SMALL_STACK_SIZE),
+            scan_pool: IoPool::new(
+                "oak-scan",
+                1,
+                stdext::SMALL_STACK_SIZE,
+                Arc::clone(&service_context),
+            ),
             // Full stack because a fetch runs a rustls handshake, zstd and tar
             // decoding, and an R subprocess.
             source_pool: IoPool::new(
                 "oak-source",
                 SOURCE_POOL_THREADS,
                 stdext::DEFAULT_STACK_SIZE,
+                Arc::clone(&service_context),
             ),
             watchdog: Watchdog::new(),
+            service_context,
         }
     }
 }
@@ -251,6 +307,7 @@ impl LspState {
 /// The auxiliary loop currently handles:
 /// - Log messages.
 /// - Diagnostics publication.
+/// - Background task panic reports.
 struct AuxiliaryState {
     client: Client,
     auxiliary_event_rx: TokioUnboundedReceiver<AuxiliaryEvent>,
@@ -315,6 +372,7 @@ impl GlobalState {
             world,
             lsp_state,
             client,
+            reported_request_panics: HashSet::new(),
             events_tx,
             events_rx,
         }
@@ -325,11 +383,15 @@ impl GlobalState {
         self.events_tx.clone()
     }
 
+    pub(crate) fn service_context(&self) -> &Arc<LspServiceContext> {
+        &self.lsp_state.service_context
+    }
+
     /// Start the main and auxiliary loops.
     ///
     /// The returned [`LoopHandles`] owns everything the loops need. Drop it to
     /// shut the loops down and release the owned state.
-    pub(crate) fn start(self) -> LoopHandles {
+    pub(crate) fn start(self, server_shutdown_tx: Sender<()>) -> LoopHandles {
         let mut aux = tokio::task::JoinSet::<()>::new();
 
         // The auxiliary loop is fully async and never blocks. Must be started
@@ -346,8 +408,40 @@ impl GlobalState {
         // thread that we're in control of.
         let (shutdown_tx, shutdown_rx) = oneshot::channel();
         let handle = Handle::current();
+        let client = self.client.clone();
+        let service_context = Arc::clone(&self.lsp_state.service_context);
+
         let main_loop = spawn!("oak-main-loop", move || {
-            handle.block_on(self.main_loop(shutdown_rx));
+            let outcome = panic::catch_unwind(Recovery::Always, {
+                let server_shutdown_tx = server_shutdown_tx.clone();
+                let handle = handle.clone();
+                let service_context = Arc::clone(&service_context);
+                move || {
+                    handle.block_on(self.main_loop(
+                        shutdown_rx,
+                        server_shutdown_tx,
+                        service_context,
+                    ))
+                }
+            });
+
+            // Handle panics that bypass `handle_event()`'s recovery boundary.
+            if let Err(payload) = outcome {
+                let message = panic::message(&payload);
+                lsp::log_error!("Panic in the main loop: {message}");
+                service_context.set_crashed();
+
+                let report = panic::catch_unwind(Recovery::Always, || {
+                    handle.block_on(report_crash(&client))
+                });
+                if let Err(payload) = report {
+                    let message = panic::message(&payload);
+                    log::error!("Panic while reporting an LSP crash: {message}");
+                }
+
+                // The runtime may be shutting down, so don't wait for channel capacity.
+                server_shutdown_tx.try_send(()).log_err();
+            }
         });
 
         LoopHandles {
@@ -361,7 +455,12 @@ impl GlobalState {
     ///
     /// This takes ownership of all global state and handles one by one LSP
     /// requests, notifications, and other internal events.
-    async fn main_loop(mut self, mut shutdown_rx: oneshot::Receiver<()>) {
+    async fn main_loop(
+        mut self,
+        mut shutdown_rx: oneshot::Receiver<()>,
+        server_shutdown_tx: Sender<()>,
+        service_context: Arc<LspServiceContext>,
+    ) {
         loop {
             tokio::select! {
                 _ = &mut shutdown_rx => {
@@ -373,8 +472,31 @@ impl GlobalState {
                         lsp::log_info!("Main loop stopping: event channel closed");
                         break;
                     };
-                    if let Err(err) = self.handle_event(event).await {
-                        lsp::log_error!("Failure while handling event:\n{err:?}")
+
+                    #[cfg(feature = "testing")]
+                    if matches!(event, Event::TestPanicMainLoop) {
+                        panic!("Test panic outside the event recovery boundary");
+                    }
+
+                    let outcome =
+                        panic::catch_unwind_async(Recovery::Always, self.handle_event(event)).await;
+
+                    match outcome {
+                        Ok(Ok(())) => {},
+                        Ok(Err(err)) => lsp::log_error!("Failure while handling event:\n{err:?}"),
+                        Err(payload) => match classify_event_unwind(payload) {
+                            EventUnwind::Cancelled => {},
+                            EventUnwind::Panicked(payload) => {
+                                // `report_crash()` reads only the `Client` handle. Drop `self` after
+                                // the panic because a handler may have partially written its state.
+                                let message = panic::message(&payload);
+                                lsp::log_error!("Panic while handling event: {message}");
+                                service_context.set_crashed();
+                                report_crash(&self.client).await;
+                                let _ = server_shutdown_tx.send(()).await;
+                                break;
+                            },
+                        },
                     }
                 }
             }
@@ -450,82 +572,111 @@ impl GlobalState {
                         LspNotification::DidCloseTextDocument(params) => {
                             state_handlers::did_close(params, &mut self.world)?;
                         },
+
+                        #[cfg(feature = "testing")]
+                        LspNotification::TestPanic => {
+                            panic!("Test panic in a notification handler");
+                        },
+                        #[cfg(feature = "testing")]
+                        LspNotification::TestCancelRTask => {
+                            crate::r_task(|| {
+                                std::panic::resume_unwind(Box::new(
+                                    salsa::Cancelled::PendingWrite,
+                                ))
+                            });
+                        },
+                        #[cfg(feature = "testing")]
+                        LspNotification::TestPanicRTask => {
+                            crate::r_task(|| panic!("Test panic in an R task"));
+                        },
                     }
                 },
 
                 LspMessage::Request(request, tx) => {
                     lsp::log_info!("{request:#?}");
 
-                    match request {
+                    let request_kind = discriminant(&request);
+                    let outcome = match request {
                         LspRequest::Initialize(params) => {
-                            respond(tx, || state_handlers::initialize(params, &mut self.lsp_state, &mut self.world, &self.events_tx), LspResponse::Initialize)?;
+                            respond_exclusive(tx, || state_handlers::initialize(params, &mut self.lsp_state, &mut self.world, &self.events_tx), LspResponse::Initialize)?
                         },
                         LspRequest::WorkspaceSymbol(params) => {
-                            respond(tx, || handlers::handle_symbol(params, &self.world), LspResponse::WorkspaceSymbol)?;
+                            respond(tx, || handlers::handle_symbol(params, &self.world), LspResponse::WorkspaceSymbol)?
                         },
                         LspRequest::DocumentSymbol(params) => {
-                            respond(tx, || handlers::handle_document_symbol(params, &self.world), LspResponse::DocumentSymbol)?;
+                            respond(tx, || handlers::handle_document_symbol(params, &self.world), LspResponse::DocumentSymbol)?
                         },
                         LspRequest::FoldingRange(params) => {
-                            respond(tx, || handlers::handle_folding_range(params, &self.world), LspResponse::FoldingRange)?;
+                            respond(tx, || handlers::handle_folding_range(params, &self.world), LspResponse::FoldingRange)?
                         },
                         LspRequest::ExecuteCommand(_params) => {
                             let response = handlers::handle_execute_command(&self.client).await;
-                            respond(tx, || response, LspResponse::ExecuteCommand)?;
+                            respond(tx, || response, LspResponse::ExecuteCommand)?
                         },
                         LspRequest::Completion(params) => {
-                            respond(tx, || handlers::handle_completion(params, &self.world), LspResponse::Completion)?;
+                            respond(tx, || handlers::handle_completion(params, &self.world), LspResponse::Completion)?
                         },
                         LspRequest::CompletionResolve(params) => {
-                            respond(tx, || handlers::handle_completion_resolve(params), LspResponse::CompletionResolve)?;
+                            respond(tx, || handlers::handle_completion_resolve(params), LspResponse::CompletionResolve)?
                         },
                         LspRequest::Hover(params) => {
-                            respond(tx, || handlers::handle_hover(params, &self.world), LspResponse::Hover)?;
+                            respond(tx, || handlers::handle_hover(params, &self.world), LspResponse::Hover)?
                         },
                         LspRequest::SignatureHelp(params) => {
-                            respond(tx, || handlers::handle_signature_help(params, &self.world), LspResponse::SignatureHelp)?;
+                            respond(tx, || handlers::handle_signature_help(params, &self.world), LspResponse::SignatureHelp)?
                         },
                         LspRequest::GotoDefinition(params) => {
-                            respond(tx, || handlers::handle_goto_definition(params, &self.world), LspResponse::GotoDefinition)?;
+                            respond(tx, || handlers::handle_goto_definition(params, &self.world), LspResponse::GotoDefinition)?
                         },
                         LspRequest::GotoImplementation(_params) => {
                             // TODO
-                            respond(tx, || Ok(None), LspResponse::GotoImplementation)?;
+                            respond(tx, || Ok(None), LspResponse::GotoImplementation)?
                         },
                         LspRequest::SelectionRange(params) => {
-                            respond(tx, || handlers::handle_selection_range(params, &self.world), LspResponse::SelectionRange)?;
+                            respond(tx, || handlers::handle_selection_range(params, &self.world), LspResponse::SelectionRange)?
                         },
                         LspRequest::References(params) => {
-                            respond(tx, || handlers::handle_references(params, &self.world), LspResponse::References)?;
+                            respond(tx, || handlers::handle_references(params, &self.world), LspResponse::References)?
                         },
                         LspRequest::PrepareRename(params) => {
-                            respond(tx, || handlers::handle_prepare_rename(params, &self.world), LspResponse::PrepareRename)?;
+                            respond(tx, || handlers::handle_prepare_rename(params, &self.world), LspResponse::PrepareRename)?
                         },
                         LspRequest::Rename(params) => {
-                            respond(tx, || handlers::handle_rename(params, &self.world), LspResponse::Rename)?;
+                            respond(tx, || handlers::handle_rename(params, &self.world), LspResponse::Rename)?
                         },
                         LspRequest::StatementRange(params) => {
-                            respond(tx, || handlers::handle_statement_range(params, &self.world), LspResponse::StatementRange)?;
+                            respond(tx, || handlers::handle_statement_range(params, &self.world), LspResponse::StatementRange)?
                         },
                         LspRequest::HelpTopic(params) => {
-                            respond(tx, || handlers::handle_help_topic(params, &self.world), LspResponse::HelpTopic)?;
+                            respond(tx, || handlers::handle_help_topic(params, &self.world), LspResponse::HelpTopic)?
                         },
                         LspRequest::OnTypeFormatting(params) => {
                             if let Some(path) = params.text_document_position.text_document.uri.to_document_path().log_err() {
                                 state_handlers::did_change_formatting_options(&path, &params.options, &mut self.world);
                             }
-                            respond(tx, || handlers::handle_indent(params, &self.world), LspResponse::OnTypeFormatting)?;
+                            respond(tx, || handlers::handle_indent(params, &self.world), LspResponse::OnTypeFormatting)?
                         },
                         LspRequest::CodeAction(params) => {
-                            respond(tx, || handlers::handle_code_action(params, &self.lsp_state, &self.world), LspResponse::CodeAction)?;
+                            respond(tx, || handlers::handle_code_action(params, &self.lsp_state, &self.world), LspResponse::CodeAction)?
                         },
                         LspRequest::VirtualDocument(params) => {
-                            respond(tx, || handlers::handle_virtual_document(params, &self.world), LspResponse::VirtualDocument)?;
+                            respond(tx, || handlers::handle_virtual_document(params, &self.world), LspResponse::VirtualDocument)?
                         },
                         LspRequest::InputBoundaries(params) => {
-                            respond(tx, || handlers::handle_input_boundaries(params), LspResponse::InputBoundaries)?;
+                            respond(tx, || handlers::handle_input_boundaries(params), LspResponse::InputBoundaries)?
+                        },
+
+                        #[cfg(feature = "testing")]
+                        LspRequest::TestPanic => {
+                            respond(tx, || -> LspResult<()> { panic!("Test panic in a request handler") }, LspResponse::TestPanic)?
                         },
                     };
+
+                    if outcome == RequestOutcome::Panicked &&
+                        self.reported_request_panics.insert(request_kind)
+                    {
+                        report_request_panic(&self.client).await;
+                    }
                 },
             },
 
@@ -623,6 +774,9 @@ impl GlobalState {
                     );
                 }
             },
+
+            #[cfg(feature = "testing")]
+            Event::TestPanicMainLoop => unreachable!(),
         }
         lsp::log_info!("Finished handling event in {}ms", loop_tick.elapsed().as_millis());
 
@@ -679,6 +833,50 @@ impl GlobalState {
     }
 }
 
+/// Send via `request::ShowMessageRequest` not `notification::ShowMessage` so
+/// that we can ensure that the message has been received on the frontend side.
+/// We are about to shut the LSP down, and sending out a fire-and-forget
+/// notification often won't get sent out before shutdown occurs. The request
+/// returns control to us when the user acknowledges the message. It doesn't
+/// matter if that takes awhile because we shut down right after, and we've
+/// already set the session's `CrashFlag`. We do bound it with a 5 second
+/// timeout just in case the user ignores the message entirely, so we can still
+/// shutdown.
+async fn report_crash(client: &Client) {
+    let user_message = concat!(
+        "The R language server has crashed and has been disabled. ",
+        "Smart features such as completions will no longer work in this session. ",
+        "Please report this crash to https://github.com/posit-dev/positron/issues ",
+        "with full logs (see https://positron.posit.co/troubleshooting.html#python-and-r-logs)."
+    );
+    let request = client.send_request::<request::ShowMessageRequest>(ShowMessageRequestParams {
+        typ: MessageType::ERROR,
+        message: String::from(user_message),
+        actions: None,
+    });
+    match tokio::time::timeout(Duration::from_secs(5), request).await {
+        Ok(result) => {
+            result.log_err();
+        },
+        Err(_) => {
+            log::warn!("Timed out waiting for frontend to acknowledge LSP crash notification");
+        },
+    }
+}
+
+async fn report_request_panic(client: &Client) {
+    client
+        .show_message(
+            MessageType::ERROR,
+            concat!(
+                "An R language server feature encountered an internal error. ",
+                "The request failed, but the language server is still running. ",
+                "See the R Kernel and R Language Server logs for the panic and backtrace."
+            ),
+        )
+        .await;
+}
+
 /// Build the LSP's [`SourceHandler`], or `None` to disable source fetching
 fn source_handler(r_home: &Path) -> Option<Arc<dyn SourceHandler>> {
     // Also reported to the LSP output channel from `handle_initialized()`.
@@ -769,6 +967,19 @@ impl GlobalState {
     }
 }
 
+enum EventUnwind {
+    Cancelled,
+    Panicked(panic::PanicPayload),
+}
+
+fn classify_event_unwind(payload: panic::PanicPayload) -> EventUnwind {
+    if payload.is::<salsa::Cancelled>() {
+        EventUnwind::Cancelled
+    } else {
+        EventUnwind::Panicked(payload)
+    }
+}
+
 /// Run each [`ScanRequest`] on `pool`. Each job runs the pure-I/O
 /// [`ScanRequest::run`] and ships the [`ScanCompleted`] back to the main loop as
 /// [`Event::OakScanCompleted`], where the scheduler then applies it.
@@ -786,6 +997,12 @@ pub(super) fn dispatch_scan_requests(
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum RequestOutcome {
+    Handled,
+    Panicked,
+}
+
 /// Respond to a request from the LSP
 ///
 /// We receive requests from the LSP client with a response channel. Once we
@@ -808,29 +1025,69 @@ fn respond<T>(
     response_tx: TokioUnboundedSender<RequestResponse>,
     response: impl FnOnce() -> LspResult<T>,
     into_lsp_response: impl FnOnce(T) -> LspResponse,
-) -> anyhow::Result<()> {
-    let response = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(response)) {
-        Ok(Ok(t)) => RequestResponse::Result(Ok(into_lsp_response(t))),
-        Ok(Err(e)) => RequestResponse::Result(Err(e)),
-        Err(err) if err.downcast_ref::<salsa::Cancelled>().is_some() => {
+) -> anyhow::Result<RequestOutcome> {
+    let (response, outcome) = match panic::catch_unwind(Recovery::Always, || {
+        catch_cancellation(response)
+    }) {
+        Ok(Some(Ok(value))) => (
+            RequestResponse::Result(Ok(into_lsp_response(value))),
+            RequestOutcome::Handled,
+        ),
+        Ok(Some(Err(err))) => (RequestResponse::Result(Err(err)), RequestOutcome::Handled),
+        Ok(None) => {
             // A salsa write cancelled an oak query while the handler ran.
             // Report `ContentModified` so the client knows the content moved
             // under us and re-requests.
-            RequestResponse::Result(Err(LspError::JsonRpc(jsonrpc::Error::content_modified())))
+            (
+                RequestResponse::Result(Err(LspError::JsonRpc(jsonrpc::Error::content_modified()))),
+                RequestOutcome::Handled,
+            )
         },
-        Err(err) => {
-            // Set global crash flag to disable the LSP
-            LSP_HAS_CRASHED.store(true, Ordering::Release);
+        Err(payload) => {
+            // The panic hook emits the backtrace to the kernel logs. Mention
+            // the panic in the LSP log too for cross-reference.
+            let message = panic::message(&payload);
+            lsp::log_error!(
+                "Panic while handling request: {message}. \
+                 See the R kernel log for the full panic backtrace."
+            );
+            (
+                RequestResponse::Result(Err(LspError::Anyhow(anyhow!(
+                    "Panic while handling request: {message}"
+                )))),
+                RequestOutcome::Panicked,
+            )
+        },
+    };
 
-            let msg = panic_message(err.as_ref());
+    send_response(response_tx, response)?;
+    Ok(outcome)
+}
 
-            // This creates an uninformative backtrace that is reported in the
-            // LSP logs. Note that the relevant backtrace is the one created by
-            // our panic hook and reported via the _kernel_ logs.
-            RequestResponse::Crashed(anyhow!("Panic occurred while handling request: {msg}"))
-        },
+/// Run a handler that holds an exclusive world-state borrow.
+///
+/// We don't recover from panics in these handlers because they could leave
+/// leave the world state half-built. Let them reach `main_loop()`'s recovery
+/// boundary, which terminates the session.
+fn respond_exclusive<T>(
+    response_tx: TokioUnboundedSender<RequestResponse>,
+    response: impl FnOnce() -> LspResult<T>,
+    into_lsp_response: impl FnOnce(T) -> LspResponse,
+) -> anyhow::Result<RequestOutcome> {
+    let response = match catch_cancellation(response) {
+        Some(Ok(value)) => RequestResponse::Result(Ok(into_lsp_response(value))),
+        Some(Err(err)) => RequestResponse::Result(Err(err)),
+        None => RequestResponse::Result(Err(LspError::JsonRpc(jsonrpc::Error::content_modified()))),
     };
 
+    send_response(response_tx, response)?;
+    Ok(RequestOutcome::Handled)
+}
+
+fn send_response(
+    response_tx: TokioUnboundedSender<RequestResponse>,
+    response: RequestResponse,
+) -> anyhow::Result<()> {
     let out = match response {
         RequestResponse::Result(Ok(_)) => Ok(()),
         RequestResponse::Result(Err(ref error)) => {
@@ -841,9 +1098,6 @@ fn respond<T>(
             lsp::log_info!("Error while handling request:\n{error:?}");
             Ok(())
         },
-        RequestResponse::Crashed(ref error) => {
-            Err(anyhow!("Crashed while handling request:\n{error:?}"))
-        },
         RequestResponse::Disabled => Err(anyhow!("Received impossible `Disabled` response state")),
     };
 
@@ -889,6 +1143,7 @@ impl AuxiliaryState {
                 AuxiliaryEvent::PublishDiagnostics(publication) => {
                     self.publish_diagnostics(publication).await
                 },
+                AuxiliaryEvent::ReportBackgroundPanic => self.report_background_panic().await,
                 AuxiliaryEvent::Shutdown => break,
             }
         }
@@ -908,6 +1163,21 @@ impl AuxiliaryState {
         }
     }
 
+    async fn report_background_panic(&self) {
+        self.client
+            .show_message(
+                MessageType::ERROR,
+                String::from(
+                    "An R language server background task encountered an internal error. \
+                     Some smart features may be temporarily unavailable. \
+                     See https://positron.posit.co/troubleshooting.html#python-and-r-logs \
+                     for full logs and report the problem at \
+                     https://github.com/posit-dev/positron/issues.",
+                ),
+            )
+            .await;
+    }
+
     /// Publish diagnostics, skipping the client round-trip when the set is
     /// identical to what we last sent for that file. Only non-empty sets are
     /// remembered, and an absent entry counts as empty. So an empty result is
@@ -972,6 +1242,21 @@ fn send_auxiliary(event: AuxiliaryEvent) {
     })
 }
 
+fn report_background_panic() {
+    let Ok(auxiliary_event_tx) = AUXILIARY_EVENT_TX.read() else {
+        log::warn!("Can't lock auxiliary event sender to report a background panic");
+        return;
+    };
+    let Some(auxiliary_event_tx) = auxiliary_event_tx.as_ref() else {
+        log::warn!("Can't report a background panic before the LSP is initialized");
+        return;
+    };
+
+    if let Err(err) = auxiliary_event_tx.send(AuxiliaryEvent::ReportBackgroundPanic) {
+        log::warn!("LSP is shut down, can't report a background panic:\n{err:?}");
+    }
+}
+
 /// Initialise the auxiliary channel for unit tests that exercise LSP
 /// handlers calling `publish_diagnostics` / `log` / similar.
 ///
@@ -1058,14 +1343,66 @@ mod tests {
     use tower_lsp_server::jsonrpc;
     use url::Url;
 
+    use super::classify_event_unwind;
     use super::respond;
+    use super::send_auxiliary;
     use super::tokio_unbounded_channel;
+    use super::AuxiliaryEvent;
+    use super::AuxiliaryState;
+    use super::EventUnwind;
+    use super::LspServiceContext;
     use crate::lsp::backend::LspError;
     use crate::lsp::backend::LspResponse;
     use crate::lsp::backend::RequestResponse;
     use crate::lsp::state::WorldState;
+    use crate::lsp::tests::utils::client::TestClient;
     use crate::lsp::traits::url::UrlExt;
 
+    #[tokio::test]
+    async fn test_background_panic_is_reported_once_per_session() {
+        let client = TestClient::new(&[]).await;
+        let auxiliary = AuxiliaryState::new(client.client());
+        let auxiliary_loop = tokio::spawn(auxiliary.start());
+        let first_session = LspServiceContext::new();
+        let second_session = LspServiceContext::new();
+
+        first_session.report_background_panic();
+        first_session.report_background_panic();
+        second_session.report_background_panic();
+        send_auxiliary(AuxiliaryEvent::Shutdown);
+        auxiliary_loop.await.unwrap();
+
+        // The socket is FIFO, so a round-trip flushes the notifications the peer
+        // has not read yet.
+        let _ = client.client().configuration(vec![]).await;
+
+        let notifications = client.notifications();
+        assert_eq!(notifications.len(), 2);
+        assert_eq!(notifications[0].0, "window/showMessage");
+        assert_eq!(notifications[1].0, "window/showMessage");
+        let message = notifications[0].1["message"].as_str().unwrap();
+        assert!(message.contains("background task encountered an internal error"));
+    }
+
+    #[test]
+    fn test_salsa_cancellation_is_not_classified_as_event_panic() {
+        let payload = Box::new(salsa::Cancelled::PendingWrite);
+        let outcome = classify_event_unwind(payload);
+
+        assert!(matches!(outcome, EventUnwind::Cancelled));
+    }
+
+    #[test]
+    fn test_genuine_panic_is_classified_as_event_panic() {
+        let payload = Box::new(String::from("oh no"));
+        let outcome = classify_event_unwind(payload);
+
+        let EventUnwind::Panicked(payload) = outcome else {
+            panic!("Expected a panic");
+        };
+        assert_eq!(crate::panic::message(&payload), "oh no");
+    }
+
     /// A `salsa::Cancelled` re-raised out of a request handler (by `r_task`,
     /// after catching it on the R thread) must not crash the LSP. `respond`
     /// recognises the payload and answers `ContentModified` so the client
diff --git a/crates/ark/src/lsp/tests.rs b/crates/ark/src/lsp/tests.rs
index a2fc33e40..f75a39993 100644
--- a/crates/ark/src/lsp/tests.rs
+++ b/crates/ark/src/lsp/tests.rs
@@ -9,4 +9,4 @@ mod source_handler;
 mod sources;
 mod state;
 mod state_handlers;
-mod utils;
+pub(crate) mod utils;
diff --git a/crates/ark/src/lsp/tests/utils/client.rs b/crates/ark/src/lsp/tests/utils/client.rs
index 7d3af937b..8eb77c5c6 100644
--- a/crates/ark/src/lsp/tests/utils/client.rs
+++ b/crates/ark/src/lsp/tests/utils/client.rs
@@ -35,6 +35,7 @@ pub(crate) struct TestClient {
     client: Client,
     settings: Arc<Mutex<HashMap<String, Value>>>,
     requests: Arc<Mutex<Vec<String>>>,
+    notifications: Arc<Mutex<Vec<(String, Value)>>>,
 
     /// Aborts the peer on drop.
     _peer: JoinSet<()>,
@@ -68,18 +69,21 @@ impl TestClient {
                 .collect(),
         ));
         let requests = Arc::new(Mutex::new(Vec::new()));
+        let notifications = Arc::new(Mutex::new(Vec::new()));
 
         let mut peer = JoinSet::new();
         peer.spawn(answer_requests(
             socket,
             Arc::clone(&settings),
             Arc::clone(&requests),
+            Arc::clone(&notifications),
         ));
 
         Self {
             client,
             settings,
             requests,
+            notifications,
             _peer: peer,
         }
     }
@@ -98,11 +102,15 @@ impl TestClient {
             .insert(section.to_string(), value);
     }
 
-    /// Methods of the requests the peer has answered, in order. Notifications
-    /// aren't recorded.
+    /// Methods of the requests the peer has answered, in order.
     pub(crate) fn answered_requests(&self) -> Vec<String> {
         self.requests.lock().unwrap().clone()
     }
+
+    /// Methods and params of the notifications the server has sent, in order.
+    pub(crate) fn notifications(&self) -> Vec<(String, Value)> {
+        self.notifications.lock().unwrap().clone()
+    }
 }
 
 /// Answer every request the server sends until it drops its [`Client`].
@@ -110,6 +118,7 @@ async fn answer_requests(
     socket: ClientSocket,
     settings: Arc<Mutex<HashMap<String, Value>>>,
     requests: Arc<Mutex<Vec<String>>>,
+    notifications: Arc<Mutex<Vec<(String, Value)>>>,
 ) {
     let (mut incoming, mut outgoing) = socket.split();
 
@@ -119,6 +128,10 @@ async fn answer_requests(
         // A notification, such as `textDocument/publishDiagnostics`, carries no
         // id and gets no reply.
         let Some(id) = id else {
+            notifications
+                .lock()
+                .unwrap()
+                .push((method.to_string(), params.unwrap_or(Value::Null)));
             continue;
         };
 
diff --git a/crates/ark/src/lsp/tests/utils/mod.rs b/crates/ark/src/lsp/tests/utils/mod.rs
index b90f16ca1..81d60e1a2 100644
--- a/crates/ark/src/lsp/tests/utils/mod.rs
+++ b/crates/ark/src/lsp/tests/utils/mod.rs
@@ -1,4 +1,4 @@
-mod client;
+pub(crate) mod client;
 mod description_writer;
 mod events;
 mod namespace_writer;
diff --git a/crates/ark/src/main.rs b/crates/ark/src/main.rs
index 59410b097..6dd23ca1b 100644
--- a/crates/ark/src/main.rs
+++ b/crates/ark/src/main.rs
@@ -7,15 +7,14 @@
 
 #![allow(unused_unsafe)]
 
-use std::cell::Cell;
 use std::env;
 
 use amalthea::kernel;
 use amalthea::kernel_spec::KernelSpec;
 use anyhow::Context;
-use ark::console::catching_panics;
 use ark::console::SessionMode;
 use ark::logger;
+use ark::panic;
 use ark::repos::DefaultRepos;
 use ark::signals::initialize_signal_block;
 use ark::start::start_kernel;
@@ -23,13 +22,8 @@ use ark::traps::register_trap_handlers;
 use crossbeam::channel::unbounded;
 use harp::command::r_home_setup;
 use notify::Watcher;
-use stdext::panic_message;
 use stdext::unwrap;
 
-thread_local! {
-    pub static ON_R_THREAD: Cell<bool> = const { Cell::new(false) };
-}
-
 fn print_usage() {
     println!("Ark {}, an R Kernel.", ark::BUILD_VERSION);
     print!(
@@ -83,7 +77,7 @@ https://github.com/posit-dev/ark/blob/main/doc/configuration.md
 }
 
 fn main() -> anyhow::Result<()> {
-    ON_R_THREAD.set(true);
+    panic::mark_r_thread();
 
     // Block signals in this thread (and any child threads).
     initialize_signal_block();
@@ -381,60 +375,7 @@ fn main() -> anyhow::Result<()> {
         String::from("--no-restore-data"),
     ]);
 
-    // This causes panics on background threads to propagate on the main
-    // thread. If we don't propagate a background thread panic, the program
-    // keeps running in an unstable state as all communications with this
-    // thread will error out or panic.
-    // https://stackoverflow.com/questions/35988775/how-can-i-cause-a-panic-on-a-thread-to-immediately-end-the-main-thread
-    //
-    // A better way to manage panics on background threads would be to ensure
-    // that we join all spawned threads up to the main thread.
-    let old_hook = std::panic::take_hook();
-    std::panic::set_hook(Box::new(move |panic_info| {
-        let info = panic_info.payload();
-
-        let loc = if let Some(location) = panic_info.location() {
-            format!("In file '{}' at line {}:", location.file(), location.line(),)
-        } else {
-            String::from("No location information:")
-        };
-
-        let msg = panic_message(info);
-
-        // Top-level-exec and try-catch errors already contain a backtrace
-        // for the R thread so don't repeat it if we see one. Only perform
-        // this check on the R thread because we do want other threads'
-        // backtraces if the panic occurred elsewhere.
-        let trace = if ON_R_THREAD.get() && msg.contains("\n{R_BACKTRACE_HEADER}\n") {
-            String::new()
-        } else {
-            format!("Backtrace:\n{}", std::backtrace::Backtrace::force_capture())
-        };
-
-        log::error!("Panic! {loc} {msg}\n{trace}");
-
-        // `Console::with()` catches panics with `catch_unwind` in release
-        // builds. Return early so the catch handler can convert the panic
-        // to an `anyhow::Error`. The backtrace is already logged above.
-        if catching_panics() {
-            return;
-        }
-
-        // We don't want the threads managed by a Tokio runtime to `abort()` the
-        // process since their panics are caught and handled in other ways.
-        // This escape hatch is a hack that will also be activated by other
-        // Tokio contexts than just the LSP.
-        if tokio::runtime::Handle::try_current().is_ok() {
-            return;
-        }
-
-        // Give some time to flush log
-        log::logger().flush();
-        std::thread::sleep(std::time::Duration::from_millis(250));
-
-        old_hook(panic_info);
-        std::process::abort();
-    }));
+    panic::install();
 
     let Some(connection_file) = connection_file else {
         return Err(anyhow::anyhow!(
diff --git a/crates/ark/src/panic.rs b/crates/ark/src/panic.rs
new file mode 100644
index 000000000..d65c2a555
--- /dev/null
+++ b/crates/ark/src/panic.rs
@@ -0,0 +1,251 @@
+//
+// panic.rs
+//
+// Copyright (C) 2026 Posit Software, PBC. All rights reserved.
+//
+//
+
+//! Logs panics before `catch_unwind()` handles them. `catch_unwind()` prevents a
+//! recovered panic from aborting the process. Kept outside `main.rs` so tests can
+//! install it.
+
+use std::cell::Cell;
+use std::future::Future;
+use std::panic::AssertUnwindSafe;
+use std::task::Poll;
+
+use stdext::panic_message;
+
+pub(crate) type PanicPayload = Box<dyn std::any::Any + Send + 'static>;
+
+/// Install the global panic hook.
+///
+/// This causes panics on background threads to propagate on the main
+/// thread. If we don't propagate a background thread panic, the program
+/// keeps running in an unstable state as all communications with this
+/// thread will error out or panic.
+/// https://stackoverflow.com/questions/35988775/how-can-i-cause-a-panic-on-a-thread-to-immediately-end-the-main-thread
+pub fn install() {
+    let old_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(move |panic_info| {
+        let info = panic_info.payload();
+
+        let loc = if let Some(location) = panic_info.location() {
+            format!("In file '{}' at line {}:", location.file(), location.line(),)
+        } else {
+            String::from("No location information:")
+        };
+
+        let msg = panic_message(info);
+
+        // Top-level-exec and try-catch errors already contain a backtrace
+        // for the R thread so don't repeat it if we see one. Only perform
+        // this check on the R thread because we do want other threads'
+        // backtraces if the panic occurred elsewhere.
+        let trace = if on_r_thread() && msg.contains("\n{R_BACKTRACE_HEADER}\n") {
+            String::new()
+        } else {
+            format!("Backtrace:\n{}", std::backtrace::Backtrace::force_capture())
+        };
+
+        log::error!("Panic! {loc} {msg}\n{trace}");
+
+        // A boundary has a `catch_unwind()` waiting for this panic. The
+        // backtrace is already logged above.
+        if recovers_panic() {
+            // Return and let the panic continue unwinding to the catch site
+            return;
+        }
+
+        // A current Tokio handle may be unrelated to the LSP, but Tokio captures task
+        // panics for its caller to handle.
+        if tokio::runtime::Handle::try_current().is_ok() {
+            return;
+        }
+
+        // Leave time for the log sink to write the flushed panic before aborting.
+        log::logger().flush();
+        std::thread::sleep(std::time::Duration::from_millis(250));
+
+        old_hook(panic_info);
+        std::process::abort();
+    }));
+}
+
+thread_local! {
+    static ON_R_THREAD: Cell<bool> = const { Cell::new(false) };
+}
+
+/// Mark the calling thread as the main R thread, so the panic hook knows
+/// whether an R backtrace is already present in the panic message.
+pub fn mark_r_thread() {
+    ON_R_THREAD.set(true);
+}
+
+fn on_r_thread() -> bool {
+    ON_R_THREAD.get()
+}
+
+/// Which builds a boundary recovers panics in.
+#[derive(Clone, Copy)]
+pub(crate) enum Recovery {
+    /// Recover in every build.
+    Always,
+    /// Recover in release builds only. A panic in an R callback aborts during
+    /// development instead of surfacing as an R error.
+    ReleaseOnly,
+}
+
+thread_local! {
+    static BOUNDARY: Cell<Option<Recovery>> = const { Cell::new(None) };
+}
+
+/// Whether a `catch_unwind()` boundary is waiting to recover this panic.
+pub(crate) fn recovers_panic() -> bool {
+    match BOUNDARY.get() {
+        None => false,
+        Some(Recovery::Always) => true,
+        Some(Recovery::ReleaseOnly) => !cfg!(debug_assertions),
+    }
+}
+
+/// Runs `f` inside a `catch_unwind()` boundary. `Err` preserves the panic payload so
+/// callers can distinguish control-flow unwinds such as `salsa::Cancelled` from genuine
+/// panics. Unwind safety is asserted on the caller's behalf.
+pub(crate) fn catch_unwind<T>(
+    recovery: Recovery,
+    f: impl FnOnce() -> T,
+) -> Result<T, PanicPayload> {
+    let _boundary = catch_boundary(recovery);
+    std::panic::catch_unwind(std::panic::AssertUnwindSafe(f))
+}
+
+/// Recover panics while polling a future. Enter the recovery boundary for each poll so
+/// unrelated work on the polling thread cannot inherit it. Preserve the panic payload so
+/// the caller can distinguish control-flow unwinds from genuine panics.
+pub(crate) async fn catch_unwind_async<T>(
+    recovery: Recovery,
+    future: impl Future<Output = T>,
+) -> Result<T, PanicPayload> {
+    let mut future = Box::pin(future);
+
+    std::future::poll_fn(move |cx| {
+        let _boundary = catch_boundary(recovery);
+
+        match std::panic::catch_unwind(AssertUnwindSafe(|| future.as_mut().poll(cx))) {
+            Ok(Poll::Pending) => Poll::Pending,
+            Ok(Poll::Ready(value)) => Poll::Ready(Ok(value)),
+            Err(payload) => Poll::Ready(Err(payload)),
+        }
+    })
+    .await
+}
+
+pub(crate) fn message(payload: &PanicPayload) -> String {
+    panic_message(payload.as_ref())
+}
+
+/// Guard that preserves a `catch_unwind()` recovery boundary for the panic hook.
+///
+/// Restores the preceding flag in `Drop::drop()` so a nested boundary cannot disable an
+/// outer boundary.
+struct CatchBoundary {
+    previous: Option<Recovery>,
+}
+
+impl Drop for CatchBoundary {
+    fn drop(&mut self) {
+        BOUNDARY.set(self.previous);
+    }
+}
+
+/// Prevent the hook from aborting a panic handled by `catch_unwind()`.
+fn catch_boundary(recovery: Recovery) -> CatchBoundary {
+    let previous = BOUNDARY.get();
+    BOUNDARY.set(Some(recovery));
+    CatchBoundary { previous }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_recovers_panic_false_without_boundary() {
+        assert!(!recovers_panic());
+    }
+
+    #[test]
+    fn test_recovers_panic_true_inside_always_boundary() {
+        let _boundary = catch_boundary(Recovery::Always);
+        assert!(recovers_panic());
+    }
+
+    #[test]
+    fn test_nested_always_boundary_recovers_inside_release_only() {
+        let _outer = catch_boundary(Recovery::ReleaseOnly);
+        {
+            let _inner = catch_boundary(Recovery::Always);
+            assert!(recovers_panic());
+        }
+        assert_eq!(recovers_panic(), !cfg!(debug_assertions));
+    }
+
+    #[test]
+    fn test_dropping_boundary_leaves_none() {
+        {
+            let _boundary = catch_boundary(Recovery::Always);
+            assert!(recovers_panic());
+        }
+        assert!(!recovers_panic());
+    }
+
+    #[test]
+    fn test_catch_unwind_passes_through_ok() {
+        let result = catch_unwind(Recovery::Always, || 1 + 1);
+        let Ok(value) = result else {
+            panic!("Expected a value");
+        };
+        assert_eq!(value, 2);
+    }
+
+    #[test]
+    fn test_catch_unwind_preserves_panic_payload() {
+        let result = catch_unwind(Recovery::Always, || panic!("oh no"));
+        let Err(payload) = result else {
+            panic!("Expected a panic payload");
+        };
+        assert_eq!(message(&payload), "oh no");
+    }
+
+    #[tokio::test]
+    async fn test_catch_unwind_async_passes_through_ok() {
+        let result = catch_unwind_async(Recovery::Always, async { 1 + 1 }).await;
+        let Ok(value) = result else {
+            panic!("Expected a value");
+        };
+        assert_eq!(value, 2);
+    }
+
+    #[tokio::test]
+    async fn test_catch_unwind_async_preserves_panic_payload() {
+        let result = catch_unwind_async(Recovery::Always, async { panic!("oh no") }).await;
+        let Err(payload) = result else {
+            panic!("Expected a panic payload");
+        };
+        assert_eq!(message(&payload), "oh no");
+    }
+
+    #[tokio::test]
+    async fn test_catch_unwind_async_preserves_salsa_cancellation() {
+        let result = catch_unwind_async(Recovery::Always, async {
+            std::panic::resume_unwind(Box::new(salsa::Cancelled::PendingWrite))
+        })
+        .await;
+
+        let Err(payload) = result else {
+            panic!("Expected a cancellation payload");
+        };
+        assert!(payload.is::<salsa::Cancelled>());
+    }
+}
diff --git a/crates/ark/src/r_task.rs b/crates/ark/src/r_task.rs
index 93321f3b0..9d14883ab 100644
--- a/crates/ark/src/r_task.rs
+++ b/crates/ark/src/r_task.rs
@@ -25,6 +25,8 @@ use uuid::Uuid;
 use crate::console::Console;
 use crate::console::ConsoleOutputCapture;
 use crate::fixtures::r_test_init;
+use crate::panic;
+use crate::panic::Recovery;
 
 /// Task channels for idle-time tasks (top-level only)
 static IDLE_TASKS: LazyLock<TaskChannels> = LazyLock::new(TaskChannels::new);
@@ -296,16 +298,24 @@ where
     // Instead of scoping the task with a thread join, we send it on the R
     // thread and block the thread until a completion channel wakes us up.
 
-    // Stores the outcome of `f`. We catch any unwind on the R thread instead of
-    // letting it escape the closure: the closure runs inside `r_sandbox`'s
-    // `try_catch`, and a Rust unwind crossing those C frames is UB. The payload
-    // is ferried back and re-raised below, on the calling thread.
+    // Stores the outcome of `f`. If the caller can recover panics, catch any unwind
+    // on the R thread instead of letting it escape through `r_sandbox`'s C frames.
+    // The payload is ferried back and re-raised below, on the calling thread.
+    //
+    // Without a recovering caller, don't install a boundary on the R thread. The
+    // production panic hook must abort rather than let the process continue after
+    // losing a thread.
+    let caller_recovers_panic = panic::recovers_panic();
     let result: SharedOption<std::thread::Result<T>> = SharedOption::default();
 
     {
         let result = Arc::clone(&result);
         let closure = move || {
-            let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
+            let caught = if caller_recovers_panic {
+                panic::catch_unwind(Recovery::Always, f)
+            } else {
+                Ok(f())
+            };
             *result.lock().unwrap() = Some(caught);
         };
 
@@ -364,9 +374,10 @@ where
         }
     }
 
-    // The closure ran to completion: it caught its own unwind, and an R-level
-    // error would have panicked above. Re-raise on this thread any panic the
-    // closure caught on the R thread.
+    // The closure ran to completion, and an R-level error would have panicked above.
+    // If the caller has a recovery boundary, re-raise any Rust panic caught on the
+    // R thread. If there is no recovery boundary, the production panic hook aborts
+    // before reaching here.
     let caught = result.lock().unwrap().take().unwrap();
     match caught {
         Ok(value) => value,
diff --git a/crates/ark/tests/integration/lsp.rs b/crates/ark/tests/integration/lsp.rs
index fe77ec2e3..fc764f538 100644
--- a/crates/ark/tests/integration/lsp.rs
+++ b/crates/ark/tests/integration/lsp.rs
@@ -24,16 +24,11 @@ fn test_lsp_init() {
     assert!(lsp.server_capabilities().completion_provider.is_some());
 }
 
-// Reproduces https://github.com/posit-dev/ark/issues/1361: an abrupt client
-// disconnect (TCP reset, e.g. the extension host being recycled) hits an
-// `unreachable!()` in our pinned tower-lsp fork's transport and panics the
-// `ark-lsp` thread.
+// An abrupt client disconnect must not panic the `ark-lsp` thread.
 #[test]
 fn test_lsp_survives_abrupt_disconnect() {
-    // The panic happens on the `ark-lsp` thread, so a plain `#[should_panic]`
-    // or `catch_unwind` on this (the main test) thread can't see it. Install
-    // a hook that records panics by thread name instead, and chain to the
-    // previous hook so panic output still gets printed.
+    // Capture `ark-lsp` panics because `catch_unwind()` on this test thread
+    // cannot observe them. Chain the prior hook so the panic remains in test output.
     let panic_message: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
     let panic_message_hook = Arc::clone(&panic_message);
     let previous_hook = std::panic::take_hook();
@@ -49,34 +44,147 @@ fn test_lsp_survives_abrupt_disconnect() {
 
     lsp.disconnect_abruptly();
 
-    // Give the LSP thread a moment to observe the reset.
+    // Wait for the `ark-lsp` thread to observe the reset.
     std::thread::sleep(Duration::from_millis(200));
 
     if let Some(message) = panic_message.lock().unwrap().take() {
         panic!("`ark-lsp` thread panicked on abrupt disconnect: {message}");
     }
 
-    // The LSP should still be usable after a client disconnects abruptly.
     let lsp2 = frontend.start_lsp();
     assert!(lsp2.server_capabilities().completion_provider.is_some());
 }
 
-// Reproduces https://github.com/ebkalderon/tower-lsp/issues/399 and #424:
-// some clients send `exit` but never close their own end of the connection
-// afterward. The server must hang up on its own rather than waiting forever
-// for more input that will never arrive.
+// `exit` must close the server side even if the client socket remains open.
 #[test]
 fn test_lsp_exits_promptly_after_exit_without_client_close() {
     let frontend = DummyArkFrontend::lock();
     let mut lsp = frontend.start_lsp();
 
-    // Sends `shutdown`/`exit` but deliberately leaves our end of the socket
-    // open, unlike the client's normal teardown on `Drop`.
+    // `shutdown()` sends `shutdown`/`exit` without closing the client socket.
     lsp.shutdown();
 
     lsp.expect_server_closes_connection(Duration::from_secs(5));
 }
 
+// A panicking request handler must return an error without ending the LSP session.
+#[test]
+fn test_lsp_panicking_request_is_task_local() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    lsp.allow_log_message("Panic while handling request");
+
+    let message = lsp.send_request_expect_error("ark/testPanic", json!({}));
+    assert!(message.contains("Panic while handling request"));
+
+    let toast = lsp.recv_show_message();
+    assert_eq!(
+        toast,
+        concat!(
+            "An R language server feature encountered an internal error. ",
+            "The request failed, but the language server is still running. ",
+            "See the R Kernel and R Language Server logs for the panic and backtrace."
+        )
+    );
+
+    // The same handler still logs and returns an error, but doesn't repeat its toast.
+    let message = lsp.send_request_expect_error("ark/testPanic", json!({}));
+    assert!(message.contains("Panic while handling request"));
+
+    let uri = lsp.open_document("test_panic_request.R", "x <- 1\n");
+    lsp.completions(&uri, 0, 0);
+    assert!(lsp.show_messages().is_empty());
+}
+
+// A notification handler panic must show a crash dialog before closing the LSP connection.
+#[test]
+fn test_lsp_panicking_notification_ends_session() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    // Allow the expected panic log because its auxiliary loop can deliver it
+    // before or after the crash dialog.
+    lsp.allow_log_message("Panic while handling event");
+
+    lsp.send_notification("ark/testPanicNotification", json!({}));
+
+    // Read `window/showMessageRequest` first because shutdown races it.
+    lsp.recv_server_request("window/showMessageRequest");
+    lsp.expect_server_closes_connection(Duration::from_secs(5));
+
+    // Skip `shutdown()` because the server has already closed the connection.
+    lsp.disconnect_abruptly();
+}
+
+// Salsa cancellation is control flow, not a crash. Its typed unwind payload must
+// survive the trip through `r_task()` and the async event boundary.
+#[test]
+fn test_lsp_cancellation_across_r_task_keeps_running() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    lsp.send_notification("ark/testCancelRTask", json!({}));
+
+    let uri = lsp.open_document("test_cancel_r_task.R", "x <- 1\n");
+    lsp.completions(&uri, 0, 0);
+    assert!(lsp.show_messages().is_empty());
+}
+
+// A genuine panic raised in `r_task()` must return to the caller's recovery
+// boundary rather than aborting the R process.
+#[test]
+fn test_lsp_panic_across_r_task_ends_session() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    lsp.allow_log_message("Panic while handling event");
+    lsp.send_notification("ark/testPanicRTask", json!({}));
+
+    lsp.recv_server_request("window/showMessageRequest");
+    lsp.expect_server_closes_connection(Duration::from_secs(5));
+    lsp.disconnect_abruptly();
+}
+
+// A panic outside the per-event boundary must still show the crash dialog before shutdown.
+#[test]
+fn test_lsp_panicking_main_loop_reports_crash() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    lsp.allow_log_message("Panic in the main loop");
+    lsp.send_notification("ark/testPanicMainLoop", json!({}));
+
+    lsp.recv_server_request("window/showMessageRequest");
+    lsp.expect_server_closes_connection(Duration::from_secs(5));
+    lsp.disconnect_abruptly();
+}
+
+// A crash is scoped to the session that crashed, so the next one must serve
+// requests rather than answer every one of them with `Disabled`.
+#[test]
+fn test_lsp_reconnect_after_crash_is_functional() {
+    let frontend = DummyArkFrontend::lock();
+    let mut lsp = frontend.start_lsp();
+
+    lsp.allow_log_message("Panic while handling event");
+    lsp.send_notification("ark/testPanicNotification", json!({}));
+
+    lsp.recv_server_request("window/showMessageRequest");
+    lsp.expect_server_closes_connection(Duration::from_secs(5));
+    lsp.disconnect_abruptly();
+
+    let mut lsp = frontend.start_lsp();
+    assert!(lsp.server_capabilities().completion_provider.is_some());
+
+    let uri = lsp.open_document("reconnect_after_crash.R", "pas");
+    let items = lsp.completions(&uri, 0, 3);
+    let labels: Vec<&str> = items.iter().map(|item| item.label.as_str()).collect();
+    assert!(labels.contains(&"paste"));
+
+    assert!(lsp.show_messages().is_empty());
+}
+
 // The two cases below test errors that don't depend on the rename
 // implementation's resolution capabilities. New-name validation always
 // applies (R language constraints), so these tests stay valid once
diff --git a/crates/ark_test/src/lsp_client.rs b/crates/ark_test/src/lsp_client.rs
index 41ac8bbfa..44c7c52aa 100644
--- a/crates/ark_test/src/lsp_client.rs
+++ b/crates/ark_test/src/lsp_client.rs
@@ -36,6 +36,10 @@ pub struct LspClient {
     server_capabilities: Option<lsp_types::ServerCapabilities>,
     /// Buffered diagnostics notifications, keyed by document URI
     diagnostics: std::collections::HashMap<lsp_types::Uri, Vec<lsp_types::Diagnostic>>,
+    /// Buffered `window/showMessage` notifications.
+    show_messages: Vec<String>,
+    /// Expected error and warning log message substrings.
+    allowed_log_messages: Vec<String>,
     /// Set by `disconnect_abruptly()` so `Drop` skips the graceful `shutdown`/`exit` sequence
     killed: bool,
 }
@@ -58,10 +62,42 @@ impl LspClient {
             open_documents: Vec::new(),
             server_capabilities: None,
             diagnostics: std::collections::HashMap::new(),
+            show_messages: Vec::new(),
+            allowed_log_messages: Vec::new(),
             killed: false,
         })
     }
 
+    /// Allow an expected server error or warning without hiding unexpected logs.
+    /// Only messages containing `substring` are ignored.
+    pub fn allow_log_message(&mut self, substring: &str) {
+        self.allowed_log_messages.push(substring.to_string());
+    }
+
+    /// Receive the next `window/showMessage` notification. Buffers diagnostics.
+    #[track_caller]
+    pub fn recv_show_message(&mut self) -> String {
+        loop {
+            if !self.show_messages.is_empty() {
+                return self.show_messages.remove(0);
+            }
+
+            match self.recv_any() {
+                LspMessage::Notification { diagnostics } => {
+                    if let Some(params) = diagnostics {
+                        self.diagnostics.insert(params.uri, params.diagnostics);
+                    }
+                },
+                other => panic!("Expected `window/showMessage`, got: {other:?}"),
+            }
+        }
+    }
+
+    /// Buffered `window/showMessage` notifications consumed while awaiting other messages.
+    pub fn show_messages(&self) -> &[String] {
+        &self.show_messages
+    }
+
     /// Sever the connection with a TCP reset instead of a graceful close.
     ///
     /// A zero `SO_LINGER` makes the close abortive, so the socket sends a `RST`
@@ -399,7 +435,7 @@ impl LspClient {
     ///
     /// Skips benign server notifications. Panics on unexpected messages.
     #[track_caller]
-    fn recv_server_request(&mut self, expected_method: &str) {
+    pub fn recv_server_request(&mut self, expected_method: &str) {
         loop {
             match self.recv_any() {
                 LspMessage::ServerRequest { id, method, params } => {
@@ -481,7 +517,7 @@ impl LspClient {
             },
 
             (false, true, false) => {
-                let diagnostics = Self::check_server_notification(&message);
+                let diagnostics = self.check_server_notification(&message);
                 LspMessage::Notification { diagnostics }
             },
 
@@ -491,6 +527,7 @@ impl LspClient {
 
     /// Check a server notification, returning parsed diagnostics if applicable.
     fn check_server_notification(
+        &mut self,
         message: &serde_json::Map<String, Value>,
     ) -> Option<lsp_types::PublishDiagnosticsParams> {
         let method = message["method"].as_str().unwrap_or("unknown");
@@ -505,10 +542,23 @@ impl LspClient {
                     let text = message["params"]["message"]
                         .as_str()
                         .unwrap_or("(no message)");
-                    panic!("LSP server {level}: {text}");
+                    if !self
+                        .allowed_log_messages
+                        .iter()
+                        .any(|allowed| text.contains(allowed))
+                    {
+                        panic!("LSP server {level}: {text}");
+                    }
                 }
                 None
             },
+            "window/showMessage" => {
+                let text = message["params"]["message"]
+                    .as_str()
+                    .unwrap_or("(no message)");
+                self.show_messages.push(text.to_string());
+                None
+            },
             "textDocument/publishDiagnostics" => {
                 let params: lsp_types::PublishDiagnosticsParams =
                     serde_json::from_value(message["params"].clone())