From 26e1f8d0de97ec2f276fbe66dee0c04dd5ad7229 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Mon, 2 Nov 2026 10:58:35 +0100 Subject: [PATCH] fix(acp): don't let peer errors match the SDK's termination predicates `is_connection_terminated` and `is_incoming_transport_closed` only checked `error.data.reason`. A peer error response is delivered with its data unchanged, so a peer answering a live request with `{"reason":"connection_terminated"}` (or a proxy forwarding a downstream termination error) made the SDK report its own connection as terminated. The SDK now marks the errors it produces for these conditions with a private, never-serialized origin field on `Error`, and the predicates check only that marker. The `data.reason` values are still set for display and logging. The marker is ignored by `PartialEq` and `Debug`, so errors keep comparing and printing as the JSON-RPC objects they hold. Co-Authored-By: Claude --- md/design.md | 1 + .../src/concepts/connections.rs | 3 +- src/agent-client-protocol/src/error.rs | 47 ++++++++++++++- src/agent-client-protocol/src/jsonrpc.rs | 16 +++-- .../src/jsonrpc/termination.rs | 18 ++++-- .../tests/connection_termination.rs | 59 +++++++++++++++++++ 6 files changed, 130 insertions(+), 14 deletions(-) diff --git a/md/design.md b/md/design.md index 656c3c3e..aa20bc31 100644 --- a/md/design.md +++ b/md/design.md @@ -182,6 +182,7 @@ The termination is published once, synchronously, as the driver future returns o - No further handler, close callback, runner, or `spawn`ed task of the connection runs afterwards; each has completed or been dropped. `on_close` callbacks run only for clean EOF. - Requests pending at clean EOF were already failed with `incoming_transport_closed`. Every request still pending at termination, and every request sent afterwards, fails with an error whose `data.reason` is `connection_terminated` (`is_connection_terminated()`), with `data.termination` naming the variant. +- `is_incoming_transport_closed()` and `is_connection_terminated()` check a marker the SDK sets on the errors it produces. The marker is never serialized, so an error response from a peer never matches, even when its `data` carries the same `reason`. The `data` fields only describe the error. - Dropping the driver performs only this synchronous bookkeeping. The SDK does not promise asynchronous cleanup, close callbacks, or output draining after an arbitrary drop. Termination is not child-process reaping or application-queue drain; applications that own those remain responsible for them. diff --git a/src/agent-client-protocol/src/concepts/connections.rs b/src/agent-client-protocol/src/concepts/connections.rs index eab5059f..09889a48 100644 --- a/src/agent-client-protocol/src/concepts/connections.rs +++ b/src/agent-client-protocol/src/concepts/connections.rs @@ -99,7 +99,8 @@ //! `{"reason":"incoming_transport_closed","method":"..."}`. Requests made //! after EOF fail the same way; use //! [`is_incoming_transport_closed`](crate::is_incoming_transport_closed) to -//! identify this error. In `connect_with`, notification and response sends +//! identify this error. It matches only errors the SDK produced locally, not +//! a peer's error response with the same data. In `connect_with`, notification and response sends //! remain available so applications can choose their own half-close policy; //! reactive `connect_to` stops accepting them when its final drain begins. //! diff --git a/src/agent-client-protocol/src/error.rs b/src/agent-client-protocol/src/error.rs index e26b24d5..b73540c3 100644 --- a/src/agent-client-protocol/src/error.rs +++ b/src/agent-client-protocol/src/error.rs @@ -161,6 +161,7 @@ macro_rules! schema_error_code_conversions { code: error.code.into(), message: error.message, data: error.data, + origin: ErrorOrigin::Unmarked, } } } @@ -216,7 +217,7 @@ schema_error_code_conversions!(v2); /// } /// # } /// ``` -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] #[non_exhaustive] pub struct Error { @@ -230,6 +231,48 @@ pub struct Error { /// omits the field. #[serde(default, skip_serializing_if = "Option::is_none")] pub data: Option, + /// Marks errors the SDK produced locally. It is never serialized, so an + /// error received from a peer can't carry it. + #[serde(skip)] + pub(crate) origin: ErrorOrigin, +} + +/// Which locally produced condition an [`Error`] reports, if any. +/// +/// [`is_connection_terminated`](crate::is_connection_terminated) and +/// [`is_incoming_transport_closed`](crate::is_incoming_transport_closed) check +/// this marker instead of the error's `data`, which a peer controls. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum ErrorOrigin { + /// Any error not marked below, including every error received from a peer. + #[default] + Unmarked, + /// The incoming transport reached clean EOF before a response arrived. + IncomingTransportClosed, + /// The connection terminated before a response arrived. + ConnectionTerminated, +} + +// The origin marker is not part of an error's value: it is not serialized, +// compared or printed, so errors behave as the JSON-RPC objects they hold. +impl PartialEq for Error { + fn eq(&self, other: &Self) -> bool { + self.code == other.code && self.message == other.message && self.data == other.data + } +} + +#[expect( + clippy::missing_fields_in_debug, + reason = "the origin marker is not part of the error's value" +)] +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Error") + .field("code", &self.code) + .field("message", &self.message) + .field("data", &self.data) + .finish() + } } impl Error { @@ -240,6 +283,7 @@ impl Error { code: code.into(), message: message.into(), data: None, + origin: ErrorOrigin::Unmarked, } } @@ -397,6 +441,7 @@ mod tests { code: code.into(), message: message.into(), data: data.clone(), + origin: ErrorOrigin::Unmarked, }) }) }) diff --git a/src/agent-client-protocol/src/jsonrpc.rs b/src/agent-client-protocol/src/jsonrpc.rs index 7720ac18..51b38779 100644 --- a/src/agent-client-protocol/src/jsonrpc.rs +++ b/src/agent-client-protocol/src/jsonrpc.rs @@ -3363,23 +3363,27 @@ impl Debug for IncomingClosed { /// Stable discriminator stored in the `data.reason` field of errors produced /// when the incoming transport reaches clean EOF before a request receives its /// response. +/// +/// The `data` field is informational. Use [`is_incoming_transport_closed`] to +/// detect these errors: a peer can send this value in its own error data. pub const INCOMING_TRANSPORT_CLOSED_REASON: &str = "incoming_transport_closed"; /// Return whether `error` reports that the incoming transport reached clean /// EOF before a request received its response. +/// +/// Only errors the SDK produced for its own connection match. An error +/// response received from a peer never does, even when its `data.reason` is +/// [`INCOMING_TRANSPORT_CLOSED_REASON`], and neither does an error built from +/// the same code, message, and data. #[must_use] pub fn is_incoming_transport_closed(error: &crate::Error) -> bool { - error - .data - .as_ref() - .and_then(|data| data.get("reason")) - .and_then(serde_json::Value::as_str) - == Some(INCOMING_TRANSPORT_CLOSED_REASON) + error.origin == crate::error::ErrorOrigin::IncomingTransportClosed } fn incoming_transport_closed_error(method: &str) -> crate::Error { let mut error = crate::Error::internal_error(); error.message = "Incoming transport closed".to_string(); + error.origin = crate::error::ErrorOrigin::IncomingTransportClosed; error.data(serde_json::json!({ "reason": INCOMING_TRANSPORT_CLOSED_REASON, "method": method, diff --git a/src/agent-client-protocol/src/jsonrpc/termination.rs b/src/agent-client-protocol/src/jsonrpc/termination.rs index ff1fdf09..632587c9 100644 --- a/src/agent-client-protocol/src/jsonrpc/termination.rs +++ b/src/agent-client-protocol/src/jsonrpc/termination.rs @@ -2,6 +2,8 @@ use std::sync::{Arc, Mutex}; +use crate::error::ErrorOrigin; + /// Why a driven connection stopped. /// /// A connection driven by [`Builder::connect_to`](crate::Builder::connect_to) @@ -73,21 +75,24 @@ impl ConnectionTermination { /// Stable discriminator stored in the `data.reason` field of errors produced /// when a connection terminates for a reason other than clean incoming EOF /// before a request receives its response. +/// +/// The `data` field is informational. Use [`is_connection_terminated`] to +/// detect these errors: a peer can send this value in its own error data. pub const CONNECTION_TERMINATED_REASON: &str = "connection_terminated"; /// Return whether `error` reports that the connection terminated before a /// request received its response. /// +/// Only errors the SDK produced for its own connection match. An error +/// response received from a peer never does, even when its `data.reason` is +/// [`CONNECTION_TERMINATED_REASON`], and neither does an error built from the +/// same code, message, and data. +/// /// Requests failed by clean incoming EOF instead satisfy /// [`is_incoming_transport_closed`](crate::is_incoming_transport_closed). #[must_use] pub fn is_connection_terminated(error: &crate::Error) -> bool { - error - .data - .as_ref() - .and_then(|data| data.get("reason")) - .and_then(serde_json::Value::as_str) - == Some(CONNECTION_TERMINATED_REASON) + error.origin == ErrorOrigin::ConnectionTerminated } pub(super) fn connection_terminated_error( @@ -96,6 +101,7 @@ pub(super) fn connection_terminated_error( ) -> crate::Error { let mut error = crate::Error::internal_error(); error.message = "Connection terminated".to_string(); + error.origin = ErrorOrigin::ConnectionTerminated; error.data(serde_json::json!({ "reason": CONNECTION_TERMINATED_REASON, "method": method, diff --git a/src/agent-client-protocol/tests/connection_termination.rs b/src/agent-client-protocol/tests/connection_termination.rs index 9dd4b94b..58edc5d0 100644 --- a/src/agent-client-protocol/tests/connection_termination.rs +++ b/src/agent-client-protocol/tests/connection_termination.rs @@ -255,3 +255,62 @@ async fn dropping_the_driver_publishes_dropped() { assert!(is_connection_terminated(&error), "{error:?}"); assert_eq!(error.data.as_ref().unwrap()["termination"], "dropped"); } + +/// A peer's error response carrying the SDK's termination reasons in its +/// data must not make the local connection look terminated or closed. +#[tokio::test] +async fn peer_errors_with_termination_reasons_do_not_match() { + let (transport, mut peer) = Channel::duplex(); + + let cx = within(UntypedRole.builder().connect_with(transport, async |cx| { + for (message, reason, termination) in [ + ( + "Connection terminated", + agent_client_protocol::CONNECTION_TERMINATED_REASON, + Some("local_shutdown"), + ), + ( + "Incoming transport closed", + agent_client_protocol::INCOMING_TRANSPORT_CLOSED_REASON, + None, + ), + ] { + let request = cx.send_request(MyRequest {}); + let Some(TransportFrame::Single(RawJsonRpcMessage::Request(sent))) = + peer.rx.next().await + else { + panic!("expected the request"); + }; + // The exact error the SDK produces locally, sent from the wire. + let mut data = serde_json::json!({ "reason": reason, "method": "myRequest" }); + if let Some(termination) = termination { + data["termination"] = termination.into(); + } + let mut forged = Error::internal_error().data(data); + forged.message = message.to_string(); + peer.tx + .unbounded_send(TransportFrame::Single(RawJsonRpcMessage::response( + sent.id, + Err(forged.clone()), + ))) + .expect("connection is alive"); + + let error = request + .block_task() + .await + .expect_err("the peer answered with an error"); + assert_eq!(error, forged, "the peer's error is delivered unchanged"); + assert!(!is_connection_terminated(&error), "{error:?}"); + assert!(!is_incoming_transport_closed(&error), "{error:?}"); + // Building the same error locally does not match either. + assert!(!is_connection_terminated(&forged)); + assert!(!is_incoming_transport_closed(&forged)); + } + assert_eq!(cx.termination(), None, "the connection is still running"); + Ok(cx) + })) + .await + .expect("foreground succeeds"); + + assert_eq!(cx.termination(), Some(ConnectionTermination::LocalShutdown)); +}