Attempt at fixing flakiness

```mac count=2
failures:
    test_kernel_request_priority_over_idle_tasks

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 0
filtered out; finished in 0.70s

stderr ───

thread 'test_kernel_request_priority_over_idle_tasks' (109970) panicked
at crates/ark/tests/data_explorer_priority.rs:72:14:
Expected Idle status, got CommMsg(JupyterMessage { zmq_identities: [],
header: JupyterHeader { msg_id: "243cf539-e28a-487d-9fc7-43e72f3d55af",
session: "71034bea-15a2-4bdf-906f-48acf4ced1e3", username: "kernel",
date: "2026-03-16T14:20:32.871678+00:00", msg_type: "comm_msg", version:
"5.3" }, parent_header: Some(JupyterHeader { msg_id:
"8d15e26e-89bb-4791-b560-ec8907f0eda0", session: "test-session",
username: "test-user", date: "2026-03-16T14:20:32.860651+00:00",
msg_type: "comm_msg", version: "5.3" }), content: CommWireMsg { comm_id:
"3ee9adb6-b60e-4b5d-b3e5-ce401e9d6bfa", data: Object {"method":
String("GetSchemaReply"), "result": Object {"columns": Array [Object
{"column_name": String("a"), "column_label": Null, "column_index":
Number(0), "type_name": String("int"), "type_display":
String("integer"), "description": Null, "children": Null, "precision":
Null, "scale": Null, "timezone": Null, "type_size": Null}]}} } })
note: run with `RUST_BACKTRACE=1` environment variable to display a
backtrace
```

Author: lionel-

crates/ark/src/r_task.rs

@@ -12,8 +12,6 @@ use std::sync::LazyLock;
 use std::sync::Mutex;
 use std::time::Duration;
 
-use amalthea::comm::comm_channel::CommMsg;
-use amalthea::wire::header::JupyterHeader;
 use crossbeam::channel::bounded;
 use crossbeam::channel::unbounded;
 use crossbeam::channel::Receiver;
@@ -24,7 +22,6 @@ use uuid::Uuid;
 use crate::console::Console;
 use crate::console::ConsoleOutputCapture;
 use crate::fixtures::r_test_init;
-use crate::request::KernelRequest;
 
 /// Task channels for interrupt-time tasks
 static INTERRUPT_TASKS: LazyLock<TaskChannels> = LazyLock::new(TaskChannels::new);
