fix: avoid app event queue stalls

refs #265

4 files changed, 157 insertions, 10 deletions

diff --git a/src/app/api.rs b/src/app/api.rs
index a867f4d7..6f88f6c3 100644
--- a/src/app/api.rs
+++ b/src/app/api.rs
@@ -322,7 +322,14 @@ impl App {
     }
 
     pub(crate) fn handle_api_request(&mut self, request: crate::api::schema::Request) -> String {
-        self.drain_internal_events();
+        self.drain_all_internal_events();
+        self.handle_api_request_after_internal_events_drained(request)
+    }
+
+    pub(crate) fn handle_api_request_after_internal_events_drained(
+        &mut self,
+        request: crate::api::schema::Request,
+    ) -> String {
         use crate::api::schema::{Method, ResponseResult, SuccessResponse};
 
         let response = match request.method {
diff --git a/src/app/mod.rs b/src/app/mod.rs
index 8a981226..bacbf0ce 100644
--- a/src/app/mod.rs
+++ b/src/app/mod.rs
@@ -93,6 +93,9 @@ pub struct App {
     pub(crate) config_reloaded_from_disk: bool,
 }
 
+pub(crate) const APP_EVENT_CHANNEL_CAPACITY: usize = 256;
+pub(crate) const APP_EVENT_DRAIN_LIMIT: usize = 64;
+
 pub(crate) enum LoopEvent {
     Timer,
     Internal(AppEvent),
@@ -221,7 +224,7 @@ impl App {
     ) -> Self {
         let (prefix_code, prefix_mods) = config.prefix_key();
         crate::kitty_graphics::set_enabled(config.experimental.kitty_graphics);
-        let (event_tx, event_rx) = mpsc::channel::<AppEvent>(64);
+        let (event_tx, event_rx) = mpsc::channel::<AppEvent>(APP_EVENT_CHANNEL_CAPACITY);
         let render_notify = Arc::new(Notify::new());
         let render_dirty = Arc::new(AtomicBool::new(false));
 
@@ -561,7 +564,8 @@ impl App {
                 needs_render = true;
             }
 
-            // Drain internal events first so API reads observe fresh pane state.
+            // Drain a bounded internal-event batch for responsiveness. API handlers
+            // perform an exhaustive drain before reading pane/runtime state.
             if self.drain_internal_events() {
                 needs_render = true;
             }
@@ -1354,6 +1358,57 @@ mod tests {
     }
 
     #[test]
+    fn internal_event_drain_limits_work_per_tick() {
+        let mut app = test_app();
+        for i in 0..=APP_EVENT_DRAIN_LIMIT {
+            app.event_tx
+                .try_send(AppEvent::UpdateReady {
+                    version: format!("2.0.{i}"),
+                    install_command: "herdr install".into(),
+                })
+                .unwrap();
+        }
+
+        assert!(app.drain_internal_events());
+
+        let expected_version = format!("2.0.{}", APP_EVENT_DRAIN_LIMIT - 1);
+        assert_eq!(
+            app.state.update_available.as_deref(),
+            Some(expected_version.as_str())
+        );
+        assert!(app.event_rx.try_recv().is_ok());
+    }
+
+    #[test]
+    fn api_request_drains_all_pending_internal_events_before_reading_state() {
+        let mut app = test_app();
+        for i in 0..=APP_EVENT_DRAIN_LIMIT {
+            app.event_tx
+                .try_send(AppEvent::UpdateReady {
+                    version: format!("3.0.{i}"),
+                    install_command: "herdr install".into(),
+                })
+                .unwrap();
+        }
+
+        let response = app.handle_api_request(crate::api::schema::Request {
+            id: "req_server_stop_after_events".into(),
+            method: crate::api::schema::Method::ServerStop(
+                crate::api::schema::EmptyParams::default(),
+            ),
+        });
+        let response: serde_json::Value = serde_json::from_str(&response).unwrap();
+
+        assert_eq!(response["result"]["type"], "ok");
+        let expected_version = format!("3.0.{APP_EVENT_DRAIN_LIMIT}");
+        assert_eq!(
+            app.state.update_available.as_deref(),
+            Some(expected_version.as_str())
+        );
+        assert!(app.event_rx.try_recv().is_err());
+    }
+
+    #[test]
     fn startup_uses_configured_agent_panel_scope() {
         let mut config = Config::default();
         config.ui.agent_panel_scope = crate::config::AgentPanelScopeConfig::Current;
@@ -2631,7 +2686,7 @@ mod tests {
             AgentState::Working
         );
 
-        for i in 0..64 {
+        for i in 0..APP_EVENT_CHANNEL_CAPACITY {
             app.event_tx
                 .try_send(AppEvent::UpdateReady {
                     version: format!("9.9.{i}"),
@@ -2667,7 +2722,13 @@ mod tests {
             .expect("state change should enqueue once queue space is available")
             .expect("app event receiver should still be alive");
 
-        app.drain_internal_events();
+        let max_drains = (APP_EVENT_CHANNEL_CAPACITY / APP_EVENT_DRAIN_LIMIT) + 2;
+        for _ in 0..max_drains {
+            if app.state.terminals.get(&terminal_id).unwrap().state == AgentState::Idle {
+                break;
+            }
+            app.drain_internal_events();
+        }
 
         assert_eq!(
             app.state.terminals.get(&terminal_id).unwrap().state,
diff --git a/src/app/runtime.rs b/src/app/runtime.rs
index 7f7306cc..284fb882 100644
--- a/src/app/runtime.rs
+++ b/src/app/runtime.rs
@@ -418,8 +418,23 @@ impl App {
     }
 
     pub(crate) fn drain_internal_events(&mut self) -> bool {
+        self.drain_internal_events_up_to(super::APP_EVENT_DRAIN_LIMIT)
+    }
+
+    pub(crate) fn drain_all_internal_events(&mut self) -> bool {
+        let mut had_event = false;
+        while self.drain_internal_events_up_to(super::APP_EVENT_DRAIN_LIMIT) {
+            had_event = true;
+        }
+        had_event
+    }
+
+    fn drain_internal_events_up_to(&mut self, limit: usize) -> bool {
         let mut had_event = false;
-        while let Ok(ev) = self.event_rx.try_recv() {
+        for _ in 0..limit {
+            let Ok(ev) = self.event_rx.try_recv() else {
+                break;
+            };
             had_event = true;
             self.handle_internal_event(ev);
         }
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 632f205d..a6d3f6a5 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -213,7 +213,8 @@ impl HeadlessServer {
                 needs_render = true;
             }
 
-            // 2. Drain internal events.
+            // 2. Drain a bounded internal-event batch. API handlers perform an
+            // exhaustive forwarding-aware drain before reading pane/runtime state.
             if self.drain_internal_events_with_forwarding() {
                 needs_render = true;
             }
@@ -974,13 +975,36 @@ impl HeadlessServer {
     /// - Detect when a toast is set on AppState and forward as
     ///   `ServerMessage::Notify` to the foreground client for terminal/system delivery.
     fn drain_internal_events_with_forwarding(&mut self) -> bool {
+        self.drain_internal_events_with_forwarding_up_to(crate::app::APP_EVENT_DRAIN_LIMIT)
+            .1
+    }
+
+    fn drain_all_internal_events_with_forwarding(&mut self) -> bool {
         let mut changed = false;
-        while let Ok(ev) = self.app.event_rx.try_recv() {
-            changed |= self.handle_internal_event_with_forwarding(ev);
+        loop {
+            let (had_event, batch_changed) =
+                self.drain_internal_events_with_forwarding_up_to(crate::app::APP_EVENT_DRAIN_LIMIT);
+            changed |= batch_changed;
+            if !had_event {
+                break;
+            }
         }
         changed
     }
 
+    fn drain_internal_events_with_forwarding_up_to(&mut self, limit: usize) -> (bool, bool) {
+        let mut had_event = false;
+        let mut changed = false;
+        for _ in 0..limit {
+            let Ok(ev) = self.app.event_rx.try_recv() else {
+                break;
+            };
+            had_event = true;
+            changed |= self.handle_internal_event_with_forwarding(ev);
+        }
+        (had_event, changed)
+    }
+
     fn drain_client_sound_config_reload_request(&mut self) {
         if !self.app.state.request_client_sound_config_reload {
             return;
@@ -1440,6 +1464,7 @@ impl HeadlessServer {
         }
 
         let changed = api::request_changes_ui(&msg.request);
+        let changed = self.drain_all_internal_events_with_forwarding() || changed;
 
         // Capture toast and effective pane states before the API call so we can
         // forward resulting client-local notifications. API requests like
@@ -1490,7 +1515,8 @@ impl HeadlessServer {
                 .unwrap_or_else(|_| "{}".to_string())
             })
         } else {
-            self.app.handle_api_request(msg.request)
+            self.app
+                .handle_api_request_after_internal_events_drained(msg.request)
         };
         let _ = msg.respond_to.send(response);
 
@@ -2241,6 +2267,44 @@ mod tests {
         }
     }
 
+    #[test]
+    fn headless_api_request_drains_all_pending_internal_events_before_reading_state() {
+        let mut server = test_headless_server();
+        for i in 0..=crate::app::APP_EVENT_DRAIN_LIMIT {
+            server
+                .app
+                .event_tx
+                .try_send(AppEvent::UpdateReady {
+                    version: format!("4.0.{i}"),
+                    install_command: "herdr install".into(),
+                })
+                .unwrap();
+        }
+
+        let (respond_to, response_rx) = std::sync::mpsc::channel();
+        assert!(
+            server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
+                request: api::schema::Request {
+                    id: "headless_stop_after_events".into(),
+                    method: api::schema::Method::ServerStop(api::schema::EmptyParams::default()),
+                },
+                respond_to,
+            })
+        );
+        let response = response_rx
+            .recv_timeout(Duration::from_millis(100))
+            .unwrap();
+        let response: serde_json::Value = serde_json::from_str(&response).unwrap();
+
+        assert_eq!(response["result"]["type"], "ok");
+        let expected_version = format!("4.0.{}", crate::app::APP_EVENT_DRAIN_LIMIT);
+        assert_eq!(
+            server.app.state.update_available.as_deref(),
+            Some(expected_version.as_str())
+        );
+        assert!(server.app.event_rx.try_recv().is_err());
+    }
+
     fn test_client_writer() -> (
         ClientWriter,
         std::sync::mpsc::Receiver<Vec<u8>>,