From aad66ebeb68eecf3e1e1d202fe0d3c6b5d3f03ca Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 24 Sep 2026 23:45:15 +0000 Subject: [PATCH] chore: sync public mirror from internal --- .repository-projection.json | 6 +- packages/local-host-rs/src/lib.rs | 1 + packages/local-host-rs/src/local_models.rs | 1 + .../local-host-rs/src/mcp/notifications.rs | 3 + .../local-host-rs/src/subagents/lifecycle.rs | 1 + .../src/tools/background_tasks.rs | 5 + packages/local-host-rs/src/ui_wake.rs | 26 ++ packages/tui-rs/src/app.rs | 267 +++++++++++-- packages/tui-rs/src/app/a2a_handoff.rs | 15 +- packages/tui-rs/src/app/command_handlers.rs | 2 + packages/tui-rs/src/app/exec_commands.rs | 2 + packages/tui-rs/src/app/input_handlers.rs | 10 +- packages/tui-rs/src/app/onboarding.rs | 9 +- packages/tui-rs/src/app/selective_summary.rs | 7 + packages/tui-rs/src/app/tests.rs | 47 ++- packages/tui-rs/src/components/operations.rs | 1 + .../tui-rs/src/components/session_switcher.rs | 12 + packages/tui-rs/src/config_watcher.rs | 10 + packages/tui-rs/src/lib.rs | 1 + packages/tui-rs/src/loop_wake.rs | 376 ++++++++++++++++++ packages/tui-rs/src/model_monitor.rs | 1 + packages/tui-rs/src/rubber_duck.rs | 1 + packages/tui-rs/src/terminal/events.rs | 5 + 23 files changed, 764 insertions(+), 45 deletions(-) create mode 100644 packages/local-host-rs/src/ui_wake.rs create mode 100644 packages/tui-rs/src/loop_wake.rs diff --git a/.repository-projection.json b/.repository-projection.json index ae658fdd1..0c748963d 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "d86f17d9fc24cbabd645f27a7dbc3d861d57acb7", + "sourceSha": "0c4999b07bdbd4132282664256240c97cca5f8e6", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "9610ff4cc0f57cd58d5d334d46b875b766f6b2e2", + "priorProjectedBase": "01cbd80c5f6e9b39311ae3ead51074bf34eee588", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "89cbdbe1d79917bad52655817aab0eb545b89183915380eaedc3217f014e6715", - "contentDigest": "bf569f56d06fbd6cd2c5de9cd3b3d699dbc019417492e2493ec4da8b9ab6202e", + "contentDigest": "9d42748e8e0da39268d1af21ef1b2d7d4f71ba4814e1541f9cf2246dee33a956", "publicationEligible": true } diff --git a/packages/local-host-rs/src/lib.rs b/packages/local-host-rs/src/lib.rs index c21c4a3c7..861a8d823 100644 --- a/packages/local-host-rs/src/lib.rs +++ b/packages/local-host-rs/src/lib.rs @@ -70,6 +70,7 @@ pub mod tool_output; pub mod tools; pub mod transcript; pub mod ui_prefs; +pub mod ui_wake; pub mod video; pub mod workflow_runtime; pub use sandbox::SandboxPolicy; diff --git a/packages/local-host-rs/src/local_models.rs b/packages/local-host-rs/src/local_models.rs index 14ebcb398..e56fba733 100644 --- a/packages/local-host-rs/src/local_models.rs +++ b/packages/local-host-rs/src/local_models.rs @@ -283,6 +283,7 @@ fn spawn_local_model_discovery_with_client_factory( { break; } + crate::ui_wake::wake(); } }) .expect("local model discovery thread should start"); diff --git a/packages/local-host-rs/src/mcp/notifications.rs b/packages/local-host-rs/src/mcp/notifications.rs index fcdc15541..8de20c39d 100644 --- a/packages/local-host-rs/src/mcp/notifications.rs +++ b/packages/local-host-rs/src/mcp/notifications.rs @@ -41,6 +41,7 @@ impl NotificationQueue { .position(|method| *method == notification.method) { self.0.lock().unwrap().invalidated[index] = true; + crate::ui_wake::wake(); return true; } let Ok(serialized) = serde_json::to_vec(¬ification) else { @@ -58,6 +59,8 @@ impl NotificationQueue { } pending.bytes += bytes; pending.events.push_back((notification, bytes)); + drop(pending); + crate::ui_wake::wake(); true } diff --git a/packages/local-host-rs/src/subagents/lifecycle.rs b/packages/local-host-rs/src/subagents/lifecycle.rs index 59612bc8f..4b009bf36 100644 --- a/packages/local-host-rs/src/subagents/lifecycle.rs +++ b/packages/local-host-rs/src/subagents/lifecycle.rs @@ -713,6 +713,7 @@ impl SubagentManager { record.id ) })?; + crate::ui_wake::wake(); record.lifecycle_notification_published = true; self.write_record(record) } diff --git a/packages/local-host-rs/src/tools/background_tasks.rs b/packages/local-host-rs/src/tools/background_tasks.rs index 8b74f8294..990642989 100644 --- a/packages/local-host-rs/src/tools/background_tasks.rs +++ b/packages/local-host-rs/src/tools/background_tasks.rs @@ -341,6 +341,7 @@ fn emit_task_lifecycle(task_id: &str, command: &str, status: &str, exit_code: Op }); } persist_running_snapshot(); + crate::ui_wake::wake(); } /// Drain process exit/stop notifications for the UI (and optional agent nudge). @@ -564,6 +565,7 @@ fn emit_monitor_matches(task_id: &str, stream: &'static str, line: &str) { }); } } + let had_events = !matched.is_empty(); for event in matched { if let Ok(mut events) = MONITOR_EVENTS.write() { if events.len() >= MAX_MONITOR_EVENTS { @@ -578,6 +580,9 @@ fn emit_monitor_matches(task_id: &str, stream: &'static str, line: &str) { history.push_back(event); } } + if had_events { + crate::ui_wake::wake(); + } } fn remove_task_monitors(task_id: &str) { diff --git a/packages/local-host-rs/src/ui_wake.rs b/packages/local-host-rs/src/ui_wake.rs new file mode 100644 index 000000000..9a0c7a891 --- /dev/null +++ b/packages/local-host-rs/src/ui_wake.rs @@ -0,0 +1,26 @@ +//! Process-wide wake hook for the Deixic Code terminal event loop. +//! +//! Producers that enqueue work from another crate, a sync thread, or a +//! file-watcher callback cannot hold the TUI's [`tokio::sync::Notify`]. +//! They call [`wake`] after the enqueue. The TUI installs one hook for the +//! lifetime of its main loop; other processes leave the hook empty. + +use std::sync::{Arc, Mutex}; + +static HOOK: Mutex>> = Mutex::new(None); + +/// Install or clear the UI wake hook. The TUI sets this while its loop runs. +pub fn set_hook(hook: Option>) { + *HOOK.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = hook; +} + +/// Wake the installed UI loop, if any. Safe to call when no hook is set. +pub fn wake() { + let hook = HOOK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + if let Some(hook) = hook { + hook(); + } +} diff --git a/packages/tui-rs/src/app.rs b/packages/tui-rs/src/app.rs index e3a0a696e..c542b3e3f 100644 --- a/packages/tui-rs/src/app.rs +++ b/packages/tui-rs/src/app.rs @@ -92,6 +92,38 @@ use crate::git; use crate::goal::GoalStore; use crate::harness::HarnessStore; use crate::keybindings::load_rust_tui_keybindings; +use crate::loop_wake::{ + LoopWake, LoopWakeCause, TerminalPollInput, await_loop_wake, terminal_poll_timeout, +}; + +struct LoopWakeHookGuard; + +impl Drop for LoopWakeHookGuard { + fn drop(&mut self) { + maestro_local_host::ui_wake::set_hook(None); + } +} + +/// Interrupts a blocking uncurses poll if the async wait is cancelled +/// (signal shutdown drops the run future at this await point). +struct UncursesPollGuard { + wake: LoopWake, + armed: bool, +} + +impl UncursesPollGuard { + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for UncursesPollGuard { + fn drop(&mut self) { + if self.armed { + self.wake.signal(); + } + } +} use crate::keybindings::{is_keybindings_config_path, summarize_keybindings_config_issues}; use crate::mailbox::MailboxStore; #[cfg(test)] @@ -656,6 +688,17 @@ pub struct App { /// Protocol-aware terminal input. Falls back to crossterm when unavailable. terminal_events: Option, + /// Wakes the main loop when a drained channel receives an item. + loop_wake: LoopWake, + + /// Crossterm fallback reader. Unused while [`Self::terminal_events`] is set. + crossterm_events: Option, + + /// Keep polling subagent lifecycle briefly after the last observed worker + /// so a completion that lands inside the 250 ms mailbox gate is not held + /// until the idle safety tick. + subagent_followup_until: Option, + /// Flag to exit the main loop. should_quit: bool, /// Relaunch in the saved workspace after this agent and writer shut down. @@ -1197,6 +1240,7 @@ impl App { app.state.textarea = startup.textarea; // Keep the reader and its buffered input across the startup handoff. app.terminal_events = terminal_events; + app.attach_terminal_waker(); app.initialize_terminal_events(); app.note_goal_paused_on_restart_if_needed(); app.note_orphan_background_tasks_if_any(); @@ -1705,6 +1749,9 @@ impl App { terminal_clear_supported, terminal_size, terminal_events: None, + loop_wake: LoopWake::new(), + crossterm_events: None, + subagent_followup_until: None, should_quit: false, resume_target: None, capabilities, @@ -1981,14 +2028,182 @@ Always use tools when they would be helpful. Be concise and direct in your respo } } - fn poll_terminal_event(&mut self, timeout: Duration) -> Result> { - if let Some(reader) = &mut self.terminal_events { + fn attach_terminal_waker(&self) { + if let Some(reader) = &self.terminal_events { + self.loop_wake.install_terminal_waker(reader.waker()); + } + } + + fn install_loop_wake_hook(&self) -> LoopWakeHookGuard { + let wake = self.loop_wake.clone(); + maestro_local_host::ui_wake::set_hook(Some(std::sync::Arc::new(move || wake.signal()))); + LoopWakeHookGuard + } + + /// Uncurses blocks in `EventSource::poll`. Producers interrupt that wait + /// through [`LoopWake::signal`]. The crossterm fallback selects on the + /// same notify, the event stream, and the timeout. + async fn poll_terminal_event(&mut self, timeout: Duration) -> Result> { + if self.terminal_events.is_some() { + return self.poll_uncurses_event(timeout).await; + } + self.poll_crossterm_event(timeout).await + } + + async fn poll_uncurses_event(&mut self, timeout: Duration) -> Result> { + // Short waits match the previous inline poll, including shutdown + // latency. Longer waits run on a blocking thread so dropping the run + // future can interrupt them through the uncurses waker. + if timeout <= crate::loop_wake::SHORT_MAINTENANCE_POLL { + let reader = self + .terminal_events + .as_mut() + .expect("uncurses reader checked"); return reader.poll(timeout).map_err(Into::into); } - if event::poll(timeout)? { - return Ok(AppTerminalEvent::from_crossterm(event::read()?)); + + let mut reader = self + .terminal_events + .take() + .expect("uncurses reader checked"); + let mut guard = UncursesPollGuard { + wake: self.loop_wake.clone(), + armed: true, + }; + let joined = tokio::task::spawn_blocking(move || { + let event = reader.poll(timeout); + (event, reader) + }) + .await; + guard.disarm(); + match joined { + Ok((event, reader)) => { + self.terminal_events = Some(reader); + event.map_err(Into::into) + } + Err(error) => Err(anyhow::anyhow!("terminal poll task failed: {error}")), + } + } + + async fn poll_crossterm_event( + &mut self, + timeout: Duration, + ) -> Result> { + use tokio_stream::StreamExt; + + let wake = self.loop_wake.clone(); + let stream = self + .crossterm_events + .get_or_insert_with(crossterm::event::EventStream::new); + match await_loop_wake(timeout, &wake, stream.next()).await { + LoopWakeCause::Terminal(Some(Ok(event))) => Ok(AppTerminalEvent::from_crossterm(event)), + LoopWakeCause::Terminal(Some(Err(error))) => Err(error.into()), + LoopWakeCause::Terminal(None) | LoopWakeCause::Producer | LoopWakeCause::Tick => { + Ok(None) + } } - Ok(None) + } + + fn terminal_poll_budget(&mut self, agent_activity: bool, needs_redraw: bool) -> Duration { + self.note_subagent_followup(); + let mut timeout = terminal_poll_timeout(TerminalPollInput { + agent_activity, + busy: self.state.busy, + pending_redraw: needs_redraw, + short_cadence: self.needs_short_cadence(), + }); + if let Some(remaining) = self.next_wait_cap() { + timeout = timeout.min(remaining); + } + timeout + } + + fn note_subagent_followup(&mut self) { + let workers_live = !self.state.busy + && matches!( + self.tool_executor.worker_activity(), + Ok((running, waiting)) if running.saturating_add(waiting) > 0 + ); + if self.state.busy || workers_live { + self.subagent_followup_until = Some(Instant::now() + Duration::from_millis(250)); + } + } + + fn needs_short_cadence(&self) -> bool { + self.presentation_animation_active() + || self.theme_query_outstanding() + || self.config_watcher.has_pending_debounce() + || self.session_switcher.content_search_pending() + || self.selective_summary_in_flight() + || self.session_transition_waiting() + || self.session_cleanup_pending() + || self.agent_note_ack_pending() + || self.subagent_followup_remaining().is_some() + } + + fn presentation_animation_active(&self) -> bool { + self.dex_hop_active() || self.dex_pet_active() || self.onboarding_animation_active() + } + + fn dex_hop_active(&self) -> bool { + self.dex_terminal == Some(crate::components::dex_companion::DexCompanionState::Finished) + && self.dex_pose_started.elapsed() < Duration::from_millis(800) + && self + .ui_prefs + .animations + .unwrap_or(self.configured_animations) + && self.ui_prefs.dex_personality() + != crate::components::dex_companion::DexPersonality::Quiet + } + + fn theme_query_outstanding(&self) -> bool { + self.terminal_events.is_some() + && self.state.theme_follower.is_some() + && !self.state.theme_reporting_available + && self.state.last_theme_query.is_some() + } + + fn agent_note_ack_pending(&self) -> bool { + !self.pending_agent_note_applications.is_empty() + || !self.pending_agent_note_consumptions.is_empty() + } + + fn subagent_followup_remaining(&self) -> Option { + let until = self.subagent_followup_until?; + let now = Instant::now(); + if now >= until { + None + } else { + Some(until.saturating_duration_since(now)) + } + } + + fn next_wait_cap(&self) -> Option { + let mut nearest: Option = None; + let consider = |nearest: &mut Option, until: Duration| { + *nearest = Some(nearest.map_or(until, |current| current.min(until))); + }; + if let Some(schedule) = &self.loop_schedule { + let until = schedule + .next_fire + .saturating_duration_since(Instant::now()) + .max(crate::loop_wake::SHORT_MAINTENANCE_POLL); + consider(&mut nearest, until); + } + if self.goal_auto_continue_armed + && !self.state.busy + && self.queued_prompts.is_empty() + && self.loop_schedule.is_none() + && self.queued_prompt_inflight.is_none() + && self.queued_prompt_active.is_none() + && !self.session_cleanup_pending() + { + consider(&mut nearest, Duration::ZERO); + } + if let Some(remaining) = self.subagent_followup_remaining() { + consider(&mut nearest, remaining); + } + nearest } fn poll_terminal_theme(&mut self) { @@ -2076,6 +2291,7 @@ Always use tools when they would be helpful. Be concise and direct in your respo } async fn run_inner(&mut self) -> Result { + let _loop_wake_hook = self.install_loop_wake_hook(); self.record_configuration_visibility(); // Optional Jane Street magic-trace slow-frame snapshots (Linux/Intel PT). if crate::magic_trace::init_from_env() { @@ -2182,11 +2398,13 @@ Always use tools when they would be helpful. Be concise and direct in your respo } } - // Poll faster while busy so the working-state sheen and spinners - // advance. Idle welcome screens stay still on the normal cadence. - let poll_timeout = terminal_poll_timeout(self.state.busy, agent_activity); + // Busy frames stay at 33 ms. Queued agent output and a pending + // redraw do not wait. A quiescent screen blocks until input, a + // producer signal, or the 5 s safety tick. Animations, an + // outstanding theme query, and other timer-driven UI keep 100 ms. + let poll_timeout = self.terminal_poll_budget(agent_activity, needs_redraw); self.poll_terminal_theme(); - if let Some(event) = self.poll_terminal_event(poll_timeout)? { + if let Some(event) = self.poll_terminal_event(poll_timeout).await? { match event { AppTerminalEvent::Key(key) if should_handle_key_event(key.kind) => { self.handle_key(key.code, key.modifiers).await?; @@ -2581,15 +2799,7 @@ Always use tools when they would be helpful. Be concise and direct in your respo } // Paint when dirty, or continuously while busy (thinking/spinner). - let dex_hop_active = self.dex_terminal - == Some(crate::components::dex_companion::DexCompanionState::Finished) - && self.dex_pose_started.elapsed() < Duration::from_millis(800) - && self - .ui_prefs - .animations - .unwrap_or(self.configured_animations) - && self.ui_prefs.dex_personality() - != crate::components::dex_companion::DexPersonality::Quiet; + let dex_hop_active = self.dex_hop_active(); if needs_redraw || self.state.busy || dex_hop_active @@ -2673,12 +2883,14 @@ Always use tools when they would be helpful. Be concise and direct in your respo /// via `poll_workspace_scan` and applies the result when it arrives. fn spawn_workspace_scan(&mut self) { let (workspace_tx, workspace_rx) = std::sync::mpsc::channel(); + let wake = self.loop_wake.clone(); std::thread::Builder::new() .name("maestro-workspace-scan".into()) .spawn(move || { let cwd = std::env::current_dir().unwrap_or_default(); let files = get_workspace_files(&cwd, 10_000); let _ = workspace_tx.send(files); + wake.signal(); }) .ok(); self.workspace_scan_rx = Some(workspace_rx); @@ -2792,7 +3004,10 @@ Always use tools when they would be helpful. Be concise and direct in your respo self.restore_request_cache(&agent); let tool_tx = agent.tool_response_sender(); self.native_agent = Some(agent); - self.native_event_rx = Some(event_rx); + self.native_event_rx = Some(crate::loop_wake::forward_unbounded( + event_rx, + self.loop_wake.clone(), + )); self.tool_response_tx = Some(tool_tx); // A session restored at startup exists before the runner does, @@ -3579,8 +3794,10 @@ Always use tools when they would be helpful. Be concise and direct in your respo self.mcp_status_refresh_in_flight = true; let executor = Arc::clone(&self.tool_executor); let tx = self.mcp_status_tx.clone(); + let wake = self.loop_wake.clone(); tokio::spawn(async move { let _ = tx.send(executor.mcp_status().await); + wake.signal(); }); false } @@ -4707,11 +4924,13 @@ Always use tools when they would be helpful. Be concise and direct in your respo }; let guardian_tx = self.guardian_tx.clone(); let request = request.clone(); + let wake = self.loop_wake.clone(); self.pending_guardian_reviews .insert(request.call_id.clone()); tokio::spawn(async move { let verdict = guardian.evaluate(context).await; let _ = guardian_tx.send((request, verdict)); + wake.signal(); }); true } @@ -5553,16 +5772,6 @@ fn should_handle_key_event(kind: KeyEventKind) -> bool { matches!(kind, KeyEventKind::Press | KeyEventKind::Repeat) } -fn terminal_poll_timeout(busy: bool, agent_activity: bool) -> Duration { - if agent_activity { - Duration::ZERO - } else if busy { - Duration::from_millis(33) - } else { - Duration::from_millis(100) - } -} - /// Combine an action-firewall reason and a sandbox-bypass warning into the /// single reason string shown on an [`ApprovalRequest`]. /// diff --git a/packages/tui-rs/src/app/a2a_handoff.rs b/packages/tui-rs/src/app/a2a_handoff.rs index 176e5bb44..41f11249a 100644 --- a/packages/tui-rs/src/app/a2a_handoff.rs +++ b/packages/tui-rs/src/app/a2a_handoff.rs @@ -37,6 +37,7 @@ impl App { computer_package: Option, ) { let tx = self.a2a_handoff_tx.clone(); + let wake = self.loop_wake.clone(); let tool_executor = Arc::clone(&self.tool_executor); let requested_peer = peer .clone() @@ -46,11 +47,15 @@ impl App { std::slice::from_ref(&(requested_peer)), )); tokio::spawn(async move { + let emit = |event| { + let _ = tx.send(event); + wake.signal(); + }; let package = match computer_package { Some(selection) => match create_computer_package(tool_executor, selection).await { Ok(package) => Some(package), Err(error) => { - let _ = tx.send(A2aHandoffEvent::Failed { + emit(A2aHandoffEvent::Failed { peer: requested_peer, task_id: None, error: format!("could not create Computer package: {error:#}"), @@ -64,7 +69,7 @@ impl App { let pending = match start_handoff(peer.clone(), text, package.as_ref()).await { Ok(pending) => pending, Err(error) => { - let _ = tx.send(A2aHandoffEvent::Failed { + emit(A2aHandoffEvent::Failed { peer: requested_peer, task_id: None, error: format!("could not send A2A handoff: {error:#}"), @@ -74,7 +79,7 @@ impl App { }; let peer = pending.peer.clone(); let task_id = pending.task.id.clone(); - let _ = tx.send(A2aHandoffEvent::Accepted { + emit(A2aHandoffEvent::Accepted { peer: peer.clone(), task_id: task_id.clone(), package_id: package.as_ref().map(|package| package.package_id.clone()), @@ -89,14 +94,14 @@ impl App { .await { Ok(completed) => { - let _ = tx.send(A2aHandoffEvent::Finished { + emit(A2aHandoffEvent::Finished { peer, task: Box::new(completed.task), ledger_warning: completed.ledger_warning, }); } Err(error) => { - let _ = tx.send(A2aHandoffEvent::Failed { + emit(A2aHandoffEvent::Failed { peer, task_id: Some(task_id), error: format!("could not follow A2A handoff: {error:#}"), diff --git a/packages/tui-rs/src/app/command_handlers.rs b/packages/tui-rs/src/app/command_handlers.rs index 08dc3c3c9..4a4acf5df 100644 --- a/packages/tui-rs/src/app/command_handlers.rs +++ b/packages/tui-rs/src/app/command_handlers.rs @@ -1875,10 +1875,12 @@ impl App { .to_string(), ); let tx = self.mcp_config_tx.clone(); + let wake = self.loop_wake.clone(); tokio::spawn(async move { let result = crate::mcp_config_cli::apply_mcp_config_async_quiet(&args).await; let _ = tx.send(result.map_err(|error| error.to_string())); + wake.signal(); }); } else { match crate::mcp_config_cli::apply_mcp_config_async(&args).await { diff --git a/packages/tui-rs/src/app/exec_commands.rs b/packages/tui-rs/src/app/exec_commands.rs index 5fd316af9..d26fc2f62 100644 --- a/packages/tui-rs/src/app/exec_commands.rs +++ b/packages/tui-rs/src/app/exec_commands.rs @@ -115,6 +115,7 @@ impl App { .map(PathBuf::from) .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); let tx = self.exec_command_tx.clone(); + let wake = self.loop_wake.clone(); self.state.status = Some(format!("Running /{name} ...")); let outcome_name = name.to_string(); @@ -130,6 +131,7 @@ impl App { source, result, }); + wake.signal(); }); if let Err(err) = spawned { self.state.error = Some(self.state.locale.format( diff --git a/packages/tui-rs/src/app/input_handlers.rs b/packages/tui-rs/src/app/input_handlers.rs index e269cd5e4..ebff649c8 100644 --- a/packages/tui-rs/src/app/input_handlers.rs +++ b/packages/tui-rs/src/app/input_handlers.rs @@ -1671,8 +1671,14 @@ impl App { .map_err(|error| format!("{error:#}")); let _ = tx.send(result); }); - self.setup_login_rx = Some(rx); - self.setup_login_url_rx = Some(url_rx); + self.setup_login_rx = Some(crate::loop_wake::forward_oneshot( + rx, + self.loop_wake.clone(), + )); + self.setup_login_url_rx = Some(crate::loop_wake::forward_unbounded( + url_rx, + self.loop_wake.clone(), + )); self.setup_login_task = Some(task); self.setup_modal.set_waiting_evalops(); } diff --git a/packages/tui-rs/src/app/onboarding.rs b/packages/tui-rs/src/app/onboarding.rs index 6c947e762..8c0203b0c 100644 --- a/packages/tui-rs/src/app/onboarding.rs +++ b/packages/tui-rs/src/app/onboarding.rs @@ -101,7 +101,7 @@ impl App { tick, ); })?; - if let Some(event) = self.poll_terminal_event(Duration::from_millis(30))? { + if let Some(event) = self.poll_terminal_event(Duration::from_millis(30)).await? { match event { AppTerminalEvent::Key(key) if should_handle_key_event(key.kind) => { if key.code == KeyCode::Char('c') @@ -169,6 +169,7 @@ impl App { self.onboarding.attempts.saturating_sub(1).min(1_000), ); let tx = self.onboarding.collection_tx.clone(); + let wake = self.loop_wake.clone(); let origin = if matches!( stage, OnboardingStage::ChecksCompleted | OnboardingStage::Completed @@ -180,6 +181,7 @@ impl App { tokio::spawn(async move { let status = crate::telemetry::record_onboarding_event(event, origin).await; let _ = tx.send(status); + wake.signal(); }); } @@ -202,8 +204,11 @@ impl App { self.onboarding.attempts = self.onboarding.attempts.saturating_add(1); self.onboarding.transition = Instant::now(); self.setup_modal.set_checking(); + let wake = self.loop_wake.clone(); self.onboarding.checks = Some(tokio::spawn(async move { - crate::onboarding_checks::run_checks(model.as_deref(), &cwd).await + let report = crate::onboarding_checks::run_checks(model.as_deref(), &cwd).await; + wake.signal(); + report })); } diff --git a/packages/tui-rs/src/app/selective_summary.rs b/packages/tui-rs/src/app/selective_summary.rs index 6d2df2527..3243a614a 100644 --- a/packages/tui-rs/src/app/selective_summary.rs +++ b/packages/tui-rs/src/app/selective_summary.rs @@ -108,6 +108,13 @@ impl App { } } + pub(super) fn selective_summary_in_flight(&self) -> bool { + matches!( + self.selective_summary.as_ref().map(|dialog| &dialog.stage), + Some(Stage::Loading(_) | Stage::Running { .. }) + ) + } + pub(super) fn poll_selective_summary(&mut self) -> bool { let Some(mut dialog) = self.selective_summary.take() else { return false; diff --git a/packages/tui-rs/src/app/tests.rs b/packages/tui-rs/src/app/tests.rs index d7bfa226e..ecf93b630 100644 --- a/packages/tui-rs/src/app/tests.rs +++ b/packages/tui-rs/src/app/tests.rs @@ -5550,16 +5550,55 @@ fn signal_shutdown_only_ends_a_started_terminal_session() { #[test] fn queued_agent_activity_skips_the_terminal_poll_delay() { - assert_eq!(terminal_poll_timeout(true, true), Duration::ZERO); - assert_eq!(terminal_poll_timeout(false, true), Duration::ZERO); assert_eq!( - terminal_poll_timeout(true, false), + terminal_poll_timeout(TerminalPollInput { + agent_activity: true, + busy: true, + pending_redraw: false, + short_cadence: false, + }), + Duration::ZERO + ); + assert_eq!( + terminal_poll_timeout(TerminalPollInput { + agent_activity: false, + busy: true, + pending_redraw: false, + short_cadence: false, + }), + Duration::from_millis(33) + ); + assert_eq!( + terminal_poll_timeout(TerminalPollInput { + agent_activity: false, + busy: true, + pending_redraw: true, + short_cadence: true, + }), Duration::from_millis(33) ); assert_eq!( - terminal_poll_timeout(false, false), + terminal_poll_timeout(TerminalPollInput { + agent_activity: false, + busy: false, + pending_redraw: true, + short_cadence: false, + }), + Duration::ZERO + ); + assert_eq!( + terminal_poll_timeout(TerminalPollInput { + agent_activity: false, + busy: false, + pending_redraw: false, + short_cadence: true, + }), Duration::from_millis(100) ); + assert_eq!( + terminal_poll_timeout(TerminalPollInput::idle()), + Duration::from_secs(5) + ); } #[tokio::test] diff --git a/packages/tui-rs/src/components/operations.rs b/packages/tui-rs/src/components/operations.rs index 071d1be9d..dda55398c 100644 --- a/packages/tui-rs/src/components/operations.rs +++ b/packages/tui-rs/src/components/operations.rs @@ -540,6 +540,7 @@ impl OperationsModal { std::thread::spawn(move || { let manager = SessionManager::with_sessions_dir(cwd, sessions_dir); let _ = tx.send(load_operations(&manager)); + maestro_local_host::ui_wake::wake(); }); } diff --git a/packages/tui-rs/src/components/session_switcher.rs b/packages/tui-rs/src/components/session_switcher.rs index 837a66ad4..4860db043 100644 --- a/packages/tui-rs/src/components/session_switcher.rs +++ b/packages/tui-rs/src/components/session_switcher.rs @@ -106,6 +106,16 @@ impl SessionSwitcher { self.visible } + /// The content search has not caught up with the visible query. + /// + /// The 150 ms debounce and the worker thread both need another loop turn. + #[must_use] + pub fn content_search_pending(&self) -> bool { + self.visible + && !self.query.trim().is_empty() + && self.content_query.as_deref() != Some(self.query.as_str()) + } + /// Refresh metadata off the input thread, coalescing repeated opens. pub fn refresh_async(&mut self) { if self.pending.is_some() { @@ -129,6 +139,7 @@ impl SessionSwitcher { .name("maestro-session-list".into()) .spawn(move || { let _ = tx.send(load()); + maestro_local_host::ui_wake::wake(); }) { Ok(_) => self.pending = Some(rx), Err(error) => { @@ -391,6 +402,7 @@ impl SessionSwitcher { let documents = collect_documents(&root, cache.as_deref()); let results = search_documents(&documents, &query, None, documents.len()); let _ = tx.send((query, results)); + maestro_local_host::ui_wake::wake(); }) { Ok(_) => self.content_pending = Some(rx), Err(error) => { diff --git a/packages/tui-rs/src/config_watcher.rs b/packages/tui-rs/src/config_watcher.rs index ed7aabad3..8d693bb56 100644 --- a/packages/tui-rs/src/config_watcher.rs +++ b/packages/tui-rs/src/config_watcher.rs @@ -142,6 +142,7 @@ impl ConfigWatcher { _ => continue, }; let _ = tx.send(config_event); + maestro_local_host::ui_wake::wake(); } } }, @@ -263,6 +264,14 @@ impl ConfigWatcher { !self.watched_paths.is_empty() } + /// A debounced change is waiting out its quiet period. + #[must_use] + pub fn has_pending_debounce(&self) -> bool { + self.debounce_states + .values() + .any(|state| state.pending_event.is_some()) + } + /// Clear all pending events pub fn clear_pending(&mut self) { while self.event_rx.try_recv().is_ok() {} @@ -283,6 +292,7 @@ impl Default for ConfigWatcher { let _ = event_tx.send(ConfigEvent::Error(format!( "Failed to initialize config watcher: {err}" ))); + maestro_local_host::ui_wake::wake(); Self { watched_paths: HashSet::new(), event_rx, diff --git a/packages/tui-rs/src/lib.rs b/packages/tui-rs/src/lib.rs index 040f6f464..d42c803da 100644 --- a/packages/tui-rs/src/lib.rs +++ b/packages/tui-rs/src/lib.rs @@ -466,6 +466,7 @@ pub use maestro_swarm as swarm; /// Main application struct and event loop. /// This is the top-level coordinator that ties everything together. mod app; +mod loop_wake; // ───────────────────────────────────────────────────────────────────────────── // RE-EXPORTS diff --git a/packages/tui-rs/src/loop_wake.rs b/packages/tui-rs/src/loop_wake.rs new file mode 100644 index 000000000..0368c4ae6 --- /dev/null +++ b/packages/tui-rs/src/loop_wake.rs @@ -0,0 +1,376 @@ +//! Scheduling for the interactive terminal loop. +//! +//! A quiescent TUI blocks until terminal input, a producer signal, or a slow +//! safety tick. Busy frames, queued agent output, and timer-driven UI keep +//! the previous short poll. + +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use tokio::sync::{Notify, mpsc, oneshot}; + +/// Safety drain when no terminal input and no producer signal arrives. +/// Matches the hosted runner's notified maintenance interval. +pub(crate) const IDLE_SAFETY_TICK: Duration = Duration::from_secs(5); +pub(crate) const SHORT_MAINTENANCE_POLL: Duration = Duration::from_millis(100); +pub(crate) const BUSY_FRAME_POLL: Duration = Duration::from_millis(33); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct TerminalPollInput { + pub(crate) agent_activity: bool, + pub(crate) busy: bool, + pub(crate) pending_redraw: bool, + /// Animation, an outstanding theme query, a debounce, or a live source + /// that cannot signal the loop itself. + pub(crate) short_cadence: bool, +} + +impl TerminalPollInput { + #[cfg(test)] + pub(crate) fn idle() -> Self { + Self { + agent_activity: false, + busy: false, + pending_redraw: false, + short_cadence: false, + } + } +} + +pub(crate) fn terminal_poll_timeout(input: TerminalPollInput) -> Duration { + // Busy frames stay at 33 ms even when a redraw is already queued. + // Agent output still skips the wait, matching the previous contract. + if input.agent_activity { + Duration::ZERO + } else if input.busy { + BUSY_FRAME_POLL + } else if input.pending_redraw { + Duration::ZERO + } else if input.short_cadence { + SHORT_MAINTENANCE_POLL + } else { + IDLE_SAFETY_TICK + } +} + +/// Shared wake for every in-process producer the main loop drains. +/// +/// `notify` wakes the crossterm `select`. `terminal_waker` interrupts a +/// blocking uncurses `EventSource::poll` from another thread. +#[derive(Clone)] +pub(crate) struct LoopWake { + notify: Arc, + terminal_waker: Arc>>, +} + +impl std::fmt::Debug for LoopWake { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.debug_struct("LoopWake").finish_non_exhaustive() + } +} + +impl LoopWake { + pub(crate) fn new() -> Self { + Self { + notify: Arc::new(Notify::new()), + terminal_waker: Arc::new(Mutex::new(None)), + } + } + + pub(crate) fn signal(&self) { + self.notify.notify_one(); + if let Ok(guard) = self.terminal_waker.lock() { + if let Some(waker) = guard.as_ref() { + let _ = waker.wake(); + } + } + } + + pub(crate) fn install_terminal_waker(&self, waker: uncurses::event::Waker) { + if let Ok(mut guard) = self.terminal_waker.lock() { + *guard = Some(waker); + } + } + + fn notified(&self) -> tokio::sync::futures::Notified<'_> { + self.notify.notified() + } +} + +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum LoopWakeCause { + Terminal(T), + Producer, + Tick, +} + +/// Wait for terminal input, a [`LoopWake`] signal, or `timeout`. +/// +/// A zero timeout still prefers terminal input that is already queued, then +/// returns. That is the non-blocking poll the busy and agent-activity paths +/// already used. +pub(crate) async fn await_loop_wake( + timeout: Duration, + wake: &LoopWake, + tty: impl Future, +) -> LoopWakeCause { + tokio::pin!(tty); + if timeout.is_zero() { + tokio::select! { + biased; + value = &mut tty => return LoopWakeCause::Terminal(value), + () = std::future::ready(()) => return LoopWakeCause::Tick, + } + } + let notified = wake.notified(); + tokio::pin!(notified); + tokio::select! { + biased; + value = &mut tty => LoopWakeCause::Terminal(value), + () = &mut notified => LoopWakeCause::Producer, + () = tokio::time::sleep(timeout) => LoopWakeCause::Tick, + } +} + +/// Forward a Tokio channel onto a receiver the loop can `try_recv`, and signal +/// the loop after each item. Runtime agent sends already wake this task. +pub(crate) fn forward_unbounded( + mut rx: mpsc::UnboundedReceiver, + wake: LoopWake, +) -> mpsc::UnboundedReceiver { + let (tx, forwarded) = mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(item) = rx.recv().await { + if tx.send(item).is_err() { + break; + } + wake.signal(); + } + wake.signal(); + }); + forwarded +} + +pub(crate) fn forward_oneshot( + rx: oneshot::Receiver, + wake: LoopWake, +) -> oneshot::Receiver { + let (tx, forwarded) = oneshot::channel(); + tokio::spawn(async move { + if let Ok(value) = rx.await { + let _ = tx.send(value); + } + wake.signal(); + }); + forwarded +} + +/// Counted stand-in for one idle main-loop wait. Tests drive it under +/// `tokio::time::pause`. The crossterm path uses [`await_loop_wake`] with the +/// same timeout decision. +#[cfg(test)] +pub(crate) async fn run_notified_terminal_pump( + stop: tokio::sync::watch::Receiver, + wake: &LoopWake, + wakes: &std::sync::atomic::AtomicUsize, + input: impl Fn() -> TerminalPollInput, +) { + use std::sync::atomic::Ordering; + + loop { + if *stop.borrow() { + break; + } + let timeout = terminal_poll_timeout(input()); + let _cause = await_loop_wake(timeout, wake, std::future::pending::<()>()).await; + if *stop.borrow() { + break; + } + wakes.fetch_add(1, Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + use super::*; + + async fn pump_for( + input: impl Fn() -> TerminalPollInput + Send + 'static, + ) -> ( + LoopWake, + std::sync::Arc, + tokio::sync::watch::Sender, + tokio::task::JoinHandle<()>, + ) { + let wake = LoopWake::new(); + let wakes = std::sync::Arc::new(AtomicUsize::new(0)); + let (stop_tx, stop_rx) = tokio::sync::watch::channel(false); + let pump_wake = wake.clone(); + let pump_wakes = std::sync::Arc::clone(&wakes); + let pump = tokio::spawn(async move { + run_notified_terminal_pump(stop_rx, &pump_wake, &pump_wakes, input).await; + }); + tokio::task::yield_now().await; + (wake, wakes, stop_tx, pump) + } + + async fn stop_pump( + wake: &LoopWake, + stop_tx: tokio::sync::watch::Sender, + pump: tokio::task::JoinHandle<()>, + ) { + let _ = stop_tx.send(true); + wake.signal(); + pump.await.expect("terminal pump"); + } + + #[tokio::test(start_paused = true)] + async fn notified_terminal_loop_has_a_bounded_idle_wake_rate() { + let (wake, wakes, stop_tx, pump) = pump_for(TerminalPollInput::idle).await; + let initial = wakes.load(Ordering::Relaxed); + assert_eq!( + initial, 0, + "an idle loop must not wake before the first tick" + ); + + // Advance in small steps so Tokio cannot collapse a missed interval + // into one callback. This measures the work a real idle minute schedules. + for _ in 0..600 { + tokio::time::advance(Duration::from_millis(100)).await; + tokio::task::yield_now().await; + } + let idle_wakes = wakes.load(Ordering::Relaxed) - initial; + assert!( + idle_wakes <= 12, + "idle terminal loop woke {idle_wakes} times in one minute" + ); + + wake.signal(); + tokio::task::yield_now().await; + assert_eq!( + wakes.load(Ordering::Relaxed), + initial + idle_wakes + 1, + "a producer signal must wake the loop without waiting for the safety tick" + ); + stop_pump(&wake, stop_tx, pump).await; + } + + #[tokio::test(start_paused = true)] + async fn short_cadence_keeps_the_100ms_maintenance_poll() { + let (wake, wakes, stop_tx, pump) = pump_for(|| TerminalPollInput { + short_cadence: true, + ..TerminalPollInput::idle() + }) + .await; + let initial = wakes.load(Ordering::Relaxed); + for _ in 0..10 { + tokio::time::advance(Duration::from_millis(100)).await; + tokio::task::yield_now().await; + } + assert_eq!(wakes.load(Ordering::Relaxed), initial + 10); + stop_pump(&wake, stop_tx, pump).await; + } + + #[tokio::test(start_paused = true)] + async fn busy_frames_keep_the_33ms_poll() { + let (wake, wakes, stop_tx, pump) = pump_for(|| TerminalPollInput { + busy: true, + ..TerminalPollInput::idle() + }) + .await; + let initial = wakes.load(Ordering::Relaxed); + for _ in 0..10 { + tokio::time::advance(BUSY_FRAME_POLL).await; + tokio::task::yield_now().await; + } + assert_eq!(wakes.load(Ordering::Relaxed), initial + 10); + stop_pump(&wake, stop_tx, pump).await; + } + + #[tokio::test(start_paused = true)] + async fn agent_activity_does_not_wait() { + assert_eq!( + terminal_poll_timeout(TerminalPollInput { + agent_activity: true, + ..TerminalPollInput::idle() + }), + Duration::ZERO + ); + let wake = LoopWake::new(); + let cause = await_loop_wake(Duration::ZERO, &wake, std::future::pending::<()>()).await; + assert_eq!(cause, LoopWakeCause::Tick); + assert_eq!(tokio::time::Instant::now().elapsed(), Duration::ZERO); + } + + #[tokio::test(start_paused = true)] + async fn tty_event_wakes_the_idle_loop_before_the_safety_tick() { + let wake = LoopWake::new(); + let (tx, rx) = oneshot::channel::<()>(); + let wait = tokio::spawn(async move { + await_loop_wake(IDLE_SAFETY_TICK, &wake, async move { rx.await.ok() }).await + }); + tokio::task::yield_now().await; + tx.send(()).expect("tty signal"); + let cause = wait.await.expect("tty wait"); + assert_eq!(cause, LoopWakeCause::Terminal(Some(()))); + } + + #[tokio::test(start_paused = true)] + async fn forwarded_channel_send_wakes_before_the_safety_tick() { + let (wake, wakes, stop_tx, pump) = pump_for(TerminalPollInput::idle).await; + let (tx, rx) = mpsc::unbounded_channel(); + let mut forwarded = forward_unbounded(rx, wake.clone()); + tokio::task::yield_now().await; + let before = wakes.load(Ordering::Relaxed); + tx.send(7_u8).expect("enqueue"); + for _ in 0..8 { + tokio::task::yield_now().await; + if wakes.load(Ordering::Relaxed) > before { + break; + } + } + assert!( + wakes.load(Ordering::Relaxed) > before, + "a channel send must wake the idle loop before the safety tick" + ); + assert_eq!(forwarded.try_recv().expect("forwarded item"), 7); + stop_pump(&wake, stop_tx, pump).await; + } + + #[cfg(target_os = "linux")] + #[tokio::test] + #[ignore = "manual one-minute process CPU sample; use --ignored --nocapture"] + async fn notified_terminal_loop_idle_cpu_probe() { + fn process_cpu() -> Duration { + let mut clock = std::mem::MaybeUninit::::uninit(); + // SAFETY: clock points to writable timespec storage and the clock id + // is the Linux process CPU clock. + assert_eq!( + unsafe { libc::clock_gettime(libc::CLOCK_PROCESS_CPUTIME_ID, clock.as_mut_ptr()) }, + 0, + "read process CPU clock" + ); + // SAFETY: a successful clock_gettime initialized the whole timespec. + let clock = unsafe { clock.assume_init() }; + Duration::new(clock.tv_sec as u64, clock.tv_nsec as u32) + } + + let (wake, wakes, stop_tx, pump) = pump_for(TerminalPollInput::idle).await; + let before_wakes = wakes.load(Ordering::Relaxed); + let before_cpu = process_cpu(); + tokio::time::sleep(Duration::from_mins(1)).await; + let cpu = process_cpu() + .checked_sub(before_cpu) + .expect("process CPU clock is monotonic"); + let idle_wakes = wakes.load(Ordering::Relaxed) - before_wakes; + println!( + "idle_60s_process_cpu_ms={} wake_count={idle_wakes}", + cpu.as_millis() + ); + stop_pump(&wake, stop_tx, pump).await; + } +} diff --git a/packages/tui-rs/src/model_monitor.rs b/packages/tui-rs/src/model_monitor.rs index 80ec672ac..b123fd46b 100644 --- a/packages/tui-rs/src/model_monitor.rs +++ b/packages/tui-rs/src/model_monitor.rs @@ -53,6 +53,7 @@ pub fn spawn_model_monitor() -> (ModelMonitor, mpsc::Receiver uncurses::event::Waker { + self.source.waker() + } + /// Poll for one application event. /// /// Query replies unsupported by the application are deliberately consumed