use std::time::{Duration, Instant};
use chrono::{DateTime, Utc};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use crate::core::a2a::dlq::{DeadLetter, DeadLetterDelivery, DlqScope, MAX_ERROR_BYTES};
use crate::core::a2a::relay::{
RELAY_MAX_PAYLOAD_BYTES, RELAY_MAX_RETRIES, RELAY_MAX_RETRY_DELAY_MS, RELAY_PEER_HEADER,
RelayPeerConfigV1, RelayPeerTableV1,
};
use crate::core::a2a_transport::TransportEnvelopeV1;
const DEFAULT_MAX_PAYLOAD_BYTES: usize = 2_000_000;
const MAX_ERROR_BODY_BYTES: usize = 64 * 1024;
const MAX_TASK_RESPONSE_BYTES: usize = 64 * 1024;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RemoteTransportConfig {
pub endpoint_url: String,
pub timeout: Duration,
pub max_payload_bytes: usize,
pub auth_token: Option<String>,
pub signing_key: Option<String>,
pub recipient_id: Option<String>,
pub tenant_id: Option<String>,
pub project_id: Option<String>,
pub retry_count: u8,
pub retry_delay: Duration,
#[serde(default)]
pub peer_id: Option<String>,
#[serde(default)]
pub local_peer_id: Option<String>,
#[serde(default)]
pub peers: Vec<RelayPeerConfigV1>,
#[serde(skip)]
pub allow_loopback_http: bool,
}
impl Default for RemoteTransportConfig {
fn default() -> Self {
Self {
endpoint_url: String::new(),
timeout: Duration::from_secs(30),
max_payload_bytes: DEFAULT_MAX_PAYLOAD_BYTES,
auth_token: None,
signing_key: None,
recipient_id: None,
tenant_id: None,
project_id: None,
retry_count: 2,
retry_delay: Duration::from_secs(1),
peer_id: None,
local_peer_id: None,
peers: Vec::new(),
allow_loopback_http: false,
}
}
}
impl RemoteTransportConfig {
pub fn validate(&self) -> Result<(), String> {
if self.timeout.is_zero() || self.max_payload_bytes == 0 {
return Err("transport bounds must be greater than zero".into());
}
if self.retry_count > RELAY_MAX_RETRIES
|| self.retry_delay.as_millis() > RELAY_MAX_RETRY_DELAY_MS as u128
{
return Err("retry policy exceeds bounded relay caps".into());
}
if self.local_peer_id.is_some() || !self.peers.is_empty() {
return self.validate_peer_mode();
}
if self.peer_id.is_some() {
return Err("singleton transport must not contain a peer_id".into());
}
let url = reqwest::Url::parse(&self.endpoint_url)
.map_err(|error| format!("invalid endpoint_url: {error}"))?;
if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
return Err("endpoint_url must be an absolute HTTP(S) URL".to_string());
}
if !url.username().is_empty() || url.password().is_some() {
return Err("endpoint_url must not contain credentials".to_string());
}
if url.scheme() == "http" && !is_loopback_host(&url) {
return Err("endpoint_url must use HTTPS except for loopback hosts".to_string());
}
if self.auth_token.as_ref().is_none_or(String::is_empty) {
return Err("auth_token is required for remote A2A delivery".to_string());
}
if self.signing_key.as_ref().is_none_or(String::is_empty) {
return Err("signing_key is required for remote A2A delivery".to_string());
}
if self.auth_token == self.signing_key {
return Err("signing_key must be distinct from auth_token".to_string());
}
if self.recipient_id.as_ref().is_none_or(String::is_empty) {
return Err("recipient_id is required for remote A2A delivery".to_string());
}
if self.tenant_id.as_ref().is_none_or(String::is_empty) {
return Err("tenant_id is required for remote A2A delivery".to_string());
}
if self.project_id.as_ref().is_none_or(String::is_empty) {
return Err("project_id is required for remote A2A delivery".to_string());
}
Ok(())
}
fn validate_peer_mode(&self) -> Result<(), String> {
if self.local_peer_id.as_deref().is_none_or(|id| {
id.is_empty() || id.len() > crate::core::a2a::relay::RELAY_MAX_ID_BYTES
}) || self.peers.is_empty()
|| self.max_payload_bytes > RELAY_MAX_PAYLOAD_BYTES
{
return Err(
"peer transport requires a bounded local identity, peer table and payload limit"
.into(),
);
}
if !self.endpoint_url.is_empty()
|| [
&self.auth_token,
&self.signing_key,
&self.recipient_id,
&self.tenant_id,
&self.project_id,
&self.peer_id,
]
.iter()
.any(|value| value.is_some())
|| self.retry_count != 0
|| !self.retry_delay.is_zero()
{
return Err(
"peer transport must not contain singleton authority or retry policy".into(),
);
}
let table = RelayPeerTableV1 {
schema_version: crate::core::a2a::relay::RELAY_PEER_TABLE_VERSION,
peers: self.peers.clone(),
};
table
.validate(self.allow_loopback_http)
.map_err(|error| error.to_string())?;
table
.reject_shared_secret()
.map_err(|error| error.to_string())
}
fn delivery_url(&self) -> Result<reqwest::Url, String> {
self.validate()?;
let mut url = reqwest::Url::parse(&self.endpoint_url)
.map_err(|error| format!("invalid endpoint_url: {error}"))?;
let path = format!("{}/a2a/deliver", url.path().trim_end_matches('/'));
url.set_path(&path);
url.set_query(None);
url.set_fragment(None);
Ok(url)
}
}
fn is_loopback_host(url: &reqwest::Url) -> bool {
url.host_str().is_some_and(|host| {
let ip_host = host
.strip_prefix('[')
.and_then(|value| value.strip_suffix(']'))
.unwrap_or(host);
host.eq_ignore_ascii_case("localhost")
|| ip_host
.parse::<std::net::IpAddr>()
.is_ok_and(|address| address.is_loopback())
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RemoteTransport {
config: RemoteTransportConfig,
#[serde(skip, default = "fixed_peer_client")]
client: reqwest::Client,
}
fn fixed_peer_client() -> reqwest::Client {
reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("initialize fixed-peer HTTP client")
}
impl RemoteTransport {
#[cfg(any(all(test, feature = "http-server"), feature = "test-support"))]
pub fn with_test_tls_root(mut self, certificate: reqwest::Certificate) -> Self {
self.client = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.add_root_certificate(certificate)
.build()
.expect("test TLS client");
self
}
pub fn new(config: RemoteTransportConfig) -> Result<Self, String> {
config.validate()?;
Ok(Self {
config,
client: fixed_peer_client(),
})
}
pub fn for_peer_table(
table: RelayPeerTableV1,
local_peer_id: &str,
timeout: Duration,
allow_loopback_http: bool,
) -> Result<Self, String> {
table
.validate(allow_loopback_http)
.map_err(|error| error.to_string())?;
Self::new(RemoteTransportConfig {
timeout,
max_payload_bytes: RELAY_MAX_PAYLOAD_BYTES,
retry_count: 0,
retry_delay: Duration::ZERO,
local_peer_id: Some(local_peer_id.to_string()),
peers: table.peers,
allow_loopback_http,
..RemoteTransportConfig::default()
})
}
pub async fn deliver(
&self,
envelope: &TransportEnvelopeV1,
) -> Result<DeliveryReceipt, TransportError> {
self.config
.validate()
.map_err(TransportError::SerializationError)?;
if self.config.local_peer_id.is_some() {
return Err(TransportError::SerializationError(
"peer transport requires explicit peer delivery".into(),
));
}
self.ensure_dlq_available().await?;
let mut signed = envelope.clone();
let expected_recipient = self
.config
.recipient_id
.as_deref()
.expect("validated transport has recipient identity");
if signed.recipient.as_deref() != Some(expected_recipient) {
return Err(TransportError::SerializationError(
"envelope recipient does not match configured remote recipient".to_string(),
));
}
signed.metadata.insert(
"tenant_id".to_string(),
self.config
.tenant_id
.clone()
.expect("validated transport has tenant scope"),
);
signed.metadata.insert(
"project_id".to_string(),
self.config
.project_id
.clone()
.expect("validated transport has project scope"),
);
let secret = self
.config
.signing_key
.as_deref()
.ok_or_else(|| TransportError::SerializationError("missing signing key".into()))?;
signed
.sign(secret.as_bytes())
.map_err(TransportError::SerializationError)?;
let body = serialize_and_validate(&signed, self.config.max_payload_bytes)?;
let envelope_id = envelope_id(&body);
let delivery_url = self
.config
.delivery_url()
.map_err(TransportError::SerializationError)?;
let started_at = Instant::now();
for attempt in 0..=self.config.retry_count {
let mut request = self
.client
.post(delivery_url.clone())
.header(reqwest::header::CONTENT_TYPE, "application/json")
.timeout(self.config.timeout)
.body(body.clone());
if let Some(token) = self.config.auth_token.as_deref() {
request = request.bearer_auth(token);
}
match request.send().await {
Ok(response) if response.status().is_success() => {
let remote_status = response.status().as_u16();
let unverified_task_response = match read_task_response(response, &signed).await
{
Ok(body) => body,
Err(error) => return Err(error),
};
return Ok(DeliveryReceipt {
envelope_id,
delivered_at: Utc::now(),
remote_status,
round_trip_ms: elapsed_millis(started_at),
unverified_task_response,
});
}
Ok(response) if response.status().is_server_error() => {
if attempt == self.config.retry_count {
return Err(self
.record_failure(
&signed,
&body,
TransportError::Exhausted(self.config.retry_count),
attempt.saturating_add(1),
)
.await);
}
}
Ok(response) => {
let status = response.status();
let error_body = read_error_body(response).await;
let error = TransportError::RemoteError(status.as_u16(), error_body);
return Err(self
.record_failure(&signed, &body, error, attempt.saturating_add(1))
.await);
}
Err(error) if error.is_timeout() && self.config.retry_count == 0 => {
return Err(self
.record_failure(
&signed,
&body,
TransportError::Timeout,
attempt.saturating_add(1),
)
.await);
}
Err(_) if attempt == self.config.retry_count => {
return Err(self
.record_failure(
&signed,
&body,
TransportError::Exhausted(self.config.retry_count),
attempt.saturating_add(1),
)
.await);
}
Err(_) => {}
}
tokio::time::sleep(self.config.retry_delay).await;
}
Err(TransportError::Exhausted(self.config.retry_count))
}
pub async fn deliver_to(
&self,
peer_id: &str,
envelope: &TransportEnvelopeV1,
) -> Result<DeliveryReceipt, TransportError> {
self.config
.validate()
.map_err(TransportError::SerializationError)?;
if self.config.local_peer_id.is_none() {
return Err(TransportError::SerializationError(
"explicit peer delivery requires peer transport".into(),
));
}
let peer = self
.config
.peers
.iter()
.find(|peer| peer.peer_id == peer_id)
.cloned()
.ok_or_else(|| {
TransportError::SerializationError(format!("unknown relay peer: {peer_id}"))
})?;
let mut signed = envelope.clone();
let record = signed
.relay_record()
.map_err(TransportError::SerializationError)?
.ok_or_else(|| {
TransportError::SerializationError(
"relay record is required for peer delivery".into(),
)
})?;
let now = Utc::now();
record
.validate_at(
now,
&record.current_peer.clone(),
Some(peer.peer_id.as_str()),
peer.max_hops,
)
.map_err(|error| TransportError::SerializationError(error.to_string()))?;
record
.verify_hop(&peer.channel_key)
.map_err(|error| TransportError::SerializationError(error.to_string()))?;
if !peer.allows(&record, signed.payload_json.len())
|| signed.content_type != record.content_type
|| signed.sender.agent_id != record.current_peer
|| self.config.local_peer_id.as_deref() != Some(record.current_peer.as_str())
|| signed.metadata.get("tenant_id").map(String::as_str)
!= Some(record.tenant_id.as_str())
|| signed.metadata.get("project_id").map(String::as_str)
!= Some(record.project_id.as_str())
|| record
.verify_payload(signed.payload_json.as_bytes())
.is_err()
{
return Err(TransportError::SerializationError(
"peer policy denied relay scope".into(),
));
}
if signed.recipient.as_deref() != Some(record.final_recipient.as_str()) {
return Err(TransportError::SerializationError(
"envelope recipient does not match signed final recipient".into(),
));
}
signed
.attach_relay_record(&record)
.map_err(TransportError::SerializationError)?;
signed.signature = None;
signed
.sign(peer.channel_key.as_bytes())
.map_err(TransportError::SerializationError)?;
let body = serialize_and_validate(&signed, peer.max_payload_bytes)?;
let delivery_id = record.delivery_id.clone();
let delivery_url = peer_delivery_url(&peer).map_err(TransportError::SerializationError)?;
self.ensure_dlq_scope(&record.tenant_id, &record.project_id)
.await?;
let started_at = Instant::now();
let source_peer = self
.config
.local_peer_id
.as_deref()
.unwrap_or(record.current_peer.as_str());
for attempt in 0..=peer.retry_count {
let request = self
.client
.post(delivery_url.clone())
.header(reqwest::header::CONTENT_TYPE, "application/json")
.header("idempotency-key", delivery_id.as_str())
.header(RELAY_PEER_HEADER, source_peer)
.timeout(self.config.timeout)
.bearer_auth(peer.bearer_token.as_str())
.body(body.clone());
match request.send().await {
Ok(response) if response.status().is_success() => {
let remote_status = response.status().as_u16();
let unverified_task_response = match read_task_response(response, &signed).await
{
Ok(body) => body,
Err(error) => return Err(error),
};
return Ok(DeliveryReceipt {
envelope_id: delivery_id,
delivered_at: Utc::now(),
remote_status,
round_trip_ms: elapsed_millis(started_at),
unverified_task_response,
});
}
Ok(response) if response.status().is_server_error() => {
if attempt == peer.retry_count {
let failure = TransportError::Exhausted(peer.retry_count);
return Err(self
.record_failure_for_peer(
&signed,
&body,
&failure,
attempt.saturating_add(1),
&peer,
)
.await);
}
}
Ok(response) => {
let status = response.status();
let error = TransportError::RemoteError(
status.as_u16(),
read_error_body(response).await,
);
return Err(self
.record_failure_for_peer(
&signed,
&body,
&error,
attempt.saturating_add(1),
&peer,
)
.await);
}
Err(error) if error.is_timeout() && peer.retry_count == 0 => {
return Err(self
.record_failure_for_peer(
&signed,
&body,
&TransportError::Timeout,
attempt.saturating_add(1),
&peer,
)
.await);
}
Err(_) if attempt == peer.retry_count => {
let failure = TransportError::Exhausted(peer.retry_count);
return Err(self
.record_failure_for_peer(
&signed,
&body,
&failure,
attempt.saturating_add(1),
&peer,
)
.await);
}
Err(_) => {}
}
tokio::time::sleep(Duration::from_millis(peer.retry_delay_ms)).await;
}
Err(TransportError::Exhausted(peer.retry_count))
}
async fn record_failure(
&self,
envelope: &TransportEnvelopeV1,
body: &[u8],
failure: TransportError,
attempts: u8,
) -> TransportError {
let envelope = envelope.clone();
let body = body.to_vec();
let config = self.config.clone();
let failure_for_store = failure.clone();
let stored = tokio::task::spawn_blocking(move || {
enqueue_permanent_failure(&envelope, &body, &failure_for_store, attempts, &config)
})
.await;
match stored {
Ok(Ok(())) => failure,
Ok(Err(error)) => TransportError::DeadLetterFailure(format!("{failure}; {error}")),
Err(error) => TransportError::DeadLetterFailure(format!(
"{failure}; dead-letter worker failed: {error}"
)),
}
}
async fn ensure_dlq_available(&self) -> Result<(), TransportError> {
let scope = DlqScope::new(
self.config
.tenant_id
.clone()
.expect("validated tenant scope"),
self.config
.project_id
.clone()
.expect("validated project scope"),
)
.map_err(|error| TransportError::DeadLetterFailure(error.to_string()))?;
self.ensure_dlq_scope_inner(scope).await
}
async fn ensure_dlq_scope(
&self,
tenant_id: &str,
project_id: &str,
) -> Result<(), TransportError> {
let scope = DlqScope::new(tenant_id, project_id)
.map_err(|error| TransportError::DeadLetterFailure(error.to_string()))?;
self.ensure_dlq_scope_inner(scope).await
}
async fn ensure_dlq_scope_inner(&self, scope: DlqScope) -> Result<(), TransportError> {
let queue = crate::core::ocla::health::dead_letter_queue().clone();
tokio::task::spawn_blocking(move || queue.ensure_writable(&scope))
.await
.map_err(|error| {
TransportError::DeadLetterFailure(format!("dead-letter worker failed: {error}"))
})?
.map_err(|error| TransportError::DeadLetterFailure(error.to_string()))
}
async fn record_failure_for_peer(
&self,
envelope: &TransportEnvelopeV1,
body: &[u8],
failure: &TransportError,
attempts: u8,
peer: &RelayPeerConfigV1,
) -> TransportError {
let envelope = envelope.clone();
let body = body.to_vec();
let config = self.config.clone();
let failure_for_store = failure.clone();
let peer_id = peer.peer_id.clone();
let endpoint_url = peer.endpoint_url.clone();
let stored = tokio::task::spawn_blocking(move || {
enqueue_permanent_failure_for_peer(
&envelope,
&body,
&failure_for_store,
attempts,
&config,
&peer_id,
&endpoint_url,
)
})
.await;
match stored {
Ok(Ok(())) => failure.clone(),
Ok(Err(error)) => TransportError::DeadLetterFailure(format!("{failure}; {error}")),
Err(error) => TransportError::DeadLetterFailure(format!(
"{failure}; dead-letter worker failed: {error}"
)),
}
}
pub async fn retry_dead_letter(
&self,
letter: &DeadLetter,
) -> Result<DeliveryReceipt, TransportError> {
if letter.peer_id != "legacy" {
let envelope: TransportEnvelopeV1 = serde_json::from_str(&letter.original_message)
.map_err(|error| TransportError::SerializationError(error.to_string()))?;
let record = envelope
.relay_record()
.map_err(TransportError::SerializationError)?
.ok_or_else(|| {
TransportError::SerializationError(
"peer dead letter is missing relay record".into(),
)
})?;
if record.delivery_id != letter.delivery_id {
return Err(TransportError::SerializationError(
"dead letter delivery ID does not match relay record".into(),
));
}
if record.tenant_id != letter.tenant_id || record.project_id != letter.project_id {
return Err(TransportError::SerializationError(
"dead letter scope does not match relay record".into(),
));
}
let peer = self
.config
.peers
.iter()
.find(|peer| peer.peer_id == letter.peer_id)
.ok_or_else(|| {
TransportError::SerializationError("dead letter peer is not configured".into())
})?;
if &peer.endpoint_url
!= match &letter.delivery {
DeadLetterDelivery::RemoteHttp { endpoint_url } => endpoint_url,
DeadLetterDelivery::LocalAgentBus => {
return Err(TransportError::SerializationError(
"dead letter is not remote".into(),
));
}
}
{
return Err(TransportError::SerializationError(
"dead letter endpoint does not match peer".into(),
));
}
let mut envelope = envelope;
envelope.sent_at = Utc::now();
envelope.signature = None;
return self.deliver_to(&letter.peer_id, &envelope).await;
}
let expected_scope = (
self.config.tenant_id.as_deref(),
self.config.project_id.as_deref(),
);
if expected_scope
!= (
Some(letter.tenant_id.as_str()),
Some(letter.project_id.as_str()),
)
{
return Err(TransportError::SerializationError(
"dead letter scope does not match configured remote transport".into(),
));
}
let DeadLetterDelivery::RemoteHttp { endpoint_url } = &letter.delivery else {
return Err(TransportError::SerializationError(
"dead letter is not a remote delivery".into(),
));
};
if endpoint_url != &self.config.endpoint_url {
return Err(TransportError::SerializationError(
"dead letter endpoint does not match configured remote transport".into(),
));
}
let mut envelope: TransportEnvelopeV1 = serde_json::from_str(&letter.original_message)
.map_err(|error| TransportError::SerializationError(error.to_string()))?;
if !envelope
.metadata
.contains_key(crate::core::a2a_transport::LEGACY_DELIVERY_ID_METADATA)
&& let Some(signature) = envelope.signature.as_deref()
{
envelope.metadata.insert(
crate::core::a2a_transport::LEGACY_DELIVERY_ID_METADATA.to_string(),
signature.to_ascii_lowercase(),
);
envelope.metadata.insert(
crate::core::a2a_transport::LEGACY_DELIVERY_SENT_AT_METADATA.to_string(),
envelope.sent_at.to_rfc3339(),
);
}
envelope.sent_at = Utc::now();
envelope.signature = None;
self.deliver(&envelope).await
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct DeliveryReceipt {
pub envelope_id: String,
pub delivered_at: DateTime<Utc>,
pub remote_status: u16,
pub round_trip_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub unverified_task_response: Option<String>,
}
impl DeliveryReceipt {
pub fn verified_task_status(
&self,
request: &super::task::TaskControlDescriptorV1,
policy: &super::task::TaskAuthorityConfigV1,
now: DateTime<Utc>,
) -> Result<super::task::TaskStatusV1, super::task_response::InvalidTaskResponse> {
use super::task_response::{InvalidTaskResponse, SignedTaskStatusV1};
if !(200..300).contains(&self.remote_status) {
return Err(InvalidTaskResponse);
}
let raw = self
.unverified_task_response
.as_deref()
.ok_or(InvalidTaskResponse)?;
let response = SignedTaskStatusV1::from_json(raw)?;
response.verify(request, policy, now)?;
Ok(response.status)
}
}
async fn read_task_response(
response: reqwest::Response,
envelope: &TransportEnvelopeV1,
) -> Result<Option<String>, TransportError> {
if envelope.content_type != crate::core::a2a_transport::TransportContentType::A2ATask {
return Ok(None);
}
let invalid =
|| TransportError::SerializationError("invalid or oversized task response".into());
if response
.content_length()
.is_some_and(|length| length > MAX_TASK_RESPONSE_BYTES as u64)
{
return Err(invalid());
}
let mut stream = response.bytes_stream();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|_| invalid())?;
if chunk.len() > MAX_TASK_RESPONSE_BYTES.saturating_sub(bytes.len()) {
return Err(invalid());
}
bytes.extend_from_slice(&chunk);
}
if bytes.is_empty() {
return Err(invalid());
}
String::from_utf8(bytes).map(Some).map_err(|_| invalid())
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, thiserror::Error)]
pub enum TransportError {
#[error("payload too large: {0} bytes")]
PayloadTooLarge(usize),
#[error("transport timed out")]
Timeout,
#[error("remote returned HTTP {0}: {1}")]
RemoteError(u16, String),
#[error("serialization failed: {0}")]
SerializationError(String),
#[error("delivery exhausted after {0} retries")]
Exhausted(u8),
#[error("delivery failed and could not be persisted to DLQ: {0}")]
DeadLetterFailure(String),
}
fn serialize_and_validate(
envelope: &TransportEnvelopeV1,
max_payload_bytes: usize,
) -> Result<Vec<u8>, TransportError> {
let body = serde_json::to_vec(envelope)
.map_err(|error| TransportError::SerializationError(error.to_string()))?;
if body.len() > max_payload_bytes {
return Err(TransportError::PayloadTooLarge(body.len()));
}
Ok(body)
}
fn peer_delivery_url(peer: &RelayPeerConfigV1) -> Result<reqwest::Url, String> {
let mut url = reqwest::Url::parse(&peer.endpoint_url)
.map_err(|error| format!("invalid peer endpoint_url: {error}"))?;
let path = format!("{}/a2a/deliver", url.path().trim_end_matches('/'));
url.set_path(&path);
url.set_query(None);
url.set_fragment(None);
Ok(url)
}
fn envelope_id(body: &[u8]) -> String {
format!("envelope:{}", blake3::hash(body).to_hex())
}
fn elapsed_millis(started_at: Instant) -> u64 {
u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX)
}
async fn read_error_body(response: reqwest::Response) -> String {
let mut stream = response.bytes_stream();
let mut body = Vec::new();
while let Some(chunk) = stream.next().await {
let Ok(chunk) = chunk else {
break;
};
let remaining = MAX_ERROR_BODY_BYTES.saturating_sub(body.len());
body.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
if body.len() == MAX_ERROR_BODY_BYTES {
break;
}
}
String::from_utf8_lossy(&body).into_owned()
}
fn enqueue_permanent_failure(
envelope: &TransportEnvelopeV1,
body: &[u8],
failure: &TransportError,
attempts: u8,
config: &RemoteTransportConfig,
) -> crate::core::ocla::types::OclaResult<()> {
let peer_id = config.peer_id.as_deref().unwrap_or("legacy");
enqueue_permanent_failure_for_peer(
envelope,
body,
failure,
attempts,
config,
peer_id,
&config.endpoint_url,
)
}
fn enqueue_permanent_failure_for_peer(
envelope: &TransportEnvelopeV1,
body: &[u8],
failure: &TransportError,
attempts: u8,
config: &RemoteTransportConfig,
peer_id: &str,
endpoint_url: &str,
) -> crate::core::ocla::types::OclaResult<()> {
let failed_at = Utc::now().to_rfc3339();
let target_agent = envelope.recipient.as_deref().unwrap_or(endpoint_url);
let mut error = failure.to_string();
if error.len() > MAX_ERROR_BYTES {
let mut boundary = MAX_ERROR_BYTES;
while !error.is_char_boundary(boundary) {
boundary -= 1;
}
error.truncate(boundary);
}
let relay_record = envelope.relay_record().ok().flatten();
let delivery_id = relay_record
.as_ref()
.map(|record| record.delivery_id.clone())
.or_else(|| envelope.stable_delivery_id().ok())
.unwrap_or_else(|| envelope_id(body));
let tenant_id = relay_record
.as_ref()
.map(|record| record.tenant_id.clone())
.or_else(|| config.tenant_id.clone())
.expect("validated tenant scope");
let project_id = relay_record
.as_ref()
.map(|record| record.project_id.clone())
.or_else(|| config.project_id.clone())
.expect("validated project scope");
crate::core::ocla::health::dead_letter_queue().enqueue(DeadLetter {
id: envelope_id(body),
peer_id: peer_id.to_string(),
delivery_id,
tenant_id,
project_id,
delivery: DeadLetterDelivery::RemoteHttp {
endpoint_url: endpoint_url.to_string(),
},
original_message: String::from_utf8_lossy(body).into_owned(),
target_agent: target_agent.to_string(),
error,
attempts,
first_failed_at: failed_at.clone(),
last_failed_at: failed_at,
})
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
#[cfg(feature = "http-server")]
use axum::{Router, http::StatusCode, routing::post};
use super::*;
#[tokio::test]
async fn relay_peer_mode_has_no_singleton_authority_or_fallback() {
let peer = RelayPeerConfigV1 {
schema_version: 1,
peer_id: "destination".into(),
endpoint_url: "https://destination.example".into(),
bearer_token: "destination-bearer".into(),
channel_key: "destination-channel".into(),
origin_public_key: crate::core::a2a::relay::test_origin_public_key("destination"),
recipient_id: "recipient".into(),
allowed_tenant_ids: vec!["tenant".into()],
allowed_project_ids: vec!["project".into()],
allowed_content_types: vec![
crate::core::a2a_transport::TransportContentType::EvidenceBundle,
],
allowed_classifications: vec![lean_ctx_protocol::DataClassification::Internal],
max_hops: 4,
max_payload_bytes: 1024,
retry_count: 0,
retry_delay_ms: 0,
};
let transport = RemoteTransport::for_peer_table(
RelayPeerTableV1 {
schema_version: 1,
peers: vec![peer],
},
"origin",
Duration::from_secs(5),
false,
)
.unwrap();
assert!(
transport.config.endpoint_url.is_empty(),
"{:?}",
transport.config.endpoint_url
);
assert!(transport.config.auth_token.is_none());
assert!(transport.config.signing_key.is_none());
assert!(transport.config.recipient_id.is_none());
assert!(transport.config.tenant_id.is_none());
assert!(transport.config.project_id.is_none());
let restored: RemoteTransport =
serde_json::from_str(&serde_json::to_string(&transport).unwrap()).unwrap();
restored.config.validate().unwrap();
for current in [&transport, &restored] {
assert!(matches!(current.deliver(&envelope("payload")).await,
Err(TransportError::SerializationError(message))
if message == "peer transport requires explicit peer delivery"));
}
let mut mixed = transport.config.clone();
mixed.auth_token = Some("legacy-bearer".into());
assert!(RemoteTransport::new(mixed).is_err());
let mut incomplete = transport.config.clone();
incomplete.local_peer_id = None;
assert!(RemoteTransport::new(incomplete).is_err());
let mut empty = transport.config.clone();
empty.peers.clear();
assert!(RemoteTransport::new(empty).is_err());
let mut malformed = restored;
malformed.config.timeout = Duration::ZERO;
assert!(matches!(
malformed
.deliver_to("destination", &envelope("payload"))
.await,
Err(TransportError::SerializationError(_))
));
}
use crate::core::a2a_transport::{AgentIdentityV1, TransportContentType};
#[cfg(feature = "http-server")]
#[tokio::test]
async fn relay_clients_do_not_follow_redirects_after_deserialization() {
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
let hits = Arc::new(AtomicUsize::new(0));
let received = hits.clone();
let target_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let target_url = format!("http://{}/sink", target_listener.local_addr().unwrap());
let target = tokio::spawn(async move {
axum::serve(
target_listener,
Router::new().route(
"/sink",
post(move || {
let received = received.clone();
async move {
received.fetch_add(1, Ordering::SeqCst);
StatusCode::OK
}
}),
),
)
.await
.unwrap();
});
for status in [
StatusCode::TEMPORARY_REDIRECT,
StatusCode::PERMANENT_REDIRECT,
] {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let endpoint = format!("http://{}", listener.local_addr().unwrap());
let location = target_url.clone();
let source = tokio::spawn(async move {
axum::serve(
listener,
Router::new().route(
"/a2a/deliver",
post(move || {
let location = location.clone();
async move { (status, [(axum::http::header::LOCATION, location)]) }
}),
),
)
.await
.unwrap();
});
let original = RemoteTransport::new(RemoteTransportConfig {
endpoint_url: endpoint.clone(),
auth_token: Some("bearer".into()),
signing_key: Some("signing-key".into()),
recipient_id: Some("recipient".into()),
tenant_id: Some("tenant".into()),
project_id: Some("project".into()),
..RemoteTransportConfig::default()
})
.unwrap();
let restored: RemoteTransport =
serde_json::from_str(&serde_json::to_string(&original).unwrap()).unwrap();
for transport in [original, restored] {
let response = transport
.client
.post(format!("{endpoint}/a2a/deliver"))
.timeout(Duration::from_secs(5))
.body("private context")
.send()
.await
.unwrap();
assert_eq!(response.status(), status);
assert_eq!(hits.load(Ordering::SeqCst), 0);
}
source.abort();
let _ = source.await;
}
target.abort();
let _ = target.await;
}
fn envelope(payload: &str) -> TransportEnvelopeV1 {
TransportEnvelopeV1 {
format_version: 1,
sent_at: Utc::now(),
sender: AgentIdentityV1 {
agent_id: "sender".to_string(),
agent_type: "test".to_string(),
daemon_fingerprint: "fingerprint".to_string(),
capabilities: Vec::new(),
},
recipient: Some("recipient".to_string()),
content_type: TransportContentType::A2AMessage,
payload_json: payload.to_string(),
signature: None,
metadata: HashMap::new(),
}
}
#[test]
fn default_config_has_bounded_transport_values() {
let config = RemoteTransportConfig {
endpoint_url: "https://agent.example/api/".to_string(),
auth_token: Some("secret".to_string()),
signing_key: Some("signing-secret".to_string()),
recipient_id: Some("recipient".to_string()),
tenant_id: Some("tenant-a".to_string()),
project_id: Some("project-a".to_string()),
..RemoteTransportConfig::default()
};
assert!(config.validate().is_ok());
assert_eq!(config.timeout, Duration::from_secs(30));
assert_eq!(config.max_payload_bytes, 2_000_000);
assert_eq!(config.retry_count, 2);
assert_eq!(config.retry_delay, Duration::from_secs(1));
assert_eq!(
config.delivery_url().expect("valid URL").as_str(),
"https://agent.example/api/a2a/deliver"
);
for peer_id in ["p", "legacy", ""] {
let mut mixed = config.clone();
mixed.peer_id = Some(peer_id.into());
let restored: RemoteTransportConfig =
serde_json::from_str(&serde_json::to_string(&mixed).unwrap()).unwrap();
assert!(restored.validate().unwrap_err().contains("peer_id"));
assert!(RemoteTransport::new(restored).is_err());
}
}
#[cfg(feature = "http-server")]
#[tokio::test]
async fn task_response_is_complete_bounded_utf8_and_explicitly_unverified() {
let mut task = envelope("{}");
task.content_type = TransportContentType::A2ATask;
for size in [1, MAX_TASK_RESPONSE_BYTES] {
let body = "x".repeat(size);
let response = reqwest::Response::from(axum::http::Response::new(body.clone()));
assert_eq!(
read_task_response(response, &task).await.unwrap(),
Some(body)
);
}
for body in [
Vec::new(),
vec![b'x'; MAX_TASK_RESPONSE_BYTES + 1],
vec![0xff],
] {
let response = reqwest::Response::from(axum::http::Response::new(body));
assert!(read_task_response(response, &task).await.is_err());
}
let response = reqwest::Response::from(axum::http::Response::new(vec![0xff]));
assert_eq!(
read_task_response(response, &envelope("{}")).await.unwrap(),
None
);
}
#[cfg(feature = "http-server")]
#[tokio::test]
async fn interrupted_task_response_is_not_accepted_as_complete() {
let mut task = envelope("{}");
task.content_type = TransportContentType::A2ATask;
let chunks = futures::stream::iter([
Ok::<_, std::io::Error>(vec![b'{']),
Err(std::io::Error::other("interrupted response")),
]);
let response = reqwest::Response::from(axum::http::Response::new(
reqwest::Body::wrap_stream(chunks),
));
assert!(read_task_response(response, &task).await.is_err());
}
#[test]
fn config_validation_rejects_unbounded_or_unsupported_values() {
let zero_timeout = RemoteTransportConfig {
endpoint_url: "https://agent.example".to_string(),
timeout: Duration::ZERO,
auth_token: Some("secret".to_string()),
signing_key: Some("signing-secret".to_string()),
..RemoteTransportConfig::default()
};
let unsupported_scheme = RemoteTransportConfig {
endpoint_url: "file:///tmp/agent".to_string(),
auth_token: Some("secret".to_string()),
signing_key: Some("signing-secret".to_string()),
..RemoteTransportConfig::default()
};
assert!(zero_timeout.validate().is_err());
assert!(unsupported_scheme.validate().is_err());
let insecure_remote = RemoteTransportConfig {
endpoint_url: "http://agent.example".to_string(),
auth_token: Some("bearer-secret".to_string()),
signing_key: Some("signing-secret".to_string()),
recipient_id: Some("recipient".to_string()),
tenant_id: Some("tenant-a".to_string()),
project_id: Some("project-a".to_string()),
..RemoteTransportConfig::default()
};
assert_eq!(
insecure_remote.validate(),
Err("endpoint_url must use HTTPS except for loopback hosts".to_string())
);
assert!(
RemoteTransportConfig {
endpoint_url: "http://[::1]:8080".to_string(),
..insecure_remote
}
.validate()
.is_ok()
);
assert!(
RemoteTransportConfig {
endpoint_url: "https://user:secret@agent.example".to_string(),
auth_token: Some("secret".to_string()),
signing_key: Some("signing-secret".to_string()),
recipient_id: Some("recipient".to_string()),
tenant_id: Some("tenant-a".to_string()),
project_id: Some("project-a".to_string()),
..RemoteTransportConfig::default()
}
.validate()
.is_err()
);
assert!(
RemoteTransportConfig {
endpoint_url: "https://agent.example".to_string(),
..RemoteTransportConfig::default()
}
.validate()
.is_err()
);
assert_eq!(
RemoteTransportConfig {
endpoint_url: "https://agent.example".to_string(),
auth_token: Some("same-secret".to_string()),
signing_key: Some("same-secret".to_string()),
..RemoteTransportConfig::default()
}
.validate(),
Err("signing_key must be distinct from auth_token".to_string())
);
}
#[tokio::test]
async fn delivery_rejects_recipient_outside_configured_authority() {
let transport = RemoteTransport::new(RemoteTransportConfig {
endpoint_url: "http://127.0.0.1:9".to_string(),
auth_token: Some("bearer-secret".to_string()),
signing_key: Some("signing-secret".to_string()),
recipient_id: Some("different-recipient".to_string()),
tenant_id: Some("tenant-a".to_string()),
project_id: Some("project-a".to_string()),
retry_count: 0,
..RemoteTransportConfig::default()
})
.expect("valid transport config");
assert_eq!(
transport.deliver(&envelope("payload")).await,
Err(TransportError::SerializationError(
"envelope recipient does not match configured remote recipient".to_string()
))
);
}
#[cfg(feature = "http-server")]
#[tokio::test]
async fn remote_retry_refreshes_timestamp_and_signature_for_authority_receiver() {
let signing_key = "retry-signing-secret";
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let app = Router::new().route(
"/a2a/deliver",
post(move |body: axum::body::Bytes| async move {
let Ok(envelope) = serde_json::from_slice::<TransportEnvelopeV1>(&body) else {
return StatusCode::BAD_REQUEST;
};
let age = Utc::now().signed_duration_since(envelope.sent_at);
if age > chrono::Duration::seconds(300)
|| age < chrono::Duration::seconds(-30)
|| !envelope.verify_signature(signing_key.as_bytes())
{
StatusCode::UNAUTHORIZED
} else {
StatusCode::OK
}
}),
);
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let transport = RemoteTransport::new(RemoteTransportConfig {
endpoint_url: format!("http://{address}"),
auth_token: Some("bearer-secret".into()),
signing_key: Some(signing_key.into()),
recipient_id: Some("recipient".into()),
tenant_id: Some("tenant-a".into()),
project_id: Some("project-a".into()),
retry_count: 0,
..RemoteTransportConfig::default()
})
.unwrap();
let mut stale = envelope("payload");
stale.sent_at = Utc::now() - chrono::Duration::minutes(10);
stale.sign(signing_key.as_bytes()).unwrap();
let letter = DeadLetter {
id: "stale-letter".into(),
peer_id: "legacy".into(),
delivery_id: stale.stable_delivery_id().unwrap(),
tenant_id: "tenant-a".into(),
project_id: "project-a".into(),
delivery: DeadLetterDelivery::RemoteHttp {
endpoint_url: format!("http://{address}"),
},
original_message: serde_json::to_string(&stale).unwrap(),
target_agent: "recipient".into(),
error: "previous failure".into(),
attempts: 1,
first_failed_at: Utc::now().to_rfc3339(),
last_failed_at: Utc::now().to_rfc3339(),
};
assert_eq!(
transport
.retry_dead_letter(&letter)
.await
.unwrap()
.remote_status,
StatusCode::OK.as_u16()
);
server.abort();
}
#[cfg(feature = "http-server")]
#[tokio::test]
async fn permanent_remote_failure_is_persisted_with_exact_scope() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let app = Router::new().route(
"/a2a/deliver",
post(|| async { (StatusCode::BAD_REQUEST, "scope denied") }),
);
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let suffix = uuid::Uuid::new_v4().to_string();
let tenant_id = format!("tenant-{suffix}");
let project_id = format!("project-{suffix}");
let scope = crate::core::a2a::dlq::DlqScope::new(&tenant_id, &project_id).unwrap();
let transport = RemoteTransport::new(RemoteTransportConfig {
endpoint_url: format!("http://{address}"),
auth_token: Some("bearer-secret".into()),
signing_key: Some("signing-secret".into()),
recipient_id: Some("recipient".into()),
tenant_id: Some(tenant_id),
project_id: Some(project_id),
retry_count: 0,
..RemoteTransportConfig::default()
})
.unwrap();
assert!(matches!(
transport.deliver(&envelope("payload")).await,
Err(TransportError::RemoteError(400, _))
));
let entries = crate::core::ocla::health::dead_letter_queue()
.peek(&scope)
.unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].target_agent, "recipient");
assert!(matches!(
entries[0].delivery,
DeadLetterDelivery::RemoteHttp { .. }
));
server.abort();
}
#[test]
fn payload_size_limit_reports_serialized_size() {
let envelope = envelope("payload");
let serialized = serde_json::to_vec(&envelope).expect("serializable envelope");
let error = serialize_and_validate(&envelope, serialized.len() - 1)
.expect_err("payload must exceed configured limit");
assert_eq!(error, TransportError::PayloadTooLarge(serialized.len()));
}
#[test]
fn transport_error_variants_are_serializable_and_distinct() {
let variants = [
TransportError::PayloadTooLarge(10),
TransportError::Timeout,
TransportError::RemoteError(400, "bad request".to_string()),
TransportError::SerializationError("invalid JSON".to_string()),
TransportError::Exhausted(2),
TransportError::DeadLetterFailure("store unavailable".to_string()),
];
for variant in variants {
let json = serde_json::to_string(&variant).expect("serialize error");
let decoded = serde_json::from_str(&json).expect("deserialize error");
assert_eq!(variant, decoded);
}
}
#[test]
fn receipt_round_trips_without_losing_delivery_fields() {
let receipt = DeliveryReceipt {
envelope_id: "envelope:abc".to_string(),
delivered_at: Utc::now(),
remote_status: 202,
round_trip_ms: 17,
unverified_task_response: None,
};
let json = serde_json::to_string(&receipt).expect("serialize receipt");
let decoded: DeliveryReceipt = serde_json::from_str(&json).expect("deserialize receipt");
assert_eq!(decoded, receipt);
}
#[test]
fn envelope_ids_are_deterministic_and_content_addressed() {
let first = serialize_and_validate(&envelope("one"), usize::MAX).expect("serialize");
let first_again = first.clone();
let second = serialize_and_validate(&envelope("two"), usize::MAX).expect("serialize");
assert_eq!(envelope_id(&first), envelope_id(&first_again));
assert_ne!(envelope_id(&first), envelope_id(&second));
}
}