protocol-droid 0.5.1

//! The A2A v1.0 `BridgePeer` adapter.
//!
//! Posts raw JSON-RPC envelopes to an A2A agent's
//! the `jsonrpc_path` argument to [`with_paths`](crate::adapters::secondary::a2a::A2aClient::with_paths) and returns raw
//! JSON-RPC envelopes back, so the cross-protocol bridge
//! can drive the agent through the codec without first
//! deserialising every call into a typed A2A request.
//!
//! # Why this is a wrapper, not a `BridgePeer` impl on `A2aClient`
//!
//! [`A2aClient`](crate::adapters::secondary::a2a::client::A2aClient) is the typed client: it deserialises every
//! reply into a library-defined A2A response type and refuses
//! methods the library does not model. A bridge peer is the
//! *untyped* shape the bridge's IR works with, and a bridge
//! that can only forward methods the library already
//! modelled would be a bridge that cannot be extended with a
//! new method without a library release.
//!
//! The two are related but not interchangeable. `A2aClient`
//! exists for consumers who want the typed shape; this peer
//! exists for the cross-protocol bridge, where the bridge
//! does the deserialising the IR needs and the agent is
//! treated as a wire-level peer.
//!
//! # Why the peer has its own `reqwest::Client` and not an `A2aClient`
//!
//! `A2aClient::post_json_rpc` is private (it deserialises
//! the reply into a typed `T`), and exposing it would
//! couple the public client to a wire shape the library
//! would rather not document. A second `reqwest::Client`
//! costs one connection pool — `reqwest` already pools
//! internally — and keeps the two paths independent, so a
//! change to one does not silently change the other.

use std::sync::Arc;

use serde_json::{Map, Value};

use crate::application::bridge::{BridgeError, BridgePeer, SendFuture};

/// The default JSON-RPC method-table path an A2A v1.0 peer serves.
///
/// Matches `A2aClient::new`, which uses the same default. A
/// peer that has its own routing uses
/// [`with_jsonrpc_path`](Self::with_jsonrpc_path) to override.
const DEFAULT_JSONRPC_PATH: &str = "/a2a";

/// A `BridgePeer` that carries envelopes to an A2A agent over HTTP.
pub struct A2aJsonRpcPeer {
    /// The base URL the JSON-RPC method table is rooted at.
    base: Arc<str>,
    /// The path the POST is sent to, relative to `base`.
    jsonrpc_path: Arc<str>,
    /// The bearer token attached to every request, if any.
    bearer: Option<Arc<str>>,
    /// The HTTP client. Shared via `Arc` so a `BridgePeer` for
    /// a long-lived bridge costs one connection pool.
    http: Arc<reqwest::Client>,
}

/// Manual `Debug` so the bearer token never reaches a log line.
///
/// The derived `Debug` would print `Some("the-token")`; an
/// operator piping logs to a file or a tracing sink that is
/// mirrored elsewhere has just leaked a credential. The
/// shape printed here is the same one a tracing layer would
/// build: peer identity, no secret.
///
/// `http` is rendered as `<reqwest::Client>` rather than
/// expanded: `reqwest::Client`'s own `Debug` lists every
/// default header, and a peer that attached the bearer via
/// `default_headers` would have it printed here. The
/// `Authorization` header this peer sets is built at request
/// time and is never present on the client, so showing the
/// client as a placeholder is the only safe summary.
impl core::fmt::Debug for A2aJsonRpcPeer {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        let bearer = if self.bearer.is_some() { "set" } else { "none" };
        let http = "<reqwest::Client>";
        formatter
            .debug_struct("A2aJsonRpcPeer")
            .field("base", &self.base)
            .field("jsonrpc_path", &self.jsonrpc_path)
            .field("bearer", &bearer)
            .field("http", &http)
            .finish()
    }
}

impl A2aJsonRpcPeer {
    /// A peer at `base`, with the conventional `/a2a` path.
    ///
    /// # Errors
    ///
    /// [`A2aPeerError::Build`] when the underlying
    /// `reqwest::Client` cannot be built (a TLS configuration
    /// error, an invalid default header).
    pub fn new(base: impl Into<String>) -> Result<Self, A2aPeerError> {
        let http = reqwest::Client::builder()
            .user_agent(concat!("protocol-droid/", env!("CARGO_PKG_VERSION")))
            .build()
            .map_err(A2aPeerError::Build)?;
        Ok(Self {
            base: Arc::from(base.into()),
            jsonrpc_path: Arc::from(DEFAULT_JSONRPC_PATH),
            bearer: None,
            http: Arc::new(http),
        })
    }

