agent_client_protocol/jsonrpc/
termination.rs

1//! Terminal lifecycle event for driven connections.
2
3use std::sync::{Arc, Mutex};
4
5/// Why a driven connection stopped.
6///
7/// A connection driven by [`Builder::connect_to`](crate::Builder::connect_to)
8/// or [`Builder::connect_with`](crate::Builder::connect_with) publishes
9/// exactly one termination, observable through
10/// [`ConnectionTo::terminated`](crate::ConnectionTo::terminated) and
11/// [`ConnectionTo::termination`](crate::ConnectionTo::termination).
12///
13/// Unlike [`ConnectionTo::incoming_closed`](crate::ConnectionTo::incoming_closed),
14/// which only reports clean incoming EOF, termination covers every way the
15/// connection driver can end.
16#[derive(Clone, Debug, PartialEq, Eq)]
17#[non_exhaustive]
18pub enum ConnectionTermination {
19    /// The incoming transport reached clean EOF and the driver finished
20    /// successfully.
21    ///
22    /// This is also reported when a `connect_with` foreground returns `Ok`
23    /// after observing clean EOF.
24    IncomingClosed,
25    /// The `connect_with` foreground returned `Ok` before incoming EOF,
26    /// initiating a local shutdown that completed successfully.
27    LocalShutdown,
28    /// The `connect_with` foreground returned an error.
29    ForegroundError(crate::Error),
30    /// The transport driver failed to read, write, or drain.
31    TransportError(crate::Error),
32    /// A message handler, close callback, runner, or spawned task failed.
33    HandlerError(crate::Error),
34    /// The driver future was dropped before it completed.
35    ///
36    /// Only synchronous bookkeeping happens on drop: pending requests are
37    /// failed and this termination is published. No asynchronous cleanup,
38    /// close callbacks, or output draining run after an arbitrary drop.
39    Dropped,
40}
41
42impl ConnectionTermination {
43    /// The error that ended the connection, if any.
44    #[must_use]
45    pub fn error(&self) -> Option<&crate::Error> {
46        match self {
47            Self::ForegroundError(error)
48            | Self::TransportError(error)
49            | Self::HandlerError(error) => Some(error),
50            Self::IncomingClosed | Self::LocalShutdown | Self::Dropped => None,
51        }
52    }
53
54    /// Whether the driver completed without error.
55    #[must_use]
56    pub fn is_clean(&self) -> bool {
57        matches!(self, Self::IncomingClosed | Self::LocalShutdown)
58    }
59
60    /// Payload-free discriminator stored in pending-request errors.
61    pub(super) fn kind(&self) -> &'static str {
62        match self {
63            Self::IncomingClosed => "incoming_closed",
64            Self::LocalShutdown => "local_shutdown",
65            Self::ForegroundError(_) => "foreground_error",
66            Self::TransportError(_) => "transport_error",
67            Self::HandlerError(_) => "handler_error",
68            Self::Dropped => "dropped",
69        }
70    }
71}
72
73/// Stable discriminator stored in the `data.reason` field of errors produced
74/// when a connection terminates for a reason other than clean incoming EOF
75/// before a request receives its response.
76pub const CONNECTION_TERMINATED_REASON: &str = "connection_terminated";
77
78/// Return whether `error` reports that the connection terminated before a
79/// request received its response.
80///
81/// Requests failed by clean incoming EOF instead satisfy
82/// [`is_incoming_transport_closed`](crate::is_incoming_transport_closed).
83#[must_use]
84pub fn is_connection_terminated(error: &crate::Error) -> bool {
85    error
86        .data
87        .as_ref()
88        .and_then(|data| data.get("reason"))
89        .and_then(serde_json::Value::as_str)
90        == Some(CONNECTION_TERMINATED_REASON)
91}
92
93pub(super) fn connection_terminated_error(
94    method: &str,
95    termination: &ConnectionTermination,
96) -> crate::Error {
97    let mut error = crate::Error::internal_error();
98    error.message = "Connection terminated".to_string();
99    error.data(serde_json::json!({
100        "reason": CONNECTION_TERMINATED_REASON,
101        "method": method,
102        "termination": termination.kind(),
103    }))
104}
105
106/// Which part of the driver produced an error.
107#[derive(Clone, Copy, Debug)]
108pub(super) enum TerminationSource {
109    Transport,
110    Handler,
111    Foreground,
112}
113
114/// Records errors by origin so the driver's final result can be classified.
115#[derive(Clone, Default)]
116pub(super) struct TerminationSources {
117    errors: Arc<Mutex<Vec<(TerminationSource, crate::Error)>>>,
118}
119
120impl TerminationSources {
121    pub(super) fn record<T>(&self, source: TerminationSource, result: &Result<T, crate::Error>) {
122        if let Err(error) = result {
123            self.errors
124                .lock()
125                .expect("termination sources mutex poisoned")
126                .push((source, error.clone()));
127        }
128    }
129
130    /// Classify the driver's returned error. Several actors may fail during
131    /// one shutdown; the source of the error actually returned wins.
132    pub(super) fn classify(&self, error: crate::Error) -> ConnectionTermination {
133        let source = {
134            let errors = self
135                .errors
136                .lock()
137                .expect("termination sources mutex poisoned");
138            errors
139                .iter()
140                .find(|(_, recorded)| *recorded == error)
141                .or_else(|| errors.first())
142                .map_or(TerminationSource::Handler, |(source, _)| *source)
143        };
144        match source {
145            TerminationSource::Transport => ConnectionTermination::TransportError(error),
146            TerminationSource::Handler => ConnectionTermination::HandlerError(error),
147            TerminationSource::Foreground => ConnectionTermination::ForegroundError(error),
148        }
149    }
150}
agent_client_protocol/jsonrpc/termination.rs

agent_client_protocol/jsonrpc/
termination.rs
