Review Console reference management

Cargo.toml

@@ -12,6 +12,7 @@ resolver = "2"
 members = [
     "crates/amalthea",
     "crates/ark",
+    "crates/ark/ark_macros",
     "crates/echo",
     "crates/harp",
     "crates/libr",
@@ -38,6 +39,7 @@ aether_syntax = { git = "https://github.com/posit-dev/air", package = "air_r_syn
 amalthea = { path = "crates/amalthea" }
 anyhow = "1.0.100"
 ark = { path = "crates/ark" }
+ark_macros = { path = "crates/ark/ark_macros" }
 ark_test = { path = "crates/ark_test" }
 assert_matches = "1.5.0"
 async-trait = "0.1.66"

crates/amalthea/src/comm/event.rs

@@ -13,19 +13,17 @@ use crate::socket::comm::CommSocket;
 
 /// Comm events sent to the frontend via Shell.
 pub enum CommEvent {
-    /// A new Comm was opened
-    Opened(CommSocket, Value),
+    /// A new Comm was opened. The optional `Sender` is a synchronisation
+    /// barrier: if provided, Shell signals it after processing the open
+    /// (sending `comm_open` on IOPub). The caller blocks on the paired
+    /// receiver to guarantee that the `comm_open` message has been sent
+    /// before any subsequent messages.
+    Opened(CommSocket, Value, Option<Sender<()>>),
 
     /// A message was received on a Comm; the first value is the comm ID, and the
     /// second value is the message.
     Message(String, CommMsg),
 
     /// A Comm was closed
     Closed(String),
-
-    /// Synchronisation barrier. Shell signals the sender after processing all
-    /// preceding events. The caller blocks on the paired receiver to guarantee
-    /// that earlier events (e.g. `Opened`) have been fully handled before
-    /// continuing.
-    Barrier(Sender<()>),
 }

crates/amalthea/src/language/shell_handler.rs

@@ -26,10 +26,10 @@ use crate::wire::kernel_info_request::KernelInfoRequest;
 use crate::wire::originator::Originator;
 
 /// Result of a `handle_comm_msg` or `handle_comm_close` call on the
-/// `ShellHandler`. `Handled` means the kernel processed the message
-/// synchronously (blocking Shell until done). `NotHandled` means amalthea
-/// should fall back to the historical `incoming_tx` path. This fallback is
-/// temporary until all comms are migrated to the blocking path.
+/// `ShellHandler`. `Handled` means the kernel dispatched the message
+/// (possibly asynchronously via a completion receiver). `NotHandled` means
+/// amalthea should fall back to the historical `incoming_tx` path. This
+/// fallback is temporary until all comms are migrated to the new path.
 pub enum CommHandled {
     Handled,
     NotHandled,
@@ -81,26 +81,30 @@ pub trait ShellHandler: Send {
     /// Docs: https://jupyter-client.readthedocs.io/en/stable/messaging.html#history
     async fn handle_history_request(&self, req: &HistoryRequest) -> crate::Result<HistoryReply>;
 
-    /// Handles a request to open a comm.
+    /// Handle a request to open a comm.
     ///
-    /// https://jupyter-client.readthedocs.io/en/stable/messaging.html#opening-a-comm
+    /// Returns `(true, Some(receiver))` if the comm was opened and the handler
+    /// was dispatched asynchronously. Shell will select-loop on the receiver
+    /// and `comm_event_rx` to drain comm events while the handler runs.
     ///
-    /// Returns true if the handler handled the request (and opened the comm), false if it did not.
-    ///
-    /// * `target` - The target name of the comm, such as `positron.variables`
-    /// * `comm` - The comm channel to use to communicate with the frontend
-    /// * `data` - The `data` payload from the `comm_open` message
-    async fn handle_comm_open(
-        &self,
+    /// Returns `(true, None)` if the comm was opened synchronously.
+    /// Returns `(false, None)` if the comm was not handled.
+    fn handle_comm_open(
+        &mut self,
         target: Comm,
         comm: CommSocket,
         data: serde_json::Value,
-    ) -> crate::Result<bool>;
+    ) -> crate::Result<(bool, Option<Receiver<()>>)>;
 
-    /// Handle an incoming comm message (RPC or data). Return
-    /// `CommHandled::Handled` if the message was processed, or
-    /// `CommHandled::NotHandled` to fall back to the existing
-    /// `incoming_tx` path.
+    /// Handle an incoming comm message (RPC or data).
+    ///
+    /// Returns `(CommHandled::Handled, Some(receiver))` if the message was
+    /// dispatched to the R thread. Shell will select-loop on the receiver
+    /// and `comm_event_rx` to drain comm events (e.g. barriers from
+    /// `comm_open_backend`) while the handler runs.
+    ///
+    /// Returns `(CommHandled::NotHandled, None)` to fall back to the
+    /// existing `incoming_tx` path.
     ///
     /// * `comm_id` - The comm's unique identifier
     /// * `comm_name` - The comm's target name (e.g. `"positron.dataExplorer"`)
@@ -113,21 +117,22 @@ pub trait ShellHandler: Send {
         _comm_name: &str,
         _msg: CommMsg,
         _originator: Originator,
-    ) -> crate::Result<CommHandled> {
-        Ok(CommHandled::NotHandled)
+    ) -> crate::Result<(CommHandled, Option<Receiver<()>>)> {
+        Ok((CommHandled::NotHandled, None))
     }
 
-    /// Handle a comm close. Return `CommHandled::Handled` if the close
-    /// was processed, or `CommHandled::NotHandled` to fall back to the
-    /// existing `incoming_tx` path.
+    /// Handle a comm close.
+    ///
+    /// Same pattern as `handle_comm_msg`: returns a completion receiver
+    /// so Shell can drain comm events while the handler runs.
     ///
     /// * `comm_id` - The comm's unique identifier
     /// * `comm_name` - The comm's target name
     fn handle_comm_close(
         &mut self,
         _comm_id: &str,
         _comm_name: &str,
-    ) -> crate::Result<CommHandled> {
-        Ok(CommHandled::NotHandled)
+    ) -> crate::Result<(CommHandled, Option<Receiver<()>>)> {
+        Ok((CommHandled::NotHandled, None))
     }
 }