@@ -457,22 +454,6 @@ pub(crate) fn spawn(task: RTask) {
 static TEST_PENDING_TASK_TX: Mutex<Option<futures::channel::oneshot::Sender<()>>> =
     Mutex::new(None);
 
-/// Clone of the kernel-request channel sender, stored at startup when
-/// `IS_TESTING` is true. Allows R-callable test helpers (e.g.
-/// `ps_test_send_kernel_comm_request`) to enqueue kernel requests directly
-/// from the R thread, bypassing the Shell thread. This is used to create
-/// deterministic contention between kernel requests and idle tasks in the
-/// event loop without any race window.
-#[cfg(debug_assertions)]
-static TEST_KERNEL_REQUEST_TX: Mutex<Option<Sender<KernelRequest>>> = Mutex::new(None);
-
-/// Store a clone of `kernel_request_tx` so test helpers can enqueue kernel
-/// requests directly from the R thread.
-#[cfg(debug_assertions)]
-pub(crate) fn set_test_kernel_request_tx(tx: Sender<KernelRequest>) {
-    *TEST_KERNEL_REQUEST_TX.lock().unwrap() = Some(tx);
-}
-
 #[cfg(debug_assertions)]
 #[harp::register]
 unsafe extern "C-unwind" fn ps_test_spawn_pending_task() -> anyhow::Result<SEXP> {
@@ -538,51 +519,3 @@ unsafe extern "C-unwind" fn ps_test_spawn_sleeping_idle_tasks(
 
     Ok(libr::R_NilValue)
 }
-
-/// Send a comm RPC request directly onto the kernel-request channel from
-/// the R thread. This bypasses the Shell thread entirely, so the request
-/// is already pending when R enters the event loop.
-///
-/// Used by integration tests to create deterministic contention between
-/// kernel requests and idle tasks, without any race window.
-#[cfg(debug_assertions)]
-#[harp::register]
-unsafe extern "C-unwind" fn ps_test_send_kernel_comm_request(
-    comm_id: SEXP,
-    data: SEXP,
-) -> anyhow::Result<SEXP> {
-    stdext::assert_testing();
-
-    let comm_id: String = harp::RObject::view(comm_id).try_into()?;
-    let data: String = harp::RObject::view(data).try_into()?;
-    let json: serde_json::Value = serde_json::from_str(&data)?;
-
-    let msg = CommMsg::Rpc {
-        id: String::from("test-rpc"),
-        parent_header: JupyterHeader::create(
-            String::from("comm_msg"),
-            String::from("test-session"),
-            String::from("test-user"),
-        ),
-        data: json,
-    };
-
-    // We don't wait on `done_rx`. When the event loop processes the
-    // request it will send on `done_tx`, which harmlessly fails since the
-    // receiver is already dropped.
-    let (done_tx, _done_rx) = bounded(0);
-
-    let tx = TEST_KERNEL_REQUEST_TX.lock().unwrap();
-    let tx = tx
-        .as_ref()
-        .ok_or_else(|| anyhow::anyhow!("kernel_request_tx not set; is the kernel running?"))?;
-
-    tx.try_send(KernelRequest::CommMsg {
-        comm_id,
-        msg,
-        done_tx,
-    })
-    .map_err(|err| anyhow::anyhow!("Failed to send kernel request: {err}"))?;
-
-    Ok(libr::R_NilValue)
-}

crates/ark/src/start.rs

@@ -59,11 +59,6 @@ pub fn start_kernel(
     let (r_request_tx, r_request_rx) = bounded::<RRequest>(1);
     let (kernel_request_tx, kernel_request_rx) = bounded::<KernelRequest>(1);
 
-    #[cfg(debug_assertions)]
-    if stdext::IS_TESTING {
-        crate::r_task::set_test_kernel_request_tx(kernel_request_tx.clone());
-    }
-
     // Async communication channel with the R thread (Console)
     let (console_notification_tx, console_notification_rx) =
         tokio::sync::mpsc::unbounded_channel::<ConsoleNotification>();

crates/ark/tests/data_explorer_priority.rs

@@ -21,21 +21,20 @@ use ark_test::DummyArkFrontend;
 /// Verify that Shell requests (`get_schema`) are prioritized over idle tasks.
 ///
 /// 1. Open a data explorer so we have a comm to send RPCs on.
-/// 2. In a single execute request, spawn 5 idle tasks that each sleep for
-///    200ms **and** enqueue a `get_schema` kernel request directly on the
-///    kernel-request channel. Because both happen inside the same execute
-///    window, the kernel request is already pending when R returns to the
-///    event loop — no race.
-/// 3. Assert that the reply arrives in well under a single sleep duration.
-///    With the priority fix the kernel request is serviced before any
-///    sleeping idle task runs, so the reply is near-instant. Without the
-///    fix `select` picks randomly, so at least one 200ms sleeper runs
-///    first.
+/// 2. Spawn 5 idle tasks that each sleep for 200ms.
+/// 3. After the execute request completes, send `get_schema` through Shell.
+///    Shell dispatches a `KernelRequest` to the R thread and blocks until
+///    it's processed, so the ordering on IOPub is deterministic:
+///    Busy -> CommMsg -> Idle.
+/// 4. With the priority fix, R processes the kernel request after finishing
+///    at most one idle task (~200ms). Without the fix, `select` picks
+///    randomly among ready channels, so multiple idle tasks could run
+///    first (~600ms+).
 #[test]
 fn test_kernel_request_priority_over_idle_tasks() {
     let frontend = DummyArkFrontend::lock();
 
-    // A small data frame is enough — the contention comes from the
+    // A small data frame is enough -- the contention comes from the
     // sleeping idle tasks, not from profile computation.
     frontend.send_execute_request(
         "test_priority_df <- data.frame(a = 1:5)",
@@ -48,35 +47,41 @@ fn test_kernel_request_priority_over_idle_tasks() {
 
     let comm_id = frontend.open_data_explorer("test_priority_df");
 
-    // Build the JSON for the `get_schema` RPC.
+    // Build the get_schema RPC data. The `id` field marks it as an RPC
+    // so Shell creates a `CommMsg::Rpc`.
     let schema_request = DataExplorerBackendRequest::GetSchema(GetSchemaParams {
         column_indices: vec![0],
     });
-    let schema_json = serde_json::to_string(&schema_request).unwrap();
+    let mut data = serde_json::to_value(&schema_request).unwrap();
+    data["id"] = serde_json::Value::String(String::from("test-rpc"));
 
-    // In a single execute request:
-    // - Spawn 5 idle tasks that each block the R thread for 200ms
-    // - Enqueue a `get_schema` kernel request directly on the
-    //   kernel-request channel via `ps_test_send_kernel_comm_request`
-    //
-    // Both are pending when R returns to the event loop.
-    let code = format!(
-        r#"
-        invisible(.Call("ps_test_spawn_sleeping_idle_tasks", 5L, 200L))
-        invisible(.Call("ps_test_send_kernel_comm_request", "{comm_id}", '{schema_json}'))
-        "#,
+    // Spawn 5 idle tasks that each block the R thread for 200ms.
+    // After this execute request completes, R enters the event loop with
+    // all 5 tasks ready on the idle-task channel.
+    frontend.send_execute_request(
+        r#"invisible(.Call("ps_test_spawn_sleeping_idle_tasks", 5L, 200L))"#,
+        ExecuteRequestOptions::default(),
     );
-    frontend.send_execute_request(&code, ExecuteRequestOptions::default());
     frontend.recv_iopub_busy();
     frontend.recv_iopub_execute_input();
     frontend.recv_iopub_idle();
     frontend.recv_shell_execute_reply();
 
-    // The kernel request was enqueued directly (bypassing Shell), so the
-    // RPC reply arrives on IOPub without the Shell's Busy/Idle wrapper.
+    // Send get_schema through Shell. Shell dispatches a `KernelRequest`
+    // to the R thread and blocks on `done_rx`, so the IOPub ordering is
+    // deterministic: Busy -> CommMsg -> Idle (no race).
+    //
+    // R is likely already executing an idle task (sleeping for 200ms).
+    // When it finishes, the priority check at the top of the event loop
+    // picks up the kernel request before `select` can hand R another
+    // idle task.
     let start = Instant::now();
+    frontend.send_shell_comm_msg(comm_id.clone(), data);
+
+    frontend.recv_iopub_busy();
     let msg = frontend.recv_iopub_comm_msg();
     let schema_latency = start.elapsed();
+    frontend.recv_iopub_idle();
 
     assert_eq!(msg.comm_id, comm_id);
     let reply: DataExplorerBackendReply = serde_json::from_value(msg.data).unwrap();
@@ -88,12 +93,13 @@ fn test_kernel_request_priority_over_idle_tasks() {
         other => panic!("Expected GetSchemaReply, got: {other:?}"),
     }
 
-    // With the priority fix the kernel request is serviced before any
-    // sleeping idle task runs, so the reply is near-instant. Without the
-    // fix, `select` picks randomly between the kernel-request channel and
-    // the idle-task channel, so at least one 200ms sleeper runs first.
+    // With the priority fix the kernel request is serviced after at most
+    // one sleeping idle task (~200ms). Without the fix, `select` picks
+    // randomly among ready channels, so multiple 200ms sleepers could
+    // run first (~600ms+). The 350ms threshold is 200ms for one idle
+    // task plus 150ms headroom for slow CI machines.
     assert!(
-        schema_latency < Duration::from_millis(150),
+        schema_latency < Duration::from_millis(350),
         "get_schema took {schema_latency:?}, which suggests kernel requests \
          are being starved by idle tasks"
     );