From 9e209b46d0da8e447ba3f182c386a273e4877e7e Mon Sep 17 00:00:00 2001 From: w0wl0lxd Date: Wed, 22 Jul 2026 03:54:35 -0400 Subject: [PATCH 1/2] feat(ui): add safe submission escape hatch --- changelog.d/submission-escape-hatch.added.md | 1 + n00n-agent/src/agent/run.rs | 123 ++++- n00n-agent/src/cancel.rs | 101 +++- n00n-agent/src/lib.rs | 3 +- n00n-storage/src/sessions.rs | 63 ++- n00n-ui/src/agent/agent_loop.rs | 75 ++- n00n-ui/src/agent/shared_queue.rs | 131 ++++- n00n-ui/src/app/mod.rs | 197 ++++++- n00n-ui/src/app/queue.rs | 167 +++++- n00n-ui/src/app/session.rs | 146 +++++- n00n-ui/src/app/session_state.rs | 1 + n00n-ui/src/app/shell.rs | 7 + n00n-ui/src/app/tests.rs | 507 ++++++++++++++++++- n00n-ui/src/chat.rs | 5 +- n00n-ui/src/components/messages/mod.rs | 17 +- n00n-ui/src/components/mod.rs | 14 +- n00n-ui/src/event_loop.rs | 295 +++++++++-- n00n-ui/src/storage_writer.rs | 261 ++++++++-- 18 files changed, 1941 insertions(+), 173 deletions(-) create mode 100644 changelog.d/submission-escape-hatch.added.md diff --git a/changelog.d/submission-escape-hatch.added.md b/changelog.d/submission-escape-hatch.added.md new file mode 100644 index 000000000..12874095c --- /dev/null +++ b/changelog.d/submission-escape-hatch.added.md @@ -0,0 +1 @@ +Made submitted messages appear before startup work and added a safe short Escape window to return unsent text and images for editing. diff --git a/n00n-agent/src/agent/run.rs b/n00n-agent/src/agent/run.rs index eb8809ae9..5a2c4f049 100644 --- a/n00n-agent/src/agent/run.rs +++ b/n00n-agent/src/agent/run.rs @@ -13,7 +13,7 @@ use super::history::{History, sanitize_cancelled_history}; use super::instructions::LoadedInstructions; use super::streaming::stream_with_retry; use super::tool_dispatch::{self, RecentCalls}; -use crate::cancel::{CancelMap, CancelToken}; +use crate::cancel::{CancelMap, CancelToken, PreDispatchGate}; use crate::mcp::McpHandle; use crate::permissions::PermissionManager; use crate::tools::{Deadline, FileReadTracker, LocalTools, ToolAudience, ToolContext}; @@ -94,7 +94,9 @@ pub struct Agent<'h> { recent_calls: RecentCalls, auto_compact: bool, loaded_instructions: LoadedInstructions, - rollback_len: usize, + pre_dispatch_rollback_len: Option, + rollback_len: Option, + pre_dispatch_gate: Option>, mcp: Option, config: AgentConfig, tool_output_lines: ToolOutputLines, @@ -138,7 +140,9 @@ impl<'h> Agent<'h> { recent_calls: RecentCalls::new(), auto_compact: compaction::auto_compact_enabled(), loaded_instructions: LoadedInstructions::new(), - rollback_len: 0, + pre_dispatch_rollback_len: None, + rollback_len: None, + pre_dispatch_gate: None, mcp: None, reauth_attempts: 0, post_tool_empty_retried: false, @@ -181,6 +185,18 @@ impl<'h> Agent<'h> { self } + #[must_use] + pub fn with_pre_dispatch_gate(mut self, gate: Arc) -> Self { + self.pre_dispatch_gate = Some(gate); + self + } + + #[must_use] + pub fn with_pre_dispatch_rollback_len(mut self, rollback_len: usize) -> Self { + self.pre_dispatch_rollback_len = Some(rollback_len); + self + } + #[must_use] pub fn with_local_tools(mut self, local_tools: LocalTools) -> Self { self.local_tools = local_tools; @@ -209,7 +225,11 @@ impl<'h> Agent<'h> { /// Returns an error if the agent loop fails due to provider errors, /// tool execution failures, or cancellation. pub async fn run(&mut self, input: AgentInput) -> Result<(), AgentError> { - self.rollback_len = self.history.len(); + let rollback_len = self.rollback_len.unwrap_or_else(|| self.history.len()); + self.rollback_len = Some(rollback_len); + let pre_dispatch_rollback_len = self + .pre_dispatch_rollback_len + .unwrap_or_else(|| rollback_len); let msg = Message::user_with_images(input.message.clone(), input.images); self.history.push(msg); self.context_size = estimate_message_tokens(self.history.as_slice()); @@ -234,7 +254,16 @@ impl<'h> Agent<'h> { .await; if matches!(result, Err(AgentError::Cancelled)) { - sanitize_cancelled_history(self.history, self.rollback_len); + if self + .pre_dispatch_gate + .as_ref() + .is_some_and(|gate| gate.is_cancelled()) + { + self.history.truncate(pre_dispatch_rollback_len); + } else { + let rollback_len = self.rollback_len.unwrap_or_else(|| self.history.len()); + sanitize_cancelled_history(self.history, rollback_len); + } } result @@ -258,8 +287,14 @@ impl<'h> Agent<'h> { } } + fn commit_pre_dispatch(&self) -> bool { + self.pre_dispatch_gate + .as_ref() + .is_none_or(|gate| gate.try_commit()) + } + async fn turn(&mut self) -> Result { - if self.cancel.is_cancelled() { + if self.cancel.is_cancelled() || !self.commit_pre_dispatch() { return Err(AgentError::Cancelled); } let response = match stream_with_retry( @@ -482,6 +517,9 @@ impl<'h> Agent<'h> { } async fn do_compact(&mut self) -> Result<(), AgentError> { + if !self.commit_pre_dispatch() { + return Err(AgentError::Cancelled); + } let (compact_provider, compact_model) = resolve_compaction_model(&self.provider, &self.model, self.timeouts); let usage = compaction::compact_history( @@ -494,7 +532,7 @@ impl<'h> Agent<'h> { .await?; let cost = usage.cost(&compact_model.pricing, false); self.record_usage(usage, cost); - self.rollback_len = self.history.len(); + self.rollback_len = Some(self.history.len()); self.event_tx.send(AgentEvent::CompactionDone)?; self.history .push(Message::synthetic(CONTINUE_AFTER_COMPACT.into())); @@ -563,13 +601,17 @@ pub fn estimate_message_tokens(messages: &[Message]) -> u32 { #[cfg(test)] mod tests { use std::collections::VecDeque; - use std::sync::{Arc, Mutex}; + use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }; use n00n_providers::provider::{BoxFuture, Provider}; use n00n_providers::{ ContentBlock, Message, Model, ProviderEvent, RequestOptions, Role, StopReason, StreamResponse, TokenUsage, }; + use n00n_storage::sessions::TranscriptEntry; use serde_json::Value; use test_case::test_case; @@ -600,6 +642,8 @@ mod tests { struct MockProvider { responses: Mutex>, requests: Arc>>>, + cancel_on_request: Option, + calls: AtomicUsize, } impl MockProvider { @@ -607,6 +651,8 @@ mod tests { Self { responses: Mutex::new(responses), requests: Arc::new(Mutex::new(Vec::new())), + cancel_on_request: None, + calls: AtomicUsize::new(0), } } @@ -616,10 +662,21 @@ mod tests { Self { responses: Mutex::new(responses), requests: Arc::clone(&requests), + cancel_on_request: None, + calls: AtomicUsize::new(0), }, requests, ) } + + fn cancel_on_request(responses: Vec, request: usize) -> Self { + Self { + responses: Mutex::new(responses), + requests: Arc::new(Mutex::new(Vec::new())), + cancel_on_request: Some(request), + calls: AtomicUsize::new(0), + } + } } impl Provider for MockProvider { @@ -634,6 +691,10 @@ mod tests { _: Option<&'a SessionRef>, ) -> BoxFuture<'a, Result> { Box::pin(async { + let request = self.calls.fetch_add(1, Ordering::Relaxed); + if self.cancel_on_request == Some(request) { + return Err(AgentError::Cancelled); + } self.requests.lock().unwrap().push(messages.to_vec()); let mut responses = self.responses.lock().unwrap(); assert!(!responses.is_empty(), "MockProvider: no more responses"); @@ -1115,6 +1176,52 @@ mod tests { }); } + #[test] + fn post_compaction_cancellation_keeps_compaction_transcript_and_cleanup() { + smol::block_on(async { + let mut history = History::new(vec![Message::user("x".repeat(680_000))]); + let (mut agent, _event_rx) = make_agent( + MockProvider::cancel_on_request(vec![text_response(StopReason::EndTurn)], 1), + &mut history, + ); + agent.model = Arc::new(small_context_model(200_000, 8_192)); + + let result = agent.run(default_input()).await; + + assert!(matches!(result, Err(AgentError::Cancelled))); + drop(agent); + assert!( + history + .transcript() + .iter() + .any(|entry| matches!(entry, TranscriptEntry::Compaction { .. })) + ); + assert_ends_with_cancel_marker(&history); + }); + } + + #[test] + fn pre_dispatch_cancellation_rolls_back_without_provider_call() { + smol::block_on(async { + let (provider, requests) = + MockProvider::recording(vec![text_response(StopReason::EndTurn)]); + let mut history = History::new(Vec::new()); + let (mut agent, _event_rx) = make_agent(provider, &mut history); + let gate = Arc::new(PreDispatchGate::new()); + assert!(gate.try_cancel()); + agent = agent + .with_pre_dispatch_gate(gate) + .with_pre_dispatch_rollback_len(0); + + let result = agent.run(default_input()).await; + + assert!(matches!(result, Err(AgentError::Cancelled))); + drop(agent); + assert_eq!(history.len(), 0); + assert_eq!(requests.lock().unwrap().len(), 0); + }); + } + #[test_case( vec![tool_call_response("nonexistent_tool_xyz", "t1"), text_response(StopReason::EndTurn)], "t1" diff --git a/n00n-agent/src/cancel.rs b/n00n-agent/src/cancel.rs index a31ecbdcc..e08cfc988 100644 --- a/n00n-agent/src/cancel.rs +++ b/n00n-agent/src/cancel.rs @@ -5,7 +5,7 @@ use std::collections::HashMap; use std::future::Future; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, Mutex}; use event_listener::Event; @@ -27,6 +27,63 @@ pub struct CancelToken(Arc); pub struct CancelTrigger(Arc); +const PRE_DISPATCH_PENDING: u8 = 0; +const PRE_DISPATCH_COMMITTED: u8 = 1; +const PRE_DISPATCH_CANCELLED: u8 = 2; + +/// One-shot atomic handoff between UI cancellation and provider dispatch. +pub struct PreDispatchGate(AtomicU8); + +impl Default for PreDispatchGate { + fn default() -> Self { + Self::new() + } +} + +impl PreDispatchGate { + #[must_use] + pub const fn new() -> Self { + Self(AtomicU8::new(PRE_DISPATCH_PENDING)) + } + + /// Claims the submission before its first provider request can start. + #[must_use] + pub fn try_cancel(&self) -> bool { + self.0 + .compare_exchange( + PRE_DISPATCH_PENDING, + PRE_DISPATCH_CANCELLED, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + } + + /// Commits provider dispatch. Repeated calls support safe provider retries. + #[must_use] + pub fn try_commit(&self) -> bool { + match self.0.compare_exchange( + PRE_DISPATCH_PENDING, + PRE_DISPATCH_COMMITTED, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) | Err(PRE_DISPATCH_COMMITTED) => true, + Err(_) => false, + } + } + + #[must_use] + pub fn is_cancelled(&self) -> bool { + self.0.load(Ordering::Acquire) == PRE_DISPATCH_CANCELLED + } + + #[must_use] + pub fn is_committed(&self) -> bool { + self.0.load(Ordering::Acquire) == PRE_DISPATCH_COMMITTED + } +} + impl CancelToken { #[must_use] pub fn new() -> (CancelTrigger, Self) { @@ -315,4 +372,46 @@ mod tests { map.cancel_or_precancel("x".to_owned()); assert!(tok2.is_cancelled()); } + + #[test] + fn pre_dispatch_cancel_and_commit_are_mutually_exclusive() { + let gate = Arc::new(PreDispatchGate::new()); + let barrier = Arc::new(std::sync::Barrier::new(3)); + let cancel = { + let gate = Arc::clone(&gate); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + gate.try_cancel() + }) + }; + let commit = { + let gate = Arc::clone(&gate); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + gate.try_commit() + }) + }; + barrier.wait(); + + let cancel_won = cancel.join().unwrap(); + let commit_won = commit.join().unwrap(); + assert_ne!(cancel_won, commit_won); + assert_eq!(gate.is_cancelled(), cancel_won); + assert_eq!(gate.is_committed(), commit_won); + } + + #[test] + fn pre_dispatch_lifecycle_is_monotonic() { + let cancelled = PreDispatchGate::new(); + assert!(cancelled.try_cancel()); + assert!(!cancelled.try_cancel()); + assert!(!cancelled.try_commit()); + + let committed = PreDispatchGate::new(); + assert!(committed.try_commit()); + assert!(committed.try_commit()); + assert!(!committed.try_cancel()); + } } diff --git a/n00n-agent/src/lib.rs b/n00n-agent/src/lib.rs index 9c0309493..93e890fca 100644 --- a/n00n-agent/src/lib.rs +++ b/n00n-agent/src/lib.rs @@ -14,7 +14,7 @@ pub use agent::{ Agent, AgentParams, AgentRunParams, History, Instructions, LoadedInstructions, SharedMessages, SharedTranscript, find_subdirectory_instructions, is_instruction_file, }; -pub use cancel::{CancelMap, CancelToken, CancelTrigger}; +pub use cancel::{CancelMap, CancelToken, CancelTrigger, PreDispatchGate}; pub use n00n_config::{AgentConfig, PermissionsConfig, ToolOutputLines}; pub mod command; pub mod diff; @@ -76,6 +76,7 @@ pub struct McpPromptRef { pub arguments: HashMap, } +#[derive(Clone)] pub struct AgentInput { pub message: String, pub mode: AgentMode, diff --git a/n00n-storage/src/sessions.rs b/n00n-storage/src/sessions.rs index 393b2fabb..a6102f8dd 100644 --- a/n00n-storage/src/sessions.rs +++ b/n00n-storage/src/sessions.rs @@ -89,6 +89,46 @@ impl std::ops::AddAssign for StoredTokenUsage { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum StoredImageMediaType { + Png, + Jpeg, + Gif, + Webp, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredImageSource { + pub media_type: StoredImageMediaType, + pub data: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredMcpPrompt { + pub qualified_name: String, + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub arguments: HashMap, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredQueuedMessage { + pub text: String, + pub images: Vec, + /// `None` means this field was absent in an older session snapshot. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mode: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plan_path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thinking: Option, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub fast: bool, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub workflow: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prompt: Option, +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct SessionMeta { #[serde(default)] @@ -105,6 +145,9 @@ pub struct SessionMeta { pub input_draft: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub queued_messages: Vec, + /// Full queued-message snapshots, including messages hidden by the paint gate. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub queued_submissions: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub subagents: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -115,6 +158,9 @@ pub struct SessionMeta { pub workflow: bool, #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub usage_by_model: HashMap, + /// Monotonic snapshot ordering used by write-behind persistence. + #[serde(default)] + pub revision: u64, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -1567,8 +1613,8 @@ mod tests { openai_response_chain_path, save_openai_response_chain, }; use super::{ - SCAN_CACHE_FILE, Session, SessionError, SessionLog, StorageError, TitleSource, - TranscriptEntry, + SCAN_CACHE_FILE, Session, SessionError, SessionLog, StorageError, StoredQueuedMessage, + TitleSource, TranscriptEntry, }; use crate::StateDir; use crate::id::N00nId; @@ -2363,6 +2409,19 @@ mod tests { assert!(!meta.workflow); } + #[test] + fn queued_message_backward_compat_defaults_semantics() { + let stored: StoredQueuedMessage = + serde_json::from_str(r#"{"text":"queued","images":[]}"#).unwrap(); + assert_eq!(stored.text, "queued"); + assert!(stored.mode.is_none()); + assert!(stored.plan_path.is_none()); + assert!(stored.thinking.is_none()); + assert!(!stored.fast); + assert!(!stored.workflow); + assert!(stored.prompt.is_none()); + } + #[test] fn session_meta_persists_through_save_load() { let tmp = TempDir::new().unwrap(); diff --git a/n00n-ui/src/agent/agent_loop.rs b/n00n-ui/src/agent/agent_loop.rs index 5cb78cdfe..852e5ee97 100644 --- a/n00n-ui/src/agent/agent_loop.rs +++ b/n00n-ui/src/agent/agent_loop.rs @@ -137,6 +137,7 @@ impl AgentLoop { image_count, input, displayed, + pre_dispatch_gate, .. } => { if !displayed { @@ -146,7 +147,8 @@ impl AgentLoop { images: input.images.clone(), }); } - self.do_agent_run(input, event_tx, run_id).await + self.do_agent_run(input, event_tx, run_id, pre_dispatch_gate) + .await } QueueItem::Compact { .. } => self.do_compact(&event_tx).await, }; @@ -184,7 +186,9 @@ impl AgentLoop { mut input: AgentInput, event_tx: EventSender, run_id: u64, + pre_dispatch_gate: Arc, ) -> Result<(), AgentError> { + let rollback_len = self.history.len(); let slot = self.model_slot.load(); let old_cwd = self.vars.apply("{cwd}").into_owned(); @@ -194,36 +198,57 @@ impl AgentLoop { } self.rebuild_tools(&slot.model, input.workflow); - for msg in std::mem::take(&mut input.preamble) { - self.history.push(msg); - } - - if let Some(ref prompt_ref) = input.prompt { + // Keep the cancellation trigger live while MCP prompt retrieval is in + // flight. Retrieval is fallible, so neither the shell preamble nor the + // MCP messages may enter history until it succeeds. + let (trigger, cancel) = CancelToken::new(); + self.set_cancel_trigger(run_id, trigger); + let prompt_messages = if let Some(ref prompt_ref) = input.prompt { let Some(ref mcp) = self.mcp_handle else { + self.clear_cancel_trigger(run_id); return Err(AgentError::Tool { tool: "mcp_prompt".into(), message: "MCP not available".into(), }); }; - let messages = mcp + match mcp .get_prompt(&prompt_ref.qualified_name, &prompt_ref.arguments) .await - .map_err(|e| AgentError::Tool { - tool: "mcp_prompt".into(), - message: e.to_string(), - })?; - for pm in messages { - let text = pm.content.text.unwrap_or_default(); - let msg = match pm.role { - PromptRole::Assistant => Message { - role: n00n_providers::Role::Assistant, - content: vec![n00n_providers::ContentBlock::Text { text }], - ..Default::default() - }, - PromptRole::User => Message::user(text), - }; - self.history.push(msg); + { + Ok(messages) => messages, + Err(error) => { + self.clear_cancel_trigger(run_id); + if cancel.is_cancelled() || pre_dispatch_gate.is_cancelled() { + return Err(AgentError::Cancelled); + } + return Err(AgentError::Tool { + tool: "mcp_prompt".into(), + message: error.to_string(), + }); + } } + } else { + Vec::new() + }; + if cancel.is_cancelled() || pre_dispatch_gate.is_cancelled() { + self.clear_cancel_trigger(run_id); + return Err(AgentError::Cancelled); + } + + for msg in std::mem::take(&mut input.preamble) { + self.history.push(msg); + } + for pm in prompt_messages { + let text = pm.content.text.unwrap_or_default(); + let msg = match pm.role { + PromptRole::Assistant => Message { + role: n00n_providers::Role::Assistant, + content: vec![n00n_providers::ContentBlock::Text { text }], + ..Default::default() + }, + PromptRole::User => Message::user(text), + }; + self.history.push(msg); } let prompt_slots = match self.lua_handle.as_ref() { @@ -238,8 +263,6 @@ impl AgentLoop { &slot.model, ); self.publish_btw_system(&prompt_slots); - let (trigger, cancel) = CancelToken::new(); - self.set_cancel_trigger(run_id, trigger); while self.answer_rx.lock().await.try_recv().is_ok() {} @@ -269,7 +292,9 @@ impl AgentLoop { .with_user_response_rx(Arc::clone(&self.answer_rx)) .with_interrupt_source(Arc::clone(&self.queue) as Arc) .with_cancel(cancel) - .with_mcp(self.mcp_handle.clone()); + .with_mcp(self.mcp_handle.clone()) + .with_pre_dispatch_gate(pre_dispatch_gate) + .with_pre_dispatch_rollback_len(rollback_len); let result = agent.run(input).await; drop(agent); diff --git a/n00n-ui/src/agent/shared_queue.rs b/n00n-ui/src/agent/shared_queue.rs index b38e9ae38..7baa837c6 100644 --- a/n00n-ui/src/agent/shared_queue.rs +++ b/n00n-ui/src/agent/shared_queue.rs @@ -8,9 +8,14 @@ use std::borrow::Cow; use std::collections::VecDeque; -use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::sync::{ + Arc, Mutex, MutexGuard, PoisonError, + atomic::{AtomicBool, Ordering}, +}; -use n00n_agent::{AgentInput, ExtractedCommand, ImageSource, InterruptPoint, InterruptSource}; +use n00n_agent::{ + AgentInput, ExtractedCommand, ImageSource, InterruptPoint, InterruptSource, PreDispatchGate, +}; use crate::components::input::Submission; use crate::components::queue_panel::QueueEntry; @@ -50,6 +55,11 @@ pub(crate) enum QueueItem { image_count: usize, input: AgentInput, run_id: u64, + submission_id: u64, + pre_dispatch_gate: Arc, + /// The UI releases this after persistence succeeds. A blocked item + /// keeps later submissions from reaching the provider first. + ready: Arc, /// `true` when the UI already drew the bubble (immediate dispatch). /// The agent then skips `QueueItemConsumed` so we don't draw it twice. /// `false` when the user typed while the agent was busy: the UI waits @@ -105,6 +115,17 @@ impl QueueItem { Self::Compact { .. } => true, } } + + fn is_ready(&self) -> bool { + match self { + Self::Message { ready, .. } => ready.load(Ordering::Acquire), + Self::Compact { .. } => true, + } + } + + fn blocks_dispatch(&self) -> bool { + matches!(self, Self::Message { ready, .. } if !ready.load(Ordering::Acquire)) + } } fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { @@ -140,6 +161,11 @@ impl QueueSender { let _ = self.notify_tx.try_send(()); } + pub(crate) fn push_front(&self, entry: QueueItem) { + lock(&self.items).push_front(entry); + let _ = self.notify_tx.try_send(()); + } + fn panel_index(items: &VecDeque, index: usize) -> Option { items .iter() @@ -187,6 +213,49 @@ impl QueueSender { lock(&self.items).clear(); } + pub(crate) fn remove_submission(&self, submission_id: u64) { + let mut items = lock(&self.items); + let before = items.len(); + items.retain(|item| { + !matches!( + item, + QueueItem::Message { + submission_id: id, + .. + } if *id == submission_id + ) + }); + let removed = items.len() != before; + drop(items); + if removed { + let _ = self.notify_tx.try_send(()); + } + } + + pub(crate) fn contains_submission(&self, submission_id: u64) -> bool { + lock(&self.items).iter().any(|item| { + matches!(item, QueueItem::Message { submission_id: id, .. } if *id == submission_id) + }) + } + + pub(crate) fn mark_submission_ready(&self, submission_id: u64, input: AgentInput) -> bool { + let mut items = lock(&self.items); + let Some(QueueItem::Message { + input: queued_input, + ready, + .. + }) = items.iter_mut().find(|item| { + matches!(item, QueueItem::Message { submission_id: id, .. } if *id == submission_id) + }) else { + return false; + }; + *queued_input = input; + ready.store(true, Ordering::Release); + drop(items); + let _ = self.notify_tx.try_send(()); + true + } + pub(crate) fn text_messages(&self) -> Vec { lock(&self.items) .iter() @@ -198,6 +267,16 @@ impl QueueSender { .collect() } + pub(crate) fn queued_inputs(&self) -> Vec { + lock(&self.items) + .iter() + .filter_map(|item| match item { + QueueItem::Message { input, .. } => Some(input.clone()), + QueueItem::Compact { .. } => None, + }) + .collect() + } + pub(crate) fn panel_len(&self) -> usize { lock(&self.items) .iter() @@ -217,14 +296,21 @@ impl QueueSender { impl QueueReceiver { pub(crate) fn pop(&self) -> Option { let mut items = lock(&self.items); - let index = items.iter().position(|item| { - matches!( - item, - QueueItem::Message { - delivery: Delivery::TurnEnd, - .. - } | QueueItem::Compact { .. } - ) + let index = items.iter().enumerate().find_map(|(index, item)| { + if !item.is_ready() + || items.iter().take(index).any(QueueItem::blocks_dispatch) + || !matches!( + item, + QueueItem::Message { + delivery: Delivery::TurnEnd, + .. + } | QueueItem::Compact { .. } + ) + { + None + } else { + Some(index) + } })?; items.remove(index) } @@ -237,13 +323,18 @@ impl QueueReceiver { impl InterruptSource for QueueReceiver { fn poll(&self, point: InterruptPoint) -> Option { let mut items = lock(&self.items); - let index = items.iter().position(|item| match item { - QueueItem::Message { delivery, .. } => match delivery { - Delivery::TurnEnd => false, - Delivery::Steering => point == InterruptPoint::ToolComplete, - Delivery::Immediate => true, - }, - QueueItem::Compact { .. } => point == InterruptPoint::Safe, + let index = items.iter().enumerate().find_map(|(index, item)| { + if !item.is_ready() || items.iter().take(index).any(QueueItem::blocks_dispatch) { + return None; + } + match item { + QueueItem::Message { delivery, .. } => match delivery { + Delivery::TurnEnd => None, + Delivery::Steering => (point == InterruptPoint::ToolComplete).then_some(index), + Delivery::Immediate => Some(index), + }, + QueueItem::Compact { .. } => (point == InterruptPoint::Safe).then_some(index), + } })?; items.remove(index).map(QueueItem::into_extracted_command) } @@ -269,6 +360,9 @@ mod tests { prompt: None, }, run_id: 0, + submission_id: 0, + pre_dispatch_gate: Arc::new(PreDispatchGate::new()), + ready: Arc::new(AtomicBool::new(true)), displayed, delivery: Delivery::TurnEnd, } @@ -301,6 +395,9 @@ mod tests { prompt: None, }, run_id: 0, + submission_id: 0, + pre_dispatch_gate: Arc::new(PreDispatchGate::new()), + ready: Arc::new(AtomicBool::new(true)), displayed: false, delivery, }; diff --git a/n00n-ui/src/app/mod.rs b/n00n-ui/src/app/mod.rs index f0ffe8ddf..d41396632 100644 --- a/n00n-ui/src/app/mod.rs +++ b/n00n-ui/src/app/mod.rs @@ -45,7 +45,8 @@ use crate::components::theme_picker::{ThemePicker, ThemePickerAction}; use crate::components::tool_display::format_turn_usage; use crate::components::usage_modal::{UsageFetchState, UsageModal}; use crate::components::{ - Action, DisplayMessage, DisplayRole, ExitRequest, Overlay, RetryInfo, Status, is_ctrl, + Action, DisplayMessage, DisplayRole, ExitRequest, Overlay, RetryInfo, Status, + SubmissionDispatch, is_ctrl, }; use crate::image; use crate::selection::{SelectionState, SelectionZone, ZoneRegistry}; @@ -54,7 +55,7 @@ use crossterm::event::{KeyCode, KeyEvent, MouseEvent}; use n00n_agent::permissions::PermissionManager; use n00n_agent::{ AgentEvent, Envelope, ImageSource, McpConfigErrors, McpPromptInfo, McpSnapshotReader, - SubagentInfo, ToolOutput, + PreDispatchGate, SubagentInfo, ToolOutput, }; use n00n_config::UiConfig; use n00n_lua::{EventHandle, HintReader, KeymapReader, LuaCommandReader}; @@ -76,6 +77,8 @@ pub(crate) use session::session_has_content; use session_state::SessionState; const CANCEL_MSG: &str = "Cancelled."; +const PERSISTENCE_FAILURE_MSG: &str = "Could not save the session. Your message was restored."; +pub(super) const SUBMISSION_ESCAPE_WINDOW: Duration = Duration::from_millis(2_500); /// Bypasses the per-run staleness filter because re-bake replies /// don't belong to any real agent run. pub(crate) const RESTORE_RUN_ID: u64 = u64::MAX; @@ -146,6 +149,28 @@ pub(super) enum PendingInput { }, } +trait SubmissionClock: Send + Sync { + fn now(&self) -> Instant; +} + +struct SystemSubmissionClock; + +impl SubmissionClock for SystemSubmissionClock { + fn now(&self) -> Instant { + Instant::now() + } +} + +struct PendingSubmission { + submission_id: u64, + run_id: u64, + submitted_at: Instant, + message: QueuedMessage, + gate: Arc, + preamble: Vec, + display_len_before: usize, +} + pub enum Msg { Key(KeyEvent), Paste(String), @@ -186,6 +211,9 @@ pub struct App { pub(crate) cmd_tx: Option>, pub(super) pending_input: PendingInput, pub(crate) run_id: u64, + next_submission_id: u64, + pending_submission: Option, + submission_clock: Arc, pub(super) retry_info: Option, pub(super) zones: ZoneRegistry, pub(super) selection_state: Option, @@ -274,6 +302,9 @@ impl App { cmd_tx: None, pending_input: PendingInput::None, run_id: 0, + next_submission_id: 0, + pending_submission: None, + submission_clock: Arc::new(SystemSubmissionClock), retry_info: None, zones: ZoneRegistry::new(), selection_state: None, @@ -974,6 +1005,9 @@ impl App { } KeyCode::Tab if !self.is_bash_input() => self.toggle_mode(), KeyCode::Esc => { + if self.try_restore_pending_submission() { + return vec![]; + } if let Some(t) = self.last_esc.take() && t.elapsed() < self.status_bar.flash_duration { @@ -1091,11 +1125,159 @@ impl App { self.submit_or_queue(sub.into()) } + fn try_restore_pending_submission(&mut self) -> bool { + if !matches!(self.status, Status::Streaming | Status::Error { .. }) { + return false; + } + let Some(pending) = self.pending_submission.as_ref() else { + return false; + }; + let failed_before_dispatch = + matches!(self.status, Status::Error { .. }) && !pending.gate.is_committed(); + if pending.run_id != self.run_id + || (!failed_before_dispatch + && self + .submission_clock + .now() + .saturating_duration_since(pending.submitted_at) + > SUBMISSION_ESCAPE_WINDOW) + || !self.input_box.is_empty() + { + return false; + } + if !pending.gate.try_cancel() { + self.pending_submission = None; + return false; + } + let submission_id = pending.submission_id; + self.restore_pending_submission(submission_id, pending.run_id) + } + + fn restore_pending_submission(&mut self, submission_id: u64, run_id: u64) -> bool { + let Some(pending) = self.pending_submission.take() else { + return false; + }; + if pending.submission_id != submission_id + || pending.run_id != run_id + || self.run_id != run_id + { + self.pending_submission = Some(pending); + return false; + } + + self.queue.remove_submission(submission_id); + self.shell.restore_results(pending.preamble); + self.main_chat() + .truncate_messages(pending.display_len_before); + self.input_box.set_submission(Submission { + text: pending.message.text, + images: pending.message.images, + }); + self.run_id += 1; + self.status = Status::Idle; + self.retry_info = None; + self.last_esc = None; + self.status_bar.clear_flash(); + true + } + + pub(crate) fn handle_submission_persistence_failure(&mut self, dispatch: &SubmissionDispatch) { + self.queue.remove_submission(dispatch.submission_id); + if dispatch.paint_required { + let Some(pending) = self.pending_submission.as_ref() else { + return; + }; + if pending.submission_id != dispatch.submission_id + || pending.run_id != dispatch.run_id + || !dispatch.gate.try_cancel() + { + return; + } + if self.restore_pending_submission(dispatch.submission_id, dispatch.run_id) { + self.flash(PERSISTENCE_FAILURE_MSG.into()); + } + return; + } + + let relevant_run = dispatch.run_id == self.run_id; + let _ = dispatch.gate.try_cancel(); + if relevant_run && self.status == Status::Streaming { + self.status = Status::error(PERSISTENCE_FAILURE_MSG.into()); + self.main_chat().push(DisplayMessage::new( + DisplayRole::Error, + PERSISTENCE_FAILURE_MSG.into(), + )); + self.fire_session_autocmd( + "TurnError", + serde_json::json!({ "message": PERSISTENCE_FAILURE_MSG }), + ); + } + } + pub(crate) fn preserve_submission_for_shutdown(&mut self, dispatch: SubmissionDispatch) { + if !dispatch.paint_required { + return; + } + let Some(pending) = self.pending_submission.as_ref() else { + return; + }; + if pending.submission_id != dispatch.submission_id + || pending.run_id != dispatch.run_id + || !Arc::ptr_eq(&pending.gate, &dispatch.gate) + { + return; + } + self.queue.preserve_submission(dispatch); + } + + pub(crate) fn stage_submission_preamble(&mut self, dispatch: &mut SubmissionDispatch) -> bool { + if !dispatch.paint_required { + if dispatch.gate.is_cancelled() { + return false; + } + dispatch.input.preamble = self.shell.drain_results(); + return true; + } + let Some(pending) = self.pending_submission.as_mut() else { + return false; + }; + if pending.submission_id != dispatch.submission_id + || pending.run_id != dispatch.run_id + || self.run_id != dispatch.run_id + || !Arc::ptr_eq(&pending.gate, &dispatch.gate) + || dispatch.gate.is_cancelled() + { + return false; + } + let preamble = self.shell.drain_results(); + dispatch.input.preamble.clone_from(&preamble); + pending.preamble = preamble; + true + } + + pub(crate) fn accepts_submission_persistence(&self, dispatch: &SubmissionDispatch) -> bool { + if dispatch.run_id != self.run_id || dispatch.gate.is_cancelled() { + return false; + } + if !dispatch.paint_required { + return true; + } + self.pending_submission.as_ref().is_some_and(|pending| { + pending.submission_id == dispatch.submission_id + && pending.run_id == dispatch.run_id + && Arc::ptr_eq(&pending.gate, &dispatch.gate) + }) + } + pub(crate) fn cancel_current_run(&mut self) -> Vec { self.handle_cancel() } fn handle_cancel(&mut self) -> Vec { + if let Some(pending) = self.pending_submission.take() + && pending.gate.try_cancel() + { + self.shell.restore_results(pending.preamble); + } let cancelled_run = self.run_id; self.run_id += 1; self.retry_info = None; @@ -1629,9 +1811,14 @@ impl App { vec![] } else { self.run_id += 1; - self.status = Status::Streaming; - self.main_chat().show_user_message(display_text); - vec![Action::SendMessage(Box::new(input))] + self.start_submission( + &QueuedMessage { + text: display_text, + images: Vec::new(), + }, + input, + true, + ) } } diff --git a/n00n-ui/src/app/queue.rs b/n00n-ui/src/app/queue.rs index 1728e06ac..c6350bbe3 100644 --- a/n00n-ui/src/app/queue.rs +++ b/n00n-ui/src/app/queue.rs @@ -3,8 +3,9 @@ use super::{Action, App, Status}; use crate::agent::shared_queue::{Delivery, QueueItem, QueueSender}; -use crate::components::queue_panel::QueueEntry; +use crate::components::{SubmissionDispatch, queue_panel::QueueEntry}; use n00n_agent::ImageSource; +use std::sync::{Arc, atomic::AtomicBool}; pub(crate) use crate::agent::shared_queue::QueuedMessage; @@ -57,6 +58,44 @@ impl MessageQueue { self.focus = None; } + pub(crate) fn remove_submission(&self, submission_id: u64) { + if let Some(shared) = self.shared.clone() { + shared.remove_submission(submission_id); + } + } + + pub(crate) fn mark_submission_ready( + &self, + submission_id: u64, + input: n00n_agent::AgentInput, + ) -> bool { + self.shared + .as_ref() + .is_some_and(|shared| shared.mark_submission_ready(submission_id, input)) + } + + pub(crate) fn preserve_submission(&self, dispatch: SubmissionDispatch) { + let Some(shared) = self.shared.clone() else { + return; + }; + if shared.contains_submission(dispatch.submission_id) { + return; + } + let image_count = dispatch.input.images.len(); + let text = dispatch.input.message.clone(); + shared.push_front(QueueItem::Message { + text, + image_count, + input: dispatch.input, + run_id: dispatch.run_id, + submission_id: dispatch.submission_id, + pre_dispatch_gate: dispatch.gate, + ready: Arc::new(AtomicBool::new(false)), + displayed: false, + delivery: Delivery::TurnEnd, + }); + } + pub(crate) fn focus(&self) -> Option { self.focus } @@ -154,6 +193,13 @@ impl MessageQueue { ) } + pub(crate) fn queued_inputs(&self) -> Vec { + self.shared.as_ref().map_or( + vec![], + super::super::agent::shared_queue::QueueSender::queued_inputs, + ) + } + fn clamp_focus(&mut self) { let len = self.len(); self.focus = match self.focus { @@ -192,6 +238,24 @@ impl App { /// Keyboard path: nobody is around to receive an `Err`, so /// rejections flash on screen instead. + /// Session API prompts do not have a user-visible optimistic bubble, so + /// they deliberately bypass the terminal paint gate. + pub(crate) fn submit_background_prompt(&mut self, msg: QueuedMessage) -> SubmitOutcome { + if msg.text.trim().is_empty() && msg.images.is_empty() { + return SubmitOutcome::Rejected(EMPTY_PROMPT_ERR); + } + if self.status == Status::Streaming { + if self.queue_and_notify(msg) { + SubmitOutcome::Queued + } else { + SubmitOutcome::Rejected(NO_QUEUE_ERR) + } + } else { + self.run_id += 1; + SubmitOutcome::Started(self.start_background_submission(&msg)) + } + } + pub(super) fn submit_or_queue(&mut self, msg: QueuedMessage) -> Vec { match self.submit_prompt(msg) { SubmitOutcome::Started(actions) => actions, @@ -207,17 +271,42 @@ impl App { /// `QueueItemConsumed` draw it once the agent picks it up. Returns /// false when there is no shared queue, meaning the message was dropped. pub(super) fn queue_and_notify(&mut self, msg: QueuedMessage) -> bool { + let input = self.build_agent_input(&msg); + let queued = self.queue_input(msg, input, Delivery::TurnEnd); + if queued { + self.save_session(); + } + queued + } + + pub(crate) fn queue_restored_submission( + &mut self, + msg: QueuedMessage, + input: n00n_agent::AgentInput, + ) -> bool { + self.queue_input(msg, input, Delivery::TurnEnd) + } + + fn queue_input( + &mut self, + msg: QueuedMessage, + input: n00n_agent::AgentInput, + delivery: Delivery, + ) -> bool { let Some(shared) = self.queue.shared.clone() else { return false; }; - let input = self.build_agent_input(&msg); + let (submission_id, pre_dispatch_gate) = self.next_submission_metadata(); shared.push(QueueItem::Message { text: msg.text, image_count: msg.images.len(), input, run_id: self.run_id, + submission_id, + pre_dispatch_gate, + ready: Arc::new(AtomicBool::new(true)), displayed: false, - delivery: Delivery::TurnEnd, + delivery, }); true } @@ -238,11 +327,15 @@ impl App { return false; }; let input = self.build_agent_input(&msg); + let (submission_id, pre_dispatch_gate) = self.next_submission_metadata(); let item = QueueItem::Message { text: msg.text, image_count: msg.images.len(), input, run_id: self.run_id, + submission_id, + pre_dispatch_gate, + ready: Arc::new(AtomicBool::new(true)), displayed: false, delivery, }; @@ -251,6 +344,7 @@ impl App { } else { shared.push(item); } + self.save_session(); true } @@ -278,11 +372,69 @@ impl App { /// Immediate path: kick off the agent and draw the bubble in the same /// frame, so the user sees their message land where it will stay. pub(super) fn start_from_queue(&mut self, msg: &QueuedMessage) -> Vec { + self.start_submission(msg, self.build_agent_input(msg), true) + } + + fn start_background_submission(&mut self, msg: &QueuedMessage) -> Vec { + self.start_submission(msg, self.build_agent_input(msg), false) + } + + pub(super) fn start_submission( + &mut self, + msg: &QueuedMessage, + input: n00n_agent::AgentInput, + paint_required: bool, + ) -> Vec { self.status = Status::Streaming; self.fire_session_autocmd("TurnStart", serde_json::json!({})); - self.main_chat() - .show_user_message_with_images(msg.text.clone(), msg.images.clone()); - vec![Action::SendMessage(Box::new(self.build_agent_input(msg)))] + let display_len_before = self.main_chat().message_count(); + if paint_required { + self.main_chat() + .show_user_message_with_images(msg.text.clone(), msg.images.clone()); + } + + let (submission_id, gate) = self.next_submission_metadata(); + let Some(shared) = self.queue.shared.clone() else { + return vec![]; + }; + shared.push(QueueItem::Message { + text: msg.text.clone(), + image_count: msg.images.len(), + input: input.clone(), + run_id: self.run_id, + submission_id, + pre_dispatch_gate: Arc::clone(&gate), + ready: Arc::new(AtomicBool::new(false)), + displayed: paint_required, + delivery: Delivery::TurnEnd, + }); + if paint_required { + self.pending_submission = Some(super::PendingSubmission { + submission_id, + run_id: self.run_id, + submitted_at: self.submission_clock.now(), + message: msg.clone(), + gate: std::sync::Arc::clone(&gate), + preamble: Vec::new(), + display_len_before, + }); + } + vec![Action::SendMessage(Box::new( + crate::components::SubmissionDispatch { + input, + submission_id, + run_id: self.run_id, + gate, + paint_required, + }, + ))] + } + fn next_submission_metadata(&mut self) -> (u64, std::sync::Arc) { + self.next_submission_id += 1; + ( + self.next_submission_id, + std::sync::Arc::new(n00n_agent::PreDispatchGate::new()), + ) } } @@ -307,6 +459,9 @@ mod tests { prompt: None, }, run_id: 0, + submission_id: 0, + pre_dispatch_gate: std::sync::Arc::new(n00n_agent::PreDispatchGate::new()), + ready: Arc::new(AtomicBool::new(true)), displayed: true, delivery: Delivery::TurnEnd, } diff --git a/n00n-ui/src/app/session.rs b/n00n-ui/src/app/session.rs index 772b95342..ac12c0555 100644 --- a/n00n-ui/src/app/session.rs +++ b/n00n-ui/src/app/session.rs @@ -1,3 +1,4 @@ +use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -5,9 +6,13 @@ use crate::chat::{Chat, DONE_TEXT, RESTORE_BATCH_SIZE, history_to_display, trans use crate::components::DisplayRole; use crate::components::rewind_picker::RewindEntry; use crate::components::{Action, LoadedSession}; +use n00n_agent::{AgentInput, AgentMode, McpPromptRef}; use n00n_providers::{Model, TokenUsage}; use n00n_storage::id::N00nId; -use n00n_storage::sessions::StoredSubagent; +use n00n_storage::sessions::{ + StoredImageMediaType, StoredImageSource, StoredMcpPrompt, StoredMode, StoredQueuedMessage, + StoredSubagent, StoredThinking, +}; use crate::AppSession; @@ -25,7 +30,93 @@ pub(crate) fn session_has_content(session: &AppSession) -> bool { || !session.meta.subagents.is_empty() || session.meta.input_draft.is_some() || !session.meta.queued_messages.is_empty() + || !session.meta.queued_submissions.is_empty() || session.meta.mode != Some(n00n_storage::sessions::StoredMode::Build) + || session.meta.plan_path.is_some() + || session.meta.plan_written + || !session.meta.session_rules.is_empty() + || session.meta.context_size != 0 + || !matches!(session.meta.thinking, None | Some(StoredThinking::Off)) + || session.meta.fast + || session.meta.workflow + || !session.meta.usage_by_model.is_empty() + || !session.transcript.is_empty() + || !session.tool_outputs.is_empty() + || session.token_usage != TokenUsage::default() +} + +fn stored_image(image: &n00n_agent::ImageSource) -> StoredImageSource { + StoredImageSource { + media_type: match image.media_type { + n00n_agent::ImageMediaType::Png => StoredImageMediaType::Png, + n00n_agent::ImageMediaType::Jpeg => StoredImageMediaType::Jpeg, + n00n_agent::ImageMediaType::Gif => StoredImageMediaType::Gif, + n00n_agent::ImageMediaType::Webp => StoredImageMediaType::Webp, + }, + data: image.data.to_string(), + } +} + +fn restored_image(image: StoredImageSource) -> n00n_agent::ImageSource { + let media_type = match image.media_type { + StoredImageMediaType::Png => n00n_agent::ImageMediaType::Png, + StoredImageMediaType::Jpeg => n00n_agent::ImageMediaType::Jpeg, + StoredImageMediaType::Gif => n00n_agent::ImageMediaType::Gif, + StoredImageMediaType::Webp => n00n_agent::ImageMediaType::Webp, + }; + n00n_agent::ImageSource::new(media_type, Arc::from(image.data)) +} + +fn stored_message(input: AgentInput) -> StoredQueuedMessage { + // Preamble contains live shell results and may include transient secrets. + let (mode, plan_path) = match input.mode { + AgentMode::Build => (Some(StoredMode::Build), None), + AgentMode::Plan(path) => ( + Some(StoredMode::Plan), + Some(path.to_string_lossy().into_owned()), + ), + }; + StoredQueuedMessage { + text: input.message, + images: input.images.iter().map(stored_image).collect(), + mode, + plan_path, + thinking: Some(input.thinking.into()), + fast: input.fast, + workflow: input.workflow, + prompt: input.prompt.map(|prompt| StoredMcpPrompt { + qualified_name: prompt.qualified_name, + arguments: prompt.arguments, + }), + } +} + +fn restored_submission(app: &App, message: StoredQueuedMessage) -> (QueuedMessage, AgentInput) { + let queued = QueuedMessage { + text: message.text, + images: message.images.into_iter().map(restored_image).collect(), + }; + let mut input = app.build_agent_input(&queued); + if let Some(mode) = message.mode { + input.mode = match mode { + StoredMode::Build => AgentMode::Build, + StoredMode::Plan => message + .plan_path + .map_or(input.mode, |path| AgentMode::Plan(PathBuf::from(path))), + }; + } + if let Some(thinking) = message.thinking { + input.thinking = thinking.into(); + } + input.fast = message.fast; + input.workflow = message.workflow; + input.prompt = message.prompt.map(|prompt| { + Box::new(McpPromptRef { + qualified_name: prompt.qualified_name, + arguments: prompt.arguments, + }) + }); + (queued, input) } impl App { @@ -34,6 +125,14 @@ impl App { } pub(crate) fn save_session(&mut self) { + let snapshot = self.session_snapshot(); + if !session_has_content(&snapshot) { + return; + } + self.storage_writer.send(Box::new(snapshot)); + } + + pub(crate) fn session_snapshot(&mut self) -> AppSession { self.state.sync_session( &self.shared_history, &self.shared_transcript, @@ -41,17 +140,17 @@ impl App { &self.permissions, ); self.sync_ephemeral_state(); - if !self.has_content() { - return; - } - self.enqueue_save(); + self.state.session.clone() } fn sync_ephemeral_state(&mut self) { let draft = self.input_box.buffer.value(); self.state.session.meta.input_draft = if draft.is_empty() { None } else { Some(draft) }; + let queued = self.queue.queued_inputs(); self.state.session.meta.queued_messages = self.queue.text_messages(); + self.state.session.meta.queued_submissions = + queued.into_iter().map(stored_message).collect(); self.state.session.meta.subagents = self .chats @@ -74,9 +173,11 @@ impl App { } } - pub(super) fn enqueue_save(&self) { - self.storage_writer - .send(Box::new(self.state.session.clone())); + pub(super) fn enqueue_save(&mut self) { + let snapshot = self.session_snapshot(); + if session_has_content(&snapshot) { + self.storage_writer.send(Box::new(snapshot)); + } } pub(super) fn reset_ui_chrome(&mut self) { @@ -123,12 +224,28 @@ impl App { self.input_box.buffer.move_to_end(); } - for text in std::mem::take(&mut self.state.session.meta.queued_messages) { - let msg = QueuedMessage { - text, - images: Vec::new(), + let queued: Vec<(QueuedMessage, AgentInput)> = + if self.state.session.meta.queued_submissions.is_empty() { + std::mem::take(&mut self.state.session.meta.queued_messages) + .into_iter() + .map(|text| { + let msg = QueuedMessage { + text, + images: Vec::new(), + }; + let input = self.build_agent_input(&msg); + (msg, input) + }) + .collect() + } else { + std::mem::take(&mut self.state.session.meta.queued_submissions) + .into_iter() + .map(|message| restored_submission(self, message)) + .collect() }; - self.queue_and_notify(msg); + self.state.session.meta.queued_messages.clear(); + for (msg, input) in queued { + self.queue_restored_submission(msg, input); } self.fire_restore_items(restore_items); @@ -252,7 +369,7 @@ impl App { } pub(crate) fn load_session(&mut self, session_id: N00nId) -> Vec { - let session = match AppSession::load(session_id, &self.storage) { + let mut session = match AppSession::load(session_id, &self.storage) { Ok(s) => s, Err(e) => { self.status_bar @@ -261,6 +378,7 @@ impl App { } }; self.save_session(); + session.meta.revision = session.meta.revision.max(self.state.session.meta.revision); let loaded = self.apply_loaded_session(session, &self.state.model.clone()); vec![Action::LoadSession(Box::new(loaded))] } diff --git a/n00n-ui/src/app/session_state.rs b/n00n-ui/src/app/session_state.rs index 29c26e3ba..f6d45d78f 100644 --- a/n00n-ui/src/app/session_state.rs +++ b/n00n-ui/src/app/session_state.rs @@ -118,6 +118,7 @@ impl SessionState { self.session.meta.thinking = Some(self.thinking.into()); self.session.meta.fast = self.fast; self.session.meta.workflow = self.workflow; + self.session.meta.revision = self.session.meta.revision.saturating_add(1); self.session.updated_at = n00n_storage::now_epoch(); self.session.update_title_if_default(); } diff --git a/n00n-ui/src/app/shell.rs b/n00n-ui/src/app/shell.rs index 636b64010..5fba42586 100644 --- a/n00n-ui/src/app/shell.rs +++ b/n00n-ui/src/app/shell.rs @@ -97,6 +97,13 @@ impl ShellState { pub fn drain_results(&mut self) -> Vec { std::mem::take(&mut self.pending_results) } + + pub fn restore_results(&mut self, results: Vec) { + if results.is_empty() { + return; + } + self.pending_results.splice(0..0, results); + } } impl App { diff --git a/n00n-ui/src/app/tests.rs b/n00n-ui/src/app/tests.rs index e04713e40..1f868307e 100644 --- a/n00n-ui/src/app/tests.rs +++ b/n00n-ui/src/app/tests.rs @@ -9,8 +9,8 @@ use arc_swap::ArcSwap; use crossterm::event::{KeyCode, KeyEvent, KeyModifiers, MouseButton, MouseEventKind}; use n00n_agent::permissions::PermissionManager; use n00n_agent::{ - ImageMediaType, McpConfigErrors, McpServerInfo, McpServerStatus, McpSnapshot, - McpSnapshotReader, ToolDoneEvent, ToolOutput, ToolStartEvent, TurnCompleteEvent, + ImageMediaType, InterruptSource, McpConfigErrors, McpPromptArg, McpServerInfo, McpServerStatus, + McpSnapshot, McpSnapshotReader, ToolDoneEvent, ToolOutput, ToolStartEvent, TurnCompleteEvent, }; use n00n_config::{PermissionsConfig, UiConfig}; use n00n_lua::{HintReader, KeymapReader, LuaCommandReader}; @@ -35,13 +35,21 @@ fn set_zone(app: &mut App, zone: SelectionZone, area: Rect) { } fn build_app(dir: StateDir, writer: Arc) -> App { + build_app_with_mcp(dir, writer, McpSnapshotReader::empty()) +} + +fn build_app_with_mcp( + dir: StateDir, + writer: Arc, + mcp_reader: McpSnapshotReader, +) -> App { let model = test_model(); App::new( &model, AppSession::new("test-model", "/tmp/test"), dir, Arc::new(ArcSwapOption::empty()), - McpSnapshotReader::empty(), + mcp_reader, McpConfigErrors::new(PathBuf::new()), LuaCommandReader::empty(), KeymapReader::empty(), @@ -176,8 +184,9 @@ fn typing_and_submit() { app.update(Msg::Key(key(KeyCode::Char('i')))); let actions = app.update(Msg::Key(key(KeyCode::Enter))); - assert!(matches!(&actions[0], Action::SendMessage(s) if s.message == "hi")); + assert!(matches!(&actions[0], Action::SendMessage(s) if s.input.message == "hi")); assert_eq!(app.status, Status::Streaming); + assert!(app.input_box.is_empty()); // Regression check: the bubble has to be on screen the same frame we // submit, otherwise it briefly sits one row too high before snapping down. assert_eq!( @@ -187,6 +196,280 @@ fn typing_and_submit() { assert_eq!(app.main_chat().last_message_text(), "hi"); } +#[derive(Clone)] +struct ManualSubmissionClock(Arc>); + +impl SubmissionClock for ManualSubmissionClock { + fn now(&self) -> Instant { + *self.0.lock().unwrap() + } +} + +impl ManualSubmissionClock { + fn advance(&self, duration: Duration) { + let mut now = self.0.lock().unwrap(); + *now += duration; + } +} + +fn install_manual_submission_clock(app: &mut App) -> ManualSubmissionClock { + let clock = ManualSubmissionClock(Arc::new(std::sync::Mutex::new(Instant::now()))); + app.submission_clock = Arc::new(clock.clone()); + clock +} + +#[test] +fn rapid_submissions_keep_fifo_while_first_waits_for_persistence() { + let mut app = test_app(); + let (shared, receiver) = shared_queue::queue(); + app.queue.set_shared(shared); + + let first = type_and_submit(&mut app, "first"); + let Action::SendMessage(first_dispatch) = first.into_iter().next().unwrap() else { + panic!("expected first submission dispatch"); + }; + app.handle_submit(Submission { + text: "second".into(), + images: Vec::new(), + }); + + assert!( + receiver + .poll(n00n_agent::InterruptPoint::ToolComplete) + .is_none() + ); + + let first_id = first_dispatch.submission_id; + assert!( + app.queue + .mark_submission_ready(first_id, first_dispatch.input) + ); + let crate::agent::shared_queue::QueueItem::Message { input, .. } = receiver.pop().unwrap() + else { + panic!("expected first queued submission"); + }; + assert_eq!(input.message, "first"); + let n00n_agent::ExtractedCommand::Interrupt(second, _) = receiver + .poll(n00n_agent::InterruptPoint::ToolComplete) + .unwrap() + else { + panic!("expected second queued submission"); + }; + assert_eq!(second.message, "second"); +} + +#[test] +fn session_api_prompt_is_explicitly_non_paint_gated() { + let mut app = test_app(); + let outcome = app.submit_background_prompt(crate::app::queue::QueuedMessage { + text: "background prompt".into(), + images: Vec::new(), + }); + let SubmitOutcome::Started(actions) = outcome else { + panic!("expected background prompt to start"); + }; + let Action::SendMessage(dispatch) = &actions[0] else { + panic!("expected submission dispatch"); + }; + + assert!(!dispatch.paint_required); + assert_eq!(app.main_chat().message_count(), 0); +} + +#[test] +fn background_persistence_failure_is_terminal_without_composer_restore() { + let mut app = test_app(); + let SubmitOutcome::Started(actions) = app.submit_background_prompt(QueuedMessage { + text: "background prompt".into(), + images: Vec::new(), + }) else { + panic!("expected background prompt to start"); + }; + let Action::SendMessage(dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected background submission dispatch"); + }; + let submission_id = dispatch.submission_id; + let gate = Arc::clone(&dispatch.gate); + assert!(matches!( + app.submit_background_prompt(QueuedMessage { + text: "queued after failure".into(), + images: Vec::new(), + }), + SubmitOutcome::Queued + )); + + app.handle_submission_persistence_failure(&dispatch); + + assert!(gate.is_cancelled()); + assert!(matches!( + &app.status, + Status::Error { message, .. } if message == PERSISTENCE_FAILURE_MSG + )); + assert_eq!(app.main_chat().last_message_text(), PERSISTENCE_FAILURE_MSG); + assert!(app.input_box.is_empty()); + assert_eq!(app.queue.text_messages(), vec!["queued after failure"]); + assert_eq!(submission_id, dispatch.submission_id); +} + +#[test] +fn escape_before_dispatch_restores_text_and_exact_images_once() { + let mut app = test_app(); + install_manual_submission_clock(&mut app); + app.input_box.set_input("describe".into()); + with_image(&mut app); + + let actions = app.update(Msg::Key(key(KeyCode::Enter))); + let Action::SendMessage(dispatch) = &actions[0] else { + panic!("expected submission dispatch"); + }; + let gate = Arc::clone(&dispatch.gate); + assert_eq!(app.main_chat().message_count(), 1); + + let cancel_actions = app.update(Msg::Key(key(KeyCode::Esc))); + + assert!(cancel_actions.is_empty()); + assert!(gate.is_cancelled()); + assert_eq!(app.status, Status::Idle); + assert_eq!(app.main_chat().message_count(), 0); + let restored = app.input_box.submit().unwrap(); + assert_eq!(restored.text, "describe"); + assert_eq!(restored.images.len(), 1); + assert_eq!(restored.images[0].media_type, ImageMediaType::Png); + assert_eq!(&*restored.images[0].data, "dGVzdA=="); + + app.handle_submission_persistence_failure(dispatch); + assert_eq!(app.main_chat().message_count(), 0); + assert!(app.input_box.is_empty()); +} + +#[test] +fn committed_submission_keeps_double_escape_cancellation() { + let mut app = test_app(); + install_manual_submission_clock(&mut app); + let actions = type_and_submit(&mut app, "sent"); + let Action::SendMessage(dispatch) = &actions[0] else { + panic!("expected submission dispatch"); + }; + assert!(dispatch.gate.try_commit()); + + let first = app.update(Msg::Key(key(KeyCode::Esc))); + assert!(first.is_empty()); + assert!(app.input_box.is_empty()); + assert_eq!(app.main_chat().last_message_text(), "sent"); + + let second = app.update(Msg::Key(key(KeyCode::Esc))); + assert!(matches!(&second[..], [Action::CancelAgent { .. }])); +} + +#[test] +fn shell_preamble_restores_once_on_cancel_and_resubmits_once() { + let mut app = test_app(); + install_manual_submission_clock(&mut app); + app.shell.push_result(Message::user("shell output".into())); + + let actions = type_and_submit(&mut app, "use context"); + let Action::SendMessage(mut dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected submission dispatch"); + }; + assert!(app.stage_submission_preamble(&mut dispatch)); + assert_eq!(dispatch.input.preamble.len(), 1); + assert!(app.shell.drain_results().is_empty()); + + app.update(Msg::Key(key(KeyCode::Esc))); + let restored_preamble = app.shell.drain_results(); + assert_eq!(restored_preamble.len(), 1); + assert_eq!(restored_preamble[0].user_text(), Some("shell output")); + app.shell.restore_results(restored_preamble); + let submission = app.input_box.submit().expect("cancel restores submission"); + let actions = app.handle_submit(submission); + let Action::SendMessage(mut dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected resubmission dispatch"); + }; + assert!(app.stage_submission_preamble(&mut dispatch)); + assert_eq!(dispatch.input.preamble.len(), 1); + assert_eq!(dispatch.input.preamble[0].user_text(), Some("shell output")); + assert!(app.shell.drain_results().is_empty()); + + app.handle_submission_persistence_failure(&dispatch); + let restored_after_failure = app.shell.drain_results(); + assert_eq!(restored_after_failure.len(), 1); + assert_eq!(restored_after_failure[0].user_text(), Some("shell output")); +} +#[test] +fn escape_during_mcp_error_restores_and_resubmits_exactly_once() { + let mut app = test_app(); + install_manual_submission_clock(&mut app); + app.shell.push_result(Message::user("mcp context".into())); + + let actions = type_and_submit(&mut app, "review"); + let Action::SendMessage(mut dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected submission dispatch"); + }; + assert!(app.stage_submission_preamble(&mut dispatch)); + app.update(agent_msg(AgentEvent::Error { + message: "MCP prompt failed".into(), + })); + + app.update(Msg::Key(key(KeyCode::Esc))); + + assert_eq!(app.main_chat().message_count(), 0); + assert_eq!(app.input_box.buffer.value(), "review"); + let restored = app.shell.drain_results(); + assert_eq!(restored.len(), 1); + app.shell.restore_results(restored); + + let submission = app.input_box.submit().expect("restored submission"); + let actions = app.handle_submit(submission); + assert_eq!(actions.len(), 1); + let Action::SendMessage(mut retry) = actions.into_iter().next().unwrap() else { + panic!("expected retry dispatch"); + }; + assert!(app.stage_submission_preamble(&mut retry)); + assert_eq!(retry.input.preamble.len(), 1); + assert!(app.shell.drain_results().is_empty()); +} + +#[test] +fn expired_escape_window_does_not_restore_without_sleeping() { + let mut app = test_app(); + let clock = install_manual_submission_clock(&mut app); + type_and_submit(&mut app, "too late"); + clock.advance(SUBMISSION_ESCAPE_WINDOW + Duration::from_millis(1)); + + let actions = app.update(Msg::Key(key(KeyCode::Esc))); + + assert!(actions.is_empty()); + assert!(app.input_box.is_empty()); + assert_eq!(app.main_chat().last_message_text(), "too late"); + assert_eq!(app.status, Status::Streaming); +} + +#[test] +fn escape_at_submission_window_boundary_restores() { + let mut app = test_app(); + let clock = install_manual_submission_clock(&mut app); + type_and_submit(&mut app, "at boundary"); + clock.advance(SUBMISSION_ESCAPE_WINDOW); + + app.update(Msg::Key(key(KeyCode::Esc))); + + assert_eq!(app.status, Status::Idle); + assert_eq!(app.input_box.buffer.value(), "at boundary"); +} + +#[test] +fn escape_one_millisecond_after_submission_window_does_not_restore() { + let mut app = test_app(); + let clock = install_manual_submission_clock(&mut app); + type_and_submit(&mut app, "after boundary"); + clock.advance(SUBMISSION_ESCAPE_WINDOW + Duration::from_millis(1)); + + app.update(Msg::Key(key(KeyCode::Esc))); + + assert_eq!(app.status, Status::Streaming); + assert!(app.input_box.is_empty()); +} + fn with_text(app: &mut App) { app.update(Msg::Key(key(KeyCode::Char('h')))); app.update(Msg::Key(key(KeyCode::Char('i')))); @@ -1855,7 +2138,7 @@ fn stale_events_ignored_after_run_id_increment() { cancel_app(&mut app); let current_run = app.run_id; let actions = type_and_submit(&mut app, "new prompt"); - assert!(matches!(&actions[0], Action::SendMessage(i) if i.message == "new prompt")); + assert!(matches!(&actions[0], Action::SendMessage(i) if i.input.message == "new prompt")); let active_run = app.run_id; app.update(agent_msg_with_run_id( @@ -2104,6 +2387,216 @@ fn reload_persists_session_with_content_to_disk() { assert_eq!(AppSession::load(id, &dir).unwrap().messages.len(), 1); } +#[test] +fn rewind_snapshot_survives_crash_restore() { + let (_tmp, dir, writer, mut app) = tempdir_app(); + app.state.session.messages = vec![ + Message::user("first".into()), + Message { + role: Role::Assistant, + content: vec![ContentBlock::Text { + text: "reply".into(), + }], + ..Default::default() + }, + Message::user("second".into()), + ]; + let id = app.state.session.id; + let before = app.state.session.meta.revision; + app.rewind_to(crate::components::rewind_picker::RewindEntry { + turn_index: 2, + prompt_preview: "2: second".into(), + prompt_text: "second".into(), + }); + assert!(app.state.session.meta.revision > before); + drain_writer(app, writer); + + let restored = AppSession::load(id, &dir).unwrap(); + assert_eq!(restored.messages.len(), 2); + assert_eq!(restored.meta.input_draft.as_deref(), Some("second")); +} + +#[test] +fn loaded_metadata_consumption_survives_crash_restore() { + let (_tmp, dir, writer, mut app) = tempdir_app(); + let mut seed = AppSession::new("test-model", "/tmp/test"); + seed.meta.input_draft = Some("draft before crash".into()); + seed.meta.queued_messages = vec!["queued before crash".into()]; + seed.messages.push(Message::user("history".into())); + let id = seed.id; + seed.save(&dir).unwrap(); + + let (shared, _receiver) = shared_queue::queue(); + app.queue.set_shared(shared); + let previous_revision = seed.meta.revision; + app.apply_loaded_session(seed, &test_model()); + assert!(app.state.session.meta.revision > previous_revision); + drain_writer(app, writer); + + let restored = AppSession::load(id, &dir).unwrap(); + assert_eq!( + restored.meta.input_draft.as_deref(), + Some("draft before crash") + ); + assert_eq!(restored.meta.queued_messages, vec!["queued before crash"]); +} + +#[test] +fn draw_failure_pending_submission_restores_fifo_and_images_after_restart() { + let (_tmp, dir, writer, mut app) = tempdir_app(); + let (shared, receiver) = shared_queue::queue(); + app.queue.set_shared(shared); + + app.input_box.set_input("first with image".into()); + with_image(&mut app); + let actions = app.update(Msg::Key(key(KeyCode::Enter))); + let Action::SendMessage(dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected paint-gated submission"); + }; + assert!( + receiver.pop().is_none(), + "paint gate must block provider dispatch" + ); + + assert!(matches!( + app.submit_prompt(QueuedMessage { + text: "second in fifo".into(), + images: Vec::new(), + }), + SubmitOutcome::Queued + )); + app.preserve_submission_for_shutdown(*dispatch); + app.save_session(); + let session_id = app.state.session.id; + drop(app); + Arc::try_unwrap(writer) + .ok() + .expect("test owns the storage writer") + .shutdown(WRITER_DRAIN_TIMEOUT); + + let writer = Arc::new(StorageWriter::new(dir.clone())); + let mut restarted = build_app(dir.clone(), Arc::clone(&writer)); + let (shared, receiver) = shared_queue::queue(); + restarted.queue.set_shared(shared); + restarted.apply_loaded_session( + AppSession::load(session_id, &dir).expect("saved session loads"), + &test_model(), + ); + + let Some(shared_queue::QueueItem::Message { text, input, .. }) = receiver.pop() else { + panic!("first pending message must be restored"); + }; + assert_eq!(text, "first with image"); + assert_eq!(input.images.len(), 1); + assert_eq!(&*input.images[0].data, "dGVzdA=="); + let Some(shared_queue::QueueItem::Message { text, .. }) = receiver.pop() else { + panic!("second queued message must be restored"); + }; + assert_eq!(text, "second in fifo"); + assert!(receiver.pop().is_none()); + + drop(restarted); + Arc::try_unwrap(writer) + .ok() + .expect("test owns the restarted storage writer") + .shutdown(WRITER_DRAIN_TIMEOUT); +} + +#[test] +fn mcp_prompt_draw_failure_survives_restart_without_text_fallback() { + let (_tmp, dir, writer, mut app) = tempdir_app(); + let mcp_reader = McpSnapshotReader::from_snapshot(McpSnapshot { + prompts: vec![n00n_agent::McpPromptInfo { + display_name: "myserver:review".into(), + qualified_name: "myserver/review".into(), + description: "Review a change".into(), + arguments: vec![McpPromptArg { + name: "diff".into(), + description: "The diff".into(), + required: true, + }], + }], + ..Default::default() + }); + app.command_palette = crate::components::command::CommandPalette::new( + Arc::from([]), + mcp_reader.clone(), + LuaCommandReader::empty(), + ); + let (shared, _receiver) = shared_queue::queue(); + app.queue.set_shared(shared); + app.state.thinking = n00n_providers::ThinkingConfig::Effort(Effort::High); + app.state.fast = true; + app.state.workflow = true; + let actions = app.execute_command(ParsedCommand { + name: "/myserver:review".into(), + args: "important diff".into(), + }); + let Action::SendMessage(dispatch) = actions.into_iter().next().unwrap() else { + panic!("expected MCP prompt dispatch"); + }; + assert_eq!(dispatch.input.message, "/myserver:review important diff"); + let prompt = dispatch + .input + .prompt + .as_ref() + .expect("MCP prompt reference"); + assert_eq!(prompt.qualified_name, "myserver/review"); + assert_eq!( + prompt.arguments.get("diff").map(String::as_str), + Some("important diff") + ); + assert_eq!( + dispatch.input.thinking, + n00n_providers::ThinkingConfig::Effort(Effort::High) + ); + assert!(dispatch.input.fast); + assert!(dispatch.input.workflow); + + app.preserve_submission_for_shutdown(*dispatch); + app.save_session(); + let session_id = app.state.session.id; + drop(app); + Arc::try_unwrap(writer) + .ok() + .expect("test owns the storage writer") + .shutdown(WRITER_DRAIN_TIMEOUT); + + let writer = Arc::new(StorageWriter::new(dir.clone())); + let mut restarted = build_app_with_mcp(dir.clone(), Arc::clone(&writer), mcp_reader); + let (shared, receiver) = shared_queue::queue(); + restarted.queue.set_shared(shared); + restarted.apply_loaded_session( + AppSession::load(session_id, &dir).expect("saved session loads"), + &test_model(), + ); + + let Some(shared_queue::QueueItem::Message { input, .. }) = receiver.pop() else { + panic!("MCP prompt must be restored"); + }; + let prompt = input + .prompt + .expect("restored MCP prompt must use get_prompt"); + assert_eq!(prompt.qualified_name, "myserver/review"); + assert_eq!( + prompt.arguments.get("diff").map(String::as_str), + Some("important diff") + ); + assert_eq!( + input.thinking, + n00n_providers::ThinkingConfig::Effort(Effort::High) + ); + assert!(input.fast); + assert!(input.workflow); + assert_eq!(input.message, "/myserver:review important diff"); + + drop(restarted); + Arc::try_unwrap(writer) + .ok() + .expect("test owns the restarted storage writer") + .shutdown(WRITER_DRAIN_TIMEOUT); +} + #[test] fn reload_leaves_empty_session_unpersisted_on_disk() { let (tmp, _dir, writer, mut app) = tempdir_app(); @@ -3040,7 +3533,7 @@ fn plan_form_menu_options( assert_eq!( actions .iter() - .any(|a| matches!(a, Action::SendMessage(i) if i.message == expected_msg)), + .any(|a| matches!(a, Action::SendMessage(i) if i.input.message == expected_msg)), has_send_message ); } @@ -3056,7 +3549,7 @@ fn plan_form_implement_toggled_parallel() { assert!( actions .iter() - .any(|a| matches!(a, Action::SendMessage(i) if i.message == expected_msg)) + .any(|a| matches!(a, Action::SendMessage(i) if i.input.message == expected_msg)) ); } diff --git a/n00n-ui/src/chat.rs b/n00n-ui/src/chat.rs index 5b845220c..7a715feb0 100644 --- a/n00n-ui/src/chat.rs +++ b/n00n-ui/src/chat.rs @@ -458,12 +458,15 @@ impl Chat { self.messages_panel.tool_done(event); } - #[cfg(test)] #[must_use] pub fn message_count(&self) -> usize { self.messages_panel.message_count() } + pub fn truncate_messages(&mut self, len: usize) { + self.messages_panel.truncate_messages(len); + } + #[cfg(test)] #[must_use] pub fn in_progress_count(&self) -> usize { diff --git a/n00n-ui/src/components/messages/mod.rs b/n00n-ui/src/components/messages/mod.rs index 413d7cc0f..2bf407b0d 100644 --- a/n00n-ui/src/components/messages/mod.rs +++ b/n00n-ui/src/components/messages/mod.rs @@ -200,6 +200,18 @@ impl MessagesPanel { self.messages.push(msg); } + pub fn message_count(&self) -> usize { + self.messages.len() + } + + pub fn truncate_messages(&mut self, len: usize) { + self.messages.truncate(len); + self.streaming_thinking.clear(); + self.streaming_text.clear(); + self.prompt_progress = None; + self.cache.invalidate_from_msg_count(); + } + pub fn set_pending_compaction_summary(&mut self, summary: String) { if let Some(message) = self.messages.iter_mut().rev().find(|message| { matches!( @@ -689,11 +701,6 @@ impl MessagesPanel { } } - #[cfg(test)] - pub fn message_count(&self) -> usize { - self.messages.len() - } - #[cfg(test)] pub fn last_message_text(&self) -> &str { self.messages.last().map_or("", |m| m.text.as_str()) diff --git a/n00n-ui/src/components/mod.rs b/n00n-ui/src/components/mod.rs index a7f0f931a..05e0cc058 100644 --- a/n00n-ui/src/components/mod.rs +++ b/n00n-ui/src/components/mod.rs @@ -32,7 +32,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use crossterm::event::{KeyCode, KeyEvent, KeyModifiers}; -use n00n_agent::AgentInput; +use n00n_agent::{AgentInput, PreDispatchGate}; use n00n_agent::{BufferSnapshot, ImageSource, ToolInput, ToolOutput}; use n00n_providers::{Message, ModelTier}; use n00n_storage::sessions::TranscriptEntry; @@ -181,8 +181,18 @@ pub struct LoadedSession { use std::path::PathBuf; +pub struct SubmissionDispatch { + pub input: AgentInput, + pub submission_id: u64, + pub run_id: u64, + pub gate: Arc, + /// User-composer submissions wait for the exact session bubble to paint. + /// Background API submissions explicitly skip that UI-only gate. + pub paint_required: bool, +} + pub enum Action { - SendMessage(Box), + SendMessage(Box), ShellCommand { id: String, command: String, diff --git a/n00n-ui/src/event_loop.rs b/n00n-ui/src/event_loop.rs index 9840fb737..daaa390bf 100644 --- a/n00n-ui/src/event_loop.rs +++ b/n00n-ui/src/event_loop.rs @@ -40,7 +40,9 @@ use crate::app::shell::{ShellEvent, spawn_shell}; use crate::app::{App, Msg, QueuedMessage, SubmitOutcome}; use crate::components::input::Submission; use crate::components::usage_modal::UsageFetchState; -use crate::components::{Action, DisplayMessage, DisplayRole, ExitRequest, Status}; +use crate::components::{ + Action, DisplayMessage, DisplayRole, ExitRequest, Status, SubmissionDispatch, +}; use crate::input::InputReader; use crate::storage_writer::StorageWriter; @@ -156,7 +158,7 @@ struct SpawnCtx { impl SpawnCtx { fn spawn_runtime(&self, session: AppSession) -> SessionRuntime { - let resumed = !session.messages.is_empty(); + let resumed = crate::app::session_has_content(&session); let permissions = Arc::new(self.permissions.fork()); let handles = AgentHandles::spawn( &self.model_slot, @@ -213,6 +215,9 @@ pub(crate) struct EventLoop<'t> { warn_rx: flume::Receiver, warn_tx: flume::Sender, ui_action_rx: Option>, + submission_persist_tx: flume::Sender, + submission_persist_rx: flume::Receiver, + post_draw_submissions: Vec<(N00nId, SubmissionDispatch)>, last_save: Instant, _model_fetch_task: smol::Task<()>, /// Set when UI state changed and a fresh frame must be painted. Draws are @@ -223,12 +228,19 @@ pub(crate) struct EventLoop<'t> { /// One item from any of the event loop's sources; `None` from `next_wake` /// means the wait timed out (animation/idle tick). +struct SubmissionPersistence { + session_id: N00nId, + dispatch: SubmissionDispatch, + result: Result<(), SessionError>, +} + enum Wake { Input(Event), InputGone, Ui(UiAction), Agent(usize, Box), Shell(usize, ShellEvent), + SubmissionPersisted(SubmissionPersistence), Warn(String), } @@ -436,6 +448,7 @@ impl<'t> EventLoop<'t> { app.flash(w); } + let (submission_persist_tx, submission_persist_rx) = flume::unbounded(); Ok(Self { terminal, sessions: runtimes, @@ -445,6 +458,9 @@ impl<'t> EventLoop<'t> { warn_rx: bg.warn_rx, warn_tx: bg.warn_tx, ui_action_rx, + submission_persist_tx, + submission_persist_rx, + post_draw_submissions: Vec::new(), last_save: Instant::now(), _model_fetch_task: bg.task, dirty: true, @@ -469,13 +485,16 @@ impl<'t> EventLoop<'t> { if let Err(e) = self.drain_channels() { break Err(e); } - let should_draw = self.dirty || self.sessions[self.focused].app.is_animating(); + let should_draw = self.dirty + || self.sessions[self.focused].app.is_animating() + || !self.post_draw_submissions.is_empty(); let app = &mut self.sessions[self.focused].app; if should_draw { - if let Err(e) = self.terminal.draw(|f| app.view(f)) { + if let Err(e) = draw_then_post_terminal(self.terminal, |f| app.view(f), || {}) { break Err(e.into()); } self.dirty = false; + self.after_terminal_draw(); } if let Some(i) = self @@ -530,6 +549,9 @@ impl<'t> EventLoop<'t> { sel = sel.recv(rx, |res| res.ok().map(Wake::Ui)); } sel = sel.recv(&self.warn_rx, |res| res.ok().map(Wake::Warn)); + sel = sel.recv(&self.submission_persist_rx, |res| { + res.ok().map(Wake::SubmissionPersisted) + }); for (i, rt) in self.sessions.iter().enumerate() { if !rt.handles.agent_rx.is_disconnected() { sel = sel.recv(&rt.handles.agent_rx, move |res| { @@ -553,6 +575,9 @@ impl<'t> EventLoop<'t> { sel = sel.recv(rx, |res| res.ok().map(Wake::Ui)); } sel = sel.recv(&self.warn_rx, |res| res.ok().map(Wake::Warn)); + sel = sel.recv(&self.submission_persist_rx, |res| { + res.ok().map(Wake::SubmissionPersisted) + }); for (i, rt) in self.sessions.iter().enumerate() { if !rt.handles.agent_rx.is_disconnected() { sel = sel.recv(&rt.handles.agent_rx, move |res| { @@ -574,6 +599,7 @@ impl<'t> EventLoop<'t> { Wake::Ui(action) => self.handle_ui_action(action), Wake::Agent(i, envelope) => self.handle_agent(i, envelope), Wake::Shell(i, event) => self.sessions[i].app.handle_shell_event(event), + Wake::SubmissionPersisted(completion) => self.handle_submission_persisted(completion), Wake::Warn(warning) => self.focused_app().flash(warning), } Ok(()) @@ -880,7 +906,7 @@ impl<'t> EventLoop<'t> { text, images: Vec::new(), }; - match self.sessions[idx].app.submit_prompt(msg) { + match self.sessions[idx].app.submit_background_prompt(msg) { SubmitOutcome::Started(actions) => { self.dispatch(idx, actions); Ok(json!("started")) @@ -1034,7 +1060,27 @@ impl<'t> EventLoop<'t> { fn dispatch(&mut self, idx: usize, actions: Vec) { for action in actions { - self.handle_action(idx, action); + match action { + Action::SendMessage(dispatch) if dispatch.paint_required => { + self.post_draw_submissions + .push((self.sessions[idx].id(), *dispatch)); + } + Action::SendMessage(dispatch) => { + self.handle_action(idx, Action::SendMessage(dispatch)); + } + action => self.handle_action(idx, action), + } + } + } + + /// The optimistic user bubble must have completed a terminal draw before + /// persistence or provider dispatch can begin. + fn after_terminal_draw(&mut self) { + let painted_session = self.sessions[self.focused].id(); + for (_, dispatch) in + take_painted_submissions(&mut self.post_draw_submissions, painted_session) + { + self.handle_action(self.focused, Action::SendMessage(Box::new(dispatch))); } } @@ -1059,36 +1105,52 @@ impl<'t> EventLoop<'t> { ); } + fn handle_submission_persisted(&mut self, completion: SubmissionPersistence) { + let Some(idx) = self.position(completion.session_id) else { + return; + }; + let rt = &mut self.sessions[idx]; + if completion.result.is_err() { + rt.app + .handle_submission_persistence_failure(&completion.dispatch); + return; + } + if !rt.app.accepts_submission_persistence(&completion.dispatch) { + rt.app + .queue + .remove_submission(completion.dispatch.submission_id); + return; + } + let submission_id = completion.dispatch.submission_id; + if !rt + .app + .queue + .mark_submission_ready(submission_id, completion.dispatch.input) + { + rt.app.queue.remove_submission(submission_id); + } + } + fn handle_action(&mut self, idx: usize, action: Action) { match action { - Action::SendMessage(input) => { + Action::SendMessage(mut dispatch) => { let rt = &mut self.sessions[idx]; - match n00n_storage::sessions::openai_response_chain_parent_exists( - &rt.app.storage, - rt.app.state.session.id, - ) { - Ok(false) => { - let mut initial_session = rt.app.state.session.clone(); - if let Err(error) = initial_session.save(&rt.app.storage) { - warn!(%error, "failed to persist session before provider request"); - } - } - Err(error) => { - warn!(%error, "failed to check session persistence before provider request"); - } - Ok(true) => {} + if !rt.app.stage_submission_preamble(&mut dispatch) { + rt.app.queue.remove_submission(dispatch.submission_id); + return; } - let mut input = *input; - input.preamble = rt.app.shell.drain_results(); - let run_id = rt.app.run_id; - rt.handles.queue.push(QueueItem::Message { - text: input.message.clone(), - image_count: input.images.len(), - input, - run_id, - displayed: true, - delivery: crate::agent::shared_queue::Delivery::TurnEnd, - }); + let session_id = rt.app.state.session.id; + let snapshot = rt.app.session_snapshot(); + let completion_tx = self.submission_persist_tx.clone(); + self.ctx + .storage_writer + .persist(Box::new(snapshot), move |result| { + let _ = completion_tx.send(SubmissionPersistence { + session_id, + dispatch: *dispatch, + result, + }); + }); } Action::CancelAgent { run_id } => { let _ = self.sessions[idx] @@ -1264,7 +1326,23 @@ impl<'t> EventLoop<'t> { } } + fn preserve_post_draw_submissions(&mut self) { + for (session_id, dispatch) in std::mem::take(&mut self.post_draw_submissions) + .into_iter() + .rev() + { + let Some(idx) = self.position(session_id) else { + warn!(%session_id, "paint-gated submission lost its session before shutdown"); + continue; + }; + self.sessions[idx] + .app + .preserve_submission_for_shutdown(dispatch); + } + } + fn shutdown(mut self) -> ShutdownReport { + self.preserve_post_draw_submissions(); let exit = self.sessions[self.focused].app.exit_request; if let Some(ref h) = self.ctx.mcp_handle { mcp::kill_process_groups(&h.reader().load().pids); @@ -1302,6 +1380,35 @@ impl<'t> EventLoop<'t> { } } +fn draw_then_post_terminal( + terminal: &mut ratatui::Terminal, + draw: impl FnOnce(&mut ratatui::Frame<'_>), + after_draw: impl FnOnce(), +) -> Result<(), B::Error> +where + B: ratatui::backend::Backend, +{ + terminal.draw(draw)?; + after_draw(); + Ok(()) +} + +fn take_painted_submissions( + pending: &mut Vec<(N00nId, T)>, + painted_session: N00nId, +) -> Vec<(N00nId, T)> { + let submissions = std::mem::take(pending); + let mut ready = Vec::new(); + for (session_id, submission) in submissions { + if session_id == painted_session { + ready.push((session_id, submission)); + } else { + pending.push((session_id, submission)); + } + } + ready +} + fn scroll_delta(kind: MouseEventKind, lines: u32) -> i32 { if kind == MouseEventKind::ScrollUp { lines as i32 @@ -1312,7 +1419,71 @@ fn scroll_delta(kind: MouseEventKind, lines: u32) -> i32 { #[cfg(test)] mod tests { - use super::{DRAIN_BUDGET, DrainScheduler}; + use super::{DRAIN_BUDGET, DrainScheduler, draw_then_post_terminal, take_painted_submissions}; + use n00n_storage::id::N00nId; + use ratatui::{ + Terminal, + backend::{Backend, ClearType, TestBackend, WindowSize}, + buffer::Cell, + layout::{Position, Size}, + widgets::Paragraph, + }; + use std::io; + + struct FailingBackend(TestBackend); + + fn infallible(result: Result) -> io::Result { + match result { + Ok(value) => Ok(value), + Err(error) => match error {}, + } + } + + impl Backend for FailingBackend { + type Error = io::Error; + fn draw<'a, I>(&mut self, _content: I) -> io::Result<()> + where + I: Iterator, + { + Err(io::Error::other("deterministic draw failure")) + } + + fn hide_cursor(&mut self) -> io::Result<()> { + infallible(self.0.hide_cursor()) + } + + fn show_cursor(&mut self) -> io::Result<()> { + infallible(self.0.show_cursor()) + } + + fn get_cursor_position(&mut self) -> io::Result { + infallible(self.0.get_cursor_position()) + } + + fn set_cursor_position>(&mut self, position: P) -> io::Result<()> { + infallible(self.0.set_cursor_position(position)) + } + + fn clear(&mut self) -> io::Result<()> { + infallible(self.0.clear()) + } + + fn clear_region(&mut self, clear_type: ClearType) -> io::Result<()> { + infallible(self.0.clear_region(clear_type)) + } + + fn size(&self) -> io::Result { + infallible(self.0.size()) + } + + fn window_size(&mut self) -> io::Result { + infallible(self.0.window_size()) + } + + fn flush(&mut self) -> io::Result<()> { + infallible(self.0.flush()) + } + } #[derive(Debug, PartialEq, Eq)] enum Source { @@ -1320,6 +1491,64 @@ mod tests { Agent(usize), } + #[test] + fn painted_submission_waits_for_its_session_after_focus_switch() { + let first = N00nId::generate(); + let second = N00nId::generate(); + let mut pending = vec![(first, "first"), (second, "second")]; + + let released = take_painted_submissions(&mut pending, second); + + assert_eq!(released, vec![(second, "second")]); + assert_eq!(pending, vec![(first, "first")]); + assert_eq!( + take_painted_submissions(&mut pending, first), + vec![(first, "first")] + ); + } + + #[test] + fn post_draw_hook_runs_after_terminal_buffer_is_painted() { + let mut terminal = Terminal::new(TestBackend::new(20, 1)).expect("test terminal"); + let painted = std::cell::Cell::new(false); + let persistence_started = std::cell::Cell::new(false); + + draw_then_post_terminal( + &mut terminal, + |frame| { + frame.render_widget(Paragraph::new("bubble"), frame.area()); + painted.set(true); + }, + || { + persistence_started.set(true); + assert!(painted.get()); + }, + ) + .expect("draw succeeds"); + + assert!(persistence_started.get()); + assert_eq!( + terminal.backend().buffer().cell((0, 0)).unwrap().symbol(), + "b" + ); + } + + #[test] + fn terminal_draw_failure_does_not_release_post_draw_work() { + let mut terminal = + Terminal::new(FailingBackend(TestBackend::new(20, 1))).expect("test terminal"); + let post_draw_ran = std::cell::Cell::new(false); + + let result = draw_then_post_terminal( + &mut terminal, + |frame| frame.render_widget(Paragraph::new("bubble"), frame.area()), + || post_draw_ran.set(true), + ); + + assert!(result.is_err()); + assert!(!post_draw_ran.get()); + } + #[test] fn drain_prioritizes_input_and_preserves_fair_bounded_progress() { let (input_tx, input_rx) = flume::unbounded(); diff --git a/n00n-ui/src/storage_writer.rs b/n00n-ui/src/storage_writer.rs index bfd7d1de2..1d4298f02 100644 --- a/n00n-ui/src/storage_writer.rs +++ b/n00n-ui/src/storage_writer.rs @@ -19,15 +19,28 @@ use tracing::warn; use crate::AppSession; -type Pending = Arc>>>; +struct PendingSnapshot { + revision: u64, + session: Box, +} + +type Pending = Arc>>; type DeleteCallback = Box) + Send>; +type PersistCallback = Box) + Send>; const RETRY_DELAY: Duration = Duration::from_secs(1); enum Op { Flush, - Delete { id: N00nId, done: DeleteCallback }, + Persist { + session: Box, + done: PersistCallback, + }, + Delete { + id: N00nId, + done: DeleteCallback, + }, } pub struct StorageWriter { @@ -47,6 +60,7 @@ impl StorageWriter { .name("storage-writer".into()) .spawn(move || { let mut logs: HashMap = HashMap::new(); + let mut durable_revisions: HashMap = HashMap::new(); let mut retry_pending = false; loop { let op = if retry_pending { @@ -60,7 +74,21 @@ impl StorageWriter { }; let Some(op) = op else { break }; match op { - Op::Flush => retry_pending = flush(&writer_pending, &mut logs, &dir), + Op::Flush => { + retry_pending = + flush(&writer_pending, &mut logs, &mut durable_revisions, &dir); + } + Op::Persist { session, done } => { + retry_pending = + flush(&writer_pending, &mut logs, &mut durable_revisions, &dir); + done(persist_session( + &writer_pending, + &mut logs, + &mut durable_revisions, + &dir, + &session, + )); + } Op::Delete { id, done } => { lock(&writer_pending).remove(&id); logs.remove(&id); @@ -68,7 +96,7 @@ impl StorageWriter { } } } - flush(&writer_pending, &mut logs, &dir); + flush(&writer_pending, &mut logs, &mut durable_revisions, &dir); let _ = done_tx.send(()); }) .expect("failed to spawn storage writer thread"); @@ -83,13 +111,34 @@ impl StorageWriter { pub fn send(&self, session: Box) { let mut pending = lock(&self.pending); let was_empty = pending.is_empty(); - pending.insert(session.id, session); + let revision = session.meta.revision; + let replace = pending + .get(&session.id) + .is_none_or(|current| current.revision <= revision); + if replace { + pending.insert(session.id, PendingSnapshot { revision, session }); + } drop(pending); if was_empty { let _ = self.ops.send(Op::Flush); } } + /// Persists this snapshot before invoking `done` on the writer thread. + pub fn persist( + &self, + session: Box, + done: impl FnOnce(Result<(), SessionError>) + Send + 'static, + ) { + let op = Op::Persist { + session, + done: Box::new(done), + }; + if let Err(flume::SendError(Op::Persist { done, .. })) = self.ops.send(op) { + done(Err(writer_gone())); + } + } + /// Delete a session's files on the writer thread, discarding any pending /// snapshot first. Runs after already-queued flushes; `done` fires on the /// writer thread, so callers never block on disk. @@ -111,7 +160,7 @@ impl StorageWriter { } } -fn lock(pending: &Pending) -> std::sync::MutexGuard<'_, HashMap>> { +fn lock(pending: &Pending) -> std::sync::MutexGuard<'_, HashMap> { pending .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -121,8 +170,14 @@ fn writer_gone() -> SessionError { n00n_storage::StorageError::Io(io::Error::other("storage writer unavailable")).into() } -fn flush(pending: &Pending, logs: &mut HashMap, dir: &StateDir) -> bool { - let batch = mem::take(&mut *lock(pending)); +fn flush( + pending: &Pending, + logs: &mut HashMap, + durable_revisions: &mut HashMap, + dir: &StateDir, +) -> bool { + let mut pending_guard = lock(pending); + let batch = mem::take(&mut *pending_guard); if batch.is_empty() { return false; } @@ -130,75 +185,127 @@ fn flush(pending: &Pending, logs: &mut HashMap, dir: &StateD Ok(d) => d, Err(e) => { warn!(error = %e, "failed to ensure sessions dir"); - let mut pending = lock(pending); - for (id, session) in batch { - pending.entry(id).or_insert(session); + for (id, snapshot) in batch { + let replace = pending_guard + .get(&id) + .is_none_or(|current| current.revision < snapshot.revision); + if replace { + pending_guard.insert(id, snapshot); + } } return true; } }; - for session in batch.into_values() { - write_session(&sessions_dir, logs, &session); + for snapshot in batch.into_values() { + if let Err(error) = write_session(&sessions_dir, logs, durable_revisions, &snapshot.session) + { + warn!(error = %error, id = %snapshot.session.id, "session write failed"); + } } false } +fn persist_session( + pending: &Pending, + logs: &mut HashMap, + durable_revisions: &mut HashMap, + dir: &StateDir, + session: &AppSession, +) -> Result<(), SessionError> { + let mut pending_guard = lock(pending); + if pending_guard + .get(&session.id) + .is_some_and(|snapshot| snapshot.revision > session.meta.revision) + { + let snapshot = pending_guard.remove(&session.id).ok_or_else(|| { + n00n_storage::StorageError::Io(io::Error::other("pending snapshot disappeared")) + })?; + write_session( + &dir.ensure_subdir(SESSIONS_DIR)?, + logs, + durable_revisions, + &snapshot.session, + )?; + } + drop(pending_guard); + write_session_if_newer(logs, durable_revisions, dir, session) +} + +fn append_or_compact_result( + log: &mut SessionLog, + sessions_dir: &Path, + session: &AppSession, +) -> Result<(), SessionError> { + match log.append(session) { + Ok(()) => Ok(()), + Err(SessionError::CursorAhead { .. }) => log.compact(sessions_dir, session), + Err(error) => Err(error), + } +} fn write_session( sessions_dir: &Path, logs: &mut HashMap, + durable_revisions: &mut HashMap, session: &AppSession, -) { +) -> Result<(), SessionError> { + if durable_revisions + .get(&session.id) + .is_some_and(|revision| *revision > session.meta.revision) + { + return Ok(()); + } if let Some(log) = logs.get_mut(&session.id) { - if !append_or_compact(log, sessions_dir, session) { - logs.remove(&session.id); + if durable_revisions.get(&session.id) == Some(&session.meta.revision) { + log.compact(sessions_dir, session)?; + } else { + append_or_compact_result(log, sessions_dir, session)?; } - return; + durable_revisions.insert(session.id, session.meta.revision); + return Ok(()); } - let mut log = match open_or_create_log(sessions_dir, session) { - Ok(l) => l, - Err(e) => { - warn!(error = %e, id = %session.id, "session log open failed"); - return; - } - }; - if append_or_compact(&mut log, sessions_dir, session) { - logs.insert(session.id, log); + let (mut log, on_disk_revision) = open_or_create_log(sessions_dir, session)?; + if on_disk_revision > session.meta.revision { + durable_revisions.insert(session.id, on_disk_revision); + return Ok(()); + } + if on_disk_revision == session.meta.revision { + log.compact(sessions_dir, session)?; + } else { + append_or_compact_result(&mut log, sessions_dir, session)?; } + logs.insert(session.id, log); + durable_revisions.insert(session.id, session.meta.revision); + Ok(()) } -/// False means the log's cursors are unusable and it must not stay cached. -fn append_or_compact(log: &mut SessionLog, sessions_dir: &Path, session: &AppSession) -> bool { - match log.append(session) { - Ok(()) => true, - Err(SessionError::CursorAhead { .. }) => match log.compact(sessions_dir, session) { - Ok(()) => true, - Err(e) => { - warn!(error = %e, id = %session.id, "compact fallback failed"); - false - } - }, - Err(e) => { - warn!(error = %e, id = %session.id, "append failed"); - true - } - } +fn write_session_if_newer( + logs: &mut HashMap, + durable_revisions: &mut HashMap, + dir: &StateDir, + session: &AppSession, +) -> Result<(), SessionError> { + let sessions_dir = dir.ensure_subdir(SESSIONS_DIR)?; + write_session(&sessions_dir, logs, durable_revisions, session) } fn open_or_create_log( sessions_dir: &Path, session: &AppSession, -) -> Result { +) -> Result<(SessionLog, u64), n00n_storage::sessions::SessionError> { let jsonl_path = sessions_dir.join(format!("{}.jsonl", session.id)); if jsonl_path.exists() { let id = session.id; - let (_loaded, log) = SessionLog::open::< + let (loaded, log) = SessionLog::open::< n00n_providers::Message, n00n_providers::TokenUsage, n00n_agent::ToolOutput, >(sessions_dir, id)?; - Ok(log) + Ok((log, loaded.meta.revision)) } else { - AppSession::migrate_to_jsonl(sessions_dir, session) + Ok(( + AppSession::migrate_to_jsonl(sessions_dir, session)?, + session.meta.revision, + )) } } @@ -234,6 +341,68 @@ mod tests { assert_eq!(AppSession::load(b_id, &dir).unwrap().title, "renamed"); } + #[test] + fn persist_reports_success_after_writing_snapshot() { + let (_tmp, dir) = state_dir(); + let writer = StorageWriter::new(dir.clone()); + let session = AppSession::new("test-model", "/tmp/persist"); + let id = session.id; + let (done_tx, done_rx) = flume::bounded(1); + writer.persist(Box::new(session), move |result| { + let _ = done_tx.send(result); + }); + + assert!(done_rx.recv_timeout(DRAIN_TIMEOUT).unwrap().is_ok()); + assert!(AppSession::load(id, &dir).is_ok()); + writer.shutdown(DRAIN_TIMEOUT); + } + + #[test] + fn equal_revision_different_snapshot_is_not_dropped() { + let (_tmp, dir) = state_dir(); + let writer = StorageWriter::new(dir.clone()); + let first = AppSession::new("test-model", "/tmp/equal"); + let id = first.id; + let (done_tx, done_rx) = flume::bounded(1); + writer.persist(Box::new(first), move |result| { + let _ = done_tx.send(result); + }); + assert!(done_rx.recv_timeout(DRAIN_TIMEOUT).unwrap().is_ok()); + + let mut second = AppSession::load(id, &dir).unwrap(); + second.title = "same revision, new snapshot".into(); + writer.send(Box::new(second)); + writer.shutdown(DRAIN_TIMEOUT); + + assert_eq!( + AppSession::load(id, &dir).unwrap().title, + "same revision, new snapshot" + ); + } + + #[test] + fn persist_cannot_overwrite_newer_periodic_snapshot() { + let (_tmp, dir) = state_dir(); + let writer = StorageWriter::new(dir.clone()); + let mut older = AppSession::new("test-model", "/tmp/race"); + older.meta.revision = 1; + older.title = "submission".into(); + let id = older.id; + let mut newer = older.clone(); + newer.meta.revision = 2; + newer.title = "periodic save".into(); + writer.send(Box::new(newer)); + + let (done_tx, done_rx) = flume::bounded(1); + writer.persist(Box::new(older), move |result| { + let _ = done_tx.send(result); + }); + + assert!(done_rx.recv_timeout(DRAIN_TIMEOUT).unwrap().is_ok()); + assert_eq!(AppSession::load(id, &dir).unwrap().title, "periodic save"); + writer.shutdown(DRAIN_TIMEOUT); + } + #[test] fn delete_discards_pending_snapshot() { let (_tmp, dir) = state_dir(); From 45d49f12460730ff23dbdeb2afbde8a52f9980de Mon Sep 17 00:00:00 2001 From: w0wl0lxd Date: Wed, 22 Jul 2026 05:32:39 -0400 Subject: [PATCH 2/2] Merge origin/main into feat/submission-escape-hatch Resolve merge conflicts by integrating main's OpenAiOptions refactoring with the PR's safe submission escape hatch logic. Changes: - n00n-ui/src/agent/agent_loop.rs: Add openai_options field and parameter, pass to resolve_compaction_model and AgentParams - n00n-ui/src/event_loop.rs: Add openai_options to EventLoopParams and SpawnCtx, use from_model_with_openai_options throughout, fix API call signatures (transition_plan, pick_model_for_lua, spawn_shell, start_btw) - Preserve PR's pre_dispatch_gate and stage_submission_preamble logic --- n00n-ui/src/agent/agent_loop.rs | 14 ++++++-- n00n-ui/src/event_loop.rs | 62 ++++++++++++++++++++++++--------- 2 files changed, 57 insertions(+), 19 deletions(-) diff --git a/n00n-ui/src/agent/agent_loop.rs b/n00n-ui/src/agent/agent_loop.rs index 852e5ee97..789d59b3f 100644 --- a/n00n-ui/src/agent/agent_loop.rs +++ b/n00n-ui/src/agent/agent_loop.rs @@ -16,7 +16,7 @@ use n00n_agent::{ PromptRole, ToolOutputLines, }; use n00n_lua::EventHandle; -use n00n_providers::{AgentError, Message, Model, TokenUsage}; +use n00n_providers::{AgentError, Message, Model, OpenAiOptions, TokenUsage}; use n00n_storage::id::SessionRef; use n00n_storage::sessions::TranscriptEntry; use serde_json::Value; @@ -46,6 +46,7 @@ pub(super) struct AgentLoop { queue: Arc, session_id: Option, timeouts: n00n_providers::Timeouts, + openai_options: OpenAiOptions, lua_handle: Option, subagent_cancels: Arc>, tools_cache: Option, @@ -81,6 +82,7 @@ impl AgentLoop { init_cancel: CancelToken, session_id: Option, timeouts: n00n_providers::Timeouts, + openai_options: OpenAiOptions, lua_handle: Option, subagent_cancels: Arc>, ) -> Self { @@ -106,6 +108,7 @@ impl AgentLoop { queue, session_id, timeouts, + openai_options, lua_handle, subagent_cancels, tools_cache: None, @@ -176,8 +179,12 @@ impl AgentLoop { async fn do_compact(&mut self, event_tx: &EventSender) -> Result<(), AgentError> { let slot = self.model_slot.load(); - let (provider, model) = - agent::resolve_compaction_model(&slot.provider, &slot.model, self.timeouts); + let (provider, model) = agent::resolve_compaction_model( + &slot.provider, + &slot.model, + self.timeouts, + self.openai_options, + ); agent::compact(&*provider, &model, &mut self.history, event_tx).await } @@ -280,6 +287,7 @@ impl AgentLoop { subagent_cancels: Arc::clone(&self.subagent_cancels), registry: Arc::clone(n00n_agent::tools::ToolRegistry::global_arc()), audience: ToolAudience::MAIN, + openai_options: self.openai_options, }, AgentRunParams { history: &mut self.history, diff --git a/n00n-ui/src/event_loop.rs b/n00n-ui/src/event_loop.rs index daaa390bf..a3fcd5d2c 100644 --- a/n00n-ui/src/event_loop.rs +++ b/n00n-ui/src/event_loop.rs @@ -25,8 +25,11 @@ use n00n_lua::{ EventHandle, HintReader, KeymapReader, LuaCommandReader, SessionReply, SessionRequest, UiAction, }; use n00n_providers::Timeouts; -use n00n_providers::provider::{Provider, fetch_all_models, from_model}; -use n00n_providers::{Message, Model}; +use n00n_providers::provider::{ + Provider, fetch_all_models, from_model_fallback_with_openai_options, + from_model_with_openai_options, +}; +use n00n_providers::{Message, Model, OpenAiOptions}; use n00n_storage::StateDir; use n00n_storage::StorageError; use n00n_storage::id::{N00nId, N00nIdParseError, SessionRef}; @@ -80,6 +83,7 @@ pub struct EventLoopParams { pub input_history_size: usize, pub permissions: Arc, pub timeouts: Timeouts, + pub openai_options: OpenAiOptions, pub exit_on_done: bool, pub lua_command_reader: LuaCommandReader, pub keymap_reader: KeymapReader, @@ -143,6 +147,7 @@ struct SpawnCtx { /// rules stay per-session. permissions: Arc, timeouts: Timeouts, + openai_options: OpenAiOptions, custom_commands: Arc<[CustomCommand]>, lua_command_reader: LuaCommandReader, keymap_reader: KeymapReader, @@ -169,6 +174,7 @@ impl SpawnCtx { &permissions, Some(SessionRef::from(session.id)), self.timeouts, + self.openai_options, self.lua_event_handle.clone(), self.mcp_handle.clone(), self.mcp_config_errors.clone(), @@ -301,7 +307,11 @@ fn merge_batch( available.store(Some(Arc::new(merged))); } -fn spawn_model_fetch(model_slot: &Arc>, timeouts: Timeouts) -> BackgroundModels { +fn spawn_model_fetch( + model_slot: &Arc>, + timeouts: Timeouts, + openai_options: OpenAiOptions, +) -> BackgroundModels { let available: Arc>> = Arc::new(ArcSwapOption::empty()); let bg = Arc::clone(&available); let (warn_tx, warn_rx) = flume::unbounded::(); @@ -318,7 +328,11 @@ fn spawn_model_fetch(model_slot: &Arc>, timeouts: Timeouts) - return; } }; - let provider = match from_model(&mut resolved, timeouts) { + let provider = match from_model_with_openai_options( + &mut resolved, + timeouts, + openai_options, + ) { Ok(p) => p, Err(e) => { warn!(spec = %spec, error = %e, "failed to create provider after discovery"); @@ -374,6 +388,7 @@ impl<'t> EventLoop<'t> { input_history_size, permissions, timeouts, + openai_options, exit_on_done, lua_command_reader, keymap_reader, @@ -388,22 +403,27 @@ impl<'t> EventLoop<'t> { crate::update::spawn_check(); }); - let storage_writer = Arc::new(StorageWriter::new(storage.clone())); + let storage_writer = Arc::new(StorageWriter::new(storage.clone())?); let cwd = std::env::current_dir().unwrap_or_else(|_| ".".into()); let (mcp_handle, mcp_config_errors) = smol::block_on(mcp::start(&cwd)); let provider: Arc = if needs_login { - Arc::from(n00n_providers::provider::from_model_fallback( - &mut model, timeouts, + Arc::from(from_model_fallback_with_openai_options( + &mut model, + timeouts, + openai_options, )) } else { - Arc::from(from_model(&mut model, timeouts).context("create provider")?) + Arc::from( + from_model_with_openai_options(&mut model, timeouts, openai_options) + .context("create provider")?, + ) }; let model_slot = Arc::new(ArcSwap::from_pointee(ModelSlot { model: model.clone(), provider, })); - let bg = spawn_model_fetch(&model_slot, timeouts); + let bg = spawn_model_fetch(&model_slot, timeouts, openai_options); let picker = Arc::new(terminal_image::picker()); @@ -414,6 +434,7 @@ impl<'t> EventLoop<'t> { input_history_size, permissions, timeouts, + openai_options, custom_commands: Arc::from(commands), lua_command_reader, keymap_reader, @@ -454,7 +475,7 @@ impl<'t> EventLoop<'t> { sessions: runtimes, focused, ctx, - input: InputReader::spawn(), + input: InputReader::spawn()?, warn_rx: bg.warn_rx, warn_tx: bg.warn_tx, ui_action_rx, @@ -690,11 +711,12 @@ impl<'t> EventLoop<'t> { let app = self.focused_app(); app.float_mgr.open(buf, config, focus, event_tx, cmd_rx); if focus { - app.transition_plan(crate::app::mode::PlanTrigger::InteractivePrompt); + app.transition_plan(&crate::app::mode::PlanTrigger::InteractivePrompt); } } UiAction::PickModel { current, reply_tx } => { - self.focused_app().pick_model_for_lua(current, reply_tx); + self.focused_app() + .pick_model_for_lua(current.as_deref(), reply_tx); self.handle_action(self.focused, Action::RefreshModels); } UiAction::Session { req, reply_tx } => { @@ -1171,7 +1193,11 @@ impl<'t> EventLoop<'t> { let loaded = *loaded; if loaded.model_spec != self.ctx.model_slot.load().model.spec() && let Ok(mut new_model) = Model::from_spec(&loaded.model_spec) - && let Ok(new_provider) = from_model(&mut new_model, self.ctx.timeouts) + && let Ok(new_provider) = from_model_with_openai_options( + &mut new_model, + self.ctx.timeouts, + self.ctx.openai_options, + ) { self.sessions[idx].app.usage_slot.store(None); self.ctx.model_slot.store(Arc::new(ModelSlot { @@ -1219,7 +1245,7 @@ impl<'t> EventLoop<'t> { visible, rt.shell_tx.clone(), cancel, - self.ctx.config.clone(), + &self.ctx.config, ); } Action::OpenEditor(path) => { @@ -1239,7 +1265,7 @@ impl<'t> EventLoop<'t> { Action::Btw(question) => { let slot = self.ctx.model_slot.load(); self.sessions[idx].app.start_btw( - question, + &question, Arc::clone(&slot.provider), slot.model.clone(), ); @@ -1255,7 +1281,11 @@ impl<'t> EventLoop<'t> { fn change_model(&mut self, spec: String) { match Model::from_spec(&spec) { - Ok(mut new_model) => match from_model(&mut new_model, self.ctx.timeouts) { + Ok(mut new_model) => match from_model_with_openai_options( + &mut new_model, + self.ctx.timeouts, + self.ctx.openai_options, + ) { Ok(new_provider) => { let app = self.focused_app(); app.update_model(&new_model);