fix(acp): don't let peer errors match the SDK's termination predicates

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.

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.
 //!

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<serde_json::Value>,
+    /// 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,
                 })
             })
         })

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,

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,

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));
+}