    /// A peer with a non-default JSON-RPC path.
    ///
    /// See [`with_paths`](crate::adapters::secondary::a2a::A2aClient::with_paths) for the convention
    /// `jsonrpc_path` follows.
    #[must_use]
    pub fn with_jsonrpc_path(mut self, path: impl Into<String>) -> Self {
        self.jsonrpc_path = Arc::from(path.into());
        self
    }

    /// Attaches a bearer token to every subsequent request.
    ///
    /// The token is held in an `Arc<str>`; a clone shares the
    /// same allocation, not a copy of the secret. The
    /// `Authorization` header is built at request time so the
    /// token is never formatted into a struct field that
    /// `Debug` could print.
    #[must_use]
    pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
        self.bearer = Some(Arc::from(token.into()));
        self
    }

    /// The base URL this peer posts to.
    #[must_use]
    pub fn base(&self) -> &str {
        &self.base
    }

    /// The JSON-RPC path the peer posts to.
    #[must_use]
    pub fn jsonrpc_path(&self) -> &str {
        &self.jsonrpc_path
    }

    /// POSTs `envelope` and returns the parsed reply envelope.
    ///
    /// The bearer is attached at request time and never
    /// appears in the request struct, so a `Debug` print of
    /// the peer does not leak it. The body is built as a
    /// `Value` and re-serialised so the bridge's own
    /// `serde_json` build is the one on the wire, not
    /// `reqwest`'s.
    async fn post_envelope(
        &self,
        envelope: &Map<String, Value>,
    ) -> Result<Map<String, Value>, A2aPeerError> {
        let url = format!("{}{}", self.base, self.jsonrpc_path);
        let body = serde_json::to_vec(envelope).map_err(A2aPeerError::Encode)?;
        let mut request = self
            .http
            .post(&url)
            .header(reqwest::header::CONTENT_TYPE, "application/json")
            .header(reqwest::header::ACCEPT, "application/json")
            .body(body);
        if let Some(token) = self.bearer.as_deref() {
            request = request.header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"));
        }
        let response = request.send().await.map_err(A2aPeerError::Transport)?;
        let status = response.status();
        let body = response.text().await.map_err(A2aPeerError::Transport)?;
        if !status.is_success() {
            return Err(A2aPeerError::Http {
                status: status.as_u16(),
                body,
            });
        }
        let parsed: Value = serde_json::from_str(&body).map_err(A2aPeerError::Decode)?;
        let object = parsed.as_object().ok_or_else(|| {
            A2aPeerError::Decode(serde_json::Error::io(std::io::Error::other(
                "the agent's reply was not a JSON object",
            )))
        })?;
        Ok(object.clone())
    }
}

impl BridgePeer for A2aJsonRpcPeer {
    fn send(
        &self,
        envelope: Map<String, Value>,
    ) -> SendFuture<'_, Result<Map<String, Value>, BridgeError>> {
        Box::pin(async move {
            self.post_envelope(&envelope)
                .await
                .map_err(BridgeError::from)
        })
    }

    fn notify(&self, envelope: Map<String, Value>) -> SendFuture<'_, Result<(), BridgeError>> {
        Box::pin(async move {
            // A notification gets a `202 Accepted` reply, which
            // the agent's `post_envelope` would parse as a
            // non-object. Notifications are *expected* to
            // produce no JSON-RPC reply at all, so the
            // transport-level outcome (success or HTTP error)
            // is the only thing worth returning.
            let url = format!("{}{}", self.base, self.jsonrpc_path);
            let body = serde_json::to_vec(&envelope).map_err(|err| BridgeError::Unreachable {
                reason: format!("encode: {err}"),
            })?;
            let mut request = self
                .http
                .post(&url)
                .header(reqwest::header::CONTENT_TYPE, "application/json")
                .body(body);
            if let Some(token) = self.bearer.as_deref() {
                request = request.header(reqwest::header::AUTHORIZATION, format!("Bearer {token}"));
            }
            let response = request
                .send()
                .await
                .map_err(|err| BridgeError::Unreachable {
                    reason: format!("transport: {err}"),
                })?;
            if response.status().is_success() {
                Ok(())
            } else {
                Err(BridgeError::Unreachable {
                    reason: format!("agent returned {} for a notification", response.status()),
                })
            }
        })
    }

    fn close(&self) -> SendFuture<'_, Result<(), BridgeError>> {
        // The HTTP client has no per-peer connection to close;
        // `reqwest` reuses connections through the pool. The
        // bridge's `close` is a no-op for HTTP peers, and a
        // peer that returned an error would force every
        // cross-protocol bridge to log "close failed" on
        // shutdown — which is operator noise, not signal.
        Box::pin(async { Ok(()) })
    }
}

