diff --git a/Cargo.lock b/Cargo.lock index e567285f..3a7eb90b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12146,6 +12146,7 @@ dependencies = [ "async-channel", "async-trait", "base64 0.22.1", + "chrono", "code_assistant_core", "command_executor", "dirs 5.0.1", diff --git a/crates/code_assistant_core/src/mocks.rs b/crates/code_assistant_core/src/mocks.rs index e0ed8f96..d3f17dfe 100644 --- a/crates/code_assistant_core/src/mocks.rs +++ b/crates/code_assistant_core/src/mocks.rs @@ -1223,12 +1223,13 @@ impl ToolTestFixture { struct NullSink; #[async_trait::async_trait] impl crate::session::wakeup::WakeupSink for NullSink { - async fn fire(&self, _session_id: &str, _message: &str) {} + async fn fire(&self, _session_id: &str, _wakeup_id: u64, _message: &str) {} } let handle = crate::session::wakeup::spawn_wakeup_scheduler(NullSink, None); self.services.wakeups = Some(crate::session::wakeup::SessionWakeups { handle, session_id: "test-session".to_string(), + used_ids: 0, }); self } diff --git a/crates/code_assistant_core/src/session/instance.rs b/crates/code_assistant_core/src/session/instance.rs index 858b7abd..b7a493a8 100644 --- a/crates/code_assistant_core/src/session/instance.rs +++ b/crates/code_assistant_core/src/session/instance.rs @@ -668,6 +668,8 @@ impl SessionInstance { pending_permission_requests: self.pending_permission_requests.snapshot(), pending_new_context_target: self.pending_new_context_target.snapshot(), pending_questions: self.pending_questions.snapshot(), + // Filled in by the SessionManager, which holds the scheduler. + armed_wakeups: Vec::new(), }) } diff --git a/crates/code_assistant_core/src/session/lifecycle.rs b/crates/code_assistant_core/src/session/lifecycle.rs index 0c17de0d..7b0d8aee 100644 --- a/crates/code_assistant_core/src/session/lifecycle.rs +++ b/crates/code_assistant_core/src/session/lifecycle.rs @@ -190,6 +190,9 @@ pub enum SessionStatus { Working, /// The agent waits out a rate limit. RateLimited, + /// The agent stopped, but a wakeup it scheduled will continue the + /// session; nothing to do for the user. + Scheduled, /// The agent stopped with an error. Failed, /// Another code-assistant process runs this session. @@ -200,13 +203,19 @@ pub enum SessionStatus { } impl SessionStatus { - pub fn resolve(activity: &SessionActivityState, awaiting: Option) -> Self { + /// `wakeup_armed`: a wakeup is still to fire in the session. + pub fn resolve( + activity: &SessionActivityState, + awaiting: Option, + wakeup_armed: bool, + ) -> Self { match awaiting { Some(AwaitingUser::Approval) => return Self::NeedsApproval, Some(AwaitingUser::Answer) => return Self::NeedsAnswer, None => {} } match activity { + SessionActivityState::Idle if wakeup_armed => Self::Scheduled, SessionActivityState::Idle => Self::Ready, SessionActivityState::AgentRunning | SessionActivityState::WaitingForResponse => { Self::Working @@ -225,7 +234,7 @@ impl SessionStatus { return false; } match self { - Self::Working | Self::RateLimited | Self::RunningElsewhere => true, + Self::Working | Self::RateLimited | Self::Scheduled | Self::RunningElsewhere => true, Self::Ready => !unread, Self::NeedsApproval | Self::NeedsAnswer | Self::Failed => false, } @@ -458,24 +467,40 @@ mod tests { }, ] { assert_eq!( - SessionStatus::resolve(&activity, Some(AwaitingUser::Approval)), + SessionStatus::resolve(&activity, Some(AwaitingUser::Approval), true), SessionStatus::NeedsApproval ); assert_eq!( - SessionStatus::resolve(&activity, Some(AwaitingUser::Answer)), + SessionStatus::resolve(&activity, Some(AwaitingUser::Answer), true), SessionStatus::NeedsAnswer ); } assert_eq!( - SessionStatus::resolve(&SessionActivityState::WaitingForResponse, None), + SessionStatus::resolve(&SessionActivityState::WaitingForResponse, None, true), SessionStatus::Working ); assert_eq!( - SessionStatus::resolve(&SessionActivityState::Idle, None), + SessionStatus::resolve(&SessionActivityState::Idle, None, false), SessionStatus::Ready ); } + #[test] + fn idle_session_with_a_wakeup_to_fire_is_scheduled() { + assert_eq!( + SessionStatus::resolve(&SessionActivityState::Idle, None, true), + SessionStatus::Scheduled + ); + let failed = SessionActivityState::Errored { + message: "x".into(), + }; + assert_eq!( + SessionStatus::resolve(&failed, None, true), + SessionStatus::Failed + ); + assert!(SessionStatus::Scheduled.should_recede(true, false)); + } + #[test] fn busy_rows_recede_and_read_ready_rows_recede() { assert!(SessionStatus::Working.should_recede(true, false)); diff --git a/crates/code_assistant_core/src/session/manager.rs b/crates/code_assistant_core/src/session/manager.rs index 2fdbc4e4..5121057e 100644 --- a/crates/code_assistant_core/src/session/manager.rs +++ b/crates/code_assistant_core/src/session/manager.rs @@ -703,6 +703,11 @@ impl SessionManager { }; snapshot.current_model = model_name_for_event; + snapshot.armed_wakeups = self + .wakeup_handle + .as_ref() + .map(|handle| handle.armed(&session_id)) + .unwrap_or_default(); snapshot.allowed_models = self .allowed_models_for_session(&session_id) .unwrap_or_default(); @@ -1309,6 +1314,15 @@ impl SessionManager { crate::session::wakeup::SessionWakeups { handle, session_id: session_id.to_string(), + used_ids: self + .active_sessions + .get(session_id) + .map(|instance| { + crate::session::wakeup::highest_wakeup_id( + &instance.session.tool_executions, + ) + }) + .unwrap_or(0), } }), pty_sessions: self @@ -1590,6 +1604,60 @@ impl SessionManager { } /// Get a session instance by ID + /// Record in the stored result of `schedule_wakeup` that wakeup + /// `wakeup_id` fired, and return the result's new UI data. `None` when + /// this process armed no such wakeup in the session. + pub fn mark_wakeup_fired( + &mut self, + session_id: &str, + wakeup_id: u64, + ) -> Result> { + use crate::tools::core::{Render, ResourcesTracker}; + use crate::tools::impls::wakeup::ScheduleWakeupOutput; + + let mut fired = None; + let session = self.persistence.update_entry(session_id, |session| { + let armed_here = |output: &ScheduleWakeupOutput| { + output.wakeup_id == wakeup_id + && output.armed_by.as_deref() + == Some(crate::session::wakeup::process_instance()) + }; + for exec in &mut session.tool_executions { + if exec.tool_name != "schedule_wakeup" { + continue; + } + let Ok(mut output) = + serde_json::from_value::(exec.result_json.clone()) + else { + continue; + }; + if armed_here(&output) { + output.fired_at = Some(chrono::Utc::now()); + exec.result_json = serde_json::to_value(&output)?; + fired = Some((exec.tool_request.id.clone(), output)); + break; + } + } + Ok(()) + })?; + if let Some(instance) = self.active_sessions.get_mut(session_id) { + instance.session = session; + } + Ok(fired.map(|(tool_id, output)| { + let mut tracker = ResourcesTracker::new(); + crate::ui::ui_events::ToolResultData { + tool_id, + status: crate::ui::ToolStatus::Success, + message: Some(output.status()), + output: Some(output.render_for_ui(&mut tracker)), + styled_output: None, + duration_seconds: None, + images: Vec::new(), + output_deferred: false, + } + })) + } + pub fn get_session(&self, session_id: &str) -> Option<&SessionInstance> { self.active_sessions.get(session_id) } @@ -2129,7 +2197,13 @@ impl SessionManager { .active_sessions .get(&metadata.id) .is_some_and(|instance| !instance.get_activity_state().is_terminal()); - let busy = running_here || self.is_agent_locked_externally(&metadata.id); + // A wakeup still to fire continues the session. + let wakeup_armed = self + .wakeup_handle + .as_ref() + .is_some_and(|handle| !handle.armed(&metadata.id).is_empty()); + let busy = + running_here || wakeup_armed || self.is_agent_locked_externally(&metadata.id); let repo_root = metadata .branch .as_ref() diff --git a/crates/code_assistant_core/src/session/mod.rs b/crates/code_assistant_core/src/session/mod.rs index 591e18cb..4b307456 100644 --- a/crates/code_assistant_core/src/session/mod.rs +++ b/crates/code_assistant_core/src/session/mod.rs @@ -73,6 +73,8 @@ pub struct SessionSnapshot { pub pending_new_context_target: Option, /// `ask_question` requests still awaiting answers. pub pending_questions: Vec, + /// Wakeups still to fire, soonest first. + pub armed_wakeups: Vec, } impl SessionSnapshot { @@ -96,6 +98,10 @@ impl SessionSnapshot { session_id: self.session_id.clone(), activity_state: self.activity_state.clone(), }, + UiEvent::UpdateArmedWakeups { + session_id: self.session_id.clone(), + wakeups: self.armed_wakeups.clone(), + }, ]; // Show the error banner when connecting to an errored session. if let instance::SessionActivityState::Errored { message } = &self.activity_state { diff --git a/crates/code_assistant_core/src/session/service.rs b/crates/code_assistant_core/src/session/service.rs index 05b105e5..7dba773f 100644 --- a/crates/code_assistant_core/src/session/service.rs +++ b/crates/code_assistant_core/src/session/service.rs @@ -906,14 +906,35 @@ impl SessionService { .await } + /// Tell frontends which wakeups are still to fire in `session_id`. + pub(crate) fn publish_armed_wakeups( + &self, + session_id: &str, + wakeups: Vec, + ) { + self.events.publish_ui( + session_id, + UiEvent::UpdateArmedWakeups { + session_id: session_id.to_string(), + wakeups, + }, + ); + } + /// Deliver a fired wakeup (see [`crate::session::wakeup`]): inject /// `message` into the session and make sure a turn runs for it — started /// immediately when the session is idle, queued as the pending message /// when an agent is currently running. A session that no longer exists - /// swallows the wakeup silently. - pub async fn inject_wakeup(&self, session_id: String, message: String) -> Result<()> { + /// swallows the wakeup silently. Before the message, the stored result + /// of the `schedule_wakeup` call records that it fired. + pub async fn inject_wakeup( + &self, + session_id: String, + wakeup_id: u64, + message: String, + ) -> Result<()> { self.call_session(session_id.clone(), move |ctx| async move { - inject_wakeup_impl(&ctx, &session_id, &message).await + inject_wakeup_impl(&ctx, &session_id, wakeup_id, &message).await }) .await } @@ -1968,11 +1989,37 @@ async fn start_turn_if_idle_impl( ))) } -async fn inject_wakeup_impl(ctx: &ServiceCtx, session_id: &str, message: &str) -> Result<()> { - // Session deleted since the wakeup was armed — drop silently. - if ctx.manager.lock().await.get_session(session_id).is_none() { - debug!("Wakeup for unknown session {session_id} dropped"); - return Ok(()); +async fn inject_wakeup_impl( + ctx: &ServiceCtx, + session_id: &str, + wakeup_id: u64, + message: &str, +) -> Result<()> { + let fired = { + let mut manager = ctx.manager.lock().await; + // Session deleted since the wakeup was armed — drop silently. + if manager.get_session(session_id).is_none() { + debug!("Wakeup for unknown session {session_id} dropped"); + return Ok(()); + } + manager.mark_wakeup_fired(session_id, wakeup_id) + }; + match fired { + Ok(Some(result)) => ctx.notify_session( + session_id, + UiEvent::UpdateToolStatus { + tool_id: result.tool_id, + status: result.status, + message: result.message, + output: result.output, + styled_output: None, + duration_seconds: None, + images: Vec::new(), + }, + ), + Ok(None) => debug!("Wakeup {wakeup_id} has no result in session {session_id}"), + // The wakeup is still delivered. + Err(e) => warn!("Failed to record that wakeup {wakeup_id} fired: {e}"), } send_or_queue_user_message_impl(ctx, session_id, message, &[]).await } @@ -2263,6 +2310,7 @@ mod tests { use crate::mocks::{MockLLMProvider, PendingLLMProvider, create_test_response_text}; use crate::persistence::FileSessionPersistence; use crate::session::SessionConfig; + use crate::session::wakeup::wakeup_message; pub(super) fn test_service_with_manager( root: &std::path::Path, @@ -2719,7 +2767,7 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let service = test_service(tmp.path()); service - .inject_wakeup("gone".to_string(), "[scheduled wakeup] x".to_string()) + .inject_wakeup("gone".to_string(), 1, wakeup_message(1, "x")) .await .unwrap(); } @@ -2739,12 +2787,134 @@ mod tests { } service - .inject_wakeup(id.clone(), "[scheduled wakeup] check".to_string()) + .inject_wakeup(id.clone(), 1, wakeup_message(1, "check")) .await .unwrap(); let pending = service.take_pending_message(id).await.unwrap(); - assert_eq!(pending.as_deref(), Some("[scheduled wakeup] check")); + assert_eq!( + pending.as_deref(), + Some(wakeup_message(1, "check").as_str()) + ); + } + + /// A stored `schedule_wakeup` result for wakeup `wakeup_id`, armed by + /// `armed_by`. + fn store_wakeup_result( + root: &std::path::Path, + session_id: &str, + wakeup_id: u64, + armed_by: &str, + ) { + let output = crate::tools::impls::wakeup::ScheduleWakeupOutput { + wakeup_id, + delay_seconds: 60, + fires_at: Some(chrono::Utc::now()), + armed_by: Some(armed_by.to_string()), + fired_at: None, + }; + FileSessionPersistence::new_with_root_dir(root.to_path_buf()) + .update_entry(session_id, |session| { + session + .tool_executions + .push(agent_core::SerializedToolExecution { + tool_request: agent_core::ToolRequest { + id: format!("wake-{wakeup_id}"), + name: "schedule_wakeup".into(), + input: serde_json::json!({}), + start_offset: None, + end_offset: None, + }, + tool_name: "schedule_wakeup".into(), + result_json: serde_json::to_value(&output)?, + }); + Ok(()) + }) + .unwrap(); + } + + fn stored_fired_at(root: &std::path::Path, session_id: &str, tool_id: &str) -> Option { + let session = FileSessionPersistence::new_with_root_dir(root.to_path_buf()) + .load_chat_session(session_id) + .unwrap() + .unwrap(); + let exec = session + .tool_executions + .iter() + .find(|exec| exec.tool_request.id == tool_id) + .unwrap(); + exec.result_json + .get("fired_at") + .map(|fired_at| fired_at.to_string()) + } + + #[tokio::test] + async fn snapshot_lists_the_wakeups_still_to_fire() { + struct NullSink; + #[async_trait::async_trait] + impl crate::session::wakeup::WakeupSink for NullSink { + async fn fire(&self, _session_id: &str, _wakeup_id: u64, _message: &str) {} + } + let tmp = tempfile::tempdir().unwrap(); + let (service, manager) = test_service_with_manager(tmp.path()); + let handle = crate::session::wakeup::spawn_wakeup_scheduler(NullSink, None); + manager.lock().await.set_wakeup_handle(handle.clone()); + let id = service.create_session(None, None).await.unwrap(); + + handle.arm( + id.clone(), + 0, + std::time::Duration::from_secs(600), + "check".into(), + ); + tokio::task::yield_now().await; + let snapshot = service.load_session(id.clone(), None).await.unwrap(); + + assert_eq!(snapshot.armed_wakeups.len(), 1); + assert_eq!(snapshot.armed_wakeups[0].prompt, "check"); + assert!(snapshot.connect_events().iter().any(|event| matches!( + event, + UiEvent::UpdateArmedWakeups { wakeups, .. } if wakeups.len() == 1 + ))); + } + + #[tokio::test] + async fn fired_wakeup_is_recorded_in_its_result_and_shown() { + let tmp = tempfile::tempdir().unwrap(); + let service = test_service(tmp.path()); + let id = service.create_session(None, None).await.unwrap(); + let here = crate::session::wakeup::process_instance(); + store_wakeup_result(tmp.path(), &id, 1, here); + // The same id, armed by a process before a restart. + store_wakeup_result(tmp.path(), &id, 2, "earlier-process"); + + let mut subscription = service.subscribe(); + let _ = service + .inject_wakeup(id.clone(), 1, wakeup_message(1, "check")) + .await; + let _ = service + .inject_wakeup(id.clone(), 2, wakeup_message(2, "check")) + .await; + + assert!(stored_fired_at(tmp.path(), &id, "wake-1").is_some()); + assert_eq!(stored_fired_at(tmp.path(), &id, "wake-2"), None); + loop { + let event = subscription.recv().await.unwrap(); + if let crate::session::event_stream::EventPayload::Ui(UiEvent::UpdateToolStatus { + tool_id, + output, + .. + }) = event.payload + { + assert_eq!(tool_id, "wake-1"); + let output = crate::tools::impls::wakeup::ScheduleWakeupOutput::from_ui_output( + &output.unwrap(), + ) + .unwrap(); + assert!(output.fired_at.is_some()); + break; + } + } } #[tokio::test] @@ -2757,7 +2927,7 @@ mod tests { // the injected message must land in the history either way — exactly // like a user message sent to an idle session. let _ = service - .inject_wakeup(id.clone(), "[scheduled wakeup] check".to_string()) + .inject_wakeup(id.clone(), 1, wakeup_message(1, "check")) .await; let snapshot = service.load_session(id, None).await.unwrap(); diff --git a/crates/code_assistant_core/src/session/wakeup.rs b/crates/code_assistant_core/src/session/wakeup.rs index f45c7b07..1927fec6 100644 --- a/crates/code_assistant_core/src/session/wakeup.rs +++ b/crates/code_assistant_core/src/session/wakeup.rs @@ -5,49 +5,139 @@ //! //! Wakeups are deliberately not persisted: they die with the process. Durable //! cross-session scheduling is an application concern (see -//! `docs/session-wakeups.md`). +//! `docs/session-wakeups.md`). What is persisted is the `schedule_wakeup` +//! result: which process armed the wakeup and, once it fired, when — so a +//! frontend can tell an armed wakeup from one lost with a restart. +//! +//! Wakeup ids are numbered per session (1, 2, 3, … across restarts, seeded +//! from the session's history), so the model sees short ids; internally a +//! wakeup is identified by session and id. use crate::session::SessionService; use crate::session::sleep_inhibitor::SleepInhibitor; -use std::collections::BTreeMap; -use std::sync::Arc; +use agent_core::SerializedToolExecution; +use std::collections::{BTreeMap, HashMap}; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, LazyLock, Mutex}; use std::time::Duration; use tokio::sync::mpsc; use tokio::time::Instant; use tracing::{debug, warn}; -/// Prefix framing an injected wakeup message, so the agent and the transcript -/// can tell it from a human message. -pub const WAKEUP_PREFIX: &str = "[scheduled wakeup]"; +/// Start of the frame of an injected wakeup message, so the agent and the +/// transcript can tell it from a human message: `[scheduled wakeup #3] …`. +const WAKEUP_FRAME: &str = "[scheduled wakeup"; + +/// The message injected when wakeup `id` fires. +pub fn wakeup_message(id: u64, prompt: &str) -> String { + format!("{WAKEUP_FRAME} #{id}] {prompt}") +} + +/// An injected wakeup message, parsed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParsedWakeupMessage<'a> { + /// `None` for messages recorded before they carried the id. + pub id: Option, + pub prompt: &'a str, +} + +/// Parse an injected wakeup message; `None` for any other message. +pub fn parse_wakeup_message(message: &str) -> Option> { + let rest = message.strip_prefix(WAKEUP_FRAME)?; + let (id, prompt) = match rest.strip_prefix(" #") { + Some(numbered) => { + let (id, prompt) = numbered.split_once(']')?; + (Some(id.parse().ok()?), prompt) + } + None => (None, rest.strip_prefix(']')?), + }; + Some(ParsedWakeupMessage { + id, + prompt: prompt.trim(), + }) +} + +/// Identifies this process among the ones that ran before: a wakeup whose +/// result names another instance was lost with that process. +pub fn process_instance() -> &'static str { + static INSTANCE: LazyLock = LazyLock::new(|| { + let started = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + format!("{:x}-{started:x}", std::process::id()) + }); + &INSTANCE +} + +/// The highest wakeup id `executions` (a session's history) already use. +pub fn highest_wakeup_id(executions: &[SerializedToolExecution]) -> u64 { + executions + .iter() + .filter(|exec| exec.tool_name == "schedule_wakeup") + .filter_map(|exec| exec.result_json.get("wakeup_id")?.as_u64()) + .max() + .unwrap_or(0) +} + +/// A wakeup still to fire, as frontends show it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArmedWakeupInfo { + pub id: u64, + pub fires_at: chrono::DateTime, + pub prompt: String, +} /// Delivers a fired wakeup into a session. The production implementation is /// [`SessionService`]; tests substitute a recorder. #[async_trait::async_trait] pub trait WakeupSink: Send + Sync + 'static { - async fn fire(&self, session_id: &str, message: &str); + async fn fire(&self, session_id: &str, wakeup_id: u64, message: &str); + + /// The wakeups still to fire in `session_id` changed (soonest first). + fn armed_changed(&self, _session_id: &str, _armed: Vec) {} } #[async_trait::async_trait] impl WakeupSink for SessionService { - async fn fire(&self, session_id: &str, message: &str) { + async fn fire(&self, session_id: &str, wakeup_id: u64, message: &str) { if let Err(e) = self - .inject_wakeup(session_id.to_string(), message.to_string()) + .inject_wakeup(session_id.to_string(), wakeup_id, message.to_string()) .await { warn!("Failed to deliver wakeup to session {session_id}: {e}"); } } + + fn armed_changed(&self, session_id: &str, armed: Vec) { + self.publish_armed_wakeups(session_id, armed); + } } struct ArmedWakeup { session_id: String, + id: u64, fire_at: Instant, + /// `fire_at` on the wall clock, for frontends. + fires_at: chrono::DateTime, prompt: String, } +impl ArmedWakeup { + fn info(&self) -> ArmedWakeupInfo { + ArmedWakeupInfo { + id: self.id, + fires_at: self.fires_at, + prompt: self.prompt.clone(), + } + } +} + +/// The wakeups still to fire per session, kept by the scheduler task. +type ArmedBySession = Arc>>>; + enum Command { - Arm { id: u64, wakeup: ArmedWakeup }, + Arm(ArmedWakeup), Cancel { session_id: String, id: u64 }, CancelSession { session_id: String }, } @@ -56,27 +146,45 @@ enum Command { #[derive(Clone)] pub struct WakeupHandle { tx: mpsc::UnboundedSender, - next_id: Arc, + /// The last id issued per session. + issued: Arc>>, + armed: ArmedBySession, } impl WakeupHandle { - /// Arm a wakeup for `session_id`, firing after `delay`. Returns the - /// wakeup id (for cancellation). - pub fn arm(&self, session_id: String, delay: Duration, prompt: String) -> u64 { - let id = self.next_id.fetch_add(1, Ordering::Relaxed); - let _ = self.tx.send(Command::Arm { + /// Arm a wakeup for `session_id`, firing after `delay`. Returns its id + /// (for cancellation): the next one of the session, above `used`, the + /// highest id its history already uses. + pub fn arm(&self, session_id: String, used: u64, delay: Duration, prompt: String) -> u64 { + let id = { + let mut issued = self.issued.lock().unwrap(); + let last = issued.entry(session_id.clone()).or_default(); + *last = (*last).max(used) + 1; + *last + }; + let _ = self.tx.send(Command::Arm(ArmedWakeup { + session_id, id, - wakeup: ArmedWakeup { - session_id, - fire_at: Instant::now() + delay, - prompt, - }, - }); + fire_at: Instant::now() + delay, + fires_at: chrono::Utc::now() + + chrono::TimeDelta::from_std(delay).unwrap_or(chrono::TimeDelta::MAX), + prompt, + })); id } - /// Cancel a single wakeup. The session id must match the one the wakeup - /// was armed for — a session cannot cancel another session's wakeups. + /// The wakeups still to fire in `session_id`, soonest first. + pub fn armed(&self, session_id: &str) -> Vec { + self.armed + .lock() + .unwrap() + .get(session_id) + .cloned() + .unwrap_or_default() + } + + /// Cancel a single wakeup of a session — a session cannot cancel another + /// session's wakeups. pub fn cancel(&self, session_id: String, id: u64) { let _ = self.tx.send(Command::Cancel { session_id, id }); } @@ -93,11 +201,15 @@ impl WakeupHandle { pub struct SessionWakeups { pub handle: WakeupHandle, pub session_id: String, + /// The highest wakeup id the session's history uses + /// ([`highest_wakeup_id`]), so ids continue across restarts. + pub used_ids: u64, } impl SessionWakeups { pub fn arm(&self, delay: Duration, prompt: String) -> u64 { - self.handle.arm(self.session_id.clone(), delay, prompt) + self.handle + .arm(self.session_id.clone(), self.used_ids, delay, prompt) } pub fn cancel(&self, id: u64) { @@ -114,21 +226,49 @@ pub fn spawn_wakeup_scheduler( sleep_inhibitor: Option>, ) -> WakeupHandle { let (tx, rx) = mpsc::unbounded_channel(); - tokio::spawn(run_scheduler(rx, sink, sleep_inhibitor)); + let armed = ArmedBySession::default(); + tokio::spawn(run_scheduler(rx, sink, armed.clone(), sleep_inhibitor)); WakeupHandle { tx, - next_id: Arc::new(AtomicU64::new(1)), + issued: Arc::default(), + armed, } } +/// Bring the shared view of `session_id`'s wakeups in line with `armed` and +/// tell the sink. +fn publish_session( + session_id: &str, + armed: &BTreeMap, + shared: &ArmedBySession, + sink: &impl WakeupSink, +) { + let mut wakeups: Vec = armed + .values() + .filter(|w| w.session_id == session_id) + .map(ArmedWakeup::info) + .collect(); + wakeups.sort_by_key(|w| w.fires_at); + { + let mut shared = shared.lock().unwrap(); + if wakeups.is_empty() { + shared.remove(session_id); + } else { + shared.insert(session_id.to_string(), wakeups.clone()); + } + } + sink.armed_changed(session_id, wakeups); +} + async fn run_scheduler( mut rx: mpsc::UnboundedReceiver, sink: impl WakeupSink, + shared: ArmedBySession, sleep_inhibitor: Option>, ) { - // BTreeMap keyed by id: iteration order is arm order, which breaks - // fire-time ties deterministically. + // Keyed by arm order, which breaks fire-time ties deterministically. let mut armed: BTreeMap = BTreeMap::new(); + let next_key = AtomicU64::new(0); let inhibitor = InhibitorGuard::new(sleep_inhibitor); loop { @@ -140,39 +280,45 @@ async fn run_scheduler( // All handles dropped; nothing can arm or fire anymore. break; }; - match command { - Command::Arm { id, wakeup } => { + let session_id = match command { + Command::Arm(wakeup) => { debug!( - "Wakeup {id} armed for session {} in {:?}", + "Wakeup {} armed for session {} in {:?}", + wakeup.id, wakeup.session_id, wakeup.fire_at.saturating_duration_since(Instant::now()) ); - armed.insert(id, wakeup); + let session_id = wakeup.session_id.clone(); + armed.insert(next_key.fetch_add(1, Ordering::Relaxed), wakeup); + session_id } Command::Cancel { session_id, id } => { - if armed.get(&id).is_some_and(|w| w.session_id == session_id) { - armed.remove(&id); - debug!("Wakeup {id} cancelled"); - } + armed.retain(|_, w| !(w.session_id == session_id && w.id == id)); + debug!("Wakeup {id} of session {session_id} cancelled"); + session_id } Command::CancelSession { session_id } => { armed.retain(|_, w| w.session_id != session_id); + session_id } - } + }; + publish_session(&session_id, &armed, &shared, &sink); } _ = sleep_until_or_never(next_deadline) => { let now = Instant::now(); let due: Vec = armed .iter() .filter(|(_, w)| w.fire_at <= now) - .map(|(id, _)| *id) + .map(|(key, _)| *key) .collect(); - for id in due { - let wakeup = armed.remove(&id).expect("due id was just collected"); - debug!("Wakeup {id} fires for session {}", wakeup.session_id); + for key in due { + let wakeup = armed.remove(&key).expect("due key was just collected"); + debug!("Wakeup {} fires for session {}", wakeup.id, wakeup.session_id); + publish_session(&wakeup.session_id, &armed, &shared, &sink); sink.fire( &wakeup.session_id, - &format!("{WAKEUP_PREFIX} {}", wakeup.prompt), + wakeup.id, + &wakeup_message(wakeup.id, &wakeup.prompt), ) .await; } @@ -227,16 +373,23 @@ impl Drop for InhibitorGuard { #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; #[derive(Default)] struct RecordingSink { fired: Mutex>, + armed: Mutex)>>, } #[async_trait::async_trait] impl WakeupSink for Arc { - async fn fire(&self, session_id: &str, message: &str) { + fn armed_changed(&self, session_id: &str, armed: Vec) { + self.armed + .lock() + .unwrap() + .push((session_id.to_string(), armed.iter().map(|w| w.id).collect())); + } + + async fn fire(&self, session_id: &str, _wakeup_id: u64, message: &str) { self.fired .lock() .unwrap() @@ -244,6 +397,45 @@ mod tests { } } + #[test] + fn wakeup_messages_parse_with_and_without_id() { + assert_eq!( + parse_wakeup_message(&wakeup_message(3, "check the build")), + Some(ParsedWakeupMessage { + id: Some(3), + prompt: "check the build" + }) + ); + // Recorded before messages carried the id. + assert_eq!( + parse_wakeup_message("[scheduled wakeup] check the build"), + Some(ParsedWakeupMessage { + id: None, + prompt: "check the build" + }) + ); + assert_eq!(parse_wakeup_message("check the build"), None); + } + + #[test] + fn ids_count_per_session_and_continue_the_history() { + let (tx, _rx) = mpsc::unbounded_channel(); + let handle = WakeupHandle { + tx, + issued: Arc::default(), + armed: Arc::default(), + }; + let arm = |session: &str, used| { + handle.arm(session.into(), used, Duration::from_secs(1), "x".into()) + }; + assert_eq!(arm("s1", 0), 1); + assert_eq!(arm("s1", 0), 2); + assert_eq!(arm("s2", 0), 1); + // A session whose history (from before a restart) used ids up to 7. + assert_eq!(arm("s3", 7), 8); + assert_eq!(arm("s3", 7), 9); + } + fn recording_scheduler() -> (WakeupHandle, Arc) { let sink = Arc::new(RecordingSink::default()); let handle = spawn_wakeup_scheduler(sink.clone(), None); @@ -259,7 +451,12 @@ mod tests { #[tokio::test(start_paused = true)] async fn fires_after_delay_with_framed_message() { let (handle, sink) = recording_scheduler(); - handle.arm("s1".into(), Duration::from_secs(300), "check build".into()); + handle.arm( + "s1".into(), + 0, + Duration::from_secs(300), + "check build".into(), + ); settle().await; tokio::time::advance(Duration::from_secs(299)).await; @@ -271,14 +468,46 @@ mod tests { let fired = sink.fired.lock().unwrap().clone(); assert_eq!( fired, - vec![("s1".to_string(), format!("{WAKEUP_PREFIX} check build"))] + vec![("s1".to_string(), wakeup_message(1, "check build"))] + ); + } + + #[tokio::test(start_paused = true)] + async fn armed_wakeups_are_reported_per_session() { + let (handle, sink) = recording_scheduler(); + handle.arm("s1".into(), 0, Duration::from_secs(20), "later".into()); + handle.arm("s1".into(), 0, Duration::from_secs(10), "sooner".into()); + handle.arm("s2".into(), 0, Duration::from_secs(30), "other".into()); + settle().await; + let ids = |session| -> Vec { handle.armed(session).iter().map(|w| w.id).collect() }; + // Soonest first. + assert_eq!(ids("s1"), vec![2, 1]); + assert_eq!(ids("s2"), vec![1]); + + tokio::time::advance(Duration::from_secs(11)).await; + settle().await; + assert_eq!(ids("s1"), vec![1]); + handle.cancel("s1".into(), 1); + settle().await; + assert!(handle.armed("s1").is_empty()); + + let reported = sink.armed.lock().unwrap().clone(); + assert_eq!( + reported, + vec![ + ("s1".to_string(), vec![1]), + ("s1".to_string(), vec![2, 1]), + ("s2".to_string(), vec![1]), + ("s1".to_string(), vec![1]), + ("s1".to_string(), vec![]), + ] ); } #[tokio::test(start_paused = true)] async fn cancelled_wakeup_does_not_fire() { let (handle, sink) = recording_scheduler(); - let id = handle.arm("s1".into(), Duration::from_secs(60), "x".into()); + let id = handle.arm("s1".into(), 0, Duration::from_secs(60), "x".into()); settle().await; handle.cancel("s1".into(), id); settle().await; @@ -288,10 +517,26 @@ mod tests { assert!(sink.fired.lock().unwrap().is_empty()); } + #[tokio::test(start_paused = true)] + async fn cancel_hits_only_that_sessions_wakeup_of_the_id() { + let (handle, sink) = recording_scheduler(); + let id = handle.arm("s1".into(), 0, Duration::from_secs(60), "a".into()); + let same_id = handle.arm("s2".into(), 0, Duration::from_secs(60), "b".into()); + assert_eq!(id, same_id); + settle().await; + handle.cancel("s1".into(), id); + settle().await; + + tokio::time::advance(Duration::from_secs(61)).await; + settle().await; + let fired = sink.fired.lock().unwrap().clone(); + assert_eq!(fired, vec![("s2".to_string(), wakeup_message(1, "b"))]); + } + #[tokio::test(start_paused = true)] async fn cancel_requires_matching_session() { let (handle, sink) = recording_scheduler(); - let id = handle.arm("s1".into(), Duration::from_secs(60), "x".into()); + let id = handle.arm("s1".into(), 0, Duration::from_secs(60), "x".into()); settle().await; handle.cancel("someone-else".into(), id); settle().await; @@ -304,9 +549,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn cancel_session_removes_all_of_that_session() { let (handle, sink) = recording_scheduler(); - handle.arm("s1".into(), Duration::from_secs(10), "a".into()); - handle.arm("s1".into(), Duration::from_secs(20), "b".into()); - handle.arm("s2".into(), Duration::from_secs(30), "c".into()); + handle.arm("s1".into(), 0, Duration::from_secs(10), "a".into()); + handle.arm("s1".into(), 0, Duration::from_secs(20), "b".into()); + handle.arm("s2".into(), 0, Duration::from_secs(30), "c".into()); settle().await; handle.cancel_session("s1".into()); settle().await; @@ -321,8 +566,8 @@ mod tests { #[tokio::test(start_paused = true)] async fn multiple_wakeups_fire_in_deadline_order() { let (handle, sink) = recording_scheduler(); - handle.arm("s1".into(), Duration::from_secs(20), "second".into()); - handle.arm("s1".into(), Duration::from_secs(10), "first".into()); + handle.arm("s1".into(), 0, Duration::from_secs(20), "second".into()); + handle.arm("s1".into(), 0, Duration::from_secs(10), "first".into()); settle().await; tokio::time::advance(Duration::from_secs(15)).await; @@ -339,9 +584,9 @@ mod tests { #[tokio::test(start_paused = true)] async fn rearming_while_sleeping_shortens_the_deadline() { let (handle, sink) = recording_scheduler(); - handle.arm("s1".into(), Duration::from_secs(3600), "slow".into()); + handle.arm("s1".into(), 0, Duration::from_secs(3600), "slow".into()); settle().await; - handle.arm("s1".into(), Duration::from_secs(5), "fast".into()); + handle.arm("s1".into(), 0, Duration::from_secs(5), "fast".into()); settle().await; tokio::time::advance(Duration::from_secs(6)).await; diff --git a/crates/code_assistant_core/src/tools/impls/wakeup.rs b/crates/code_assistant_core/src/tools/impls/wakeup.rs index 4d4a199c..5fc1c33e 100644 --- a/crates/code_assistant_core/src/tools/impls/wakeup.rs +++ b/crates/code_assistant_core/src/tools/impls/wakeup.rs @@ -2,6 +2,7 @@ //! of its own session. See `crate::session::wakeup` and //! `docs/session-wakeups.md`. +use crate::session::wakeup::process_instance; use crate::tools::ToolServicesAccess; use crate::tools::core::{ Render, ResourcesTracker, Tool, ToolContext, ToolResult, ToolSpec, capabilities, @@ -20,10 +21,28 @@ pub struct ScheduleWakeupInput { pub prompt: String, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ScheduleWakeupOutput { pub wakeup_id: u64, pub delay_seconds: u64, + /// Wall-clock time the wakeup fires; absent in results recorded before + /// it was introduced. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fires_at: Option>, + /// The process that armed it ([`process_instance`]): a wakeup armed by + /// another process that never fired was lost with it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub armed_by: Option, + /// When it fired; written into the stored result by the scheduler. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fired_at: Option>, +} + +impl ScheduleWakeupOutput { + /// Parse the UI output ([`Render::render_for_ui`]) back into the result. + pub fn from_ui_output(output: &str) -> Option { + serde_json::from_str(output).ok() + } } impl Render for ScheduleWakeupOutput { @@ -34,6 +53,8 @@ impl Render for ScheduleWakeupOutput { ) } + // Leaves out `fired_at`: the text is part of the conversation the model + // sees and must not change after the fact (prompt cache). fn render(&self, _tracker: &mut ResourcesTracker) -> String { format!( "Wakeup #{} armed: this session will be woken with your prompt in {} seconds. \ @@ -41,6 +62,11 @@ impl Render for ScheduleWakeupOutput { self.wakeup_id, self.delay_seconds ) } + + /// The result as JSON, so frontends can show the fire time. + fn render_for_ui(&self, _tracker: &mut ResourcesTracker) -> String { + serde_json::to_string(self).unwrap_or_default() + } } impl ToolResult for ScheduleWakeupOutput { @@ -60,7 +86,7 @@ impl Tool for ScheduleWakeupTool { ToolSpec { name: "schedule_wakeup".into(), description: "Schedule a wakeup for this session: after the given delay, a new turn \ - is started with your prompt injected as a message (prefixed '[scheduled wakeup]'). \ + is started with your prompt injected as a message (prefixed '[scheduled wakeup #]'). \ Use it to check back on long-running or external work instead of polling. \ Wakeups are not persisted: they are lost when the application exits." .into(), @@ -112,11 +138,17 @@ impl Tool for ScheduleWakeupTool { .wakeups .as_ref() .ok_or_else(|| anyhow!("Wakeups are unavailable in this context"))?; - let wakeup_id = wakeups.arm(Duration::from_secs(input.delay_seconds), prompt.to_string()); + let delay = Duration::from_secs(input.delay_seconds); + let wakeup_id = wakeups.arm(delay, prompt.to_string()); Ok(ScheduleWakeupOutput { wakeup_id, delay_seconds: input.delay_seconds, + fires_at: chrono::TimeDelta::from_std(delay) + .ok() + .map(|delay| chrono::Utc::now() + delay), + armed_by: Some(process_instance().to_string()), + fired_at: None, }) } } @@ -220,6 +252,28 @@ mod tests { let output = tool.execute(&mut context, &mut input).await.unwrap(); assert!(output.wakeup_id >= 1); assert_eq!(output.delay_seconds, 60); + let fires_in = output.fires_at.unwrap() - chrono::Utc::now(); + assert!((55..=60).contains(&fires_in.num_seconds())); + } + + #[test] + fn ui_output_round_trips() { + let output = ScheduleWakeupOutput { + wakeup_id: 3, + delay_seconds: 1200, + fires_at: Some(chrono::Utc::now()), + armed_by: Some("instance".into()), + fired_at: Some(chrono::Utc::now()), + }; + let ui = output.render_for_ui(&mut ResourcesTracker::new()); + assert_eq!(ScheduleWakeupOutput::from_ui_output(&ui), Some(output)); + } + + #[test] + fn results_recorded_without_fire_time_still_parse() { + let parsed = + ScheduleWakeupOutput::from_ui_output(r#"{"wakeup_id":1,"delay_seconds":60}"#).unwrap(); + assert_eq!(parsed.fires_at, None); } #[tokio::test] diff --git a/crates/code_assistant_core/src/ui/ui_events.rs b/crates/code_assistant_core/src/ui/ui_events.rs index 7c87e86f..df79a6ba 100644 --- a/crates/code_assistant_core/src/ui/ui_events.rs +++ b/crates/code_assistant_core/src/ui/ui_events.rs @@ -262,6 +262,12 @@ pub enum UiEvent { session_id: String, activity_state: SessionActivityState, }, + /// The wakeups still to fire in a session changed (soonest first). + /// Relevant for every view's sidebar, not only the session's own. + UpdateArmedWakeups { + session_id: String, + wakeups: Vec, + }, /// Update pending message display UpdatePendingMessage { message: Option }, /// Display an error message to the user diff --git a/crates/ui_acp/src/ui.rs b/crates/ui_acp/src/ui.rs index dff2745d..fdea32ee 100644 --- a/crates/ui_acp/src/ui.rs +++ b/crates/ui_acp/src/ui.rs @@ -327,6 +327,10 @@ impl ToolCallState { } else { content.push(text_content(output)); } + } else if self.tool_name.as_deref() == Some("schedule_wakeup") { + content.push(text_content( + render_wakeup_output(&output).unwrap_or(output), + )); } else { content.push(text_content(output)); } @@ -436,6 +440,25 @@ fn text_content(text: String) -> acp::ToolCallContent { /// Render SubAgentOutput JSON as markdown for ACP display. /// Returns None if the JSON is not valid SubAgentOutput. +/// `schedule_wakeup` reports its result as JSON for frontends; say it in words. +fn render_wakeup_output(json_str: &str) -> Option { + use code_assistant_core::tools::impls::wakeup::ScheduleWakeupOutput; + let output = ScheduleWakeupOutput::from_ui_output(json_str)?; + Some(match output.fires_at { + Some(fires_at) => format!( + "Wakeup #{} armed, fires at {}", + output.wakeup_id, + fires_at + .with_timezone(&chrono::Local) + .format("%Y-%m-%d %H:%M:%S") + ), + None => format!( + "Wakeup #{} armed, fires in {}s", + output.wakeup_id, output.delay_seconds + ), + }) +} + fn render_sub_agent_output_as_markdown(json_str: &str) -> Option { use code_assistant_core::agent::sub_agent::{SubAgentOutput, SubAgentToolStatus}; @@ -944,6 +967,7 @@ impl UserInterface for ACPUserUI { | UiEvent::UpdateSessionLifecycle { .. } | UiEvent::ClearMessages | UiEvent::UpdateSessionActivityState { .. } + | UiEvent::UpdateArmedWakeups { .. } | UiEvent::UpdatePendingMessage { .. } | UiEvent::ClearError | UiEvent::UpdateCurrentModel { .. } diff --git a/crates/ui_gpui/Cargo.toml b/crates/ui_gpui/Cargo.toml index 42f9c46e..a7251b47 100644 --- a/crates/ui_gpui/Cargo.toml +++ b/crates/ui_gpui/Cargo.toml @@ -33,6 +33,9 @@ async-trait = "0.1" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +# Wakeup fire times in local time +chrono = "0.4" + anyhow = "1.0" tracing = "0.1" dirs = "5.0" diff --git a/crates/ui_gpui/assets/icons/alarm_clock.svg b/crates/ui_gpui/assets/icons/alarm_clock.svg new file mode 100644 index 00000000..af459591 --- /dev/null +++ b/crates/ui_gpui/assets/icons/alarm_clock.svg @@ -0,0 +1 @@ + diff --git a/crates/ui_gpui/assets/icons/alarm_clock_check.svg b/crates/ui_gpui/assets/icons/alarm_clock_check.svg new file mode 100644 index 00000000..e937ecd4 --- /dev/null +++ b/crates/ui_gpui/assets/icons/alarm_clock_check.svg @@ -0,0 +1 @@ + diff --git a/crates/ui_gpui/assets/icons/alarm_clock_off.svg b/crates/ui_gpui/assets/icons/alarm_clock_off.svg new file mode 100644 index 00000000..b5e06986 --- /dev/null +++ b/crates/ui_gpui/assets/icons/alarm_clock_off.svg @@ -0,0 +1 @@ + diff --git a/crates/ui_gpui/src/app/event_bridge.rs b/crates/ui_gpui/src/app/event_bridge.rs index ff3d5a1d..6d17ff27 100644 --- a/crates/ui_gpui/src/app/event_bridge.rs +++ b/crates/ui_gpui/src/app/event_bridge.rs @@ -101,6 +101,7 @@ impl Gpui { let forward = match &ui_event { // Sidebar state: relevant for every session, always. UiEvent::UpdateSessionActivityState { .. } + | UiEvent::UpdateArmedWakeups { .. } | UiEvent::UpdateSessionMetadata { .. } | UiEvent::UpdateSessionLifecycle { .. } | UiEvent::UpdateChatList { .. } @@ -464,6 +465,7 @@ mod tests { pending_permission_requests, pending_new_context_target: None, pending_questions: Vec::new(), + armed_wakeups: Vec::new(), } } diff --git a/crates/ui_gpui/src/app/event_loop.rs b/crates/ui_gpui/src/app/event_loop.rs index 3b8bfa3b..b66853b3 100644 --- a/crates/ui_gpui/src/app/event_loop.rs +++ b/crates/ui_gpui/src/app/event_loop.rs @@ -8,7 +8,7 @@ use code_assistant_core::ui::ui_events::ToolResultData; use code_assistant_core::ui::{DisplayFragment, UiEvent}; use std::collections::HashMap; -use super::super::blocks::{MessageContainer, MessageRole}; +use super::super::blocks::{MessageContainer, MessageRole, WakeupLink}; use gpui_kit::Entity; use tracing::{debug, trace, warn}; @@ -154,7 +154,9 @@ impl Gpui { new_message }); - queue.push(new_message); + queue.push(new_message.clone()); + drop(queue); + self.recognize_wakeup(&new_message, cx); } // Sync ListState and reset pending message @@ -352,6 +354,7 @@ impl Gpui { // This ensures the virtual list has fine-grained items (one per // LLM request/node) so that GPUI's list virtualization can skip // off-screen items efficiently. + let mut added = Vec::new(); for message_data in messages { let current_container = { let mut queue = self.message_queue.lock().unwrap(); @@ -381,10 +384,14 @@ impl Gpui { message_data.fragments, cx, ); + added.push(current_container); } // Apply tool results to update tool blocks with their execution results self.apply_tool_results(&tool_results, cx); + for container in &added { + self.recognize_wakeup(container, cx); + } // Ensure we always end with an Assistant container // This is crucial for sessions that are waiting for responses or actively running agents @@ -461,6 +468,7 @@ impl Gpui { let session_id = self.current_session_id.lock().unwrap().clone(); + let mut added = Vec::new(); for message_data in messages { let current_container = { let mut queue = self.message_queue.lock().unwrap(); @@ -489,10 +497,14 @@ impl Gpui { message_data.fragments, cx, ); + added.push(current_container); } // Apply tool results self.apply_tool_results(&tool_results, cx); + for container in &added { + self.recognize_wakeup(container, cx); + } self.notify_messages_appended(old_len, cx); } @@ -672,6 +684,20 @@ impl Gpui { sidebar.update_session_lifecycle(session_id, lifecycle, cx); }); } + UiEvent::UpdateArmedWakeups { + session_id, + wakeups, + } => { + let next = wakeups.first().map(|wakeup| wakeup.fires_at); + self.update_project_sidebar(cx, |sidebar, cx| { + sidebar.update_session_wakeup(session_id.clone(), next, cx); + }); + if self.is_current_session(&session_id) { + self.update_messages_view(cx, |view, cx| { + view.set_armed_wakeups(wakeups, cx); + }); + } + } UiEvent::UpdateSessionActivityState { session_id, activity_state, @@ -1020,6 +1046,7 @@ impl Gpui { } // Process message data + let mut added = Vec::new(); for message_data in messages { let current_container = { let mut queue = self.message_queue.lock().unwrap(); @@ -1048,10 +1075,14 @@ impl Gpui { message_data.fragments, cx, ); + added.push(current_container); } // Apply tool results self.apply_tool_results(&tool_results, cx); + for container in &added { + self.recognize_wakeup(container, cx); + } // Ensure we end with an Assistant container for the edit response { @@ -1296,4 +1327,30 @@ impl Gpui { } } } + + /// If `message` is a wakeup message, mark it as one, and mark the + /// `schedule_wakeup` block it answers as fired when that is in the turn + /// the message follows. Expects the tool results before `message` to be + /// applied: the block is found by the wakeup id in its result. + fn recognize_wakeup(&self, message: &Entity, cx: &mut gpui_kit::AsyncApp) { + let Some(mut wakeup) = cx.update_entity(message, |message, cx| message.parse_wakeup(cx)) + else { + return; + }; + let queue = self.message_queue.lock().unwrap().clone(); + let Some(position) = queue.iter().position(|c| c == message) else { + return; + }; + for container in queue[..position].iter().rev() { + match cx.update_entity(container, |c, cx| c.link_wakeup(&wakeup, cx)) { + WakeupLink::Linked => { + wakeup.linked = true; + break; + } + WakeupLink::TurnBoundary => break, + WakeupLink::NotHere => {} + } + } + cx.update_entity(message, |message, _| message.set_wakeup(wakeup)); + } } diff --git a/crates/ui_gpui/src/blocks/container.rs b/crates/ui_gpui/src/blocks/container.rs index 9534fc76..a026d4a3 100644 --- a/crates/ui_gpui/src/blocks/container.rs +++ b/crates/ui_gpui/src/blocks/container.rs @@ -9,6 +9,7 @@ use code_assistant_core::persistence::{BranchInfo, NodeId}; use crate::shared::image; use agent_core::ui::ContextBoundary; +use code_assistant_core::tools::impls::wakeup::ScheduleWakeupOutput; use code_assistant_core::ui::ToolStatus; use code_assistant_core::ui::ui_events::ToolResultData; use gpui_kit::{Context, Entity, prelude::*}; @@ -27,6 +28,29 @@ pub enum HiddenToolBlockType { Thinking, } +/// A user message the wakeup scheduler injected (`[scheduled wakeup #3] …`). +#[derive(Debug, Clone, PartialEq)] +pub struct WakeupMessage { + /// The wakeup's id; `None` in messages recorded before they carried it. + pub id: Option, + /// The prompt the agent passed to `schedule_wakeup`. + pub prompt: String, + /// The `schedule_wakeup` block it answers is in the same turn: that block + /// shows the wakeup fired, and this message is not shown at all. + pub linked: bool, +} + +/// Outcome of looking for the `schedule_wakeup` block a wakeup answers. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum WakeupLink { + /// The block was found and marked as fired. + Linked, + /// Not in this container, keep looking further back. + NotHere, + /// A user message: the turn starts here, stop looking. + TurnBoundary, +} + /// Container for all elements within a single message (one LLM request/response). /// Each MessageContainer maps to exactly one item in the virtualized list, /// enabling efficient scroll virtualization. @@ -52,6 +76,8 @@ pub struct MessageContainer { session_id: Arc>>, /// Monotonic block identifier source for stable per-block view state. next_block_id: Arc>, + /// Set when this is a wakeup message rather than one the user wrote. + wakeup: Arc>>, } impl MessageContainer { @@ -67,6 +93,7 @@ impl MessageContainer { branch_info: Arc::new(Mutex::new(None)), session_id: Arc::new(Mutex::new(None)), next_block_id: Arc::new(Mutex::new(0)), + wakeup: Arc::new(Mutex::new(None)), } } @@ -174,6 +201,9 @@ impl MessageContainer { /// `": "` with the text of its text blocks, for assertions. #[cfg(test)] pub(crate) fn transcript_line(&self, cx: &gpui_kit::App) -> String { + if let Some(wakeup) = self.wakeup() { + return format!("wakeup: {}", wakeup.prompt); + } let role = match self.role { MessageRole::User => "user", MessageRole::Assistant => "assistant", @@ -198,6 +228,78 @@ impl MessageContainer { .find_map(|block| block.read(cx).tool_block(tool_id).cloned()) } + /// Mark this user message as an injected wakeup. + pub fn set_wakeup(&self, wakeup: WakeupMessage) { + *self.wakeup.lock().unwrap() = Some(wakeup); + } + + pub fn wakeup(&self) -> Option { + self.wakeup.lock().unwrap().clone() + } + + /// The wakeup, not yet linked, if this is a user message the wakeup + /// scheduler injected. + pub fn parse_wakeup(&self, cx: &gpui_kit::App) -> Option { + if !self.is_user_message() { + return None; + } + let first = self.elements().into_iter().next()?; + let BlockData::TextBlock(text) = first.read(cx).block.as_ref() else { + return None; + }; + let parsed = code_assistant_core::session::wakeup::parse_wakeup_message(&text.content)?; + Some(WakeupMessage { + id: parsed.id, + prompt: parsed.prompt.to_string(), + linked: false, + }) + } + + /// Look for the `schedule_wakeup` block `wakeup` answers — by its id, or + /// for messages without one, the not yet fired block with its prompt — + /// and mark it as fired. + pub fn link_wakeup(&self, wakeup: &WakeupMessage, cx: &mut Context) -> WakeupLink { + if self.is_user_message() { + return WakeupLink::TurnBoundary; + } + let answers = |tool: &ToolUseBlock| { + if tool.name != "schedule_wakeup" { + return false; + } + match wakeup.id { + Some(id) => tool + .output + .as_deref() + .and_then(ScheduleWakeupOutput::from_ui_output) + .is_some_and(|output| output.wakeup_id == id), + None => { + !tool.wakeup_fired + && tool + .parameters + .iter() + .any(|p| p.name == "prompt" && p.value.trim() == wakeup.prompt) + } + } + }; + for element in self.elements().iter().rev() { + let linked = element.update(cx, |view, cx| { + let matches = view.block.as_tool().is_some_and(answers); + if matches { + view.block_mut() + .as_tool_mut() + .expect("a tool block") + .wakeup_fired = true; + cx.notify(); + } + matches + }); + if linked { + return WakeupLink::Linked; + } + } + WakeupLink::NotHere + } + pub fn is_user_message(&self) -> bool { self.role == MessageRole::User } @@ -400,6 +502,7 @@ impl MessageContainer { duration_seconds, images: Vec::new(), revision: 0, + wakeup_fired: false, }); let view = cx.new(|cx| { BlockView::new( @@ -726,6 +829,7 @@ impl MessageContainer { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, }; tool.parameters.push(ParameterBlock { diff --git a/crates/ui_gpui/src/blocks/data.rs b/crates/ui_gpui/src/blocks/data.rs index fc0c9803..3177d7f8 100644 --- a/crates/ui_gpui/src/blocks/data.rs +++ b/crates/ui_gpui/src/blocks/data.rs @@ -74,6 +74,10 @@ pub struct ToolUseBlock { /// Counts mutable accesses (see [`BlockData::as_tool_mut`]), so caches /// derived from the block can tell cheaply that nothing changed. pub revision: u64, + /// A `schedule_wakeup` block whose wakeup message followed it: the block + /// shows that it fired, the message itself is not shown (see + /// `MessageContainer::link_wakeup`). + pub wakeup_fired: bool, } /// Parameter for a tool diff --git a/crates/ui_gpui/src/blocks/mod.rs b/crates/ui_gpui/src/blocks/mod.rs index e8b40c4c..9309214b 100644 --- a/crates/ui_gpui/src/blocks/mod.rs +++ b/crates/ui_gpui/src/blocks/mod.rs @@ -219,6 +219,9 @@ pub struct BlockView { /// Timer that clears [`Self::copied_feedback_until`] after the checkmark /// feedback window elapses. Dropped/replaced on each copy. copied_feedback_task: Option>, + /// Timer that renders an inline tool block again whose description + /// changes with time (`ToolBlockRenderer::refresh_after`). + refresh_task: Option>, /// Current whole-line selection within this block's diff card body, if /// any. Drag-selecting rows sets it; the header's Copy button and /// Cmd/Ctrl-C read from it. Only meaningful for file-editing tool blocks. @@ -298,6 +301,7 @@ impl BlockView { write_file_diff_mode, copied_feedback_until: None, copied_feedback_task: None, + refresh_task: None, diff_selection: None, diff_dragging: false, focus_handle: cx.focus_handle(), diff --git a/crates/ui_gpui/src/blocks/render.rs b/crates/ui_gpui/src/blocks/render.rs index 1f7ec122..ec3a0235 100644 --- a/crates/ui_gpui/src/blocks/render.rs +++ b/crates/ui_gpui/src/blocks/render.rs @@ -159,7 +159,9 @@ impl BlockView { let theme = cx.theme().clone(); // Icon - let icon = file_icons::get().get_tool_icon(&block.name); + let icon = renderer + .icon(block) + .or_else(|| file_icons::get().get_tool_icon(&block.name)); let (icon_color, desc_color) = match block.status { ToolStatus::Error => (theme.danger, theme.danger), ToolStatus::Running | ToolStatus::Pending | ToolStatus::Success => { @@ -181,6 +183,20 @@ impl BlockView { // Optional right-aligned pill (e.g. the MCP server name). let header_tag = renderer.header_tag(block); + // A description that changes with time (a countdown) renders again + // when it is due to. + if self.refresh_task.is_none() + && let Some(after) = renderer.refresh_after(block) + { + self.refresh_task = Some(cx.spawn(async move |view, cx| { + cx.background_executor().timer(after).await; + let _ = view.update(cx, |view, cx| { + view.refresh_task = None; + cx.notify(); + }); + })); + } + // Determine expansion state — purely based on ToolBlockState, no is_generating override let is_expanded = block.state == ToolBlockState::Expanded; let has_output = block.output.as_ref().is_some_and(|o| !o.is_empty()) diff --git a/crates/ui_gpui/src/lib.rs b/crates/ui_gpui/src/lib.rs index 7e307d3b..35307af6 100644 --- a/crates/ui_gpui/src/lib.rs +++ b/crates/ui_gpui/src/lib.rs @@ -575,6 +575,7 @@ impl Gpui { tbr_registry.register(Arc::new(tool_cards::code_card::CodeCardRenderer)); tbr_registry.register(Arc::new(tool_cards::session_card::SessionCardRenderer)); tbr_registry.register(Arc::new(tool_cards::browser_renderer::BrowserToolRenderer)); + tbr_registry.register(Arc::new(tool_cards::wakeup_renderer::WakeupToolRenderer)); // MCP tools have dynamic `mcp____` names; one inline // fallback renderer handles all of them. tbr_registry.set_mcp_fallback(Arc::new(McpToolRenderer::new())); diff --git a/crates/ui_gpui/src/main_screen/tests/mod.rs b/crates/ui_gpui/src/main_screen/tests/mod.rs index f1290f7d..1a2abce3 100644 --- a/crates/ui_gpui/src/main_screen/tests/mod.rs +++ b/crates/ui_gpui/src/main_screen/tests/mod.rs @@ -6,3 +6,4 @@ mod composer; mod end_to_end; mod sidebar; mod stored_outputs; +mod wakeups; diff --git a/crates/ui_gpui/src/main_screen/tests/wakeups.rs b/crates/ui_gpui/src/main_screen/tests/wakeups.rs new file mode 100644 index 00000000..5b912946 --- /dev/null +++ b/crates/ui_gpui/src/main_screen/tests/wakeups.rs @@ -0,0 +1,186 @@ +//! A fired wakeup is no message from the user: it shows as the +//! `schedule_wakeup` block of its turn having fired. + +use crate::test_support::{MainScreenTest, TestCore}; +use agent_core::{SerializedToolExecution, ToolRequest}; +use code_assistant_core::mocks::MockLLMProvider; +use code_assistant_core::session::lifecycle::SessionStatus; +use code_assistant_core::session::wakeup::ArmedWakeupInfo; +use code_assistant_core::session::wakeup::wakeup_message; +use code_assistant_core::ui::UiEvent; +use gpui_kit::TestAppContext; +use llm::{ContentBlock, Message}; + +/// A session whose agent scheduled a wakeup and ended its turn, then +/// `between` was said, then the wakeup fired. +fn core_with_fired_wakeup(between: &[Message]) -> (TestCore, String) { + core_with_wakeup_message(between, &wakeup_message(1, "check the build")) +} + +/// Like [`core_with_fired_wakeup`], with the wakeup message as `fired`. +fn core_with_wakeup_message(between: &[Message], fired: &str) -> (TestCore, String) { + let core = TestCore::new(MockLLMProvider::new(Vec::new()).into_factory()); + let session_id = core.create_session(); + let input = serde_json::json!({ "delay_seconds": 60, "prompt": "check the build" }); + let execution = SerializedToolExecution { + tool_request: ToolRequest { + id: "wake".into(), + name: "schedule_wakeup".into(), + input: input.clone(), + start_offset: None, + end_offset: None, + }, + tool_name: "schedule_wakeup".into(), + result_json: serde_json::json!({ + "wakeup_id": 1, "delay_seconds": 60, "fires_at": "2026-10-12T12:00:00Z", + }), + }; + core.store() + .update_entry(&session_id, |session| { + session.add_message(Message::new_user("build it")); + session.add_message(Message::new_assistant_content(vec![ + ContentBlock::new_tool_use("wake", "schedule_wakeup", input.clone()), + ])); + session.add_message(Message::new_user_content(vec![ + ContentBlock::new_tool_result("wake", ""), + ])); + session.add_message(Message::new_assistant("I'll check back.")); + for message in between { + session.add_message(message.clone()); + } + session.add_message(Message::new_user(fired)); + session.tool_executions = vec![execution.clone()]; + Ok(()) + }) + .unwrap(); + (core, session_id) +} + +fn show(cx: &mut TestAppContext, core: TestCore, session_id: String) -> MainScreenTest { + let mut test = MainScreenTest::with_core(cx, core); + test.gpui.cmd_load_session(session_id, None); + test.wait_until("the wakeup block is shown", |test| { + test.tool_block("wake") + .is_some_and(|tool| tool.output.is_some()) + }); + test +} + +#[gpui_kit::test] +fn wakeup_of_the_same_turn_shows_on_its_block(cx: &mut TestAppContext) { + let (core, session_id) = core_with_fired_wakeup(&[]); + let mut test = show(cx, core, session_id); + + assert!(test.tool_block("wake").unwrap().wakeup_fired); + assert_eq!( + test.transcript(), + vec![ + "user: build it", + "assistant: ", + "assistant: I'll check back." + ] + ); +} + +#[gpui_kit::test] +fn wakeup_after_the_user_spoke_shows_on_its_own(cx: &mut TestAppContext) { + let (core, session_id) = core_with_fired_wakeup(&[ + Message::new_user("anything else?"), + Message::new_assistant("No."), + ]); + let mut test = show(cx, core, session_id); + + assert!(!test.tool_block("wake").unwrap().wakeup_fired); + assert_eq!( + test.transcript().last().map(String::as_str), + Some("wakeup: check the build") + ); +} + +#[gpui_kit::test] +fn wakeup_message_without_id_is_linked_by_its_prompt(cx: &mut TestAppContext) { + let (core, session_id) = core_with_wakeup_message(&[], "[scheduled wakeup] check the build"); + let mut test = show(cx, core, session_id); + + assert!(test.tool_block("wake").unwrap().wakeup_fired); + assert_eq!(test.transcript().len(), 3); +} + +#[gpui_kit::test] +fn wakeup_message_of_another_id_is_not_linked(cx: &mut TestAppContext) { + let (core, session_id) = core_with_wakeup_message(&[], &wakeup_message(2, "check the build")); + let mut test = show(cx, core, session_id); + + assert!(!test.tool_block("wake").unwrap().wakeup_fired); + assert_eq!( + test.transcript().last().map(String::as_str), + Some("wakeup: check the build") + ); +} + +fn metadata(session_id: &str) -> code_assistant_core::persistence::ChatMetadata { + let now = std::time::SystemTime::now(); + code_assistant_core::persistence::ChatMetadata { + id: session_id.to_string(), + name: "s".into(), + created_at: now, + updated_at: now, + message_count: 1, + total_usage: Default::default(), + last_usage: Default::default(), + tokens_limit: None, + tool_syntax: code_assistant_core::types::ToolSyntax::Native, + initial_project: String::new(), + branch: None, + plan_collapsed: false, + is_resumable: false, + } +} + +fn armed(test: &mut MainScreenTest, session_id: &str, wakeups: Vec) { + test.receive( + session_id, + UiEvent::UpdateArmedWakeups { + session_id: session_id.to_string(), + wakeups, + }, + ); +} + +#[gpui_kit::test] +fn a_wakeup_to_fire_shows_in_the_sidebar_and_below_the_transcript(cx: &mut TestAppContext) { + let mut test = MainScreenTest::new(cx); + test.push(UiEvent::UpdateChatList { + sessions: vec![metadata("viewed"), metadata("other")], + }); + test.view_session("viewed"); + let status = |test: &mut MainScreenTest, id: &str| { + test.main_screen.read_with(&test.cx, |screen, cx| { + screen.project_sidebar.read(cx).session_status(id, cx) + }) + }; + let shown = |test: &mut MainScreenTest| { + let view = test.gpui.messages_view.lock().unwrap().clone().unwrap(); + test.cx.update(|_, cx| view.read(cx).armed_wakeups().len()) + }; + let wakeup = ArmedWakeupInfo { + id: 1, + fires_at: chrono::Utc::now() + chrono::Duration::minutes(2), + prompt: "check the build".into(), + }; + + armed(&mut test, "other", vec![wakeup.clone()]); + assert_eq!(status(&mut test, "other"), Some(SessionStatus::Scheduled)); + assert_eq!(status(&mut test, "viewed"), Some(SessionStatus::Ready)); + // Another session's wakeups don't show below this transcript. + assert_eq!(shown(&mut test), 0); + + armed(&mut test, "viewed", vec![wakeup]); + assert_eq!(status(&mut test, "viewed"), Some(SessionStatus::Scheduled)); + assert_eq!(shown(&mut test), 1); + + // Fired or cancelled. + armed(&mut test, "viewed", Vec::new()); + assert_eq!(status(&mut test, "viewed"), Some(SessionStatus::Ready)); + assert_eq!(shown(&mut test), 0); +} diff --git a/crates/ui_gpui/src/messages/activity_indicator.rs b/crates/ui_gpui/src/messages/activity_indicator.rs index dae9e759..6cf0483b 100644 --- a/crates/ui_gpui/src/messages/activity_indicator.rs +++ b/crates/ui_gpui/src/messages/activity_indicator.rs @@ -2,7 +2,7 @@ //! //! Shows braille spinners for loading states and pending message cards. -use super::{BRAILLE_FRAMES, MessagesView}; +use super::{BRAILLE_FRAMES, MessagesView, WAKEUP_SPINNER_FRAME_MS, wall_clock_millis}; use code_assistant_core::session::instance::SessionActivityState; use gpui_kit::component::ActiveTheme; use gpui_kit::{Context, div, prelude::*, rems}; @@ -141,3 +141,51 @@ pub fn render_activity_indicator( } } } + +/// Render the wakeups still to fire in the viewed session, with a slow +/// spinner: without it, an idle session that goes on later would look +/// finished once its `schedule_wakeup` block scrolled out of view. +pub fn render_armed_wakeups( + view: &MessagesView, + cx: &mut Context, +) -> gpui_kit::AnyElement { + use crate::tool_cards::wakeup_renderer::countdown; + + let wakeups = view.armed_wakeups(); + let Some(next) = wakeups.first() else { + return div().into_any_element(); + }; + let now = chrono::Local::now(); + let countdown = countdown(next.fires_at.with_timezone(&chrono::Local), now); + let text = match wakeups.len() { + 1 => format!("Wakeup {countdown}"), + n => format!("{n} wakeups · next {countdown}"), + }; + let color = cx.theme().muted_foreground; + let frame = (wall_clock_millis() / WAKEUP_SPINNER_FRAME_MS) as usize % BRAILLE_FRAMES.len(); + div() + .id("armed-wakeups") + .w_full() + .px_3() + .pb_1() + .child( + div() + .flex() + .flex_row() + .items_center() + .gap_1p5() + .py_1p5() + .px_3() + .child( + div() + .w(gpui_kit::px(14.)) + .flex() + .justify_center() + .text_size(rems(0.875)) + .text_color(color) + .child(BRAILLE_FRAMES[frame]), + ) + .child(div().text_size(rems(0.8125)).text_color(color).child(text)), + ) + .into_any_element() +} diff --git a/crates/ui_gpui/src/messages/message_item.rs b/crates/ui_gpui/src/messages/message_item.rs index 4dfb2f75..741280b8 100644 --- a/crates/ui_gpui/src/messages/message_item.rs +++ b/crates/ui_gpui/src/messages/message_item.rs @@ -3,7 +3,9 @@ use super::MessagesView; use super::branch_switcher::BranchSwitcherElement; use gpui_kit::component::{ActiveTheme, Icon}; -use gpui_kit::{Context, CursorStyle, Entity, SharedString, Window, div, prelude::*, rems, rgb}; +use gpui_kit::{ + Context, CursorStyle, Entity, SharedString, Window, div, prelude::*, px, rems, rgb, +}; /// Render a single message at the given index. /// Called by the list's render callback — only for visible items. @@ -28,6 +30,18 @@ pub fn render_message( // Update the message container with current project msg.read(cx).set_current_project(current_project); + // A wakeup message is not the user's: its `schedule_wakeup` block shows + // it fired, or, without one in the turn, it reads like a tool line. + if let Some(wakeup) = msg.read(cx).wakeup() { + let row = div().id(("message-item", msg.entity_id())).w_full(); + if wakeup.linked { + return row.into_any_element(); + } + return row + .child(render_wakeup_message(&wakeup.prompt, cx)) + .into_any_element(); + } + let is_user_message = msg.read(cx).is_user_message(); let node_id = msg.read(cx).node_id(); let branch_info = msg.read(cx).branch_info(); @@ -169,6 +183,50 @@ pub fn render_message( } } +/// A wakeup message without its `schedule_wakeup` block in the turn before: +/// laid out like an expanded inline tool block. +fn render_wakeup_message(prompt: &str, cx: &Context) -> gpui_kit::AnyElement { + use super::super::shared::file_icons; + let theme = cx.theme(); + div() + .px_3() + .pb_1() + .mt_0p5() + .child( + div() + .flex() + .flex_row() + .items_center() + .gap_1p5() + .py_1p5() + .px_3() + .child(file_icons::render_icon_container( + &file_icons::get().get_type_icon(file_icons::TOOL_WAKEUP_FIRED), + 14.0, + theme.muted_foreground, + "⏰", + )) + .child( + div() + .text_size(rems(0.8125)) + .text_color(theme.muted_foreground) + .child("Woke up"), + ), + ) + .child( + div() + .pl(px(8.)) + .ml(px(8.)) + .border_l_2() + .border_color(theme.border) + .py(px(4.)) + .text_size(rems(0.8125)) + .text_color(theme.muted_foreground) + .child(prompt.to_string()), + ) + .into_any_element() +} + /// Group consecutive image blocks into horizontal galleries for user messages fn group_user_message_elements( elements: Vec>, diff --git a/crates/ui_gpui/src/messages/mod.rs b/crates/ui_gpui/src/messages/mod.rs index 39badf1f..4411d2e3 100644 --- a/crates/ui_gpui/src/messages/mod.rs +++ b/crates/ui_gpui/src/messages/mod.rs @@ -7,6 +7,7 @@ use crate::Gpui; use crate::shared::frame_profile; use code_assistant_core::session::instance::SessionActivityState; +use code_assistant_core::session::wakeup::ArmedWakeupInfo; use gpui_kit::component::scroll::ScrollableElement; use gpui_kit::component::{ActiveTheme, Icon}; @@ -30,6 +31,15 @@ use super::blocks::MessageContainer; /// Braille spinner frames for the activity indicator. const BRAILLE_FRAMES: &[&str] = &["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; +/// Frame duration of the slow spinner of a session waiting for a wakeup. +const WAKEUP_SPINNER_FRAME_MS: u128 = 500; + +fn wall_clock_millis() -> u128 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() +} /// Maximum width of the message content area. On wide viewports, messages /// stay centered rather than stretching edge-to-edge for comfortable reading. @@ -86,6 +96,11 @@ pub struct MessagesView { last_animation_offset: Rc>, /// Task that ticks the braille spinner animation. _spinner_task: Option>, + /// The viewed session's wakeups still to fire, soonest first; shown + /// below the transcript while the agent is not running. + armed_wakeups: Vec, + /// Renders the wakeup countdown again when its text changes. + wakeup_refresh_task: Option>, // -- Edge auto-scroll state (drag-to-select near a viewport edge) -- /// Bounds of the message list viewport in window coordinates, captured @@ -176,6 +191,8 @@ impl MessagesView { animation_active, last_animation_offset, _spinner_task: Some(spinner_task), + armed_wakeups: Vec::new(), + wakeup_refresh_task: None, list_bounds: Rc::new(Cell::new(Bounds::default())), edge_scroll_velocity: Rc::new(Cell::new(0.0)), edge_drag_active: Rc::new(Cell::new(false)), @@ -847,6 +864,18 @@ impl MessagesView { /// the agent ended in a state that suggests an unfinished iteration /// (e.g. crashed mid-tool-call). When true and the session is idle, the /// MessagesView shows a floating Resume button above the input area. + pub fn set_armed_wakeups(&mut self, wakeups: Vec, cx: &mut Context) { + if self.armed_wakeups != wakeups { + self.armed_wakeups = wakeups; + self.wakeup_refresh_task = None; + cx.notify(); + } + } + + pub(crate) fn armed_wakeups(&self) -> &[ArmedWakeupInfo] { + &self.armed_wakeups + } + pub fn set_is_resumable(&self, is_resumable: bool) { self.is_resumable.set(is_resumable); } @@ -909,9 +938,36 @@ impl Render for MessagesView { ) }); + // Wakeups still to fire, while nothing else shows that the session + // goes on. + let agent_running = self + .activity_state + .lock() + .ok() + .and_then(|g| g.clone()) + .is_some_and(|s| !s.is_terminal()); + let has_wakeups = !self.armed_wakeups.is_empty() && !agent_running; + if has_wakeups && self.wakeup_refresh_task.is_none() { + let next = self.armed_wakeups[0].fires_at.with_timezone(&chrono::Local); + // The next spinner frame, or earlier when the countdown changes. + let next_frame = Duration::from_millis( + (WAKEUP_SPINNER_FRAME_MS - wall_clock_millis() % WAKEUP_SPINNER_FRAME_MS) as u64, + ); + let after = crate::tool_cards::wakeup_renderer::countdown_changes_after(next) + .map_or(next_frame, |countdown| countdown.min(next_frame)); + self.wakeup_refresh_task = Some(cx.spawn(async move |view, cx| { + cx.background_executor().timer(after).await; + let _ = view.update(cx, |view, cx| { + view.wakeup_refresh_task = None; + cx.notify(); + }); + })); + } + let pending_index = total_items; let activity_index = total_items + if has_pending { 1 } else { 0 }; - let item_count = activity_index + if has_activity { 1 } else { 0 }; + let wakeups_index = activity_index + if has_activity { 1 } else { 0 }; + let item_count = wakeups_index + if has_wakeups { 1 } else { 0 }; // Build the virtualized list. The render callback is only invoked for // items that are within the visible viewport + overdraw zone. @@ -927,6 +983,8 @@ impl Render for MessagesView { message_item::render_message(this, index, window, cx) } else if index == pending_index && has_pending { activity_indicator::render_pending_message(this, cx) + } else if index == wakeups_index && has_wakeups { + activity_indicator::render_armed_wakeups(this, cx) } else { // Activity indicator (last item) activity_indicator::render_activity_indicator(this, cx) diff --git a/crates/ui_gpui/src/shared/file_icons.rs b/crates/ui_gpui/src/shared/file_icons.rs index ae025e2c..55a341b8 100644 --- a/crates/ui_gpui/src/shared/file_icons.rs +++ b/crates/ui_gpui/src/shared/file_icons.rs @@ -52,6 +52,9 @@ pub const TOOL_UPDATE_PLAN: &str = "todo_list"; // todo_list.svg pub const TOOL_SPAWN_AGENT: &str = "rerun"; // rerun.svg - for spawning sub-agents pub const TOOL_VIEW_DOCUMENTS: &str = "file_generic"; // file_generic.svg - for viewing documents pub const TOOL_MCP: &str = "mcp"; // link.svg - generic icon for MCP server tools +pub const TOOL_SCHEDULE_WAKEUP: &str = "alarm_clock"; // alarm_clock.svg +pub const TOOL_WAKEUP_FIRED: &str = "alarm_clock_check"; // alarm_clock_check.svg +pub const TOOL_CANCEL_WAKEUP: &str = "alarm_clock_off"; // alarm_clock_off.svg pub const TOOL_GENERIC: &str = "file_code"; // file_code.svg const FILE_TYPES_ASSET: &str = "icons/file_icons/file_types.json"; @@ -145,6 +148,9 @@ impl FileIcons { TOOL_SPAWN_AGENT => Some("icons/rerun.svg"), TOOL_VIEW_DOCUMENTS => Some("icons/file_generic.svg"), TOOL_MCP => Some("icons/link.svg"), + TOOL_SCHEDULE_WAKEUP => Some("icons/alarm_clock.svg"), + TOOL_WAKEUP_FIRED => Some("icons/alarm_clock_check.svg"), + TOOL_CANCEL_WAKEUP => Some("icons/alarm_clock_off.svg"), TOOL_GENERIC => Some("icons/file_code.svg"), // For file_types.json types we missed _ => None, @@ -251,6 +257,8 @@ impl FileIcons { "update_plan" => TOOL_UPDATE_PLAN, "spawn_agent" => TOOL_SPAWN_AGENT, "view_documents" => TOOL_VIEW_DOCUMENTS, + "schedule_wakeup" => TOOL_SCHEDULE_WAKEUP, + "cancel_wakeup" => TOOL_CANCEL_WAKEUP, _ => TOOL_GENERIC, }; diff --git a/crates/ui_gpui/src/sidebar/mod.rs b/crates/ui_gpui/src/sidebar/mod.rs index 0a73219d..6cdce4a3 100644 --- a/crates/ui_gpui/src/sidebar/mod.rs +++ b/crates/ui_gpui/src/sidebar/mod.rs @@ -85,6 +85,8 @@ pub struct SessionSidebar { selected_session_id: Option, focus_handle: FocusHandle, activity_states: HashMap, + /// When each session's next wakeup fires, for sessions with one to fire. + next_wakeups: HashMap>, awaiting_user: HashMap, _item_subscriptions: Vec, _picker_subscription: Subscription, @@ -116,6 +118,7 @@ impl SessionSidebar { selected_session_id: None, focus_handle: cx.focus_handle(), activity_states: HashMap::new(), + next_wakeups: HashMap::new(), awaiting_user: HashMap::new(), _item_subscriptions: Vec::new(), _picker_subscription: picker_subscription, @@ -175,6 +178,7 @@ impl SessionSidebar { .unwrap_or_default(); let activity = self.activity_states.get(&session.id).cloned(); let awaiting = self.awaiting_user.get(&session.id).copied(); + let next_wakeup = self.next_wakeups.get(&session.id).copied(); let entity = match existing.remove(&session.id) { Some(entity) => { entity.update(cx, |item, cx| { @@ -193,6 +197,7 @@ impl SessionSidebar { item.update_activity_state(state, cx); } item.set_awaiting_user(awaiting, cx); + item.set_next_wakeup(next_wakeup, cx); item.set_show_project(show_project, cx); }); self._item_subscriptions @@ -291,6 +296,33 @@ impl SessionSidebar { cx.notify(); } + /// The status the row of `session_id` shows. + #[cfg(test)] + pub(crate) fn session_status( + &self, + session_id: &str, + cx: &App, + ) -> Option { + Some(self.items.get(session_id)?.read(cx).status()) + } + + /// When the session's next wakeup fires; `None` when none is to fire. + pub fn update_session_wakeup( + &mut self, + session_id: String, + next_wakeup: Option>, + cx: &mut Context, + ) { + if let Some(item) = self.items.get(&session_id) { + item.update(cx, |item, cx| item.set_next_wakeup(next_wakeup, cx)); + } + match next_wakeup { + Some(at) => self.next_wakeups.insert(session_id, at), + None => self.next_wakeups.remove(&session_id), + }; + cx.notify(); + } + /// The sessions blocked on the user (open permission request or /// questions), and what for. pub fn set_awaiting_user( diff --git a/crates/ui_gpui/src/sidebar/session_item.rs b/crates/ui_gpui/src/sidebar/session_item.rs index 45f41c24..622ac252 100644 --- a/crates/ui_gpui/src/sidebar/session_item.rs +++ b/crates/ui_gpui/src/sidebar/session_item.rs @@ -65,6 +65,8 @@ pub struct SessionListItem { is_hovered: bool, activity_state: SessionActivityState, awaiting_user: Option, + /// When the session's next wakeup fires, if one is still to fire. + next_wakeup: Option>, /// Whether the subtitle names the project; not inside a project folder. show_project: bool, focus_handle: FocusHandle, @@ -84,6 +86,7 @@ impl SessionListItem { is_hovered: false, activity_state: SessionActivityState::Idle, awaiting_user: None, + next_wakeup: None, show_project: true, focus_handle: cx.focus_handle(), } @@ -128,6 +131,17 @@ impl SessionListItem { } } + pub fn set_next_wakeup( + &mut self, + next_wakeup: Option>, + cx: &mut Context, + ) { + if self.next_wakeup != next_wakeup { + self.next_wakeup = next_wakeup; + cx.notify(); + } + } + pub fn set_show_project(&mut self, show_project: bool, cx: &mut Context) { if self.show_project != show_project { self.show_project = show_project; @@ -136,7 +150,11 @@ impl SessionListItem { } pub(super) fn status(&self) -> SessionStatus { - SessionStatus::resolve(&self.activity_state, self.awaiting_user) + SessionStatus::resolve( + &self.activity_state, + self.awaiting_user, + self.next_wakeup.is_some(), + ) } /// A settled session is never unread: the user put it away. @@ -186,6 +204,15 @@ impl SessionListItem { }) } + /// "at 14:31" today, "on Oct 13 at 09:00" on another day. + fn format_wakeup_time(at: chrono::DateTime) -> String { + if at.date_naive() == chrono::Local::now().date_naive() { + format!("at {}", at.format("%H:%M")) + } else { + format!("on {}", at.format("%b %-d at %H:%M")) + } + } + pub(super) fn format_relative_date(timestamp: SystemTime) -> String { match timestamp.elapsed() { Ok(duration) => { @@ -292,15 +319,25 @@ impl Render for SessionListItem { None => self.metadata.updated_at, }); - let (status_label, status_color): (Option<&'static str>, Hsla) = match status { - SessionStatus::NeedsApproval => (Some("Needs approval"), cx.theme().warning), - SessionStatus::NeedsAnswer => (Some("Needs answer"), cx.theme().warning), - SessionStatus::Working => (Some("Working"), cx.theme().muted_foreground), - SessionStatus::RateLimited => (Some("Rate limited"), cx.theme().warning), - SessionStatus::Failed => (Some("Failed"), cx.theme().danger), - SessionStatus::RunningElsewhere => { - (Some("Running elsewhere"), cx.theme().muted_foreground) - } + let (status_label, status_color): (Option, Hsla) = match status { + SessionStatus::NeedsApproval => (Some("Needs approval".into()), cx.theme().warning), + SessionStatus::NeedsAnswer => (Some("Needs answer".into()), cx.theme().warning), + SessionStatus::Working => (Some("Working".into()), cx.theme().muted_foreground), + SessionStatus::RateLimited => (Some("Rate limited".into()), cx.theme().warning), + SessionStatus::Scheduled => ( + self.next_wakeup.map(|at| { + format!( + "Wakeup {}", + Self::format_wakeup_time(at.with_timezone(&chrono::Local)) + ) + }), + cx.theme().muted_foreground, + ), + SessionStatus::Failed => (Some("Failed".into()), cx.theme().danger), + SessionStatus::RunningElsewhere => ( + Some("Running elsewhere".into()), + cx.theme().muted_foreground, + ), SessionStatus::Ready => (None, cx.theme().muted_foreground), }; let git = self.git_glyph(cx); @@ -316,7 +353,7 @@ impl Render for SessionListItem { parts.push(self.metadata.initial_project.clone()); } if let Some(label) = status_label { - parts.push(label.to_string()); + parts.push(label); subtitle_color = status_color; } else if let Some(branch) = &self.metadata.branch { match &self.lifecycle.pull_request { @@ -416,6 +453,12 @@ impl Render for SessionListItem { .path("icons/lock.svg") .text_color(status_color), ), + SessionStatus::Scheduled => el.child( + gpui_kit::svg() + .size(px(13.)) + .path("icons/alarm_clock.svg") + .text_color(status_color), + ), SessionStatus::Working | SessionStatus::RateLimited => el.child( gpui_kit::svg() .size(px(12.)) diff --git a/crates/ui_gpui/src/test_support/main_screen.rs b/crates/ui_gpui/src/test_support/main_screen.rs index 2bb92243..42b9ca4d 100644 --- a/crates/ui_gpui/src/test_support/main_screen.rs +++ b/crates/ui_gpui/src/test_support/main_screen.rs @@ -131,6 +131,7 @@ impl MainScreenTest { .iter() .map(|message| message.read(cx)) .filter(|message| !message.is_empty()) + .filter(|message| message.wakeup().is_none_or(|wakeup| !wakeup.linked)) .map(|message| message.transcript_line(cx)) .collect() }) diff --git a/crates/ui_gpui/src/tool_cards/diff_card.rs b/crates/ui_gpui/src/tool_cards/diff_card.rs index 29999cb5..31e379db 100644 --- a/crates/ui_gpui/src/tool_cards/diff_card.rs +++ b/crates/ui_gpui/src/tool_cards/diff_card.rs @@ -1743,6 +1743,7 @@ mod tests { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, }; assert_eq!(extract_path_or_paths(&tool), "src/main.rs"); } @@ -1766,6 +1767,7 @@ mod tests { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, }; assert_eq!(extract_path_or_paths(&tool), "a.rs, b.rs"); } diff --git a/crates/ui_gpui/src/tool_cards/diff_prepare.rs b/crates/ui_gpui/src/tool_cards/diff_prepare.rs index 83b2efcf..77666ee2 100644 --- a/crates/ui_gpui/src/tool_cards/diff_prepare.rs +++ b/crates/ui_gpui/src/tool_cards/diff_prepare.rs @@ -203,6 +203,7 @@ pub(crate) mod tests { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, } } diff --git a/crates/ui_gpui/src/tool_cards/mod.rs b/crates/ui_gpui/src/tool_cards/mod.rs index 4045d64b..94b93155 100644 --- a/crates/ui_gpui/src/tool_cards/mod.rs +++ b/crates/ui_gpui/src/tool_cards/mod.rs @@ -26,6 +26,7 @@ pub mod mcp_tool; pub mod session_card; pub mod sub_agent_card; pub mod terminal_card; +pub mod wakeup_renderer; use crate::blocks::{BlockView, ToolUseBlock}; use gpui_kit::component::text::TextViewState; @@ -119,6 +120,18 @@ pub trait ToolBlockRenderer: Send + Sync { None } + /// Icon of an inline block's header, when it depends on the block's + /// state; `None` uses the tool's fixed icon. + fn icon(&self, _tool: &ToolUseBlock) -> Option { + None + } + + /// For an inline block whose description changes with time (a + /// countdown): render it again after this long. + fn refresh_after(&self, _tool: &ToolUseBlock) -> Option { + None + } + /// Render the tool block content. /// /// For **Inline** renderers this returns the expanded output area @@ -234,6 +247,7 @@ mod tests { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, } } diff --git a/crates/ui_gpui/src/tool_cards/session_card.rs b/crates/ui_gpui/src/tool_cards/session_card.rs index 668b7384..89141434 100644 --- a/crates/ui_gpui/src/tool_cards/session_card.rs +++ b/crates/ui_gpui/src/tool_cards/session_card.rs @@ -405,6 +405,7 @@ mod tests { duration_seconds: None, images: Vec::new(), revision: 0, + wakeup_fired: false, } } diff --git a/crates/ui_gpui/src/tool_cards/wakeup_renderer.rs b/crates/ui_gpui/src/tool_cards/wakeup_renderer.rs new file mode 100644 index 00000000..1c641327 --- /dev/null +++ b/crates/ui_gpui/src/tool_cards/wakeup_renderer.rs @@ -0,0 +1,396 @@ +//! Inline renderer for `schedule_wakeup` / `cancel_wakeup`. +//! +//! A scheduled wakeup counts down to when it fires. From the call on, the +//! line starts the same ("Wakeup in 20 min"); the fire time is appended once +//! known. Once it fired — the stored result says so, or for results recorded +//! before that, the wakeup message arrived in the same turn +//! (`ToolUseBlock::wakeup_fired`) — the line says when. A wakeup armed by an +//! earlier process that never fired was lost with it. Expanding the line +//! shows the prompt. + +use super::inline_renderer::{render_inline_output, render_inline_output_text}; +use super::{CardRenderContext, ToolBlockRenderer, ToolBlockStyle}; +use crate::blocks::{BlockView, ToolUseBlock}; +use crate::shared::file_icons; +use chrono::{DateTime, Local}; +use code_assistant_core::session::wakeup::process_instance; +use code_assistant_core::tools::impls::wakeup::ScheduleWakeupOutput; +use gpui_kit::{AnyElement, Context, SharedString, Window}; +use std::time::Duration; + +pub struct WakeupToolRenderer; + +impl ToolBlockRenderer for WakeupToolRenderer { + fn supported_tools(&self) -> Vec { + vec!["schedule_wakeup".to_string(), "cancel_wakeup".to_string()] + } + + fn style(&self) -> ToolBlockStyle { + ToolBlockStyle::Inline + } + + fn describe(&self, tool: &ToolUseBlock) -> String { + if tool.name == "cancel_wakeup" { + return match param(tool, "wakeup_id") { + Some(id) => format!("Cancel wakeup #{id}"), + None => "Cancel wakeup".to_string(), + }; + } + describe_schedule(tool, Local::now(), process_instance()) + } + + fn icon(&self, tool: &ToolUseBlock) -> Option { + let icon = match state(tool, process_instance()) { + WakeupState::Fired { .. } => file_icons::TOOL_WAKEUP_FIRED, + WakeupState::Lost { .. } => file_icons::TOOL_CANCEL_WAKEUP, + WakeupState::Requested | WakeupState::Armed { .. } => return None, + }; + file_icons::get().get_type_icon(icon) + } + + fn refresh_after(&self, tool: &ToolUseBlock) -> Option { + let WakeupState::Armed { + fires_at: Some(fires_at), + .. + } = state(tool, process_instance()) + else { + return None; + }; + countdown_changes_after(fires_at) + } + + fn render( + &self, + tool: &ToolUseBlock, + _is_generating: bool, + theme: &gpui_kit::component::theme::Theme, + _card_ctx: Option<&CardRenderContext>, + _window: &mut Window, + _cx: &mut Context, + ) -> Option { + match param(tool, "prompt") { + Some(prompt) if tool.name == "schedule_wakeup" => { + render_inline_output_text(prompt.trim(), false, tool, theme) + } + _ => render_inline_output(tool, theme), + } + } +} + +fn param<'a>(tool: &'a ToolUseBlock, name: &str) -> Option<&'a str> { + tool.parameters + .iter() + .find(|p| p.name == name) + .map(|p| p.value.as_str()) +} + +/// Where a scheduled wakeup stands. Fire times are absent in results +/// recorded before they carried them. +enum WakeupState { + /// Called, not executed yet (or failed). + Requested, + Armed { + fires_at: Option>, + delay_seconds: u64, + }, + Fired { + at: Option>, + }, + /// Armed by an earlier process, which exited before it fired. + Lost { + fires_at: Option>, + }, +} + +fn state(tool: &ToolUseBlock, this_process: &str) -> WakeupState { + let Some(output) = tool + .output + .as_deref() + .and_then(ScheduleWakeupOutput::from_ui_output) + else { + return WakeupState::Requested; + }; + let local = |time: chrono::DateTime| time.with_timezone(&Local); + let fires_at = output.fires_at.map(local); + if output.fired_at.is_some() || tool.wakeup_fired { + WakeupState::Fired { + at: output.fired_at.map(local).or(fires_at), + } + } else if output + .armed_by + .is_some_and(|armed_by| armed_by != this_process) + { + WakeupState::Lost { fires_at } + } else { + WakeupState::Armed { + fires_at, + delay_seconds: output.delay_seconds, + } + } +} + +fn describe_schedule(tool: &ToolUseBlock, now: DateTime, this_process: &str) -> String { + match state(tool, this_process) { + WakeupState::Requested => match param(tool, "delay_seconds").and_then(|d| d.parse().ok()) { + Some(delay) => { + format!("Wakeup in {}", format_remaining(Duration::from_secs(delay))) + } + None => "Wakeup".to_string(), + }, + WakeupState::Armed { + fires_at: None, + delay_seconds, + } => format!( + "Wakeup in {}", + format_remaining(Duration::from_secs(delay_seconds)) + ), + WakeupState::Armed { + fires_at: Some(fires_at), + .. + } => format!("Wakeup {}", countdown(fires_at, now)), + WakeupState::Fired { at: Some(at) } => format!("Woke up {}", format_when(at, now)), + WakeupState::Fired { at: None } => "Woke up".to_string(), + WakeupState::Lost { + fires_at: Some(fires_at), + } => format!( + "Wakeup lost with a restart · was due {}", + format_when(fires_at, now) + ), + WakeupState::Lost { fires_at: None } => "Wakeup lost with a restart".to_string(), + } +} + +/// "in 20 min · at 14:31" until `fires_at`, "at 14:31" once it is due (but +/// not fired yet: it is on its way, or was cancelled). +pub(crate) fn countdown(fires_at: DateTime, now: DateTime) -> String { + let when = format_when(fires_at, now); + match (fires_at - now).to_std() { + Ok(remaining) if !remaining.is_zero() => { + format!("in {} · {when}", format_remaining(remaining)) + } + _ => when, + } +} + +/// How long until [`countdown`] for `fires_at` shows a different text; +/// `None` once it no longer changes. +pub(crate) fn countdown_changes_after(fires_at: DateTime) -> Option { + let remaining = (fires_at - Local::now()).to_std().ok()?; + Some(until_display_changes(remaining)) +} + +/// "at 14:32" today, "on Oct 13 at 09:00" on any other day. +fn format_when(time: DateTime, now: DateTime) -> String { + if time.date_naive() == now.date_naive() { + format!("at {}", time.format("%H:%M")) + } else { + format!("on {}", time.format("%b %-d at %H:%M")) + } +} + +/// A countdown: seconds in the last minute, whole minutes rounded up before +/// that — "2 min" until one minute is left, then "60 s". +fn format_remaining(remaining: Duration) -> String { + let seconds = remaining.as_secs_f64().ceil() as u64; + if seconds <= 60 { + format!("{seconds} s") + } else { + format_minutes(seconds.div_ceil(60)) + } +} + +fn format_minutes(minutes: u64) -> String { + match (minutes / 60, minutes % 60) { + (0, m) => format!("{m} min"), + (h, 0) => format!("{h} h"), + (h, m) => format!("{h} h {m} min"), + } +} + +/// How long until [`format_remaining`] shows a different value. +fn until_display_changes(remaining: Duration) -> Duration { + // A little past the boundary, so the next render is on its far side. + const MARGIN: Duration = Duration::from_millis(20); + let millis = remaining.as_millis() as u64; + if millis <= 60_000 { + return Duration::from_millis(millis % 1000) + MARGIN; + } + // Minutes round up: the value drops when crossing a full minute. + let into_minute = millis % 60_000; + let next = if into_minute == 0 { + 60_000 + } else { + into_minute + }; + Duration::from_millis(next) + MARGIN +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::tool_cards::tests::make_tool; + use chrono::TimeZone; + + const HERE: &str = "this-process"; + + fn now() -> DateTime { + Local.with_ymd_and_hms(2026, 10, 12, 14, 12, 0).unwrap() + } + + fn scheduled(delay_seconds: u64, fires_at: DateTime) -> ToolUseBlock { + with_output(delay_seconds, fires_at, |_| {}) + } + + /// A `schedule_wakeup` block armed by this process, its result adjusted + /// by `adjust`. + fn with_output( + delay_seconds: u64, + fires_at: DateTime, + adjust: impl FnOnce(&mut ScheduleWakeupOutput), + ) -> ToolUseBlock { + let mut output = ScheduleWakeupOutput { + wakeup_id: 1, + delay_seconds, + fires_at: Some(fires_at.to_utc()), + armed_by: Some(HERE.to_string()), + fired_at: None, + }; + adjust(&mut output); + let mut tool = make_tool( + "schedule_wakeup", + &[ + ("delay_seconds", &delay_seconds.to_string()), + ("prompt", "check the build"), + ], + ); + tool.output = Some(serde_json::to_string(&output).unwrap()); + tool + } + + #[test] + fn before_execution_shows_the_requested_delay() { + let tool = make_tool("schedule_wakeup", &[("delay_seconds", "1200")]); + assert_eq!(describe_schedule(&tool, now(), HERE), "Wakeup in 20 min"); + let bare = make_tool("schedule_wakeup", &[]); + assert_eq!(describe_schedule(&bare, now(), HERE), "Wakeup"); + } + + #[test] + fn armed_wakeup_counts_down() { + let tool = scheduled(1200, now() + chrono::Duration::seconds(1190)); + assert_eq!( + describe_schedule(&tool, now(), HERE), + "Wakeup in 20 min · at 14:31" + ); + let tool = scheduled(30, now() + chrono::Duration::seconds(30)); + assert_eq!( + describe_schedule(&tool, now(), HERE), + "Wakeup in 30 s · at 14:12" + ); + } + + #[test] + fn other_days_show_the_date() { + let tool = scheduled(86_400, now() + chrono::Duration::days(1)); + assert_eq!( + describe_schedule(&tool, now(), HERE), + "Wakeup in 24 h · on Oct 13 at 14:12" + ); + } + + #[test] + fn executing_keeps_the_start_of_the_line() { + for delay in [30, 90, 1200, 5400] { + let requested = make_tool("schedule_wakeup", &[("delay_seconds", &delay.to_string())]); + // Executed a moment after the call. + let armed = scheduled(delay, now() + chrono::Duration::seconds(delay as i64)); + let before = describe_schedule(&requested, now(), HERE); + let after = + describe_schedule(&armed, now() + chrono::Duration::milliseconds(200), HERE); + assert!(after.starts_with(&before), "{before:?} → {after:?}"); + } + } + + #[test] + fn due_wakeup_without_message_claims_nothing() { + let tool = scheduled(60, now() - chrono::Duration::seconds(5)); + assert_eq!(describe_schedule(&tool, now(), HERE), "Wakeup at 14:11"); + } + + #[test] + fn fired_wakeup_says_when() { + let fired_at = now() - chrono::Duration::seconds(3); + let tool = with_output(60, now() - chrono::Duration::seconds(5), |output| { + output.fired_at = Some(fired_at.to_utc()); + }); + assert_eq!(describe_schedule(&tool, now(), HERE), "Woke up at 14:11"); + // Also when it was armed by an earlier process. + assert_eq!(describe_schedule(&tool, now(), "later"), "Woke up at 14:11"); + } + + #[test] + fn wakeup_message_of_the_turn_marks_older_results_fired() { + let mut tool = with_output(60, now() - chrono::Duration::seconds(5), |output| { + output.armed_by = None; + }); + tool.wakeup_fired = true; + assert_eq!(describe_schedule(&tool, now(), HERE), "Woke up at 14:11"); + } + + #[test] + fn wakeup_of_an_earlier_process_was_lost() { + let tool = scheduled(1200, now() + chrono::Duration::seconds(600)); + assert_eq!( + describe_schedule(&tool, now(), "later"), + "Wakeup lost with a restart · was due at 14:22" + ); + } + + #[test] + fn results_without_process_count_as_armed() { + let tool = with_output(1200, now() + chrono::Duration::seconds(600), |output| { + output.armed_by = None; + }); + assert_eq!( + describe_schedule(&tool, now(), "later"), + "Wakeup in 10 min · at 14:22" + ); + } + + #[test] + fn results_without_fire_time_fall_back_to_the_delay() { + let mut tool = make_tool("schedule_wakeup", &[]); + tool.output = Some(r#"{"wakeup_id":1,"delay_seconds":5400}"#.to_string()); + assert_eq!( + describe_schedule(&tool, now(), HERE), + "Wakeup in 1 h 30 min" + ); + } + + #[test] + fn cancel_names_the_wakeup() { + let tool = make_tool("cancel_wakeup", &[("wakeup_id", "3")]); + assert_eq!(WakeupToolRenderer.describe(&tool), "Cancel wakeup #3"); + } + + #[test] + fn minutes_round_up_until_the_seconds_take_over() { + let s = Duration::from_secs; + assert_eq!(format_remaining(s(120)), "2 min"); + assert_eq!(format_remaining(s(61)), "2 min"); + assert_eq!(format_remaining(s(60)), "60 s"); + assert_eq!(format_remaining(s(1190)), "20 min"); + assert_eq!(format_remaining(s(1210)), "21 min"); + } + + #[test] + fn refresh_follows_the_displayed_unit() { + let ms = Duration::from_millis; + assert_eq!(until_display_changes(ms(30_400)), ms(420)); + // 20 min 10 s shows "21 min" until 20 min remain. + assert_eq!(until_display_changes(ms(1_210_000)), ms(10_020)); + assert_eq!(until_display_changes(ms(1_200_000)), ms(60_020)); + // 1 min 10 s shows "2 min" until the seconds take over. + assert_eq!(until_display_changes(ms(70_000)), ms(10_020)); + } +} diff --git a/docs/session-lifecycle.md b/docs/session-lifecycle.md index c4ef6f78..c266c222 100644 --- a/docs/session-lifecycle.md +++ b/docs/session-lifecycle.md @@ -17,6 +17,7 @@ Two questions are kept apart on purpose. | `NeedsApproval` | an open `RequestToolPermission` | warning icon and label | | `Working` | `AgentRunning`, `WaitingForResponse` | spinner, recedes | | `RateLimited` | `RateLimited { .. }` | spinner, recedes | +| `Scheduled` | `Idle` with a wakeup still to fire | alarm clock and fire time, recedes | | `Failed` | `Errored { .. }` | danger icon and label | | `RunningElsewhere` | `RunningExternally` (foreign agent lock) | lock icon, recedes | | `Ready` | `Idle` | no label | diff --git a/docs/session-wakeups.md b/docs/session-wakeups.md index afb23f57..9fe078bd 100644 --- a/docs/session-wakeups.md +++ b/docs/session-wakeups.md @@ -76,15 +76,18 @@ way other cross-component handles are, at service construction time. On fire, the scheduler goes through the `SessionService` command bus: -1. Build the wakeup message: a user-role message with unambiguous framing, - `[scheduled wakeup] {prompt}` (constant `WAKEUP_PREFIX`), so the agent - and the transcript can tell it from a human message. -2. If the session is idle: `add_user_message` + `start_agent_for_session` +1. Record in the stored `schedule_wakeup` result that it fired + (`fired_at`, `SessionManager::mark_wakeup_fired`) and send the result's + update to frontends. +2. Build the wakeup message: a user-role message with unambiguous framing, + `[scheduled wakeup #{id}] {prompt}` (`wakeup_message`), so the agent and + the transcript can tell it from a human message. +3. If the session is idle: `add_user_message` + `start_agent_for_session` (the `send_user_message_impl` path). -3. If a turn is running: write into the `pending_message` slot +4. If a turn is running: write into the `pending_message` slot (`queue_structured_user_message` path) — the running turn picks it up at its next iteration; no second task is spawned. -4. If the session no longer exists: drop silently. +5. If the session no longer exists: drop silently. The turn then ends naturally (`LoopFlow::GetUserInput` → task returns → `Idle`), keeping the session idle-but-alive for the next wakeup. @@ -99,12 +102,29 @@ wake the parent — if that turns out useful it needs its own semantics. - Input: `delay_seconds: u64` (clamped to a sane range), `prompt: String`. - Effect: arms a wakeup for the calling session (session id from `ToolContext`), returns the wakeup id and resolved fire time. +- Ids are numbered per session and continue across restarts (seeded from + the highest id in the session's history), so the model sees short ids; + internally a wakeup is identified by session and id. +- The result records the process that armed it (`armed_by`, + `process_instance()`). A frontend shows a wakeup armed by another process + that never fired as lost with a restart. The text the model sees leaves + out `fired_at`, so the conversation never changes after the fact. `cancel_wakeup` — input: wakeup id. Cancels if still armed. The tool reaches the scheduler handle via the `ToolContext` extensions (same access pattern as `update_plan` reaching app services). +### Frontends + +The scheduler keeps, per session, the wakeups still to fire +(`WakeupHandle::armed`) and reports every change +(`WakeupSink::armed_changed` → `UiEvent::UpdateArmedWakeups`, also part of +the session snapshot). The GPUI frontend shows an idle session with one as +`SessionStatus::Scheduled` in the sidebar and a countdown line below the +transcript, so a session that goes on later does not look finished. Such a +session never settles automatically. + ### Interactions - **SleepInhibitor**: an armed wakeup holds the wake-lock refcount (armed →