1use std::time::{Duration, Instant};
2
3use chrono::{DateTime, Utc};
4use futures::StreamExt;
5use serde::{Deserialize, Serialize};
6
7use crate::core::a2a::dlq::DeadLetter;
8use crate::core::a2a_transport::TransportEnvelopeV1;
9
10const DEFAULT_MAX_PAYLOAD_BYTES: usize = 2_000_000;
11const MAX_ERROR_BODY_BYTES: usize = 64 * 1024;
12
13#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
14pub struct RemoteTransportConfig {
15 pub endpoint_url: String,
16 pub timeout: Duration,
17 pub max_payload_bytes: usize,
18 pub auth_token: Option<String>,
19 pub retry_count: u8,
20 pub retry_delay: Duration,
21}
22
23impl Default for RemoteTransportConfig {
24 fn default() -> Self {
25 Self {
26 endpoint_url: String::new(),
27 timeout: Duration::from_secs(30),
28 max_payload_bytes: DEFAULT_MAX_PAYLOAD_BYTES,
29 auth_token: None,
30 retry_count: 2,
31 retry_delay: Duration::from_secs(1),
32 }
33 }
34}
35
36impl RemoteTransportConfig {
37 pub fn validate(&self) -> Result<(), String> {
38 let url = reqwest::Url::parse(&self.endpoint_url)
39 .map_err(|error| format!("invalid endpoint_url: {error}"))?;
40 if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
41 return Err("endpoint_url must be an absolute HTTP(S) URL".to_string());
42 }
43 if self.timeout.is_zero() {
44 return Err("timeout must be greater than zero".to_string());
45 }
46 if self.max_payload_bytes == 0 {
47 return Err("max_payload_bytes must be greater than zero".to_string());
48 }
49 if self.auth_token.as_ref().is_some_and(String::is_empty) {
50 return Err("auth_token must not be empty".to_string());
51 }
52 Ok(())
53 }
54
55 fn delivery_url(&self) -> Result<reqwest::Url, String> {
56 self.validate()?;
57 let mut url = reqwest::Url::parse(&self.endpoint_url)
58 .map_err(|error| format!("invalid endpoint_url: {error}"))?;
59 let path = format!("{}/a2a/deliver", url.path().trim_end_matches('/'));
60 url.set_path(&path);
61 url.set_query(None);
62 url.set_fragment(None);
63 Ok(url)
64 }
65}
66
67#[derive(Debug, Clone, Serialize, Deserialize)]
68pub struct RemoteTransport {
69 config: RemoteTransportConfig,
70 #[serde(skip, default = "reqwest::Client::new")]
71 client: reqwest::Client,
72}
73
74impl RemoteTransport {
75 pub fn new(config: RemoteTransportConfig) -> Result<Self, String> {
76 config.validate()?;
77 Ok(Self {
78 config,
79 client: reqwest::Client::new(),
80 })
81 }
82
83 pub async fn deliver(
84 &self,
85 envelope: &TransportEnvelopeV1,
86 ) -> Result<DeliveryReceipt, TransportError> {
87 let body = serialize_and_validate(envelope, self.config.max_payload_bytes)?;
88 let envelope_id = envelope_id(&body);
89 let delivery_url = self
90 .config
91 .delivery_url()
92 .map_err(TransportError::SerializationError)?;
93 let started_at = Instant::now();
94
95 for attempt in 0..=self.config.retry_count {
96 let mut request = self
97 .client
98 .post(delivery_url.clone())
99 .header(reqwest::header::CONTENT_TYPE, "application/json")
100 .timeout(self.config.timeout)
101 .body(body.clone());
102 if let Some(token) = self.config.auth_token.as_deref() {
103 request = request.bearer_auth(token);
104 }
105
106 match request.send().await {
107 Ok(response) if response.status().is_success() => {
108 return Ok(DeliveryReceipt {
109 envelope_id,
110 delivered_at: Utc::now(),
111 remote_status: response.status().as_u16(),
112 round_trip_ms: elapsed_millis(started_at),
113 });
114 }
115 Ok(response) if response.status().is_server_error() => {
116 if attempt == self.config.retry_count {
117 return Err(TransportError::Exhausted(self.config.retry_count));
118 }
119 }
120 Ok(response) => {
121 let status = response.status();
122 let error_body = read_error_body(response).await;
123 if status.is_client_error() {
124 enqueue_permanent_failure(
125 &envelope_id,
126 envelope,
127 &body,
128 status.as_u16(),
129 &error_body,
130 attempt.saturating_add(1),
131 &self.config.endpoint_url,
132 );
133 }
134 return Err(TransportError::RemoteError(status.as_u16(), error_body));
135 }
136 Err(error) if error.is_timeout() && self.config.retry_count == 0 => {
137 return Err(TransportError::Timeout);
138 }
139 Err(_) if attempt == self.config.retry_count => {
140 return Err(TransportError::Exhausted(self.config.retry_count));
141 }
142 Err(_) => {}
143 }
144
145 tokio::time::sleep(self.config.retry_delay).await;
146 }
147
148 Err(TransportError::Exhausted(self.config.retry_count))
149 }
150}
151
152#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
153pub struct DeliveryReceipt {
154 pub envelope_id: String,
155 pub delivered_at: DateTime<Utc>,
156 pub remote_status: u16,
157 pub round_trip_ms: u64,
158}
159
160#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, thiserror::Error)]
161pub enum TransportError {
162 #[error("payload too large: {0} bytes")]
163 PayloadTooLarge(usize),
164 #[error("transport timed out")]
165 Timeout,
166 #[error("remote returned HTTP {0}: {1}")]
167 RemoteError(u16, String),
168 #[error("serialization failed: {0}")]
169 SerializationError(String),
170 #[error("delivery exhausted after {0} retries")]
171 Exhausted(u8),
172}
173
174fn serialize_and_validate(
175 envelope: &TransportEnvelopeV1,
176 max_payload_bytes: usize,
177) -> Result<Vec<u8>, TransportError> {
178 let body = serde_json::to_vec(envelope)
179 .map_err(|error| TransportError::SerializationError(error.to_string()))?;
180 if body.len() > max_payload_bytes {
181 return Err(TransportError::PayloadTooLarge(body.len()));
182 }
183 Ok(body)
184}
185
186fn envelope_id(body: &[u8]) -> String {
187 format!("envelope:{}", blake3::hash(body).to_hex())
188}
189
190fn elapsed_millis(started_at: Instant) -> u64 {
191 u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX)
192}
193
194async fn read_error_body(response: reqwest::Response) -> String {
195 let mut stream = response.bytes_stream();
196 let mut body = Vec::new();
197 while let Some(chunk) = stream.next().await {
198 let Ok(chunk) = chunk else {
199 break;
200 };
201 let remaining = MAX_ERROR_BODY_BYTES.saturating_sub(body.len());
202 body.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
203 if body.len() == MAX_ERROR_BODY_BYTES {
204 break;
205 }
206 }
207 String::from_utf8_lossy(&body).into_owned()
208}
209
210fn enqueue_permanent_failure(
211 envelope_id: &str,
212 envelope: &TransportEnvelopeV1,
213 body: &[u8],
214 status: u16,
215 error_body: &str,
216 attempts: u8,
217 endpoint_url: &str,
218) {
219 let failed_at = Utc::now().to_rfc3339();
220 let target_agent = envelope.recipient.as_deref().unwrap_or(endpoint_url);
221 crate::core::ocla::health::dead_letter_queue().enqueue(DeadLetter {
222 id: envelope_id.to_string(),
223 original_message: String::from_utf8_lossy(body).into_owned(),
224 target_agent: target_agent.to_string(),
225 error: format!("HTTP {status}: {error_body}"),
226 attempts,
227 first_failed_at: failed_at.clone(),
228 last_failed_at: failed_at,
229 });
230}
231
232#[cfg(test)]
233mod tests {
234 use std::collections::HashMap;
235
236 use super::*;
237 use crate::core::a2a_transport::{AgentIdentityV1, TransportContentType};
238
239 fn envelope(payload: &str) -> TransportEnvelopeV1 {
240 TransportEnvelopeV1 {
241 format_version: 1,
242 sent_at: Utc::now(),
243 sender: AgentIdentityV1 {
244 agent_id: "sender".to_string(),
245 agent_type: "test".to_string(),
246 daemon_fingerprint: "fingerprint".to_string(),
247 capabilities: Vec::new(),
248 },
249 recipient: Some("recipient".to_string()),
250 content_type: TransportContentType::A2AMessage,
251 payload_json: payload.to_string(),
252 signature: None,
253 metadata: HashMap::new(),
254 }
255 }
256
257 #[test]
258 fn default_config_has_bounded_transport_values() {
259 let config = RemoteTransportConfig {
260 endpoint_url: "https://agent.example/api/".to_string(),
261 ..RemoteTransportConfig::default()
262 };
263
264 assert!(config.validate().is_ok());
265 assert_eq!(config.timeout, Duration::from_secs(30));
266 assert_eq!(config.max_payload_bytes, 2_000_000);
267 assert_eq!(config.retry_count, 2);
268 assert_eq!(config.retry_delay, Duration::from_secs(1));
269 assert_eq!(
270 config.delivery_url().expect("valid URL").as_str(),
271 "https://agent.example/api/a2a/deliver"
272 );
273 }
274
275 #[test]
276 fn config_validation_rejects_unbounded_or_unsupported_values() {
277 let zero_timeout = RemoteTransportConfig {
278 endpoint_url: "https://agent.example".to_string(),
279 timeout: Duration::ZERO,
280 ..RemoteTransportConfig::default()
281 };
282 let unsupported_scheme = RemoteTransportConfig {
283 endpoint_url: "file:///tmp/agent".to_string(),
284 ..RemoteTransportConfig::default()
285 };
286
287 assert!(zero_timeout.validate().is_err());
288 assert!(unsupported_scheme.validate().is_err());
289 }
290
291 #[test]
292 fn payload_size_limit_reports_serialized_size() {
293 let envelope = envelope("payload");
294 let serialized = serde_json::to_vec(&envelope).expect("serializable envelope");
295 let error = serialize_and_validate(&envelope, serialized.len() - 1)
296 .expect_err("payload must exceed configured limit");
297
298 assert_eq!(error, TransportError::PayloadTooLarge(serialized.len()));
299 }
300
301 #[test]
302 fn transport_error_variants_are_serializable_and_distinct() {
303 let variants = [
304 TransportError::PayloadTooLarge(10),
305 TransportError::Timeout,
306 TransportError::RemoteError(400, "bad request".to_string()),
307 TransportError::SerializationError("invalid JSON".to_string()),
308 TransportError::Exhausted(2),
309 ];
310
311 for variant in variants {
312 let json = serde_json::to_string(&variant).expect("serialize error");
313 let decoded = serde_json::from_str(&json).expect("deserialize error");
314 assert_eq!(variant, decoded);
315 }
316 }
317
318 #[test]
319 fn receipt_round_trips_without_losing_delivery_fields() {
320 let receipt = DeliveryReceipt {
321 envelope_id: "envelope:abc".to_string(),
322 delivered_at: Utc::now(),
323 remote_status: 202,
324 round_trip_ms: 17,
325 };
326
327 let json = serde_json::to_string(&receipt).expect("serialize receipt");
328 let decoded: DeliveryReceipt = serde_json::from_str(&json).expect("deserialize receipt");
329 assert_eq!(decoded, receipt);
330 }
331
332 #[test]
333 fn envelope_ids_are_deterministic_and_content_addressed() {
334 let first = serialize_and_validate(&envelope("one"), usize::MAX).expect("serialize");
335 let first_again = first.clone();
336 let second = serialize_and_validate(&envelope("two"), usize::MAX).expect("serialize");
337
338 assert_eq!(envelope_id(&first), envelope_id(&first_again));
339 assert_ne!(envelope_id(&first), envelope_id(&second));
340 }
341}