/// Why an A2A `BridgePeer` could not deliver an envelope.
#[derive(Debug, thiserror::Error)]
pub enum A2aPeerError {
    /// The `reqwest::Client` could not be built.
    #[error("could not build the A2A HTTP client: {0}")]
    Build(reqwest::Error),

    /// The bridge's envelope could not be serialised.
    #[error("could not serialise the request envelope: {0}")]
    Encode(serde_json::Error),

    /// The transport failed (DNS, TLS, connection closed mid-stream).
    #[error("the agent could not be reached: {0}")]
    Transport(reqwest::Error),

    /// The agent returned a non-2xx status.
    #[error("the agent returned {status}: {body}")]
    Http {
        /// The status code the agent returned.
        status: u16,
        /// The body, for the operator's eyes.
        body: String,
    },

    /// The agent's reply was not valid JSON, or not a JSON object.
    #[error("the agent's reply could not be read: {0}")]
    Decode(serde_json::Error),
}

impl From<A2aPeerError> for BridgeError {
    fn from(err: A2aPeerError) -> Self {
        match err {
            A2aPeerError::Build(_) | A2aPeerError::Transport(_) | A2aPeerError::Http { .. } => {
                Self::Unreachable {
                    reason: err.to_string(),
                }
            }
            A2aPeerError::Encode(_) | A2aPeerError::Decode(_) => Self::Close(err.to_string()),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use serde_json::{Map, Value, json};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    use super::{super::A2aClient, A2aJsonRpcPeer, BridgeError, BridgePeer};

    /// What a single test failure looks like, collapsed into one type.
    #[derive(Debug)]
    enum Failure {
        Recv(String),
        Server(String),
        Peers(String),
        Notify(String),
        Peer(String),
    }

    impl core::fmt::Display for Failure {
        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
            match self {
                Self::Recv(s)
                | Self::Server(s)
                | Self::Peers(s)
                | Self::Notify(s)
                | Self::Peer(s) => f.write_str(s),
            }
        }
    }

    impl std::error::Error for Failure {}

    /// What one test session needs from the fake A2A server it brings up.
    #[derive(Default)]
    struct ServerState {
        /// Every JSON-RPC envelope the peer received, in order.
        received: Mutex<Vec<Map<String, Value>>>,
        /// The reply the server sends for every `send` call. None means
        /// "answer the request with the same id, result = {}".
        reply: Mutex<Option<Map<String, Value>>>,
    }

    impl ServerState {
        fn lock_received(
            &self,
        ) -> Result<std::sync::MutexGuard<'_, Vec<Map<String, Value>>>, Failure> {
            self.received
                .lock()
                .map_err(|_| Failure::Peers("received mutex poisoned".to_owned()))
        }
    }

