fix: address CodeRabbit security and correctness findings

Redact sensitive command data, harden Unicode and timestamp rendering, and clamp popup geometry for small terminals.

Make editor handoff and JSON-RPC requests failure-safe, preserve event ordering, and fix state, arithmetic, diagnostics, and protocol robustness issues.

Author: zatevakhin

Cargo.lock

@@ -31,6 +31,7 @@ similar = "2"
 fuzzy-matcher = "0.3"
 syntect = { version = "5", default-features = false, features = ["default-syntaxes", "default-themes", "regex-onig"] }
 unicode-width = "0.2"
+chrono = { version = "0.4.45", default-features = false, features = ["std"] }
 toml = "1"
 dirs = "6"
 fs2 = "0.4"

Cargo.toml

@@ -10,6 +10,7 @@ pub(super) fn file_index<C: AcpConnection>(ctx: CommandContext<'_, C>) {
 
 pub(super) fn command<C: AcpConnection>(ctx: CommandContext<'_, C>, command: &Command) {
     ctx.events.error(format!(
-        "unsupported in the current ACP subset: {command:?}"
+        "unsupported in the current ACP subset: {}",
+        command.label()
     ));
 }

src/acp/commands/unsupported.rs

@@ -191,7 +191,7 @@ fn model_from_option(option: &Value, model_id: &str) -> Option<Model> {
             let (provider, model) = model_id
                 .split_once('/')
                 .map(|(provider, model)| (provider.to_string(), model.to_string()))
-                .unwrap_or_else(|| ("unknown".to_string(), label.clone()));
+                .unwrap_or_else(|| ("unknown".to_string(), model_id.to_string()));
             Model {
                 id: model_id.to_string(),
                 label,
@@ -236,6 +236,18 @@ mod tests {
         assert_eq!(entry["node_id"], "node-1");
     }
 
+    #[test]
+    fn model_option_without_provider_uses_identifier_not_friendly_label() {
+        let option = json!({
+            "options": [{ "value": "stable-id", "name": "Friendly Label" }]
+        });
+        let model = model_from_option(&option, "stable-id").expect("model");
+
+        assert_eq!(model.label, "Friendly Label");
+        assert_eq!(model.provider, "unknown");
+        assert_eq!(model.model, "stable-id");
+    }
+
     #[test]
     fn option_parser_reads_grouped_models_and_profiles() {
         let option = json!({

src/acp/configuration.rs

@@ -161,14 +161,34 @@ async fn router_keeps_noop_and_exact_unsupported_semantics() {
     let unsupported = Command::ClearApiToken {
         provider: "openai".into(),
     };
-    commands::dispatch(context(&connection, &state, &events), unsupported.clone())
+    commands::dispatch(context(&connection, &state, &events), unsupported)
         .await
         .expect("api token unsupported");
     assert!(matches!(
         rx.try_recv().expect("unsupported error"),
         ServerChannelMsg::Acp(AcpAppEvent::Error { message })
-            if message == format!("unsupported in the current ACP subset: {unsupported:?}")
+            if message == "unsupported in the current ACP subset: ClearApiToken"
     ));
+
+    let api_key = "sentinel-secret-api-key";
+    let unsupported = Command::SetApiToken {
+        provider: "openai".into(),
+        api_key: api_key.into(),
+    };
+    assert!(!format!("{unsupported:?}").contains(api_key));
+    commands::dispatch(context(&connection, &state, &events), unsupported)
+        .await
+        .expect("set api token unsupported");
+    let ServerChannelMsg::Acp(AcpAppEvent::Error { message }) =
+        rx.try_recv().expect("unsupported error")
+    else {
+        panic!("expected unsupported error");
+    };
+    assert_eq!(
+        message,
+        "unsupported in the current ACP subset: SetApiToken"
+    );
+    assert!(!message.contains(api_key));
     assert!(connection.messages().is_empty());
 }
 

src/acp/contract_tests.rs

@@ -17,7 +17,7 @@ pub(super) enum Translation {
         message_id: Option<String>,
         thinking: bool,
     },
-    ToolStart(AcpSessionUpdate),
+    ToolBoundary(AcpSessionUpdate),
     AgentMode(String),
     ConfigOptions(Vec<acp::SessionConfigOption>),
     Ignore,
@@ -43,9 +43,11 @@ pub(super) fn translate(notification: acp::SessionNotification) -> (String, Tran
             thinking: true,
         },
         acp::SessionUpdate::ToolCall(tool_call) => {
-            Translation::ToolStart(tool_start_update(&tool_call))
+            Translation::ToolBoundary(tool_start_update(&tool_call))
+        }
+        acp::SessionUpdate::ToolCallUpdate(update) => {
+            Translation::ToolBoundary(tool_call_update(update))
         }
-        acp::SessionUpdate::ToolCallUpdate(update) => Translation::Update(tool_call_update(update)),
         acp::SessionUpdate::CurrentModeUpdate(update) => {
             Translation::AgentMode(update.current_mode_id.to_string())
         }
@@ -69,7 +71,7 @@ pub(super) async fn apply(
 ) {
     match translation {
         Translation::Update(update) => emit_or_buffer(state, events, &session_id, update).await,
-        Translation::ToolStart(update) => match state.replay.route(&session_id, update).await {
+        Translation::ToolBoundary(update) => match state.replay.route(&session_id, update).await {
             None => {}
             Some(update) => {
                 flush_assistant(state, events, &session_id).await;
@@ -248,6 +250,8 @@ fn value_to_text(value: &Value) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::runtime_events::ServerChannelMsg;
+    use tokio::sync::mpsc;
 
     fn notification(update: acp::SessionUpdate) -> acp::SessionNotification {
         acp::SessionNotification::new("session-1", update)
@@ -322,7 +326,7 @@ mod tests {
         )));
         assert!(matches!(
             started,
-            Translation::ToolStart(AcpSessionUpdate::ToolCallStart {
+            Translation::ToolBoundary(AcpSessionUpdate::ToolCallStart {
                 tool_call_id: Some(id),
                 name,
                 arguments: Some(arguments),
@@ -339,7 +343,7 @@ mod tests {
         )));
         assert!(matches!(
             pending,
-            Translation::Update(AcpSessionUpdate::ToolCallStart {
+            Translation::ToolBoundary(AcpSessionUpdate::ToolCallStart {
                 tool_call_id: Some(id),
                 name,
                 arguments: Some(arguments),
@@ -365,7 +369,7 @@ mod tests {
         )));
         assert!(matches!(
             completed,
-            Translation::Update(AcpSessionUpdate::ToolCallEnd {
+            Translation::ToolBoundary(AcpSessionUpdate::ToolCallEnd {
                 tool_call_id: Some(id),
                 name,
                 is_error: false,
@@ -388,7 +392,7 @@ mod tests {
                     ]),
             ),
         )));
-        let Translation::Update(AcpSessionUpdate::ToolCallEnd {
+        let Translation::ToolBoundary(AcpSessionUpdate::ToolCallEnd {
             tool_call_id: Some(id),
             name,
             is_error,
@@ -466,6 +470,126 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn live_tool_updates_flush_assistant_content_at_each_boundary() {
+        let state = Arc::new(RuntimeState::new(None));
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let events = EventSink::new(tx);
+
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::AssistantChunk {
+                text: "before start".into(),
+                message_id: Some("a1".into()),
+                thinking: false,
+            },
+        )
+        .await;
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::ToolBoundary(AcpSessionUpdate::ToolCallStart {
+                tool_call_id: Some("tool-1".into()),
+                name: "shell".into(),
+                arguments: None,
+            }),
+        )
+        .await;
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::AssistantChunk {
+                text: "before end".into(),
+                message_id: Some("a2".into()),
+                thinking: false,
+            },
+        )
+        .await;
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::ToolBoundary(AcpSessionUpdate::ToolCallEnd {
+                tool_call_id: Some("tool-1".into()),
+                name: "shell".into(),
+                is_error: false,
+                result: None,
+            }),
+        )
+        .await;
+
+        let updates = (0..6)
+            .map(|_| match rx.try_recv().expect("session update") {
+                ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate { update, .. }) => update,
+                other => panic!("unexpected event: {other:?}"),
+            })
+            .collect::<Vec<_>>();
+        assert!(matches!(
+            &updates[..],
+            [
+                AcpSessionUpdate::AssistantContentDelta { content: first_delta, .. },
+                AcpSessionUpdate::AssistantMessage { content: first_final, .. },
+                AcpSessionUpdate::ToolCallStart { .. },
+                AcpSessionUpdate::AssistantContentDelta { content: second_delta, .. },
+                AcpSessionUpdate::AssistantMessage { content: second_final, .. },
+                AcpSessionUpdate::ToolCallEnd { .. },
+            ] if first_delta == "before start"
+                && first_final == "before start"
+                && second_delta == "before end"
+                && second_final == "before end"
+        ));
+    }
+
+    #[tokio::test]
+    async fn usage_updates_do_not_flush_assistant_content() {
+        let state = Arc::new(RuntimeState::new(None));
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let events = EventSink::new(tx);
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::AssistantChunk {
+                text: "buffered".into(),
+                message_id: None,
+                thinking: false,
+            },
+        )
+        .await;
+        apply(
+            &state,
+            &events,
+            "session-1".into(),
+            Translation::Update(AcpSessionUpdate::UsageUpdate {
+                used: 1,
+                size: 2,
+                cost_usd: None,
+            }),
+        )
+        .await;
+
+        assert!(matches!(
+            rx.try_recv(),
+            Ok(ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate {
+                update: AcpSessionUpdate::AssistantContentDelta { .. },
+                ..
+            }))
+        ));
+        assert!(matches!(
+            rx.try_recv(),
+            Ok(ServerChannelMsg::Acp(AcpAppEvent::SessionUpdate {
+                update: AcpSessionUpdate::UsageUpdate { .. },
+                ..
+            }))
+        ));
+        assert!(rx.try_recv().is_err());
+        assert!(state.assistants.flush("session-1").await.is_some());
+    }
+
     #[test]
     fn usage_translation_keeps_only_usd_cost() {
         for (cost, expected) in [

src/acp/notification.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicI64, Ordering};
+use std::time::Duration;
 
 use agent_client_protocol as acp_sdk;
 use serde::{Deserialize, Serialize};
@@ -10,6 +11,8 @@ use tokio_tungstenite::tungstenite::Message;
 
 use super::super::connection::internal_error;
 
+const RESPONSE_TIMEOUT: Duration = Duration::from_secs(60);
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub(in crate::acp) struct Envelope {
     pub(in crate::acp) jsonrpc: String,
@@ -97,6 +100,16 @@ impl Peer {
         &self,
         method: &str,
         params: Value,
+    ) -> Result<Value, acp_sdk::Error> {
+        self.request_with_timeout(method, params, RESPONSE_TIMEOUT)
+            .await
+    }
+
+    async fn request_with_timeout(
+        &self,
+        method: &str,
+        params: Value,
+        timeout: Duration,
     ) -> Result<Value, acp_sdk::Error> {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (tx, rx) = oneshot::channel();
@@ -111,8 +124,16 @@ impl Peer {
             self.pending.lock().await.remove(&id);
             return Err(err);
         }
-        rx.await
-            .map_err(|_| internal_error(format!("ACP WebSocket request dropped: {method}")))?
+        match tokio::time::timeout(timeout, rx).await {
+            Ok(result) => result
+                .map_err(|_| internal_error(format!("ACP WebSocket request dropped: {method}")))?,
+            Err(_) => {
+                self.pending.lock().await.remove(&id);
+                Err(internal_error(format!(
+                    "ACP WebSocket request timed out: {method}"
+                )))
+            }
+        }
     }
 
     pub(in crate::acp) fn notify(&self, method: &str, params: Value) -> Result<(), acp_sdk::Error> {
@@ -339,6 +360,41 @@ mod tests {
         assert_eq!(peer.pending_len().await, 0);
     }
 
+    #[tokio::test]
+    async fn timeout_removes_pending_request_and_ignores_late_response() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let peer = Peer::new(tx);
+        let request_peer = peer.clone();
+        let task = tokio::spawn(async move {
+            request_peer
+                .request_with_timeout("querymt/slow", json!({}), Duration::from_millis(10))
+                .await
+        });
+        let Message::Text(text) = rx.recv().await.expect("request frame") else {
+            panic!("expected text frame");
+        };
+        let wire: Envelope = serde_json::from_str(&text).expect("request envelope");
+
+        let error = task
+            .await
+            .expect("request task")
+            .expect_err("timeout error")
+            .to_string();
+        assert!(error.contains("timed out"));
+        assert!(error.contains("querymt/slow"));
+        assert_eq!(peer.pending_len().await, 0);
+
+        peer.resolve(Envelope {
+            method: None,
+            params: Value::Null,
+            result: Some(json!("late")),
+            error: None,
+            ..wire
+        })
+        .await;
+        assert_eq!(peer.pending_len().await, 0);
+    }
+
     #[tokio::test]
     async fn fail_all_drains_each_pending_request_once_and_late_responses_are_ignored() {
         let (tx, mut rx) = mpsc::unbounded_channel();