diff --git a/Cargo.toml b/Cargo.toml index 2679e102..e4a4a0e1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,9 @@ repository = "https://github.com/agentclientprotocol/rust-sdk" [workspace.dependencies] # Internal crates agent-client-protocol = { path = "src/agent-client-protocol", version = "3.6.0" } -agent-client-protocol-derive = { path = "src/agent-client-protocol-derive", version = "3.6.0" } +# Exact pin: the derive output calls `agent_client_protocol::__private`, which +# only matches the derive crate released with the same core version. +agent-client-protocol-derive = { path = "src/agent-client-protocol-derive", version = "=3.6.0" } agent-client-protocol-polyfill = { path = "src/agent-client-protocol-polyfill", version = "3.6.0" } agent-client-protocol-rmcp = { path = "src/agent-client-protocol-rmcp", version = "4.0.6" } agent-client-protocol-test = { path = "src/agent-client-protocol-test", default-features = false } diff --git a/md/migration-v5.md b/md/migration-v5.md index a6702e37..3de0cf6a 100644 --- a/md/migration-v5.md +++ b/md/migration-v5.md @@ -533,6 +533,7 @@ and `"1"` stay distinct, and integers keep the full 64-bit signed range. | Before | After | | --- | --- | | `RawJsonRpcMessage::Request(schema::v1::Request)` | `RawJsonRpcMessage::Request(RawJsonRpcRequest)`, with the same `id`, `method` and `params` fields | +| `RawJsonRpcMessage::Notification(schema::v1::Notification)` | `RawJsonRpcMessage::Notification(RawJsonRpcNotification)`, with the same `method` and `params` fields | | `RawJsonRpcResponse`, an alias of `schema::v1::Response>` | `RawJsonRpcResponse`, the SDK's own enum with the same `Result { id, result }` and `Error { id, error }` variants and `new` | | `RawJsonRpcMessage::response_id()` returned `Option<&v1::RequestId>` | returns `Option<&RawJsonRpcId>` | @@ -727,6 +728,76 @@ callbacks receive (`ConnectionTo` for `Raw`, `v1::ConnectionTo` for `V1`, `v2::ConnectionTo` for `V2`), so code generic over a builder's context changes `Context: ConnectionContext` to `P: Protocol`. +## Schema enums and SDK helpers + +### v1 extension methods keep their `_` prefix + +Parsing a v1 dispatch enum (`schema::v1::ClientRequest`, `AgentRequest`, +`ClientNotification` or `AgentNotification`) from an extension method used to +strip the leading `_`: a parsed `_vendor/x` reported `method()` as `vendor/x` +and was sent on as `vendor/x`, which the peer answered with Method not found. +The `ExtMethodRequest` and `ExtNotification` variants now keep the method +exactly as it was on the wire, as v2 already did. Code that matched on +`ext.method` without the prefix matches the full name instead: + +```rust,ignore +// Before +v1::ClientRequest::ExtMethodRequest(ext) if &*ext.method == "vendor/x" => ... +// After +v1::ClientRequest::ExtMethodRequest(ext) if &*ext.method == "_vendor/x" => ... +``` + +### Removed `util` helpers + +`util::internal_error(message)` and `util::parse_error(message)` duplicated the +`Error` constructors and are removed. Build the error directly: + +```rust,ignore +// Before +Err(util::internal_error(format!("lost {id}"))) +result.map_err(util::internal_error)?; + +// After +Err(Error::internal_error().data(format!("lost {id}"))) +result.map_err(Error::into_internal_error)?; // any `std::error::Error` +``` + +`util::TypeNotification` is removed. It duplicated +`MatchDispatch::if_notification` for notifications. Wrap the message in +`Dispatch::Notification` and use `MatchDispatch`, which returns a parse error +for malformed params instead of logging and dropping them: + +```rust,ignore +// Before +TypeNotification::new(message) + .handle_if(async |notif: SessionNotification| { ... }) + .await + .otherwise(async |untyped| { ... }) + .await + +// After +MatchDispatch::new(Dispatch::Notification(message)) + .if_notification(async |notif: SessionNotification| { ... }) + .await + .otherwise(async |dispatch| { ... }) + .await +``` + +`agent-client-protocol` now depends on the `agent-client-protocol-derive` +version released with it exactly (`=`), because the derive output calls +hidden SDK internals. Depend on the derives through `agent_client_protocol` +rather than adding the derive crate separately. + +### MCP-over-ACP error codes are `ErrorCode`s + +`mcp_server::MCP_SERVER_UNAVAILABLE` (`-33001`) and +`mcp_server::MCP_BACKEND_FAILURE` (`-33002`) are now `ErrorCode` constants +instead of `i32`, and `mcp_server::MCP_RESOURCE_LIMIT_EXCEEDED` (`-33000`) is +new. `Error::new(MCP_BACKEND_FAILURE, ..)` still compiles. Compare with +`error.code == MCP_BACKEND_FAILURE` directly instead of through +`i32::from(error.code)`, and use `.value()` where you need an `i32` (for +example, against a JSON `code`). + ## Proxy method names The proxy protocol now uses the method names defined by the @@ -740,8 +811,9 @@ extensions. | `_proxy/successor` | `proxy/successor` | The message shapes are unchanged. `proxy/successor` still flattens the inner -ACP message into its params and sends optional metadata as `_meta` (`meta` is -also accepted on receive). +ACP message into its params and sends optional metadata as `_meta`. The +non-spec `meta` key is no longer read on receive; see +[`SuccessorMessage` and `InitializeProxyRequest` are non-exhaustive](#successormessage-and-initializeproxyrequest-are-non-exhaustive). ### Upgrade conductors and proxies together @@ -757,6 +829,30 @@ Code that uses the SDK types (`schema::v1::InitializeProxyRequest`, formerly the `Proxy` and `Conductor` roles) needs no changes for the rename. Update any hard-coded method strings, such as in raw JSON-RPC handlers, test fixtures, or trace filters. +### `SuccessorMessage` and `InitializeProxyRequest` are non-exhaustive + +`schema::SuccessorMessage`, `schema::v1::InitializeProxyRequest` and +`schema::v2::InitializeProxyRequest` are `#[non_exhaustive]`, so the proxy +protocol can add fields in a minor release. Their fields stay public for +reading, but outside the SDK you build them with constructors: + +```rust,ignore +// Before +let message = SuccessorMessage { message: request, meta: None }; +let init = v1::InitializeProxyRequest { initialize }; + +// After +let message = SuccessorMessage::new(request); +let message = SuccessorMessage::new(request).meta(meta); +let init = v1::InitializeProxyRequest::new(initialize); // or initialize.into() +``` + +`SuccessorMessage::meta` is now `Option>` +instead of `Option`, because ACP `_meta` is always an object. A +`proxy/successor` request whose `_meta` is not an object is accepted and +its `_meta` is ignored, and the non-spec `meta` key is ignored instead of +being read as `_meta`. + ### `ConductorHostRole` is sealed `agent-client-protocol-conductor::ConductorHostRole` can no longer be diff --git a/src/agent-client-protocol-conductor/src/conductor.rs b/src/agent-client-protocol-conductor/src/conductor.rs index 2dfb88eb..f1d2bf75 100644 --- a/src/agent-client-protocol-conductor/src/conductor.rs +++ b/src/agent-client-protocol-conductor/src/conductor.rs @@ -461,9 +461,9 @@ impl ConductorImpl { conductor_tx: conductor_tx.clone(), initialization: InitializationState::Pending(self.instantiator), proxies: Vec::default(), - successor: Arc::new(agent_client_protocol::util::internal_error( - "successor not initialized", - )), + successor: Arc::new( + agent_client_protocol::Error::internal_error().data("successor not initialized"), + ), trace_handle, host: self.host.clone(), }; @@ -496,7 +496,7 @@ impl ConductorImpl { message, }) .await - .map_err(agent_client_protocol::util::internal_error) + .map_err(agent_client_protocol::Error::into_internal_error) } async fn incoming_message_from_agent( @@ -509,7 +509,7 @@ impl ConductorImpl { message, }) .await - .map_err(agent_client_protocol::util::internal_error) + .map_err(agent_client_protocol::Error::into_internal_error) } } @@ -753,10 +753,7 @@ where if source_component_index == 0 { client_connection.send_request_to(Client, request) } else { - self.proxies[source_component_index - 1].send_request(SuccessorMessage { - message: request, - meta: None, - }) + self.proxies[source_component_index - 1].send_request(SuccessorMessage::new(request)) } } @@ -789,10 +786,8 @@ where target_proxy = source_component_index - 1, "Sending notification wrapped as SuccessorMessage to proxy" ); - self.proxies[source_component_index - 1].send_notification(SuccessorMessage { - message: notification, - meta: None, - }) + self.proxies[source_component_index - 1] + .send_notification(SuccessorMessage::new(notification)) } } @@ -1040,7 +1035,10 @@ where message: dispatch.map(|r, cx| (r.message, cx), |n| n.message), }) .await - .map_err(agent_client_protocol::util::internal_error) + .map_err(|e| { + agent_client_protocol::Error::internal_error() + .data(e.to_string()) + }) }) .await .otherwise(async |dispatch| { @@ -1064,10 +1062,9 @@ where ), message: dispatch, }; - conductor_tx - .send(message) - .await - .map_err(agent_client_protocol::util::internal_error) + conductor_tx.send(message).await.map_err(|e| { + agent_client_protocol::Error::internal_error().data(e.to_string()) + }) }) .await }, @@ -1554,7 +1551,10 @@ impl ConductorHostRole for Agent { message: dispatch, }) .await - .map_err(agent_client_protocol::util::internal_error) + .map_err(|e| { + agent_client_protocol::Error::internal_error() + .data(e.to_string()) + }) } }, agent_client_protocol::on_receive_dispatch!(), diff --git a/src/agent-client-protocol-conductor/src/lib.rs b/src/agent-client-protocol-conductor/src/lib.rs index e6e25b64..79c0e1b7 100644 --- a/src/agent-client-protocol-conductor/src/lib.rs +++ b/src/agent-client-protocol-conductor/src/lib.rs @@ -133,9 +133,7 @@ impl InstantiateProxiesAndAgent for CommandLineComponents { } } - Err(agent_client_protocol::util::internal_error( - "no agent component in list", - )) + Err(agent_client_protocol::Error::internal_error().data("no agent component in list")) }) } } diff --git a/src/agent-client-protocol-conductor/src/trace.rs b/src/agent-client-protocol-conductor/src/trace.rs index 8b64fa96..c8646659 100644 --- a/src/agent-client-protocol-conductor/src/trace.rs +++ b/src/agent-client-protocol-conductor/src/trace.rs @@ -11,11 +11,11 @@ use std::time::Instant; use agent_client_protocol::schema::SuccessorMessage; use agent_client_protocol::schema::v1::{ - MessageMcpNotification, MessageMcpRequest, MessageMcpResponse, Notification as RpcNotification, + MessageMcpNotification, MessageMcpRequest, MessageMcpResponse, }; use agent_client_protocol::{ - DynConnectTo, JsonRpcMessage, RawJsonRpcId, RawJsonRpcMessage, RawJsonRpcParams, - RawJsonRpcRequest, RawJsonRpcResponse as RpcResponse, Role, UntypedMessage, + DynConnectTo, JsonRpcMessage, RawJsonRpcId, RawJsonRpcMessage, RawJsonRpcNotification, + RawJsonRpcParams, RawJsonRpcRequest, RawJsonRpcResponse as RpcResponse, Role, UntypedMessage, }; use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; @@ -479,7 +479,7 @@ impl TraceHandle { incoming, message: message.clone(), }) - .map_err(agent_client_protocol::util::internal_error) + .map_err(agent_client_protocol::Error::into_internal_error) } /// Create a tracing bridge that wraps a proxy component. @@ -732,7 +732,7 @@ impl MessageInfo { Self::from_untyped_request(Successor(false), Some(req.id), Protocol::Acp, untyped) } - fn from_notification(notification: RpcNotification) -> Self { + fn from_notification(notification: RawJsonRpcNotification) -> Self { let untyped = UntypedMessage::parse_message( ¬ification.method, ¶ms_from_transport(notification.params), diff --git a/src/agent-client-protocol-conductor/tests/request_cancellation.rs b/src/agent-client-protocol-conductor/tests/request_cancellation.rs index cfe0062f..7c376ed9 100644 --- a/src/agent-client-protocol-conductor/tests/request_cancellation.rs +++ b/src/agent-client-protocol-conductor/tests/request_cancellation.rs @@ -512,11 +512,10 @@ async fn prompt_cancellation_cascades_through_real_proxy_chain() -> Result<(), E if i32::from(permission_error.code) == -32800 { responder.respond_with_result(Err(Error::request_cancelled())) } else { - responder.respond_with_result(Err( - agent_client_protocol::util::internal_error(format!( - "unexpected permission error: {permission_error:?}" - )), - )) + responder + .respond_with_result(Err(agent_client_protocol::Error::internal_error( + ) + .data(format!("unexpected permission error: {permission_error:?}")))) } })?; Ok(()) diff --git a/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs b/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs index b3fecd57..4ee93f47 100644 --- a/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs +++ b/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs @@ -89,13 +89,13 @@ async fn test_standalone_server_list_tools() -> Result<(), agent_client_protocol let client = MinimalClientHandler .serve((server_read, server_write)) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // List tools let tools_result = client .list_tools(None) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // Verify we got both tools assert_eq!(tools_result.tools.len(), 2); @@ -109,7 +109,7 @@ async fn test_standalone_server_list_tools() -> Result<(), agent_client_protocol client .cancel() .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; Ok(()) }, ) @@ -131,7 +131,7 @@ async fn test_standalone_server_call_echo_tool() -> Result<(), agent_client_prot let client = MinimalClientHandler .serve((server_read, server_write)) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // Call the echo tool let result = client @@ -144,7 +144,7 @@ async fn test_standalone_server_call_echo_tool() -> Result<(), agent_client_prot ), ) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // Verify the result let text = result @@ -159,7 +159,7 @@ async fn test_standalone_server_call_echo_tool() -> Result<(), agent_client_prot client .cancel() .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; Ok(()) }, ) @@ -181,7 +181,7 @@ async fn test_standalone_server_call_add_tool() -> Result<(), agent_client_proto let client = MinimalClientHandler .serve((server_read, server_write)) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // Call the add tool let result = client @@ -194,7 +194,7 @@ async fn test_standalone_server_call_add_tool() -> Result<(), agent_client_proto ), ) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // The add tool returns structured output (AddOutput) // Check that we get the expected result @@ -212,7 +212,7 @@ async fn test_standalone_server_call_add_tool() -> Result<(), agent_client_proto client .cancel() .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; Ok(()) }, ) @@ -234,7 +234,7 @@ async fn test_standalone_server_tool_not_found() -> Result<(), agent_client_prot let client = MinimalClientHandler .serve((server_read, server_write)) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // Call a non-existent tool let result = client @@ -247,7 +247,7 @@ async fn test_standalone_server_tool_not_found() -> Result<(), agent_client_prot client .cancel() .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; Ok(()) }, ) @@ -289,13 +289,13 @@ async fn test_standalone_server_with_disabled_tools() -> Result<(), agent_client let client = MinimalClientHandler .serve((server_read, server_write)) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; // List tools - should only show "add" let tools_result = client .list_tools(None) .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; assert_eq!(tools_result.tools.len(), 1); assert_eq!(tools_result.tools[0].name.as_ref(), "add"); @@ -316,7 +316,7 @@ async fn test_standalone_server_with_disabled_tools() -> Result<(), agent_client client .cancel() .await - .map_err(agent_client_protocol::util::internal_error)?; + .map_err(agent_client_protocol::Error::into_internal_error)?; Ok(()) }, ) diff --git a/src/agent-client-protocol-derive/src/lib.rs b/src/agent-client-protocol-derive/src/lib.rs index b46ab4ee..c0c8c2aa 100644 --- a/src/agent-client-protocol-derive/src/lib.rs +++ b/src/agent-client-protocol-derive/src/lib.rs @@ -98,7 +98,7 @@ pub fn derive_json_rpc_request(input: TokenStream) -> TokenStream { if #method_arg != #method { return ::core::result::Result::Err(#krate::Error::method_not_found()); } - #krate::util::json_cast_params(#params_arg) + #krate::__private::json_cast_params(#params_arg) } } @@ -168,7 +168,7 @@ pub fn derive_json_rpc_notification(input: TokenStream) -> TokenStream { if #method_arg != #method { return ::core::result::Result::Err(#krate::Error::method_not_found()); } - #krate::util::json_cast_params(#params_arg) + #krate::__private::json_cast_params(#params_arg) } } @@ -216,7 +216,7 @@ pub fn derive_json_rpc_response_payload(input: TokenStream) -> TokenStream { } fn from_value(#method_arg: &str, #value_arg: #krate::__private::serde_json::Value) -> ::core::result::Result { - #krate::util::json_cast(#value_arg) + #krate::__private::json_cast(#value_arg) } } }; diff --git a/src/agent-client-protocol/src/jsonrpc.rs b/src/agent-client-protocol/src/jsonrpc.rs index 755cf005..e798bc55 100644 --- a/src/agent-client-protocol/src/jsonrpc.rs +++ b/src/agent-client-protocol/src/jsonrpc.rs @@ -1,8 +1,6 @@ //! Core JSON-RPC server support. -use agent_client_protocol_schema::v1::{ - JsonRpcMessage as VersionedJsonRpcMessage, Notification as RpcNotification, SessionId, -}; +use agent_client_protocol_schema::v1::{JsonRpcMessage as VersionedJsonRpcMessage, SessionId}; // Types re-exported from crate root use serde::{Deserialize, Serialize}; @@ -86,12 +84,21 @@ use crate::{Agent, ConnectTo, RoleId}; /// Raw messages are the same for every ACP protocol version: IDs are /// [`RawJsonRpcId`]s, and method params and response errors are not /// interpreted at the transport boundary. -#[derive(Debug, Clone)] +/// +/// Serializing adds the `"jsonrpc": "2.0"` member, and deserializing requires +/// it. Requests and notifications keep only `id`, `method` and `params`: +/// other members of the envelope are dropped. Error responses keep unknown +/// members of the error object in [`RawJsonRpcError::extra`]. +#[derive(Debug, Clone, PartialEq, Eq)] +#[allow( + clippy::exhaustive_enums, + reason = "JSON-RPC defines exactly these message kinds" +)] pub enum RawJsonRpcMessage { /// A JSON-RPC request with an id and expected response. Request(RawJsonRpcRequest), /// A JSON-RPC notification without a response. - Notification(RpcNotification), + Notification(RawJsonRpcNotification), /// A JSON-RPC response to a prior request. Response(RawJsonRpcResponse), } @@ -140,7 +147,11 @@ impl TransportFrame { } /// A JSON-RPC request with uninterpreted params. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +/// +/// This is the request without the `jsonrpc` member, which +/// [`RawJsonRpcMessage`] adds and checks. Other members of the request object +/// are not kept: a relay that parses and re-serializes a request drops them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[allow( clippy::exhaustive_structs, reason = "JSON-RPC defines exactly these request members" @@ -155,10 +166,51 @@ pub struct RawJsonRpcRequest { pub params: Option, } +/// A JSON-RPC notification with uninterpreted params. +/// +/// This is the notification without the `jsonrpc` member, which +/// [`RawJsonRpcMessage`] adds and checks. Other members of the notification +/// object are not kept: a relay that parses and re-serializes a notification +/// drops them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[allow( + clippy::exhaustive_structs, + reason = "JSON-RPC defines exactly these notification members" +)] +pub struct RawJsonRpcNotification { + /// The notification method. + pub method: Arc, + /// The params, if any. `None` omits the member. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub params: Option, +} + +impl From for RawJsonRpcMessage { + fn from(request: RawJsonRpcRequest) -> Self { + Self::Request(request) + } +} + +impl From for RawJsonRpcMessage { + fn from(notification: RawJsonRpcNotification) -> Self { + Self::Notification(notification) + } +} + +impl From for RawJsonRpcMessage { + fn from(response: RawJsonRpcResponse) -> Self { + Self::Response(response) + } +} + /// Raw JSON-RPC request or notification parameters. /// /// JSON-RPC params, when present, must be either an array or an object. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Eq)] +#[allow( + clippy::exhaustive_enums, + reason = "JSON-RPC params are either an array or an object" +)] pub enum RawJsonRpcParams { /// Positional JSON-RPC params. Array(Vec), @@ -234,7 +286,7 @@ impl RawJsonRpcMessage { /// Build a raw JSON-RPC notification message. pub fn notification(method: String, params: serde_json::Value) -> Result { - Ok(Self::Notification(RpcNotification { + Ok(Self::Notification(RawJsonRpcNotification { method: Arc::from(method), params: RawJsonRpcParams::from_value(params)?, })) @@ -310,7 +362,7 @@ impl<'de> Deserialize<'de> for RawJsonRpcMessage { Ok(Self::Request(request)) } else { let notification = serde_json::from_value::< - VersionedJsonRpcMessage>, + VersionedJsonRpcMessage, >(value) .map_err(serde::de::Error::custom)? .into_inner(); @@ -6905,7 +6957,7 @@ impl SentRequest { /// // Respond to original request /// responder.respond(ValidateResponse { is_valid: true, error: None }) /// } else { - /// responder.respond_with_error(agent_client_protocol::util::internal_error("validation failed")) + /// responder.respond_with_error(agent_client_protocol::Error::internal_error().data("validation failed")) /// } /// })?; /// @@ -7505,6 +7557,41 @@ mod tests { #[cfg(feature = "unstable_protocol_v2")] use crate::Proxy; + #[test] + fn raw_notification_round_trips_and_drops_unknown_envelope_members() { + let message: RawJsonRpcMessage = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "method": "_vendor/ping", + "params": { "n": 1 }, + "extra": true, + })) + .unwrap(); + let mut params = serde_json::Map::new(); + params.insert("n".into(), 1.into()); + assert_eq!( + message, + RawJsonRpcMessage::Notification(RawJsonRpcNotification { + method: "_vendor/ping".into(), + params: Some(RawJsonRpcParams::Object(params)), + }) + ); + assert_eq!( + serde_json::to_value(&message).unwrap(), + serde_json::json!({ + "jsonrpc": "2.0", + "method": "_vendor/ping", + "params": { "n": 1 }, + }) + ); + + let without_params = + RawJsonRpcMessage::notification("ping".into(), serde_json::Value::Null).unwrap(); + assert_eq!( + serde_json::to_value(&without_params).unwrap(), + serde_json::json!({ "jsonrpc": "2.0", "method": "ping" }) + ); + } + #[test] fn protected_cleanup_keeps_scoped_runners_polled_on_every_shutdown_path() { #[derive(Clone, Copy, Debug)] @@ -7934,14 +8021,13 @@ mod tests { ProtocolMode::v2_proxy(), ); - let request = crate::schema::SuccessorMessage { - message: UntypedMessage::new( + let request = crate::schema::SuccessorMessage::new( + UntypedMessage::new( "initialize", serde_json::json!({ "protocolVersion": crate::schema::ProtocolVersion::V1 }), ) .expect("test initialize request should serialize"), - meta: None, - }; + ); let sent = connection.send_request_to(Agent, request); let (transport_tx, mut transport_rx) = mpsc::unbounded(); diff --git a/src/agent-client-protocol/src/jsonrpc/raw_error.rs b/src/agent-client-protocol/src/jsonrpc/raw_error.rs index fcfc6828..4cfb5a6c 100644 --- a/src/agent-client-protocol/src/jsonrpc/raw_error.rs +++ b/src/agent-client-protocol/src/jsonrpc/raw_error.rs @@ -11,7 +11,7 @@ use super::RawJsonRpcId; /// Raw transports and relays preserve unknown fields and distinguish omitted /// `data` from explicit JSON null. Convert to [`crate::Error`] only when /// dispatching an ACP response; other protocols have their own error domains. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[non_exhaustive] pub struct RawJsonRpcError { /// The peer's numeric error code, not an ACP [`crate::ErrorCode`]. @@ -30,7 +30,7 @@ pub struct RawJsonRpcError { /// /// Errors are boxed so their extensible representation does not enlarge every /// request, notification, and queued frame. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(untagged)] #[allow( clippy::exhaustive_enums, diff --git a/src/agent-client-protocol/src/jsonrpc/v2_connection.rs b/src/agent-client-protocol/src/jsonrpc/v2_connection.rs index a74f8d02..2e25f14b 100644 --- a/src/agent-client-protocol/src/jsonrpc/v2_connection.rs +++ b/src/agent-client-protocol/src/jsonrpc/v2_connection.rs @@ -66,7 +66,7 @@ impl ConnectionTo { /// Ok(None) /// } /// InitializeOutcome::LowerVersion(mismatch) => Ok(Some(mismatch)), - /// _ => Err(agent_client_protocol::util::internal_error("unexpected outcome")), + /// _ => Err(agent_client_protocol::Error::internal_error().data("unexpected outcome")), /// } /// }) /// .await?; diff --git a/src/agent-client-protocol/src/lib.rs b/src/agent-client-protocol/src/lib.rs index ea4321c2..290541d9 100644 --- a/src/agent-client-protocol/src/lib.rs +++ b/src/agent-client-protocol/src/lib.rs @@ -188,9 +188,9 @@ pub use jsonrpc::{ ConnectionTo, Dispatch, DynamicHandlerGuard, HandleConnectionClose, HandleDispatchFrom, Handled, INCOMING_TRANSPORT_CLOSED_REASON, IntoHandled, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, Lines, NullClose, NullHandler, PreparedRequest, Protocol, Raw, - RawJsonRpcError, RawJsonRpcId, RawJsonRpcMessage, RawJsonRpcParams, RawJsonRpcRequest, - RawJsonRpcResponse, Responder, ResponseRouter, SentRequest, TransportFrame, UntypedMessage, V1, - is_connection_terminated, is_incoming_transport_closed, + RawJsonRpcError, RawJsonRpcId, RawJsonRpcMessage, RawJsonRpcNotification, RawJsonRpcParams, + RawJsonRpcRequest, RawJsonRpcResponse, Responder, ResponseRouter, SentRequest, TransportFrame, + UntypedMessage, V1, is_connection_terminated, is_incoming_transport_closed, run::{ChainRun, NullRun, RunWithConnectionTo}, }; pub use jsonrpc::{RequestCancellation, RequestCancellationHandle, is_cancel_request_notification}; @@ -207,6 +207,7 @@ pub use component::{ConnectTo, ConnectionDriver, DynConnectTo}; /// Implementation details used by the derive macros. #[doc(hidden)] pub mod __private { + pub use crate::util::{json_cast, json_cast_params}; pub use serde; pub use serde_json; } diff --git a/src/agent-client-protocol/src/mcp_server/active_session.rs b/src/agent-client-protocol/src/mcp_server/active_session.rs index d3213403..228db160 100644 --- a/src/agent-client-protocol/src/mcp_server/active_session.rs +++ b/src/agent-client-protocol/src/mcp_server/active_session.rs @@ -729,7 +729,7 @@ mod tests { assert!(!task.is_finished()); let _sent = release_tx.send(()); let TransportFrame::Single(response) = peer.rx.next().await.unwrap() else { panic!("single frame") }; - assert_eq!(serde_json::to_value(response).unwrap()["error"]["code"], MCP_SERVER_UNAVAILABLE); + assert_eq!(serde_json::to_value(response).unwrap()["error"]["code"], MCP_SERVER_UNAVAILABLE.value()); let _sent = stop_tx.send(()); task.await.unwrap().unwrap(); }).await.expect("provider removal cleanup timed out"); @@ -814,7 +814,7 @@ mod tests { json!({"supportedVersions":[MCP_VERSION, 42]}), ] { let error = project_discovery_versions(McpOutcome::Result(malformed)).unwrap_err(); - assert_eq!(i32::from(error.code), MCP_BACKEND_FAILURE); + assert_eq!(error.code, MCP_BACKEND_FAILURE); } let error = McpError::new(-32000, "opaque backend error").data(Value::Null); let preserved = project_discovery_versions(McpOutcome::Error(error.clone())).unwrap(); diff --git a/src/agent-client-protocol/src/mcp_server/mod.rs b/src/agent-client-protocol/src/mcp_server/mod.rs index 44dccff7..02876c93 100644 --- a/src/agent-client-protocol/src/mcp_server/mod.rs +++ b/src/agent-client-protocol/src/mcp_server/mod.rs @@ -78,12 +78,15 @@ pub use service::{ McpOperationCancellation, McpOutcome, McpRequest, McpRequestContext, McpService, }; -/// The declared MCP provider is no longer available. +/// `-33000`: an MCP-over-ACP binding resource limit was exceeded. #[cfg(feature = "unstable_mcp_over_acp")] -pub const MCP_SERVER_UNAVAILABLE: i32 = -33001; -/// The MCP backend failed independently of an MCP application error. +pub const MCP_RESOURCE_LIMIT_EXCEEDED: crate::ErrorCode = crate::ErrorCode::new(-33000); +/// `-33001`: the declared MCP server registration is unavailable. #[cfg(feature = "unstable_mcp_over_acp")] -pub const MCP_BACKEND_FAILURE: i32 = -33002; +pub const MCP_SERVER_UNAVAILABLE: crate::ErrorCode = crate::ErrorCode::new(-33001); +/// `-33002`: the MCP backend or transport failed without a valid MCP outcome. +#[cfg(feature = "unstable_mcp_over_acp")] +pub const MCP_BACKEND_FAILURE: crate::ErrorCode = crate::ErrorCode::new(-33002); #[cfg(feature = "schemars")] #[cfg_attr(docsrs, doc(cfg(feature = "schemars")))] pub use tool::McpTool; diff --git a/src/agent-client-protocol/src/role.rs b/src/agent-client-protocol/src/role.rs index a637b133..f28f7f26 100644 --- a/src/agent-client-protocol/src/role.rs +++ b/src/agent-client-protocol/src/role.rs @@ -9,7 +9,7 @@ use std::{any::TypeId, fmt::Debug, future::Future, hash::Hash}; use serde::{Deserialize, Serialize}; use crate::schema::{METHOD_SUCCESSOR_MESSAGE, SuccessorMessage}; -use crate::util::json_cast; +use crate::util::json_cast_params; use crate::{Builder, ConnectionTo, Dispatch, Handled, JsonRpcMessage, UntypedMessage}; /// Roles for the ACP protocol. @@ -93,11 +93,7 @@ impl RemoteStyle { ) -> Result { match self { RemoteStyle::Counterpart | RemoteStyle::Predecessor => msg.to_untyped_message(), - RemoteStyle::Successor => SuccessorMessage { - message: msg, - meta: None, - } - .to_untyped_message(), + RemoteStyle::Successor => SuccessorMessage::new(msg).to_untyped_message(), } } } @@ -222,7 +218,7 @@ where "Response variant cannot be unwrapped as SuccessorMessage", ) })?; - let SuccessorMessage { message, meta } = json_cast(untyped_message.params())?; + let SuccessorMessage { message, meta } = json_cast_params(untyped_message.params())?; let successor_dispatch = dispatch.try_map_message(|_| Ok(message))?; tracing::trace!( unwrapped_method = %successor_dispatch.method(), @@ -245,7 +241,9 @@ where ); Ok(Handled::No { message: successor_dispatch.try_map_message(|message| { - SuccessorMessage { message, meta }.to_untyped_message() + SuccessorMessage::new(message) + .meta(meta) + .to_untyped_message() })?, retry, }) diff --git a/src/agent-client-protocol/src/schema/mod.rs b/src/agent-client-protocol/src/schema/mod.rs index 7f4d9eea..a88c2f7a 100644 --- a/src/agent-client-protocol/src/schema/mod.rs +++ b/src/agent-client-protocol/src/schema/mod.rs @@ -115,11 +115,10 @@ macro_rules! impl_jsonrpc_request_enum { } fn method(&self) -> &str { - match self { - $( $(#[$meta])* Self::$variant(_) => $method, )* - Self::$ext_variant(ext) => &ext.method, - _ => "_unknown", - } + // The schema enum's inherent `method()` (path syntax prefers + // inherent items) covers every variant, including ones this + // macro does not list. + <$enum>::method(self) } fn to_untyped_message(&self) -> Result<$crate::UntypedMessage, $crate::Error> { @@ -133,11 +132,11 @@ macro_rules! impl_jsonrpc_request_enum { match method { $( $(#[$meta])* $method => $crate::util::json_cast_params(params).map(Self::$variant), )* _ => { - if let Some(custom_method) = method.strip_prefix('_') { + if method.starts_with('_') { $crate::util::json_cast_params(params).map( |ext_req: $crate::schema::v1::ExtRequest| { Self::$ext_variant($crate::schema::v1::ExtRequest::new( - custom_method.to_string(), + method.to_string(), ext_req.params, )) }, @@ -178,11 +177,10 @@ macro_rules! impl_jsonrpc_notification_enum { } fn method(&self) -> &str { - match self { - $( $(#[$meta])* Self::$variant(_) => $method, )* - Self::$ext_variant(ext) => &ext.method, - _ => "_unknown", - } + // The schema enum's inherent `method()` (path syntax prefers + // inherent items) covers every variant, including ones this + // macro does not list. + <$enum>::method(self) } fn to_untyped_message(&self) -> Result<$crate::UntypedMessage, $crate::Error> { @@ -196,11 +194,11 @@ macro_rules! impl_jsonrpc_notification_enum { match method { $( $(#[$meta])* $method => $crate::util::json_cast_params(params).map(Self::$variant), )* _ => { - if let Some(custom_method) = method.strip_prefix('_') { + if method.starts_with('_') { $crate::util::json_cast_params(params).map( |ext_notif: $crate::schema::v1::ExtNotification| { Self::$ext_variant($crate::schema::v1::ExtNotification::new( - custom_method.to_string(), + method.to_string(), ext_notif.params, )) }, @@ -333,6 +331,7 @@ pub mod v1 { response = InitializeResponse, crate = crate )] + #[non_exhaustive] pub struct InitializeProxyRequest { /// The underlying protocol-v1 initialize request data. #[serde(flatten)] @@ -370,6 +369,7 @@ pub mod v2 { response = InitializeResponse, crate = crate )] + #[non_exhaustive] pub struct InitializeProxyRequest { /// The underlying protocol-v2 initialize request data. #[serde(flatten)] diff --git a/src/agent-client-protocol/src/schema/proxy_protocol.rs b/src/agent-client-protocol/src/schema/proxy_protocol.rs index 762ef21f..6094c0c1 100644 --- a/src/agent-client-protocol/src/schema/proxy_protocol.rs +++ b/src/agent-client-protocol/src/schema/proxy_protocol.rs @@ -2,6 +2,7 @@ //! //! These types are intended to become part of the ACP protocol specification. +use crate::schema::IntoOption; use crate::{JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, UntypedMessage}; use serde::{Deserialize, Serialize}; @@ -15,19 +16,62 @@ pub const METHOD_SUCCESSOR_MESSAGE: &str = "proxy/successor"; /// A message being sent to the successor component. /// /// Used in `proxy/successor` when the proxy wants to forward a message downstream. +/// The inner message's `method` and `params` are flattened into the envelope's +/// params, next to the optional `_meta` object. +/// +/// Build one with [`SuccessorMessage::new`] and [`SuccessorMessage::meta`]. #[derive(Debug, Clone, Serialize, Deserialize)] +#[non_exhaustive] pub struct SuccessorMessage { /// The message to be sent to the successor component. #[serde(flatten)] pub message: M, - /// Optional `_meta` metadata. + /// Metadata for the envelope itself, sent as `_meta`. The inner message + /// keeps its own `_meta`. A received `_meta` that is not an object is + /// ignored. #[serde( rename = "_meta", - alias = "meta", - skip_serializing_if = "Option::is_none" + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "deserialize_meta_or_none" )] - pub meta: Option, + pub meta: Option>, +} + +/// Read `_meta` as an object, treating any other value as absent, the same way +/// the schema crate's `DefaultOnError` handles `_meta`. +fn deserialize_meta_or_none<'de, D>( + deserializer: D, +) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + match Option::::deserialize(deserializer)? { + Some(serde_json::Value::Object(map)) => Ok(Some(map)), + _ => Ok(None), + } +} + +impl SuccessorMessage { + /// Wrap `message` for the successor, without envelope metadata. + #[must_use] + pub fn new(message: M) -> Self { + Self { + message, + meta: None, + } + } + + /// Set the envelope's `_meta` object, replacing any existing metadata. + #[must_use] + pub fn meta( + mut self, + meta: impl IntoOption>, + ) -> Self { + self.meta = meta.into_option(); + self + } } impl JsonRpcMessage for SuccessorMessage { @@ -42,10 +86,7 @@ impl JsonRpcMessage for SuccessorMessage { fn to_untyped_message(&self) -> Result { UntypedMessage::new( METHOD_SUCCESSOR_MESSAGE, - SuccessorMessage { - message: self.message.to_untyped_message()?, - meta: self.meta.clone(), - }, + SuccessorMessage::new(self.message.to_untyped_message()?).meta(self.meta.clone()), ) } @@ -58,10 +99,7 @@ impl JsonRpcMessage for SuccessorMessage { return Err(crate::Error::method_not_found()); } let inner = M::parse_message(&outer.message.method, &outer.message.params)?; - Ok(SuccessorMessage { - message: inner, - meta: outer.meta, - }) + Ok(SuccessorMessage::new(inner).meta(outer.meta)) } } diff --git a/src/agent-client-protocol/src/schema/v2_impls.rs b/src/agent-client-protocol/src/schema/v2_impls.rs index 94d849e8..94ef0ed6 100644 --- a/src/agent-client-protocol/src/schema/v2_impls.rs +++ b/src/agent-client-protocol/src/schema/v2_impls.rs @@ -86,11 +86,10 @@ macro_rules! impl_v2_jsonrpc_request_enum { } fn method(&self) -> &str { - match self { - $( $(#[$meta])* Self::$variant(_) => $method, )* - Self::$ext_variant(ext) => &ext.method, - _ => "_unknown", - } + // The schema enum's inherent `method()` (path syntax prefers + // inherent items) covers every variant, including ones this + // macro does not list. + <$enum>::method(self) } fn to_untyped_message(&self) -> Result { @@ -138,11 +137,10 @@ macro_rules! impl_v2_jsonrpc_notification_enum { } fn method(&self) -> &str { - match self { - $( $(#[$meta])* Self::$variant(_) => $method, )* - Self::$ext_variant(ext) => &ext.method, - _ => "_unknown", - } + // The schema enum's inherent `method()` (path syntax prefers + // inherent items) covers every variant, including ones this + // macro does not list. + <$enum>::method(self) } fn to_untyped_message(&self) -> Result { diff --git a/src/agent-client-protocol/src/util.rs b/src/agent-client-protocol/src/util.rs index ba7c81f9..ca9ff7fa 100644 --- a/src/agent-client-protocol/src/util.rs +++ b/src/agent-client-protocol/src/util.rs @@ -6,7 +6,7 @@ use futures::{ }; mod typed; -pub use typed::{MatchDispatch, MatchDispatchFrom, TypeNotification}; +pub use typed::{MatchDispatch, MatchDispatchFrom}; /// Format only code-derived metadata at ordinary log levels. /// @@ -108,6 +108,12 @@ pub(crate) fn redacted_json_line(line: &str) -> impl std::fmt::Display + '_ { /// Cast from `N` to `M` by serializing/deserialization to/from JSON. /// +/// Failures are Parse errors (`-32700`). Don't use this for the params of an +/// incoming request or notification: a peer that sends malformed params must +/// get Invalid params (`-32602`), which [`json_cast_params`] returns. Use +/// `json_cast` for values the SDK or your own code produced, such as response +/// results. +/// /// Returned diagnostics can contain the input JSON and payload-bearing parse /// reasons. Treat them as sensitive, as with [`json_cast_params`]. pub fn json_cast(params: N) -> Result @@ -161,16 +167,11 @@ where Ok(m) } -/// Creates an internal error with the given message -pub fn internal_error(message: impl ToString) -> crate::Error { +/// An internal error (`-32603`) whose data is `message`. +pub(crate) fn internal_error(message: impl ToString) -> crate::Error { crate::Error::internal_error().data(message.to_string()) } -/// Creates a parse error with the given message -pub fn parse_error(message: impl ToString) -> crate::Error { - crate::Error::parse_error().data(message.to_string()) -} - /// Rewrite the params of an `initialize` request for the protocol version /// `selected` that a version-agnostic forwarder, such as the conductor, /// selected for it. diff --git a/src/agent-client-protocol/src/util/typed.rs b/src/agent-client-protocol/src/util/typed.rs index 86ab013c..0edf650b 100644 --- a/src/agent-client-protocol/src/util/typed.rs +++ b/src/agent-client-protocol/src/util/typed.rs @@ -15,10 +15,11 @@ //! processing messages that don't need peer transforms. //! //! [`HandleDispatchFrom`]: crate::HandleDispatchFrom +//! [`UntypedMessage`]: crate::UntypedMessage use crate::{ ConnectionTo, Dispatch, HandleDispatchFrom, Handled, JsonRpcNotification, JsonRpcRequest, - JsonRpcResponse, Responder, ResponseRouter, UntypedMessage, + JsonRpcResponse, Responder, ResponseRouter, role::{HasPeer, Role, handle_incoming_dispatch}, }; @@ -61,7 +62,7 @@ fn preserve_retry( /// .otherwise(|message| async move { /// match message { /// Dispatch::Request(_, responder) => { -/// responder.respond_with_error(agent_client_protocol::util::internal_error("unknown method")) +/// responder.respond_with_error(agent_client_protocol::Error::method_not_found()) /// } /// Dispatch::Notification(_) | Dispatch::Response(_, _) => Ok(()), /// } @@ -451,7 +452,7 @@ impl MatchDispatch { /// .otherwise(|message| async move { /// // Fallback for unrecognized messages /// match message { -/// Dispatch::Request(_, responder) => responder.respond_with_error(agent_client_protocol::util::internal_error("unknown method")), +/// Dispatch::Request(_, responder) => responder.respond_with_error(agent_client_protocol::Error::method_not_found()), /// Dispatch::Notification(_) | Dispatch::Response(_, _) => Ok(()), /// } /// }) @@ -814,190 +815,3 @@ impl MatchDispatchFrom { } } } - -/// Builder for pattern-matching on untyped JSON-RPC notifications. -/// -/// Similar to [`MatchDispatch`] but specifically for notifications (fire-and-forget messages with no response). -/// -/// # Pattern -/// -/// The typical pattern is: -/// 1. Create a `TypeNotification` from an untyped message -/// 2. Chain `.handle_if()` calls for each type you want to try -/// 3. End with `.otherwise()` for messages that don't match any type -/// -/// # Example -/// -/// ``` -/// # use agent_client_protocol::UntypedMessage; -/// # use agent_client_protocol::schema::v1::SessionNotification; -/// # use agent_client_protocol::util::TypeNotification; -/// # async fn example(message: UntypedMessage) -> Result<(), agent_client_protocol::Error> { -/// TypeNotification::new(message) -/// .handle_if(|notif: SessionNotification| async move { -/// // Handle session notifications -/// println!("Session update: {:?}", notif); -/// Ok(()) -/// }) -/// .await -/// .otherwise(|untyped: UntypedMessage| async move { -/// // Fallback for unrecognized notifications -/// println!("Unknown notification: {}", untyped.method); -/// Ok(()) -/// }) -/// .await -/// # } -/// ``` -/// -/// Since notifications don't expect responses, handlers only receive the parsed -/// notification (not a request context). -#[must_use] -#[derive(Debug)] -pub struct TypeNotification { - state: Option, -} - -#[derive(Debug)] -enum TypeNotificationState { - Unhandled(String, serde_json::Value), - Handled(Result<(), crate::Error>), -} - -impl TypeNotification { - /// Create a new pattern matcher for the given untyped notification message. - pub fn new(request: UntypedMessage) -> Self { - let UntypedMessage { method, params } = request; - Self { - state: Some(TypeNotificationState::Unhandled(method, params)), - } - } - - /// Try to handle the message as type `N`. - /// - /// If the message can be parsed as `N`, the handler `op` is called with the parsed - /// notification. If parsing fails, the malformed matching notification is consumed - /// and logged without invoking the handler or sending a response. If the message was - /// already handled by a previous `handle_if`, this call has no effect. - /// - /// Returns `self` to allow chaining multiple `handle_if` calls. - pub async fn handle_if( - mut self, - op: impl AsyncFnOnce(N) -> Result<(), crate::Error>, - ) -> Self { - self.state = Some(match self.state.take().expect("valid state") { - TypeNotificationState::Unhandled(method, params) => { - if N::matches_method(&method) { - match N::parse_message(&method, ¶ms) { - Ok(request) => TypeNotificationState::Handled(op(request).await), - Err(error) => { - tracing::warn!( - error = ?crate::util::redacted_error(error.code), - method, - "Invalid notification params; ignoring notification without replying" - ); - TypeNotificationState::Handled(Ok(())) - } - } - } else { - TypeNotificationState::Unhandled(method, params) - } - } - - TypeNotificationState::Handled(err) => TypeNotificationState::Handled(err), - }); - self - } - - /// Handle messages that didn't match any previous `handle_if` call. - /// - /// This is the fallback handler that receives the original untyped message if none - /// of the typed handlers matched. You must call this method to complete the pattern - /// matching chain and get the final result. - pub async fn otherwise( - mut self, - op: impl AsyncFnOnce(UntypedMessage) -> Result<(), crate::Error>, - ) -> Result<(), crate::Error> { - match self.state.take().expect("valid state") { - TypeNotificationState::Unhandled(method, params) => { - match UntypedMessage::new(&method, params) { - Ok(m) => op(m).await, - Err(error) => { - tracing::warn!( - error = ?crate::util::redacted_error(error.code), - method, - "Invalid untyped notification; ignoring notification without replying" - ); - Ok(()) - } - } - } - TypeNotificationState::Handled(r) => r, - } - } -} - -#[cfg(test)] -mod type_notification_tests { - use std::{cell::Cell, rc::Rc}; - - use serde::{Deserialize, Serialize}; - - use super::{TypeNotification, UntypedMessage}; - - #[derive(Clone, Debug, Serialize, Deserialize)] - struct TestNotification { - required: String, - } - - impl crate::JsonRpcMessage for TestNotification { - fn matches_method(method: &str) -> bool { - method == "test/notification" - } - - fn method(&self) -> &'static str { - "test/notification" - } - - fn to_untyped_message(&self) -> Result { - UntypedMessage::new(self.method(), self) - } - - fn parse_message(method: &str, params: &impl Serialize) -> Result { - if !Self::matches_method(method) { - return Err(crate::Error::method_not_found()); - } - crate::util::json_cast_params(params) - } - } - - impl crate::JsonRpcNotification for TestNotification {} - - #[test] - fn invalid_matching_notification_is_consumed_without_fallback() { - futures::executor::block_on(async { - let handler_called = Rc::new(Cell::new(false)); - let fallback_called = Rc::new(Cell::new(false)); - let handler_called_in_callback = Rc::clone(&handler_called); - let fallback_called_in_callback = Rc::clone(&fallback_called); - let message = - UntypedMessage::new("test/notification", serde_json::json!({ "wrong": "field" })) - .expect("test notification should serialize"); - - TypeNotification::new(message) - .handle_if(move |_: TestNotification| async move { - handler_called_in_callback.set(true); - Ok(()) - }) - .await - .otherwise(move |_| async move { - fallback_called_in_callback.set(true); - Ok(()) - }) - .await - .expect("invalid notification should be ignored"); - - assert!(!handler_called.get()); - assert!(!fallback_called.get()); - }); - } -} diff --git a/src/agent-client-protocol/tests/jsonrpc_connection_builder.rs b/src/agent-client-protocol/tests/jsonrpc_connection_builder.rs index b819bfa1..fb3ca1e3 100644 --- a/src/agent-client-protocol/tests/jsonrpc_connection_builder.rs +++ b/src/agent-client-protocol/tests/jsonrpc_connection_builder.rs @@ -201,9 +201,8 @@ async fn test_multiple_handlers_different_methods() { .await .map_err( |e| -> agent_client_protocol::Error { - agent_client_protocol::util::internal_error(format!( - "Foo request failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Foo request failed: {e:?}")) }, )?; assert_eq!(foo_response.result, "foo: test1"); @@ -215,9 +214,8 @@ async fn test_multiple_handlers_different_methods() { .await .map_err( |e| -> agent_client_protocol::Error { - agent_client_protocol::util::internal_error(format!( - "Bar request failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Bar request failed: {e:?}")) }, )?; assert_eq!(bar_response.result, "bar: test2"); @@ -337,9 +335,8 @@ async fn test_handler_priority_ordering() { })) .await .map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Track request failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Track request failed: {e:?}")) })?; // First handler should have handled it @@ -501,9 +498,8 @@ async fn test_fallthrough_behavior() { })) .await .map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Method2 request failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Method2 request failed: {e:?}")) })?; assert_eq!(response.result, "method2: fallthrough"); @@ -674,9 +670,8 @@ async fn test_handler_claims_notification() { event: "test_event".to_string(), }) .map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Failed to send notification: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Failed to send notification: {e:?}")) })?; // Give server time to process diff --git a/src/agent-client-protocol/tests/jsonrpc_hello.rs b/src/agent-client-protocol/tests/jsonrpc_hello.rs index f454b7ef..90233d94 100644 --- a/src/agent-client-protocol/tests/jsonrpc_hello.rs +++ b/src/agent-client-protocol/tests/jsonrpc_hello.rs @@ -146,9 +146,8 @@ async fn test_hello_world() { }; let response = recv(cx.send_request(request)).await.map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Request failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Request failed: {e:?}")) })?; assert_eq!(response.echo, "pong: hello world"); @@ -242,18 +241,16 @@ async fn test_notification() { message: "test log 1".to_string(), }) .map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Failed to send notification: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Failed to send notification: {e:?}")) })?; cx.send_notification(LogNotification { message: "test log 2".to_string(), }) .map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Failed to send notification: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Failed to send notification: {e:?}")) })?; // Give the server time to process notifications @@ -319,9 +316,8 @@ async fn test_multiple_sequential_requests() { }; let response = recv(cx.send_request(request)).await.map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Request {i} failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Request {i} failed: {e:?}")) })?; assert_eq!(response.echo, format!("pong: message {i}")); @@ -390,9 +386,8 @@ async fn test_concurrent_requests() { // Now await all responses for (i, response_future) in responses { let response = recv(response_future).await.map_err(|e| { - agent_client_protocol::util::internal_error(format!( - "Request {i} failed: {e:?}" - )) + agent_client_protocol::Error::internal_error() + .data(format!("Request {i} failed: {e:?}")) })?; assert_eq!(response.echo, format!("pong: concurrent message {i}")); diff --git a/src/agent-client-protocol/tests/jsonrpc_request_cancellation.rs b/src/agent-client-protocol/tests/jsonrpc_request_cancellation.rs index 429bba99..29e38824 100644 --- a/src/agent-client-protocol/tests/jsonrpc_request_cancellation.rs +++ b/src/agent-client-protocol/tests/jsonrpc_request_cancellation.rs @@ -421,6 +421,57 @@ async fn wrapped_cancel_request_cancels_wrapped_request() { .await; } +#[tokio::test(flavor = "current_thread")] +async fn malformed_successor_request_is_rejected_as_invalid_params() { + use tokio::io::{AsyncWriteExt, BufReader}; + use tokio::task::LocalSet; + + let local = LocalSet::new(); + + local + .run_until(async { + let (mut client_writer, server_reader) = tokio::io::duplex(4096); + let (server_writer, client_reader) = tokio::io::duplex(4096); + + let server_transport = agent_client_protocol::ByteStreams::new( + server_writer.compat_write(), + server_reader.compat(), + ); + let server = Builder::new(WrappedHost).on_receive_request_from( + WrappedSuccessor, + async |_request: SimpleRequest, + _responder: Responder, + _cx: ConnectionTo| + -> Result<(), agent_client_protocol::Error> { + panic!("a malformed envelope must not reach the handler") + }, + agent_client_protocol::on_receive_request!(), + ); + + tokio::task::spawn_local(async move { + drop(server.connect_to(server_transport).await); + }); + + let mut client_reader = BufReader::new(client_reader); + + // The envelope's params have no inner `method`, so they are + // invalid params for `proxy/successor`, not a parse error. + client_writer + .write_all( + br#"{"jsonrpc":"2.0","id":7,"method":"proxy/successor","params":{"bogus":true}} +"#, + ) + .await + .unwrap(); + client_writer.flush().await.unwrap(); + + let response = read_jsonrpc_response_line(&mut client_reader).await; + assert_eq!(response["id"], 7); + assert_eq!(response["error"]["code"], -32602, "{response}"); + }) + .await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn prepared_cancellation_handle_sends_successor_wrapped_cancel_and_delivers_callback() { use agent_client_protocol::{RawJsonRpcMessage, TransportFrame, UntypedMessage}; diff --git a/src/agent-client-protocol/tests/log_privacy.rs b/src/agent-client-protocol/tests/log_privacy.rs index 94734473..2982f956 100644 --- a/src/agent-client-protocol/tests/log_privacy.rs +++ b/src/agent-client-protocol/tests/log_privacy.rs @@ -6,8 +6,8 @@ use std::{ }; use agent_client_protocol::{ - ConnectionTo, Error, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, - Responder, UntypedMessage, role::UntypedRole, util::TypeNotification, + ConnectionTo, Error, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, Responder, + UntypedMessage, role::UntypedRole, }; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -78,46 +78,6 @@ struct Request { #[derive(Clone, Debug, Serialize, Deserialize, JsonRpcResponse)] struct Response {} -#[test] -fn type_notification_warnings_do_not_echo_private_params_or_parse_reasons() { - let capture = LogCapture::new(); - tracing::subscriber::with_default(capture.subscriber(), || { - futures::executor::block_on(async { - for params in [ - json!({ "extra": PRIVATE_MARKER }), - json!({ "required": PRIVATE_MARKER, "kind": "Known" }), - json!({ "required": 1, "kind": PRIVATE_MARKER }), - ] { - // The full error remains available to callers; neither its JSON - // nor a Serde reason quoting an invalid value is safe for WARN. - let error = - Notice::parse_message("_test/private_notification", ¶ms).unwrap_err(); - assert!(format!("{error:?}").contains(PRIVATE_MARKER)); - - TypeNotification::new( - UntypedMessage::new("_test/private_notification", params).unwrap(), - ) - .handle_if(async |_: Notice| { - panic!("malformed notification must not reach its handler"); - }) - .await - .otherwise(async |_| { - panic!("malformed matching notification must not reach fallback"); - }) - .await - .unwrap(); - } - }); - }); - - let logs = capture.text(); - assert_eq!(logs.matches("Invalid notification params;").count(), 3); - assert!(logs.contains("_test/private_notification")); - assert!(logs.contains("-32602")); - assert!(logs.contains("Invalid params")); - assert!(!logs.contains(PRIVATE_MARKER), "{logs}"); -} - fn private_handler_error() -> Error { // Both fields are application-controlled, even for a standard error code. Error::new(-32602, PRIVATE_MARKER).data(json!({ "private": PRIVATE_MARKER })) diff --git a/src/agent-client-protocol/tests/mcp_message_deserialization.rs b/src/agent-client-protocol/tests/mcp_message_deserialization.rs index 841a1260..d8304831 100644 --- a/src/agent-client-protocol/tests/mcp_message_deserialization.rs +++ b/src/agent-client-protocol/tests/mcp_message_deserialization.rs @@ -76,7 +76,7 @@ async fn unresolved_server_is_a_binding_failure_not_a_peer_mcp_error() { .await .expect_err("an unresolved server must fail the outer ACP operation"); assert_eq!( - i32::from(response.code), + response.code, agent_client_protocol::mcp_server::MCP_SERVER_UNAVAILABLE ); Ok::<(), Error>(()) diff --git a/src/agent-client-protocol/tests/meta_propagation.rs b/src/agent-client-protocol/tests/meta_propagation.rs index 9bf9eaa2..c07a2ccb 100644 --- a/src/agent-client-protocol/tests/meta_propagation.rs +++ b/src/agent-client-protocol/tests/meta_propagation.rs @@ -29,10 +29,6 @@ fn prompt_request(meta: Meta) -> PromptRequest { .meta(meta) } -fn trace_context_meta_value() -> Value { - Value::Object(trace_context_meta()) -} - #[test] fn prompt_request_meta_round_trips_with_root_trace_context_keys() -> Result<(), agent_client_protocol::Error> { @@ -54,17 +50,18 @@ fn prompt_request_meta_round_trips_with_root_trace_context_keys() #[test] fn successor_message_meta_serializes_as_reserved_meta_field() -> Result<(), agent_client_protocol::Error> { - let envelope_meta = trace_context_meta_value(); + let envelope_meta = trace_context_meta(); let inner_meta = trace_context_meta(); - let message = SuccessorMessage { - message: prompt_request(inner_meta.clone()), - meta: Some(envelope_meta.clone()), - }; + let message = + SuccessorMessage::new(prompt_request(inner_meta.clone())).meta(envelope_meta.clone()); let untyped = message.to_untyped_message()?; assert_eq!(untyped.method(), "proxy/successor"); - assert_eq!(untyped.params()["_meta"], envelope_meta); + assert_eq!( + untyped.params()["_meta"], + Value::Object(envelope_meta.clone()) + ); assert_eq!( untyped.params()["params"]["_meta"], Value::Object(inner_meta.clone()) @@ -80,14 +77,11 @@ fn successor_message_meta_serializes_as_reserved_meta_field() } #[test] -fn successor_message_accepts_legacy_meta_alias() -> Result<(), agent_client_protocol::Error> { - let envelope_meta = trace_context_meta_value(); - let mut params = SuccessorMessage { - message: prompt_request(Meta::new()), - meta: Some(envelope_meta.clone()), - } - .to_untyped_message()? - .params; +fn successor_message_ignores_non_spec_meta_key() -> Result<(), agent_client_protocol::Error> { + let mut params = SuccessorMessage::new(prompt_request(Meta::new())) + .meta(trace_context_meta()) + .to_untyped_message()? + .params; let params_object = params.as_object_mut().expect("params should be an object"); let meta = params_object @@ -96,11 +90,27 @@ fn successor_message_accepts_legacy_meta_alias() -> Result<(), agent_client_prot params_object.insert("meta".into(), meta); let parsed = SuccessorMessage::::parse_message("proxy/successor", ¶ms)?; - assert_eq!(parsed.meta, Some(envelope_meta)); + assert_eq!(parsed.meta, None); Ok(()) } +#[test] +fn successor_message_ignores_non_object_meta() { + let mut params = SuccessorMessage::new(prompt_request(Meta::new())) + .to_untyped_message() + .expect("successor message should serialize") + .params; + params + .as_object_mut() + .expect("params should be an object") + .insert("_meta".into(), Value::String("not an object".into())); + + let parsed = SuccessorMessage::::parse_message("proxy/successor", ¶ms) + .expect("a non-object `_meta` should be ignored"); + assert_eq!(parsed.meta, None); +} + #[test] #[cfg(feature = "unstable_mcp_over_acp")] fn native_mcp_over_acp_message_meta_serializes_as_reserved_meta_field() diff --git a/src/agent-client-protocol/tests/protocol_v2.rs b/src/agent-client-protocol/tests/protocol_v2.rs index 9d6cd6f4..8fcbc918 100644 --- a/src/agent-client-protocol/tests/protocol_v2.rs +++ b/src/agent-client-protocol/tests/protocol_v2.rs @@ -446,9 +446,8 @@ impl ConnectTo for FutureInitializeV2Client { if let Some(task) = agent_task { task.abort(); } - Err(agent_client_protocol::util::internal_error( - "v2 agent did not respond to initialize", - )) + Err(agent_client_protocol::Error::internal_error() + .data("v2 agent did not respond to initialize")) } } @@ -536,9 +535,8 @@ async fn assert_malformed_initialize_rejected(params: Map) -> Res } agent_task.abort(); - Err(agent_client_protocol::util::internal_error( - "agent did not respond to malformed initialize", - )) + Err(agent_client_protocol::Error::internal_error() + .data("agent did not respond to malformed initialize")) } async fn assert_v2_client_rejects_v1_selection(agent: impl ConnectTo) -> Result<(), Error> { @@ -2374,9 +2372,8 @@ async fn protocol_router_v2_only_rejects_v1_client() -> Result<(), Error> { } agent_task.abort(); - Err(agent_client_protocol::util::internal_error( - "protocol router did not reject v1 initialize", - )) + Err(agent_client_protocol::Error::internal_error() + .data("protocol router did not reject v1 initialize")) } #[tokio::test(flavor = "current_thread")] @@ -2437,7 +2434,7 @@ async fn protocol_router_rejection_flushes_over_byte_streams() -> Result<(), Err agent_task .await - .map_err(agent_client_protocol::util::internal_error)??; + .map_err(agent_client_protocol::Error::into_internal_error)??; Ok(()) } @@ -2495,7 +2492,7 @@ async fn protocol_router_rejection_flushes_with_trailing_malformed_input() -> Re agent_task .await - .map_err(agent_client_protocol::util::internal_error)??; + .map_err(agent_client_protocol::Error::into_internal_error)??; Ok(()) } @@ -2549,7 +2546,7 @@ async fn protocol_router_ignores_retained_standalone_response_before_initialize( .map_err(Error::into_internal_error)?; agent_task .await - .map_err(agent_client_protocol::util::internal_error)??; + .map_err(agent_client_protocol::Error::into_internal_error)??; Ok(()) } @@ -2596,7 +2593,7 @@ async fn protocol_router_preserves_same_version_initialize_frames() -> Result<() .map_err(Error::into_internal_error)?; router_task .await - .map_err(agent_client_protocol::util::internal_error)??; + .map_err(agent_client_protocol::Error::into_internal_error)??; } Ok(()) @@ -2786,9 +2783,8 @@ async fn protocol_router_routes_future_protocol_version_to_v2() -> Result<(), Er } agent_task.abort(); - Err(agent_client_protocol::util::internal_error( - "protocol router did not respond to initialize", - )) + Err(agent_client_protocol::Error::internal_error() + .data("protocol router did not respond to initialize")) } #[tokio::test(flavor = "current_thread")] diff --git a/src/agent-client-protocol/tests/schema_v1_enum_methods.rs b/src/agent-client-protocol/tests/schema_v1_enum_methods.rs new file mode 100644 index 00000000..a2b6e615 --- /dev/null +++ b/src/agent-client-protocol/tests/schema_v1_enum_methods.rs @@ -0,0 +1,46 @@ +//! Method names reported by the v1 schema dispatch enums. + +use agent_client_protocol::{Error, ErrorCode, JsonRpcMessage, schema::v1}; +use serde_json::json; + +#[test] +fn v1_extension_enum_parsing_preserves_method_prefix() -> Result<(), Error> { + let params = json!({ "payload": true }); + + let request = v1::ClientRequest::parse_message("_vendor/request", ¶ms)?; + assert_eq!(request.method(), "_vendor/request"); + let untyped_request = request.to_untyped_message()?; + assert_eq!(untyped_request.method(), "_vendor/request"); + assert_eq!(untyped_request.params(), ¶ms); + + let request = v1::AgentRequest::parse_message("_vendor/request", ¶ms)?; + assert_eq!(request.method(), "_vendor/request"); + + let notification = v1::AgentNotification::parse_message("_vendor/notify", ¶ms)?; + assert_eq!(notification.method(), "_vendor/notify"); + let untyped_notification = notification.to_untyped_message()?; + assert_eq!(untyped_notification.method(), "_vendor/notify"); + assert_eq!(untyped_notification.params(), ¶ms); + + let notification = v1::ClientNotification::parse_message("_vendor/notify", ¶ms)?; + assert_eq!(notification.method(), "_vendor/notify"); + + Ok(()) +} + +#[test] +fn v1_enum_parsing_rejects_unknown_non_extension_methods() { + let error = v1::ClientRequest::parse_message("vendor/request", &json!({})).unwrap_err(); + assert_eq!(error.code, ErrorCode::METHOD_NOT_FOUND); +} + +#[test] +fn v1_enum_method_matches_schema_method_names() -> Result<(), Error> { + let request = v1::ClientRequest::PromptRequest(v1::PromptRequest::new( + v1::SessionId::new("session-1"), + vec![], + )); + assert_eq!(JsonRpcMessage::method(&request), request.method()); + assert_eq!(request.to_untyped_message()?.method(), "session/prompt"); + Ok(()) +} diff --git a/src/agent-client-protocol/tests/session_ordering.rs b/src/agent-client-protocol/tests/session_ordering.rs index 0c5714cd..b63b0763 100644 --- a/src/agent-client-protocol/tests/session_ordering.rs +++ b/src/agent-client-protocol/tests/session_ordering.rs @@ -545,15 +545,10 @@ async fn v2_proxy_session_start_installs_routing_before_later_message() { forwarded.id, Ok(serde_json::to_value(setup_response).expect("session response should serialize")), ); - let update = SuccessorMessage { - message: v2::UpdateSessionNotification::new( - notification_session_id, - v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running( - v2::RunningStateUpdate::new(), - )), - ), - meta: None, - } + let update = SuccessorMessage::new(v2::UpdateSessionNotification::new( + notification_session_id, + v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running(v2::RunningStateUpdate::new())), + )) .to_untyped_message()?; let (method, params) = update.into_parts(); let notification = RawJsonRpcMessage::notification(method, params)?; @@ -700,15 +695,10 @@ async fn v2_proxy_fork_installs_response_id_routing_before_later_message() { forwarded.id, Ok(serde_json::to_value(setup_response).expect("fork response should serialize")), ); - let update = SuccessorMessage { - message: v2::UpdateSessionNotification::new( - notification_session_id, - v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running( - v2::RunningStateUpdate::new(), - )), - ), - meta: None, - } + let update = SuccessorMessage::new(v2::UpdateSessionNotification::new( + notification_session_id, + v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running(v2::RunningStateUpdate::new())), + )) .to_untyped_message()?; let (method, params) = update.into_parts(); let notification = RawJsonRpcMessage::notification(method, params)?; @@ -848,15 +838,10 @@ async fn v2_proxy_resume_forwards_replay_before_its_response() { )?; assert_eq!(successor.message.session_id, session_id); - let update = SuccessorMessage { - message: v2::UpdateSessionNotification::new( - notification_session_id, - v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running( - v2::RunningStateUpdate::new(), - )), - ), - meta: None, - } + let update = SuccessorMessage::new(v2::UpdateSessionNotification::new( + notification_session_id, + v2::SessionUpdate::StateUpdate(v2::StateUpdate::Running(v2::RunningStateUpdate::new())), + )) .to_untyped_message()?; let (method, params) = update.into_parts(); let notification = RawJsonRpcMessage::notification(method, params)?;