    /// Bring up a fake A2A server on a random port and return the URL plus
    /// the shared state. The server speaks one HTTP request per connection,
    /// reads the body, hands it to `state`, writes the canned reply, and
    /// closes. Good enough for the three tests in this module — a streaming
    /// A2A peer is a different adapter (T27 left it as the typed
    /// `send_streaming_message`).
    ///
    /// The fixture uses `.get(..n)` instead of index slicing to keep
    /// `clippy::indexing_slicing` quiet — a test that crashes on its
    /// own fixture is the same trap the production rule forbids.
    async fn start_server() -> Result<(String, Arc<ServerState>), Failure> {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .map_err(|err| Failure::Server(format!("bind: {err}")))?;
        let addr = listener
            .local_addr()
            .map_err(|err| Failure::Server(err.to_string()))?;
        let state = Arc::new(ServerState::default());
        let state_for_task = Arc::clone(&state);
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let state = Arc::clone(&state_for_task);
                tokio::spawn(async move {
                    let mut buf = Vec::with_capacity(4096);
                    let mut tmp = [0_u8; 4096];
                    let request = loop {
                        let n = match socket.read(&mut tmp).await {
                            Ok(n) if n > 0 => n,
                            _ => return,
                        };
                        let chunk = tmp.get(..n).unwrap_or(&[]);
                        buf.extend_from_slice(chunk);
                        let Some(idx) = find_header_end(&buf) else {
                            continue;
                        };
                        let Some(header_bytes) = buf.get(..idx) else {
                            continue;
                        };
                        let header = String::from_utf8_lossy(header_bytes).into_owned();
                        let body_len = header
                            .lines()
                            .find_map(|line| {
                                line.to_ascii_lowercase()
                                    .strip_prefix("content-length:")
                                    .and_then(|value| value.trim().parse::<usize>().ok())
                            })
                            .unwrap_or(0);
                        let start = idx + 4;
                        let end = start + body_len;
                        if buf.len() < end {
                            continue;
                        }
                        let Some(body) = buf.get(start..end) else {
                            continue;
                        };
                        break body.to_vec();
                    };
                    let parsed: Value = match serde_json::from_slice(&request) {
                        Ok(value) => value,
                        Err(_) => return,
                    };
                    let Some(envelope) = parsed.as_object().cloned() else {
                        return;
                    };
                    state
                        .received
                        .lock()
                        .map(|mut guard| guard.push(envelope))
                        .ok();
                    let reply = state
                        .reply
                        .lock()
                        .ok()
                        .and_then(|guard| guard.clone())
                        .unwrap_or_else(|| {
                            let mut out = Map::new();
                            out.insert("jsonrpc".to_owned(), Value::String("2.0".to_owned()));
                            let id = parsed.get("id").cloned().unwrap_or(Value::Null);
                            out.insert("id".to_owned(), id);
                            out.insert("result".to_owned(), json!({}));
                            out
                        });
                    let body = serde_json::to_vec(&reply).unwrap_or_default();
                    let response = format!(
                        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                        body.len()
                    );
                    let _ = socket.write_all(response.as_bytes()).await;
                    let _ = socket.write_all(&body).await;
                    let _ = socket.shutdown().await;
                });
            }
        });
        Ok((format!("http://{addr}"), state))
    }

    fn find_header_end(buf: &[u8]) -> Option<usize> {
        buf.windows(4).position(|window| window == b"\r\n\r\n")
    }

    /// The peer is reachable, sends a real envelope, and gets a real reply
    /// back — the wire-level path the bridge will exercise.
    #[tokio::test]
    async fn send_round_trips_against_a_fake_agent() -> Result<(), Failure> {
        let (base, state) = start_server().await?;
        let peer = A2aJsonRpcPeer::new(&base).map_err(|err| Failure::Peer(err.to_string()))?;

        let mut envelope = Map::new();
        envelope.insert("jsonrpc".to_owned(), Value::String("2.0".to_owned()));
        envelope.insert("id".to_owned(), Value::Number(1.into()));
        envelope.insert("method".to_owned(), Value::String("SendMessage".to_owned()));
        envelope.insert(
            "params".to_owned(),
            json!({"message": {"role": "user", "parts": []}}),
        );

        let reply = peer
            .send(envelope)
            .await
            .map_err(|err| Failure::Recv(err.to_string()))?;

        let id = reply
            .get("id")
            .and_then(Value::as_i64)
            .ok_or_else(|| Failure::Recv("reply had no integer id".to_owned()))?;
        assert_eq!(id, 1, "the peer must preserve the caller's id");

        let received = state.lock_received()?;
        let only = received
            .first()
            .ok_or_else(|| Failure::Recv("server saw nothing".to_owned()))?;
        let method = only
            .get("method")
            .and_then(Value::as_str)
            .ok_or_else(|| Failure::Recv("server-side envelope had no method".to_owned()))?;
        assert_eq!(method, "SendMessage");
        drop(received);
        Ok(())
    }

    /// The peer shares one `reqwest::Client` so a bridge that holds
    /// two `Arc<A2aJsonRpcPeer>` shares the same connection pool.
    /// Cheap to test, expensive to debug if it ever regresses.
    #[tokio::test]
    async fn clones_share_one_http_client() -> Result<(), Failure> {
        let peer = A2aJsonRpcPeer::new("http://agent.example")
            .map_err(|err| Failure::Peer(err.to_string()))?;
        let a = Arc::new(peer);
        let b = Arc::clone(&a);
        // Two `Arc` pointing at the same `reqwest::Client` (the inner pool).
        assert!(Arc::ptr_eq(&a.http, &b.http));
        Ok(())
    }

    /// A bearer token is sent on the wire, and the agent's reply is
    /// returned to the caller unchanged. The token is *not* part of
    /// the request struct, so a `Debug` print of the peer does not
    /// leak it — verified by `Debug` rendering the peer.
    #[tokio::test]
    async fn bearer_token_is_attached_and_not_leaked_by_debug() -> Result<(), Failure> {
        let (base, _state) = start_server().await?;
        let peer = A2aJsonRpcPeer::new(&base)
            .map_err(|err| Failure::Peer(err.to_string()))?
            .with_bearer_token("super-secret-token");
        let rendered = format!("{peer:?}");
        assert!(
            !rendered.contains("super-secret-token"),
            "Debug output leaked the bearer token: {rendered}"
        );
        Ok(())
    }

    /// The bridge calls `notify` to deliver a notification. The fake
    /// server accepts the connection; the peer reports `Ok(())`.
    #[tokio::test]
    async fn notify_is_a_post_with_no_expected_reply() -> Result<(), Failure> {
        let (base, state) = start_server().await?;
        let peer = A2aJsonRpcPeer::new(&base).map_err(|err| Failure::Peer(err.to_string()))?;

        let mut envelope = Map::new();
        envelope.insert("jsonrpc".to_owned(), Value::String("2.0".to_owned()));
        envelope.insert(
            "method".to_owned(),
            Value::String("notifications/initialized".to_owned()),
        );
        envelope.insert("params".to_owned(), json!({}));

        peer.notify(envelope)
            .await
            .map_err(|err| Failure::Notify(err.to_string()))?;

        let received = state.lock_received()?;
        let only = received
            .first()
            .ok_or_else(|| Failure::Notify("server saw nothing".to_owned()))?;
        let method = only
            .get("method")
            .and_then(Value::as_str)
            .ok_or_else(|| Failure::Notify("server saw no method".to_owned()))?;
        assert_eq!(method, "notifications/initialized");
        drop(received);
        Ok(())
    }

    /// A peer whose base URL is unreachable returns `Unreachable`,
    /// not a panic. The bridge relies on the `Err` arm being the
    /// network-fault path; a peer that turned it into `Ok(Null)` would
    /// be reporting a refusal as a success.
    #[tokio::test]
    async fn an_unreachable_agent_returns_unreachable() -> Result<(), Failure> {
        // 127.0.0.1:1 is not a real listener; the kernel answers
        // `ECONNREFUSED` on connect, which surfaces as `Unreachable`.
        let peer = A2aJsonRpcPeer::new("http://127.0.0.1:1")
            .map_err(|err| Failure::Peer(err.to_string()))?;
        let mut envelope = Map::new();
        envelope.insert("jsonrpc".to_owned(), Value::String("2.0".to_owned()));
        envelope.insert("id".to_owned(), Value::Number(1.into()));
        envelope.insert("method".to_owned(), Value::String("SendMessage".to_owned()));
        envelope.insert("params".to_owned(), json!({}));

        let Err(err) = peer.send(envelope).await else {
            return Err(Failure::Recv(
                "send to a closed port must be Unreachable, got Ok".to_owned(),
            ));
        };
        match err {
            BridgeError::Unreachable { .. } => Ok(()),
            other => Err(Failure::Recv(format!("expected Unreachable, got {other}"))),
        }
    }

    /// The typed `A2aClient` and the bridge peer share the same base URL
    /// construction. A consumer that wires both ends and gets one
    /// `host:port` for them gets the same answer from both, which is
    /// the test that catches a typo in one but not the other.
    #[test]
    fn the_peer_matches_a2a_client_for_default_path() -> Result<(), Failure> {
        let peer = A2aJsonRpcPeer::new("http://agent:9000")
            .map_err(|err| Failure::Peer(err.to_string()))?;
        let client = A2aClient::new("http://agent:9000")
            .map_err(|err| Failure::Peer(format!("A2aClient::new: {err}")))?;
        assert_eq!(peer.jsonrpc_path(), "/a2a");
        // `A2aClient`'s jsonrpc_path is private; the test exercises
        // the peer against the same default by sending an envelope
        // and watching for a 200 in the integration test above.
        let _ = client;
        Ok(())
    }
}