diff --git a/Cargo.lock b/Cargo.lock index 6b4c6950b3f34b..d865984b4bf6ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -323,9 +323,9 @@ dependencies = [ [[package]] name = "agent-client-protocol" -version = "1.3.0" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d386d6a58f4dbe0ebfa98e915caa54005dabdefd419de3d4678b28cd5752444" +checksum = "6d87bc7769eba641753ba5dc52f73ec3765d51022c6753bf040967125ddc86a8" dependencies = [ "agent-client-protocol-derive", "agent-client-protocol-schema", @@ -347,19 +347,19 @@ dependencies = [ [[package]] name = "agent-client-protocol-derive" -version = "1.3.0" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97b94934d118a69e921d14e94d4af5b3076563318c52662c89a0ecb3f139e1da" +checksum = "3abd4080f51e4f24f5042beb7fb7a66ede29a2dc1c2582c329532e1c27264ddc" dependencies = [ "quote", - "syn 2.0.117", + "syn 3.0.3", ] [[package]] name = "agent-client-protocol-schema" -version = "1.4.0" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06679e1542356341f4550ccfb16338b64f37f6af70de2105446ef6fbb078234c" +checksum = "d5c231915b4ab578c722eca2d1bd7df4d300bfd6cac3b8e9f0d1e3ddc95b187c" dependencies = [ "anyhow", "derive_more", @@ -18143,6 +18143,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "0.1.2" diff --git a/Cargo.toml b/Cargo.toml index 5741492485d8be..634401ed88de50 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -512,7 +512,7 @@ accesskit = { version = "0.24.0", features = ["enumn"] } accesskit_macos = "0.26.0" accesskit_unix = "0.21.0" accesskit_windows = "0.33.1" -agent-client-protocol = { version = "=1.3.0", features = ["unstable"] } +agent-client-protocol = { version = "=2.0.0", features = ["unstable"] } aho-corasick = "1.1" alacritty_terminal = { git = "https://github.com/zed-industries/alacritty", rev = "4c129667ce56611becdc82de6e28218c80e2e88f" } any_vec = "0.14" diff --git a/crates/agent_servers/src/acp.rs b/crates/agent_servers/src/acp.rs index ec3b196272e695..2c77dfc16fcfc7 100644 --- a/crates/agent_servers/src/acp.rs +++ b/crates/agent_servers/src/acp.rs @@ -81,17 +81,30 @@ pub struct AcpDebugMessage { } impl AcpDebugMessage { - fn parse(direction: AcpDebugMessageDirection, line: &str) -> Option { + fn parse_line(direction: AcpDebugMessageDirection, line: &str) -> Vec { if direction == AcpDebugMessageDirection::Stderr { - return Some(Self { + return vec![Self { direction, message: AcpDebugMessageContent::Stderr { line: Arc::from(line), }, - }); + }]; } - let value: serde_json::Value = serde_json::from_str(line).ok()?; + let Ok(value) = serde_json::from_str(line) else { + return Vec::new(); + }; + + match value { + serde_json::Value::Array(entries) => entries + .into_iter() + .filter_map(|entry| Self::parse_value(direction, entry)) + .collect(), + value => Self::parse_value(direction, value).into_iter().collect(), + } + } + + fn parse_value(direction: AcpDebugMessageDirection, value: serde_json::Value) -> Option { let object = value.as_object()?; let parsed_id = object @@ -178,26 +191,29 @@ impl AcpDebugLog { } fn record_line(&self, direction: AcpDebugMessageDirection, line: &str) { - let Some(message) = AcpDebugMessage::parse(direction, line) else { + let messages = AcpDebugMessage::parse_line(direction, line); + if messages.is_empty() { return; - }; - self.record_message(message); + } + self.record_messages(messages); } - fn record_message(&self, message: AcpDebugMessage) { + fn record_messages(&self, messages: Vec) { let mut state = self .state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - if state.messages.len() == MAX_DEBUG_BACKLOG_MESSAGES { - state.messages.pop_front(); - } - state.messages.push_back(message.clone()); - state.subscribers.retain(|sender| !sender.is_closed()); - for sender in &state.subscribers { - sender.try_send(message.clone()).log_err(); + for message in messages { + if state.messages.len() == MAX_DEBUG_BACKLOG_MESSAGES { + state.messages.pop_front(); + } + state.messages.push_back(message.clone()); + + for sender in &state.subscribers { + sender.try_send(message.clone()).log_err(); + } } } @@ -322,7 +338,6 @@ where Notif: Send + 'static, { notification: Notif, - connection: ConnectionTo, handler: fn(Notif, &mut AsyncApp, &ClientContext), } @@ -334,17 +349,12 @@ where let Self { notification, handler, - .. } = *self; handler(notification, cx, ctx); } fn reject(self: Box) { - let Self { connection, .. } = *self; log::error!("ACP foreground dispatch queue closed while handling inbound notification"); - connection - .send_error_notification(dispatch_queue_closed_error()) - .log_err(); } } @@ -370,14 +380,12 @@ fn enqueue_request( fn enqueue_notification( dispatch_tx: &mpsc::UnboundedSender, notification: Notif, - connection: ConnectionTo, handler: fn(Notif, &mut AsyncApp, &ClientContext), ) where Notif: Send + 'static, { let work: ForegroundWork = Box::new(NotificationForegroundWork { notification, - connection, handler, }); if let Err(err) = dispatch_tx.unbounded_send(work) { @@ -685,8 +693,8 @@ fn connect_client_future( macro_rules! on_notification { ($handler:ident) => {{ let dispatch_tx = dispatch_tx.clone(); - async move |notif, connection| { - enqueue_notification(&dispatch_tx, notif, connection, $handler); + async move |notif, _connection| { + enqueue_notification(&dispatch_tx, notif, $handler); Ok(()) } }}; @@ -3150,6 +3158,78 @@ mod tests { ); } + #[test] + fn debug_log_records_each_json_rpc_batch_entry() { + let debug_log = AcpDebugLog::default(); + debug_log.record_line( + AcpDebugMessageDirection::Incoming, + r#"{"jsonrpc":"2.0","method":"legacy/update"}"#, + ); + debug_log.record_line( + AcpDebugMessageDirection::Incoming, + r#"[ + {"jsonrpc":"2.0","method":"session/update","params":{"value":1}}, + null, + [{"jsonrpc":"2.0","method":"nested/update"}], + {"jsonrpc":"2.0","id":1,"method":"session/one","params":{"value":2}}, + {"jsonrpc":"2.0","id":{"invalid":true},"method":"invalid/id"} + ]"#, + ); + debug_log.record_line( + AcpDebugMessageDirection::Outgoing, + r#"[ + {"jsonrpc":"2.0","id":1,"result":{"accepted":true}}, + {"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"Invalid Request"}} + ]"#, + ); + + let (messages, _receiver) = debug_log.subscribe(); + let mut messages = messages.iter(); + + assert!(matches!( + messages.next(), + Some(AcpDebugMessage { + direction: AcpDebugMessageDirection::Incoming, + message: AcpDebugMessageContent::Notification { method, .. }, + }) if method.as_ref() == "legacy/update" + )); + assert!(matches!( + messages.next(), + Some(AcpDebugMessage { + direction: AcpDebugMessageDirection::Incoming, + message: AcpDebugMessageContent::Notification { method, .. }, + }) if method.as_ref() == "session/update" + )); + assert!(matches!( + messages.next(), + Some(AcpDebugMessage { + direction: AcpDebugMessageDirection::Incoming, + message: AcpDebugMessageContent::Request { id, method, .. }, + }) if id == &acp::RequestId::Number(1) && method.as_ref() == "session/one" + )); + assert!(matches!( + messages.next(), + Some(AcpDebugMessage { + direction: AcpDebugMessageDirection::Outgoing, + message: AcpDebugMessageContent::Response { + id, + result: Ok(Some(_)), + }, + }) if id == &acp::RequestId::Number(1) + )); + assert!(matches!( + messages.next(), + Some(AcpDebugMessage { + direction: AcpDebugMessageDirection::Outgoing, + message: AcpDebugMessageContent::Response { + id, + result: Err(_), + }, + }) if id == &acp::RequestId::Null + )); + assert!(messages.next().is_none()); + } + #[test] fn session_directories_use_ordered_paths_when_supported() { let work_dirs = PathList::new(&[