fix: filter empty provider messages, fix apply_patch hunk headers, persist latest-query usage separately, and fix TUI queued/resume state tracking

crates/cli/src/prompt_command.rs

@@ -93,11 +93,11 @@ pub(crate) async fn run_prompt(
             turn_id: None,
             cwd: cwd.clone(),
             agent_scope: devo_core::tools::ToolAgentScope::Parent,
-            agent_context_mode: devo_protocol::AgentContextMode::CodingAgent,
             collaboration_mode: devo_protocol::CollaborationMode::Build,
             agent_coordinator: None,
             client_filesystem: None,
             client_terminal: None,
+            file_read_ledger: std::sync::Arc::new(devo_core::tools::FileReadLedger::new()),
             local_web_search: None,
             hooks: (!app_config.hooks.is_empty()).then(|| devo_core::HookRuntimeContext {
                 runner: devo_core::HookRunner::new(app_config.hooks.clone()),

crates/client/src/client_core.rs

@@ -46,7 +46,7 @@ pub const ACP_PROMPT_COMPLETED_NOTIFICATION_METHOD: &str = "_devo/acp_prompt/com
 const SERVER_RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
 
 /// Synthetic notifications emitted when falling back to detached `session/prompt`.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct ServerNotificationMessage {
     pub method: String,
     pub params: serde_json::Value,
@@ -412,6 +412,13 @@ impl ServerClientCore {
         self.notifications_rx.recv().await
     }
 
+    pub(crate) async fn recv_client_event(&mut self) -> Result<Option<crate::ClientEvent>> {
+        let Some(notification) = self.recv_notification().await else {
+            return Ok(None);
+        };
+        crate::events::client_event_from_notification(&notification)
+    }
+
     pub(crate) async fn recv_event(&mut self) -> Result<Option<(String, ServerEvent)>> {
         let Some(notification) = self.recv_notification().await else {
             return Ok(None);

crates/client/src/events.rs

@@ -0,0 +1,151 @@
+use anyhow::{Context, Result};
+use devo_protocol::{
+    ACP_SESSION_UPDATE_METHOD, AcpSessionNotification, AcpSessionUpdate, DEVO_TURN_USAGE_META,
+    ServerEvent, SessionCompactionFailedPayload, SessionEventPayload, TurnEventPayload,
+    TurnUsageUpdatedPayload,
+};
+
+use crate::client_core::ServerNotificationMessage;
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum ClientEvent {
+    TurnStarted(TurnEventPayload),
+    TurnCompleted(TurnEventPayload),
+    TurnFailed(TurnEventPayload),
+    TurnUsageUpdated(TurnUsageUpdatedPayload),
+    SessionCompactionStarted(SessionEventPayload),
+    SessionCompactionCompleted(SessionEventPayload),
+    SessionCompactionFailed(SessionCompactionFailedPayload),
+    Other(ServerNotificationMessage),
+}
+
+pub fn client_event_from_notification(
+    notification: &ServerNotificationMessage,
+) -> Result<Option<ClientEvent>> {
+    if notification.method == ACP_SESSION_UPDATE_METHOD {
+        let acp_notification: AcpSessionNotification =
+            match serde_json::from_value(notification.params.clone()) {
+                Ok(notification) => notification,
+                Err(_) => return Ok(Some(ClientEvent::Other(notification.clone()))),
+            };
+        if let AcpSessionUpdate::UsageUpdate { meta, .. } = acp_notification.update
+            && let Some(payload) = meta
+                .as_ref()
+                .and_then(|meta| meta.get(DEVO_TURN_USAGE_META))
+        {
+            return Ok(Some(ClientEvent::TurnUsageUpdated(
+                serde_json::from_value(payload.clone()).context("decode Devo usage metadata")?,
+            )));
+        }
+        return Ok(None);
+    }
+
+    let event: ServerEvent = match serde_json::from_value(notification.params.clone()) {
+        Ok(event) => event,
+        Err(_) => return Ok(Some(ClientEvent::Other(notification.clone()))),
+    };
+    Ok(Some(match event {
+        ServerEvent::TurnStarted(payload) => ClientEvent::TurnStarted(payload),
+        ServerEvent::TurnCompleted(payload) => ClientEvent::TurnCompleted(payload),
+        ServerEvent::TurnFailed(payload) => ClientEvent::TurnFailed(payload),
+        ServerEvent::TurnUsageUpdated(payload) => ClientEvent::TurnUsageUpdated(payload),
+        ServerEvent::SessionCompactionStarted(payload) => {
+            ClientEvent::SessionCompactionStarted(payload)
+        }
+        ServerEvent::SessionCompactionCompleted(payload) => {
+            ClientEvent::SessionCompactionCompleted(payload)
+        }
+        ServerEvent::SessionCompactionFailed(payload) => {
+            ClientEvent::SessionCompactionFailed(payload)
+        }
+        _ => ClientEvent::Other(notification.clone()),
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use devo_protocol::{SessionId, TurnId, TurnUsage, TurnUsageUpdatedPayload};
+
+    use super::ClientEvent;
+    use super::client_event_from_notification;
+    use crate::ServerNotificationMessage;
+
+    #[test]
+    fn client_event_usage_normalizes_devo_notification() {
+        let session_id = SessionId::new();
+        let turn_id = TurnId::new();
+        let payload = TurnUsageUpdatedPayload {
+            session_id,
+            turn_id,
+            usage: TurnUsage {
+                input_tokens: 700,
+                output_tokens: 30,
+                cache_creation_input_tokens: None,
+                cache_read_input_tokens: None,
+                reasoning_output_tokens: None,
+                total_tokens: Some(730),
+            },
+            total_input_tokens: 1_300,
+            total_output_tokens: 80,
+            total_tokens: 1_380,
+            total_cache_read_tokens: 0,
+            last_query_input_tokens: 700,
+            context_window: Some(200_000),
+        };
+        let notification = ServerNotificationMessage {
+            method: "turn/usage/updated".to_string(),
+            params: serde_json::to_value(devo_protocol::ServerEvent::TurnUsageUpdated(
+                payload.clone(),
+            ))
+            .expect("serialize usage event"),
+        };
+
+        assert_eq!(
+            client_event_from_notification(&notification).expect("decode client event"),
+            Some(ClientEvent::TurnUsageUpdated(payload))
+        );
+    }
+
+    #[test]
+    fn client_event_usage_normalizes_acp_metadata() {
+        let session_id = SessionId::new();
+        let turn_id = TurnId::new();
+        let payload = TurnUsageUpdatedPayload {
+            session_id,
+            turn_id,
+            usage: TurnUsage {
+                input_tokens: 700,
+                output_tokens: 30,
+                cache_creation_input_tokens: None,
+                cache_read_input_tokens: None,
+                reasoning_output_tokens: None,
+                total_tokens: Some(730),
+            },
+            total_input_tokens: 1_300,
+            total_output_tokens: 80,
+            total_tokens: 1_380,
+            total_cache_read_tokens: 0,
+            last_query_input_tokens: 700,
+            context_window: Some(200_000),
+        };
+        let notification = ServerNotificationMessage {
+            method: devo_protocol::ACP_SESSION_UPDATE_METHOD.to_string(),
+            params: serde_json::json!({
+                "sessionId": session_id,
+                "update": {
+                    "sessionUpdate": "usage_update",
+                    "used": 1_380,
+                    "size": 200_000,
+                    "_meta": {
+                        "devo/turnUsage": payload,
+                    },
+                },
+            }),
+        };
+
+        assert_eq!(
+            client_event_from_notification(&notification).expect("decode client event"),
+            Some(ClientEvent::TurnUsageUpdated(payload))
+        );
+    }
+}

crates/client/src/lib.rs

@@ -8,12 +8,14 @@ mod acp_fs;
 mod acp_permissions;
 mod acp_terminal;
 mod client_core;
+mod events;
 mod protocol_trace;
 mod stdio;
 mod websocket;
 
 pub use client_core::ACP_PROMPT_COMPLETED_NOTIFICATION_METHOD;
 pub use client_core::ACP_PROMPT_STARTED_NOTIFICATION_METHOD;
 pub use client_core::ServerNotificationMessage;
+pub use events::{ClientEvent, client_event_from_notification};
 pub use stdio::*;
 pub use websocket::*;

crates/client/src/stdio.rs

@@ -33,7 +33,6 @@ pub use crate::client_core::ServerNotificationMessage;
 
 const SERVER_CHILD_STDIN_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
 const SERVER_CHILD_EXIT_TIMEOUT: Duration = Duration::from_millis(500);
-const STDIO_INITIALIZE_TIMEOUT: Duration = Duration::from_secs(3);
 
 #[derive(Debug, Clone)]
 pub struct StdioServerClientConfig {
@@ -123,9 +122,7 @@ impl StdioServerClient {
         tracing::info!("initializing stdio server client");
         self.core
             .set_client_capabilities(client_capabilities.clone());
-        let result = timeout(STDIO_INITIALIZE_TIMEOUT, self.core.initialize())
-            .await
-            .context("timed out waiting for initialize response from server")??;
+        let result = self.core.initialize().await?;
         tracing::info!("stdio server client initialized");
         Ok(result)
     }
@@ -373,6 +370,10 @@ impl StdioServerClient {
         self.core.recv_notification().await
     }
 
+    pub async fn recv_client_event(&mut self) -> Result<Option<crate::ClientEvent>> {
+        self.core.recv_client_event().await
+    }
+
     pub async fn recv_event(&mut self) -> Result<Option<(String, ServerEvent)>> {
         self.core.recv_event().await
     }
@@ -898,7 +899,7 @@ mod tests {
             approval_policy: Some("on-request".to_string()),
             cwd: Some(PathBuf::from("workspace")),
             collaboration_mode: devo_protocol::CollaborationMode::Plan,
-            execution_mode: devo_protocol::TurnExecutionMode::Research,
+            execution_mode: devo_protocol::TurnExecutionMode::Regular,
         };
         let expected_params = serde_json::to_value(&params).expect("serialize turn params");
         let mut stdout_lines = BufReader::new(stdout).lines();

crates/client/src/websocket.rs

@@ -351,6 +351,10 @@ impl WebSocketServerClient {
         self.core.recv_notification().await
     }
 
+    pub async fn recv_client_event(&mut self) -> Result<Option<crate::ClientEvent>> {
+        self.core.recv_client_event().await
+    }
+
     pub async fn recv_event(&mut self) -> Result<Option<(String, ServerEvent)>> {
         self.core.recv_event().await
     }

crates/code-search/src/lib.rs

@@ -17,6 +17,7 @@ mod ranking;
 mod refresh;
 mod semantic;
 mod service;
+mod singleflight;
 mod tokens;
 mod types;
 mod watch;

crates/code-search/src/service.rs

@@ -153,11 +153,17 @@ impl CodeSearchService {
         })
     }
 
+    /// Prepares the default retrieval index without running a query.
+    pub fn prewarm(
+        &self,
+        root: &Path,
+        content: ContentFilter,
+    ) -> Result<IndexStats, CodeSearchError> {
+        let root = canonical_root(root)?;
+        self.index(root, content).map(|index| index.stats())
+    }
+
     /// Returns a warm or refreshed index for a root/content pair.
-    ///
-    /// The fast path is a clean watcher inside the safety interval. If that does
-    /// not hold, the service performs a manifest walk, checks memory, checks disk,
-    /// then asks `IndexRefresh` to reuse or re-embed at file granularity.
     fn index(
         &self,
         root: PathBuf,
@@ -168,34 +174,44 @@ impl CodeSearchService {
             return Ok(index);
         }
 
-        // A watcher-clean index is an optimization only. Once the watcher is
-        // dirty, unavailable, or beyond the safety interval, the manifest walk is
-        // the source of truth for reuse decisions.
-        let files = discover_files(&root, content)?;
-        if let Some(index) = self.matching_memory_index(&key, &root, &files)? {
+        let build_key = format!("{}|{key}", self.cache_dir.display());
+        let index =
+            crate::singleflight::run(build_key, || self.build_or_load_index(&key, &root, content))?;
+        self.store_index(key, &root, Arc::clone(&index))?;
+        Ok(index)
+    }
+
+    fn build_or_load_index(
+        &self,
+        key: &str,
+        root: &Path,
+        content: ContentFilter,
+    ) -> Result<Arc<SearchIndex>, CodeSearchError> {
+        let files = discover_files(root, content)?;
+        if let Some(index) = self.matching_memory_index(key, root, &files)? {
             return Ok(index);
         }
 
-        let cache_path = cache_file_path(&self.cache_dir, &root, content, self.provider.model_id());
+        let cache_path = cache_file_path(&self.cache_dir, root, content, self.provider.model_id());
         let previous_payload = load_payload(&cache_path).filter(|cache| {
             cache
                 .payload
-                .is_valid_for(&root, content, self.provider.model_id())
+                .is_valid_for(root, content, self.provider.model_id())
         });
         let outcome = IndexRefresh::refresh(
-            &root,
+            root,
             content,
             files,
             previous_payload,
             self.provider.as_ref(),
         )?;
         save_payload(&cache_path, &outcome.payload, &outcome.embeddings)?;
-        let index = Arc::new(SearchIndex::from_cached(crate::cache::CachedIndex {
-            payload: outcome.payload,
-            embeddings: outcome.embeddings,
-        })?);
-        self.store_index(key, &root, Arc::clone(&index))?;
-        Ok(index)
+        Ok(Arc::new(SearchIndex::from_cached(
+            crate::cache::CachedIndex {
+                payload: outcome.payload,
+                embeddings: outcome.embeddings,
+            },
+        )?))
     }
 
     /// Returns `true` when the next `search` or `find_related` call for this
@@ -821,15 +837,16 @@ mod tests {
             "should need build before first search"
         );
 
-        service
-            .search(SearchRequest {
-                root: temp.path().to_path_buf(),
-                query: "alpha".to_string(),
-                content: ContentFilter::Code,
-                top_k: 1,
-                filters: SearchFilters::empty(),
-            })
-            .expect("search");
+        let stats = service
+            .prewarm(temp.path(), ContentFilter::Code)
+            .expect("prewarm");
+        assert_eq!(
+            stats,
+            IndexStats {
+                indexed_files: 1,
+                total_chunks: 1,
+            }
+        );
 
         assert!(
             !service.needs_index_build(temp.path(), ContentFilter::Code),