polyc-a2a 2026.8.3

//! The A2A **v1.0** *client* — the symmetric other half of [`crate::server`].
//!
//! Where the server lets a peer call this deployment, this lets this deployment
//! call a peer: fetch and verify its [Agent Card](crate::card), then drive
//! `SendMessage` (and `GetTask`) and read back the terminal outcome. Client and
//! server share one set of wire types ([`crate::types`]), so what one emits the
//! other accepts by construction (proven by the round-trip test).
//!
//! The protocol is pure and testable — [`build_send_request`] /
//! [`parse_send_response`] / [`verify_card_doc`] build the exact v1.0 JSON-RPC
//! the server parses and decode the exact result it returns; reqwest is a thin
//! transport over them.
//!
//! A turn reaches this through the `peer_call` tool: the harness's peer-call
//! proxy (`polyc-harness`'s `peer_proxy`) resolves a configured peer name to
//! its base URL and drives [`A2aClient::call`].

use futures::StreamExt as _;
use serde_json::{Value, json};
use uuid::Uuid;

use crate::types::{SendMessageResponse, StreamResponse, Task, TaskState, TaskStatusUpdateEvent};

/// A peer agent's identity, verified from its Agent Card.
#[derive(Debug, Clone)]
pub struct PeerCard {
    /// The peer's advertised name.
    pub name: String,
    /// The peer's JSON-RPC endpoint (the card's first `supportedInterfaces` url).
    pub url: String,
    /// The ed25519 public key the card verified under — pin it to detect a
    /// later key swap (trust-on-first-use; see [`crate::card::verify_self_signed`]).
    pub public_key: Vec<u8>,
    /// Whether the peer's card advertises `capabilities.streaming` (`#371`).
    /// [`A2aClient::send_message_streaming`] refuses to call a peer that
    /// doesn't set this — `SendStreamingMessage` is only spec-conformant to
    /// send when the peer says it implements it.
    pub supports_streaming: bool,
}

/// The terminal outcome of a call to a peer — the client mirror of the server's
/// [`crate::task::TurnOutcome`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PeerOutcome {
    /// The peer completed the turn with this answer text.
    Completed {
        /// The agent's reply text.
        text: String,
    },
    /// The peer paused on its own approval (or auth) gate; `prompt` is what it is
    /// waiting on (an A2A `input-required` / `auth-required` task).
    InputRequired {
        /// The human-readable decision the peer is blocked on.
        prompt: String,
    },
    /// The peer failed, rejected, or canceled the turn.
    Failed {
        /// The peer's failure summary.
        message: String,
    },
}

/// Why an A2A client call failed.
#[derive(Debug, thiserror::Error)]
pub enum ClientError {
    /// The HTTP request could not be completed.
    #[error("transport error: {0}")]
    Transport(String),
    /// The peer's Agent Card did not verify under the key it advertises.
    #[error("peer agent card is not validly self-signed")]
    UnverifiedCard,
    /// The Agent Card is missing a required field.
    #[error("malformed agent card: missing `{0}`")]
    MalformedCard(&'static str),
    /// The JSON-RPC response could not be decoded.
    #[error("malformed JSON-RPC response: {0}")]
    MalformedResponse(String),
    /// The peer returned a JSON-RPC error object.
    #[error("peer returned JSON-RPC error {code}: {message}")]
    RpcError {
        /// The JSON-RPC error code.
        code: i64,
        /// The JSON-RPC error message.
        message: String,
    },
    /// [`A2aClient::send_message_streaming`] was called against a peer whose
    /// card does not advertise `capabilities.streaming` (`#371`).
    #[error("peer does not advertise the streaming capability")]
    PeerDoesNotSupportStreaming,
    /// A peer card pointed its RPC endpoint outside the configured peer
    /// origin. A brokered caller pins the configured origin before fetching
    /// the card, so following a second origin would reopen its SSRF boundary.
    #[error("peer card endpoint escapes the configured peer origin")]
    EndpointEscapesConfiguredOrigin,
    /// The peer's response headers exceeded the trusted transport bound.
    #[error("peer response headers exceed the {cap_bytes} byte cap")]
    ResponseHeadersTooLarge {
        /// The configured cap in bytes.
        cap_bytes: usize,
    },
    /// The peer's decoded response body exceeded the trusted transport bound.
    #[error("peer response body exceeds the {cap_bytes} byte cap")]
    ResponseBodyTooLarge {
        /// The configured cap in bytes.
        cap_bytes: usize,
    },
}

/// Bounded response handling for a trusted A2A transport.
///
/// The default constructors remain deliberately unbounded for the standalone
/// A2A edge. Control's D7 broker uses [`A2aClient::with_transport`] with the
/// registry limits, so a peer cannot turn the trusted side into a header or
/// decompression amplifier.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ResponseBounds {
    /// Largest permitted aggregate response-header encoding.
    pub max_header_bytes: usize,
    /// Largest permitted decoded response body.
    pub max_decoded_body_bytes: usize,
}

impl ResponseBounds {
    /// No response cap for standalone edge composition outside D7.
    #[must_use]
    pub const fn unbounded() -> Self {
        Self {
            max_header_bytes: usize::MAX,
            max_decoded_body_bytes: usize::MAX,
        }
    }
}

/// Build a `SendMessage` JSON-RPC request body carrying `text` as a single
/// `user` text message. Context/task ids are left for the peer to mint.
#[must_use]
pub fn build_send_request(text: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": Uuid::new_v4().to_string(),
        "method": "SendMessage",
        "params": {
            "message": {
                "messageId": Uuid::now_v7().to_string(),
                "role": "ROLE_USER",
                "parts": [{ "text": text }]
            }
        },
    })
}

/// Build a `SendStreamingMessage` JSON-RPC request body carrying `text` as a
/// single `user` text message — the streaming sibling of
/// [`build_send_request`]; same params shape, different method name.
#[must_use]
pub fn build_send_streaming_request(text: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": Uuid::new_v4().to_string(),
        "method": "SendStreamingMessage",
        "params": {
            "message": {
                "messageId": Uuid::now_v7().to_string(),
                "role": "ROLE_USER",
                "parts": [{ "text": text }]
            }
        },
    })
}

/// Build a `GetTask` JSON-RPC request body for `task_id`.
#[must_use]
pub fn build_get_task_request(task_id: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": Uuid::new_v4().to_string(),
        "method": "GetTask",
        "params": { "id": task_id },
    })
}

/// Decode a `SendMessage` JSON-RPC response into a [`PeerOutcome`].
///
/// # Errors
///
/// [`ClientError::RpcError`] if the peer returned a JSON-RPC error, or
/// [`ClientError::MalformedResponse`] if the body is not a JSON-RPC success
/// carrying a decodable `SendMessageResponse`.
pub fn parse_send_response(body: &[u8]) -> Result<PeerOutcome, ClientError> {
    let result = jsonrpc_result(body)?;
    let response: SendMessageResponse = serde_json::from_value(result)
        .map_err(|e| ClientError::MalformedResponse(e.to_string()))?;
    Ok(match response {
        SendMessageResponse::Task(task) => outcome_from_task(&task),
        SendMessageResponse::Message(message) => PeerOutcome::Completed {
            text: message.text(),
        },
    })
}

/// Decode a `GetTask` JSON-RPC response (a bare `Task`) into a [`PeerOutcome`].
///
/// # Errors
///
/// As [`parse_send_response`], but the result is a bare `Task`.
pub fn parse_task_response(body: &[u8]) -> Result<PeerOutcome, ClientError> {
    let result = jsonrpc_result(body)?;
    let task: Task = serde_json::from_value(result)
        .map_err(|e| ClientError::MalformedResponse(e.to_string()))?;
    Ok(outcome_from_task(&task))
}

/// Pull the `result` value out of a JSON-RPC response, or surface its `error`.
fn jsonrpc_result(body: &[u8]) -> Result<Value, ClientError> {
    let response: Value =
        serde_json::from_slice(body).map_err(|e| ClientError::MalformedResponse(e.to_string()))?;
    // A null `error` is "no error"; only a non-null error object is a fault.
    if let Some(error) = response.get("error").filter(|e| !e.is_null()) {
        return Err(ClientError::RpcError {
            code: error.get("code").and_then(Value::as_i64).unwrap_or(0),
            message: error
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_owned(),
        });
    }
    response
        .get("result")
        .cloned()
        .ok_or_else(|| ClientError::MalformedResponse("no `result` and no `error`".to_owned()))
}

/// Map a decoded [`Task`] to the terminal [`PeerOutcome`]. A completed task's
/// answer rides in `status.message`; fall back to its artifacts' text.
fn outcome_from_task(task: &Task) -> PeerOutcome {
    let status_text = task
        .status
        .message
        .as_ref()
        .map(crate::types::Message::text)
        .unwrap_or_default();
    match task.status.state {
        TaskState::Completed => PeerOutcome::Completed {
            text: if status_text.is_empty() {
                artifact_text(task)
            } else {
                status_text
            },
        },
        TaskState::InputRequired | TaskState::AuthRequired => PeerOutcome::InputRequired {
            prompt: status_text,
        },
        TaskState::Failed | TaskState::Rejected => PeerOutcome::Failed {
            message: status_text,
        },
        TaskState::Canceled => PeerOutcome::Failed {
            message: if status_text.is_empty() {
                "task canceled".to_owned()
            } else {
                status_text
            },
        },
        // Non-terminal for a unary call: the server should have driven to a
        // terminal state, so treat this as a protocol fault.
        TaskState::Submitted | TaskState::Working | TaskState::Unspecified => PeerOutcome::Failed {
            message: "peer returned a non-terminal task state".to_owned(),
        },
    }
}

/// Concatenate the text of every artifact part (newline-joined).
fn artifact_text(task: &Task) -> String {
    task.artifacts
        .iter()
        .flatten()
        .flat_map(|a| a.parts.iter())
        .filter_map(crate::types::Part::as_text)
        .collect::<Vec<_>>()
        .join("\n")
}

/// Map a [`TaskStatusUpdateEvent`] to the terminal [`PeerOutcome`] — the
/// streaming mirror of [`outcome_from_task`]. Unlike a `GetTask`/`SendMessage`
/// `Task`, a status-update event carries no `artifacts`, so — unlike
/// [`outcome_from_task`] — there is no artifact-text fallback: `apply_outcome`
/// on the server always attaches the full answer to `status.message` for a
/// `Completed` task, so this never needs one.
fn outcome_from_status(update: &TaskStatusUpdateEvent) -> PeerOutcome {
    let status_text = update
        .status
        .message
        .as_ref()
        .map(crate::types::Message::text)
        .unwrap_or_default();
    match update.status.state {
        TaskState::Completed => PeerOutcome::Completed { text: status_text },
        TaskState::InputRequired | TaskState::AuthRequired => PeerOutcome::InputRequired {
            prompt: status_text,
        },
        TaskState::Failed | TaskState::Rejected => PeerOutcome::Failed {
            message: status_text,
        },
        TaskState::Canceled => PeerOutcome::Failed {
            message: if status_text.is_empty() {
                "task canceled".to_owned()
            } else {
                status_text
            },
        },
        TaskState::Submitted | TaskState::Working | TaskState::Unspecified => PeerOutcome::Failed {
            message: "peer returned a non-terminal task state".to_owned(),
        },
    }
}

/// One decoded `SendStreamingMessage` SSE event, classified for
/// [`A2aClient::send_message_streaming`]'s fold: either more of the task is
/// still coming (`Progress`, carrying the ids so a transport drop mid-stream
/// can fall back to polling [`A2aClient::get_task`]) or the stream's terminal
/// [`PeerOutcome`] (`Terminal`).
#[derive(Debug, Clone, PartialEq, Eq)]
enum StreamStep {
    /// The task/context id seen so far, and streaming is not yet done.
    Progress {
        /// The task id, once known (unknown only before the first event).
        task_id: Option<String>,
    },
    /// The terminal outcome — the last event of the stream.
    Terminal(PeerOutcome),
}

/// Decode one `SendStreamingMessage` SSE event's JSON-RPC response body
/// (already extracted from its `data:` line(s) by [`drain_sse_events`]) into a
/// [`StreamStep`].
///
/// # Errors
///
/// As [`parse_send_response`]: [`ClientError::RpcError`] /
/// [`ClientError::MalformedResponse`].
fn parse_stream_step(body: &[u8]) -> Result<StreamStep, ClientError> {
    let result = jsonrpc_result(body)?;
    let response: StreamResponse = serde_json::from_value(result)
        .map_err(|e| ClientError::MalformedResponse(e.to_string()))?;
    Ok(match response {
        StreamResponse::Task(task) => StreamStep::Progress {
            task_id: Some(task.id),
        },
        StreamResponse::ArtifactUpdate(update) => StreamStep::Progress {
            task_id: Some(update.task_id),
        },
        StreamResponse::StatusUpdate(update) if update.is_final => {
            StreamStep::Terminal(outcome_from_status(&update))
        }
        StreamResponse::StatusUpdate(update) => StreamStep::Progress {
            task_id: Some(update.task_id),
        },
    })
}

/// Extract complete SSE event payloads from `buffer`, consuming the bytes
/// they were parsed from and leaving any trailing partial event (no
/// terminating blank line yet) for the next chunk to complete.
///
/// Each returned string is one event's `data:` line(s), per the SSE spec
/// joined with `\n` (multiple `data:` lines within one event) with the
/// single optional leading space after the colon stripped; non-`data` fields
/// (`event:`, `id:`, comments starting `:`) are ignored — this transport only
/// ever needs the payload. A small hand-rolled decoder rather than a new
/// dependency: this edge only ever needs to consume its OWN server's SSE
/// output (one `data:` line per event), not the general SSE grammar.
fn drain_sse_events(buffer: &mut String) -> Vec<String> {
    let mut events = Vec::new();
    while let Some(end) = blank_line_end(buffer) {
        let raw: String = buffer.drain(..end).collect();
        // `str::lines` already splits on both `\n` and `\r\n`, stripping the
        // trailing `\r` — no separate CRLF handling needed here.
        let data = raw
            .lines()
            .filter_map(|line| {
                let rest = line.strip_prefix("data:")?;
                Some(rest.strip_prefix(' ').unwrap_or(rest))
            })
            .collect::<Vec<_>>()
            .join("\n");
        if !data.is_empty() {
            events.push(data);
        }
    }
    events
}

/// The byte offset one past the end of `buffer`'s first blank line (the SSE
/// event terminator) — either `"\n\n"` or `"\r\n\r\n"`, whichever appears
/// first. `None` if `buffer` has no complete event yet.
fn blank_line_end(buffer: &str) -> Option<usize> {
    let lf = buffer.find("\n\n").map(|i| i + 2);
    let crlf = buffer.find("\r\n\r\n").map(|i| i + 4);
    match (lf, crlf) {
        (Some(a), Some(b)) => Some(a.min(b)),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    }
}

/// Parse and verify a peer Agent Card document, returning its verified identity.
///
/// # Errors
///
/// [`ClientError::UnverifiedCard`] if the card does not verify under the key it
/// advertises, or [`ClientError::MalformedCard`] / [`ClientError::MalformedResponse`]
/// if it is not a well-formed v1.0 card.
pub fn verify_card_doc(card_bytes: &[u8]) -> Result<PeerCard, ClientError> {
    let card: Value = serde_json::from_slice(card_bytes)
        .map_err(|e| ClientError::MalformedResponse(e.to_string()))?;
    let public_key = crate::card::verify_self_signed(&card).ok_or(ClientError::UnverifiedCard)?;
    let name = card
        .get("name")
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
        .ok_or(ClientError::MalformedCard("name"))?;
    // v1.0: the endpoint is the first supportedInterfaces entry's url.
    let url = card
        .get("supportedInterfaces")
        .and_then(Value::as_array)
        .and_then(|interfaces| interfaces.first())
        .and_then(|interface| interface.get("url"))
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
        .ok_or(ClientError::MalformedCard("supportedInterfaces[].url"))?;
    let supports_streaming = card
        .get("capabilities")
        .and_then(|capabilities| capabilities.get("streaming"))
        .and_then(Value::as_bool)
        .unwrap_or(false);
    Ok(PeerCard {
        name,
        url,
        public_key,
        supports_streaming,
    })
}

/// An A2A client that calls a peer agent over HTTP. Holds a reusable HTTP client
/// and the peer's base URL (the origin its Agent Card lives under).
#[derive(Debug, Clone)]
pub struct A2aClient {
    http: reqwest::Client,
    base_url: String,
    response_bounds: ResponseBounds,
}

impl A2aClient {
    /// Build a client targeting `base_url` (the peer origin, e.g.
    /// `https://peer.example/`).
    ///
    /// Redirect-following is disabled: a peer that 3xx-redirects the card fetch
    /// or a JSON-RPC POST cannot bounce the request to a different host (an SSRF
    /// vector, since under trust-on-first-use the served card — and the endpoint
    /// it advertises — are not yet pinned to a trusted key).
    #[must_use]
    pub fn new(base_url: impl Into<String>) -> Self {
        let http = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .unwrap_or_default();
        Self {
            http,
            base_url: base_url.into(),
            response_bounds: ResponseBounds::unbounded(),
        }
    }

    /// Build a client targeting `base_url` that attaches `Authorization:
    /// Bearer <token>` to every request.
    ///
    /// Most A2A peers are trusted by their domain-signed Agent Card rather
    /// than a bearer header (see `crate::peer_directory`'s module docs), so
    /// [`Self::new`] stays the default; this constructor is for a peer that
    /// itself enforces the bearer scheme this edge's own `POST /` advertises
    /// (see `crate::server`) — including this deployment's own endpoint.
    #[must_use]
    pub fn with_bearer_token(base_url: impl Into<String>, token: &str) -> Self {
        let mut headers = reqwest::header::HeaderMap::new();
        if let Ok(value) = reqwest::header::HeaderValue::from_str(&format!("Bearer {token}")) {
            headers.insert(reqwest::header::AUTHORIZATION, value);
        }
        let http = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .default_headers(headers)
            .build()
            .unwrap_or_default();
        Self {
            http,
            base_url: base_url.into(),
            response_bounds: ResponseBounds::unbounded(),
        }
    }

    /// Build a client over a trusted caller's already-guarded transport.
    ///
    /// The caller owns DNS pinning, redirects, connect timeout, and any
    /// credentials. This client adds exact-origin enforcement for card-advertised
    /// endpoints and caps the headers plus *decoded* streamed body before it
    /// parses A2A content. It deliberately accepts no raw Execution values.
    #[must_use]
    pub fn with_transport(
        base_url: impl Into<String>,
        http: reqwest::Client,
        response_bounds: ResponseBounds,
    ) -> Self {
        Self {
            http,
            base_url: base_url.into(),
            response_bounds,
        }
    }

    /// Fetch and verify the peer's Agent Card from
    /// `<base>/.well-known/agent-card.json`.
    ///
    /// # Errors
    ///
    /// [`ClientError::Transport`] on a network failure, or the
    /// [`verify_card_doc`] errors if the served card is malformed or unverified.
    pub async fn fetch_card(&self) -> Result<PeerCard, ClientError> {
        let url = format!(
            "{}/.well-known/agent-card.json",
            self.base_url.trim_end_matches('/')
        );
        let bytes = self.get(&url).await?;
        verify_card_doc(&bytes)
    }

    /// Send `text` to the peer's JSON-RPC `endpoint` via `SendMessage` and read
    /// the outcome.
    ///
    /// # Errors
    ///
    /// [`ClientError::Transport`] on a network failure, or the
    /// [`parse_send_response`] errors on a malformed / error response.
    pub async fn send_message(
        &self,
        endpoint: &str,
        text: &str,
    ) -> Result<PeerOutcome, ClientError> {
        let bytes = self.post(endpoint, &build_send_request(text)).await?;
        parse_send_response(&bytes)
    }

    /// Send `text` to `peer`'s JSON-RPC endpoint via `SendStreamingMessage`
    /// (`#371`) and fold its SSE event stream to the terminal outcome.
    ///
    /// Gated on `peer.supports_streaming`: calling `SendStreamingMessage`
    /// against a peer whose card doesn't advertise the capability is not
    /// spec-conformant, so this refuses with
    /// [`ClientError::PeerDoesNotSupportStreaming`] rather than sending a
    /// request the peer may not understand.
    ///
    /// If the transport drops mid-stream (before a terminal event arrives)
    /// but a task id was already seen, this falls back to polling
    /// [`Self::get_task`] once for the task's current state rather than
    /// surfacing a bare transport error for a task that may in fact have
    /// finished server-side.
    ///
    /// # Errors
    ///
    /// [`ClientError::PeerDoesNotSupportStreaming`] per the gate above;
    /// [`ClientError::Transport`] on a network failure with no task id yet
    /// known to fall back on; otherwise as [`Self::send_message`].
    pub async fn send_message_streaming(
        &self,
        peer: &PeerCard,
        text: &str,
    ) -> Result<PeerOutcome, ClientError> {
        if !peer.supports_streaming {
            return Err(ClientError::PeerDoesNotSupportStreaming);
        }

        let response = self
            .http
            .post(&peer.url)
            .header(reqwest::header::ACCEPT, "text/event-stream")
            .json(&build_send_streaming_request(text))
            .send()
            .await
            .and_then(reqwest::Response::error_for_status)
            .map_err(|e| ClientError::Transport(e.to_string()))?;

        let mut bytes_stream = response.bytes_stream();
        let mut buffer = String::new();
        let mut last_task_id: Option<String> = None;

        loop {
            let Some(chunk) = bytes_stream.next().await else {
                break;
            };
            let Ok(chunk) = chunk else {
                break; // transport dropped mid-stream — fall back below.
            };
            buffer.push_str(&String::from_utf8_lossy(&chunk));
            for event in drain_sse_events(&mut buffer) {
                match parse_stream_step(event.as_bytes())? {
                    StreamStep::Progress { task_id } => {
                        last_task_id = task_id.or(last_task_id);
                    }
                    StreamStep::Terminal(outcome) => return Ok(outcome),
                }
            }
        }

        // The stream ended (transport drop or a clean close) without a
        // terminal event. If we at least learned the task id, the task may
        // well have finished server-side regardless — poll for its actual
        // state rather than reporting a bare transport failure.
        match last_task_id {
            Some(task_id) => self.get_task(&peer.url, &task_id).await,
            None => Err(ClientError::Transport(
                "stream ended before any task id or terminal event was seen".to_owned(),
            )),
        }
    }

    /// Fetch a task from the peer's JSON-RPC `endpoint` via `GetTask`.
    ///
    /// # Errors
    ///
    /// As [`Self::send_message`], decoding a bare `Task`.
    pub async fn get_task(
        &self,
        endpoint: &str,
        task_id: &str,
    ) -> Result<PeerOutcome, ClientError> {
        let bytes = self
            .post(endpoint, &build_get_task_request(task_id))
            .await?;
        parse_task_response(&bytes)
    }

    /// Discover then call in one shot: fetch + verify the card, then `SendMessage`
    /// to the endpoint the card advertises. Returns the verified peer identity
    /// alongside the outcome.
    ///
    /// Security: the POST target is the card's own advertised url. Under
    /// trust-on-first-use that is only as trustworthy as the (unpinned) card, so
    /// a hostile card could point it at an internal address. Redirect-following
    /// is off (see [`Self::new`]); a caller that has pinned a peer's key should
    /// verify `card.public_key` against the pin before trusting the outcome.
    ///
    /// # Errors
    ///
    /// Propagates [`Self::fetch_card`] / [`Self::send_message`] errors.
    pub async fn call(&self, text: &str) -> Result<(PeerCard, PeerOutcome), ClientError> {
        let card = self.fetch_card().await?;
        if !same_origin(&self.base_url, &card.url) {
            return Err(ClientError::EndpointEscapesConfiguredOrigin);
        }
        let outcome = self.send_message(&card.url, text).await?;
        Ok((card, outcome))
    }

    async fn get(&self, url: &str) -> Result<Vec<u8>, ClientError> {
        let response = self
            .http
            .get(url)
            .send()
            .await
            .and_then(reqwest::Response::error_for_status)
            .map_err(|e| ClientError::Transport(e.to_string()))?;
        self.read_response(response).await
    }

    async fn post(&self, url: &str, body: &Value) -> Result<Vec<u8>, ClientError> {
        let response = self
            .http
            .post(url)
            .json(body)
            .send()
            .await
            .and_then(reqwest::Response::error_for_status)
            .map_err(|e| ClientError::Transport(e.to_string()))?;
        self.read_response(response).await
    }

    async fn read_response(&self, response: reqwest::Response) -> Result<Vec<u8>, ClientError> {
        let header_bytes = response
            .headers()
            .iter()
            .map(|(name, value)| name.as_str().len() + value.as_bytes().len() + 4)
            .sum::<usize>();
        if header_bytes > self.response_bounds.max_header_bytes {
            return Err(ClientError::ResponseHeadersTooLarge {
                cap_bytes: self.response_bounds.max_header_bytes,
            });
        }
        if response
            .content_length()
            .is_some_and(|bytes| bytes > self.response_bounds.max_decoded_body_bytes as u64)
        {
            return Err(ClientError::ResponseBodyTooLarge {
                cap_bytes: self.response_bounds.max_decoded_body_bytes,
            });
        }

        let mut stream = response.bytes_stream();
        let mut body = Vec::new();
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.map_err(|error| ClientError::Transport(error.to_string()))?;
            if body.len().saturating_add(chunk.len()) > self.response_bounds.max_decoded_body_bytes
            {
                return Err(ClientError::ResponseBodyTooLarge {
                    cap_bytes: self.response_bounds.max_decoded_body_bytes,
                });
            }
            body.extend_from_slice(&chunk);
        }
        Ok(body)
    }
}

/// A signed card may select a route beneath its configured origin, but it may
/// not replace the trusted registry's scheme, host, or port. The custom
/// reqwest resolver is pinned to that origin, and this check prevents a card
/// from smuggling a second destination through the otherwise valid payload.
fn same_origin(base: &str, endpoint: &str) -> bool {
    let Ok(base) = reqwest::Url::parse(base) else {
        return false;
    };
    let Ok(endpoint) = reqwest::Url::parse(endpoint) else {
        return false;
    };
    base.scheme() == endpoint.scheme()
        && base.host_str() == endpoint.host_str()
        && base.port_or_known_default() == endpoint.port_or_known_default()
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use std::future::Future;
    use std::pin::Pin;
    use std::sync::Arc;

    use futures::Stream;
    use polyc_crypto::Signer;
    use tokio::net::TcpListener;
    use tokio_util::sync::CancellationToken;

    use super::*;
    use crate::card::{CardConfig, signed_card};
    use crate::server::AppState;
    use crate::store::test_double::InMemoryTaskStore;
    use crate::task::{
        TurnOutcome, TurnRequest, TurnRunner, TurnStreamEvent, UnconfiguredApprovalResponder,
    };

    fn rpc_result(result: Value) -> Vec<u8> {
        serde_json::to_vec(&json!({ "jsonrpc": "2.0", "id": "1", "result": result })).unwrap()
    }

    #[test]
    fn build_send_request_is_v1_shape() {
        let req = build_send_request("hello");
        assert_eq!(req["method"], "SendMessage");
        assert_eq!(req["params"]["message"]["role"], "ROLE_USER");
        assert_eq!(req["params"]["message"]["parts"][0]["text"], "hello");
        assert!(req["params"]["message"]["parts"][0].get("kind").is_none());
    }

    #[test]
    fn parse_completed_wrapped_task() {
        let result = json!({ "task": {
            "id": "t", "contextId": "c",
            "status": { "state": "TASK_STATE_COMPLETED", "message": {
                "messageId": "r", "role": "ROLE_AGENT", "parts": [{ "text": "the answer is 42" }]
            }},
        }});
        assert_eq!(
            parse_send_response(&rpc_result(result)).unwrap(),
            PeerOutcome::Completed {
                text: "the answer is 42".to_owned()
            }
        );
    }

    #[test]
    fn parse_completed_from_artifacts_fallback() {
        // No status.message, answer only in artifacts.
        let result = json!({ "task": {
            "id": "t", "contextId": "c",
            "status": { "state": "TASK_STATE_COMPLETED" },
            "artifacts": [{ "artifactId": "a", "parts": [{ "text": "from artifact" }] }],
        }});
        assert_eq!(
            parse_send_response(&rpc_result(result)).unwrap(),
            PeerOutcome::Completed {
                text: "from artifact".to_owned()
            }
        );
    }

    #[test]
    fn parse_input_required_task() {
        let result = json!({ "task": {
            "id": "t", "contextId": "c",
            "status": { "state": "TASK_STATE_INPUT_REQUIRED", "message": {
                "messageId": "r", "role": "ROLE_AGENT", "parts": [{ "text": "approve?" }]
            }},
        }});
        assert_eq!(
            parse_send_response(&rpc_result(result)).unwrap(),
            PeerOutcome::InputRequired {
                prompt: "approve?".to_owned()
            }
        );
    }

    #[test]
    fn parse_immediate_message_reply() {
        let result = json!({ "message": {
            "messageId": "r", "role": "ROLE_AGENT", "parts": [{ "text": "hi there" }]
        }});
        assert_eq!(
            parse_send_response(&rpc_result(result)).unwrap(),
            PeerOutcome::Completed {
                text: "hi there".to_owned()
            }
        );
    }

    #[test]
    fn parse_rpc_error_and_null_error() {
        let err = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": "1", "error": { "code": -32001, "message": "task not found" }
        }))
        .unwrap();
        assert!(matches!(
            parse_send_response(&err).unwrap_err(),
            ClientError::RpcError { code: -32001, .. }
        ));
        // explicit null error alongside a result is NOT a fault
        let ok = serde_json::to_vec(&json!({
            "jsonrpc": "2.0", "id": "1", "error": null,
            "result": { "message": { "messageId": "r", "role": "ROLE_AGENT", "parts": [{ "text": "ok" }] }}
        }))
        .unwrap();
        assert_eq!(
            parse_send_response(&ok).unwrap(),
            PeerOutcome::Completed {
                text: "ok".to_owned()
            }
        );
    }

    #[test]
    fn verify_card_doc_reads_supported_interfaces_url() {
        let signer = Signer::from_seed(9);
        let card = signed_card(
            &CardConfig {
                name: "Peer".to_owned(),
                description: "d".to_owned(),
                url: "http://peer/".to_owned(),
                version: "1".to_owned(),
            },
            &signer,
        );
        let peer = verify_card_doc(&serde_json::to_vec(&card).unwrap()).expect("verifies");
        assert_eq!(peer.name, "Peer");
        assert_eq!(peer.url, "http://peer/");
        assert_eq!(peer.public_key, signer.public_key_bytes());
        assert!(
            peer.supports_streaming,
            "the default card advertises `capabilities.streaming: true` (`#371`)"
        );

        // Tamper a signed field → verification fails.
        let mut tampered = card;
        tampered["name"] = json!("Imposter");
        assert!(matches!(
            verify_card_doc(&serde_json::to_vec(&tampered).unwrap()).unwrap_err(),
            ClientError::UnverifiedCard
        ));
    }

    struct StubRunner(TurnOutcome);
    impl TurnRunner for StubRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let outcome = self.0.clone();
            Box::pin(async move { outcome })
        }
    }

    const TOKEN: &str = "round-trip-bearer-token";

    #[tokio::test]
    async fn client_round_trips_against_the_real_server() {
        // The real v1.0 server, served on a socket, driven by the real client —
        // the symmetric-edge proof, end to end over HTTP, including the
        // bearer-token gate on `POST /`.
        let signer = Signer::from_seed(7);
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let base = format!("http://{addr}/");
        let card = signed_card(
            &CardConfig {
                name: "Peer".to_owned(),
                description: "a peer agent".to_owned(),
                url: base.clone(),
                version: "1".to_owned(),
            },
            &signer,
        );
        let state = AppState {
            card: Arc::new(card),
            runner: Arc::new(StubRunner(TurnOutcome::Completed {
                text: "42".to_owned(),
            })),
            approvals: Arc::new(UnconfiguredApprovalResponder),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", TOKEN).unwrap(),
        };
        let app = crate::router(state);
        let shutdown = CancellationToken::new();
        let server = tokio::spawn({
            let shutdown = shutdown.clone();
            async move {
                axum::serve(listener, app)
                    .with_graceful_shutdown(async move { shutdown.cancelled().await })
                    .await
                    .unwrap();
            }
        });

        let client = A2aClient::with_bearer_token(base, TOKEN);
        let peer = client.fetch_card().await.expect("fetch card");
        assert_eq!(peer.name, "Peer");
        assert_eq!(peer.public_key, signer.public_key_bytes());

        let outcome = client.send_message(&peer.url, "hi").await.expect("send");
        assert_eq!(
            outcome,
            PeerOutcome::Completed {
                text: "42".to_owned()
            }
        );

        let (peer2, outcome2) = client.call("hi again").await.expect("call");
        assert_eq!(peer2.name, "Peer");
        assert_eq!(
            outcome2,
            PeerOutcome::Completed {
                text: "42".to_owned()
            }
        );

        shutdown.cancel();
        let _ = server.await;
    }

    /// A [`TurnRunner`] that streams a fixed sequence of [`TurnStreamEvent`]s —
    /// proves the real SSE wire round-trips text chunks, not just a
    /// single-shot outcome (`#371`).
    struct StreamingStubRunner(Vec<TurnStreamEvent>);
    impl TurnRunner for StreamingStubRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let outcome = self
                .0
                .iter()
                .find_map(|e| match e {
                    TurnStreamEvent::Outcome(o) => Some(o.clone()),
                    TurnStreamEvent::DurablyReceived | TurnStreamEvent::TextDelta(_) => None,
                })
                .unwrap_or(TurnOutcome::Completed {
                    text: String::new(),
                });
            Box::pin(async move { outcome })
        }

        fn run_turn_streaming<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send + 'a>> {
            Box::pin(futures::stream::iter(self.0.clone()))
        }
    }

    /// The `SendStreamingMessage` sibling of `client_round_trips_against_the_real_server`
    /// (`#371`): the real server's SSE transport, decoded by the real client's
    /// hand-rolled decoder, folds streamed text chunks to the same terminal
    /// outcome a unary `SendMessage` would report.
    #[tokio::test]
    async fn client_streams_send_streaming_message_against_the_real_server() {
        let signer = Signer::from_seed(13);
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let base = format!("http://{addr}/");
        let card = signed_card(
            &CardConfig {
                name: "Peer".to_owned(),
                description: "a peer agent".to_owned(),
                url: base.clone(),
                version: "1".to_owned(),
            },
            &signer,
        );
        let state = AppState {
            card: Arc::new(card),
            runner: Arc::new(StreamingStubRunner(vec![
                TurnStreamEvent::TextDelta("Hello, ".to_owned()),
                TurnStreamEvent::TextDelta("world.".to_owned()),
                TurnStreamEvent::Outcome(TurnOutcome::Completed {
                    text: "Hello, world.".to_owned(),
                }),
            ])),
            approvals: Arc::new(UnconfiguredApprovalResponder),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", TOKEN).unwrap(),
        };
        let app = crate::router(state);
        let shutdown = CancellationToken::new();
        let server = tokio::spawn({
            let shutdown = shutdown.clone();
            async move {
                axum::serve(listener, app)
                    .with_graceful_shutdown(async move { shutdown.cancelled().await })
                    .await
                    .unwrap();
            }
        });

        let client = A2aClient::with_bearer_token(base, TOKEN);
        let peer = client.fetch_card().await.expect("fetch card");
        assert!(
            peer.supports_streaming,
            "the served card advertises streaming"
        );

        let outcome = client
            .send_message_streaming(&peer, "hi")
            .await
            .expect("stream");
        assert_eq!(
            outcome,
            PeerOutcome::Completed {
                text: "Hello, world.".to_owned()
            }
        );

        shutdown.cancel();
        let _ = server.await;
    }

    /// The bearer-auth gate (`server::authenticate`) applies before EITHER
    /// dispatch path ever sees the body — proven against a real socket so a
    /// regression that only guards `rpc::handle`'s unary route can't slip
    /// through unnoticed for the new SSE route (`#371`).
    #[tokio::test]
    async fn bearer_auth_rejects_both_the_unary_and_streaming_routes() {
        let signer = Signer::from_seed(17);
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let base = format!("http://{addr}/");
        let card = signed_card(
            &CardConfig {
                name: "Peer".to_owned(),
                description: "a peer agent".to_owned(),
                url: base.clone(),
                version: "1".to_owned(),
            },
            &signer,
        );
        let state = AppState {
            card: Arc::new(card),
            runner: Arc::new(StubRunner(TurnOutcome::Completed {
                text: "42".to_owned(),
            })),
            approvals: Arc::new(UnconfiguredApprovalResponder),
            store: Arc::new(InMemoryTaskStore::new()),
            turn_limit: polyc_runtime::admission::AdmissionGate::new(64),
            peers: crate::server::PeerAuthenticator::single("test-peer", TOKEN).unwrap(),
        };
        let app = crate::router(state);
        let shutdown = CancellationToken::new();
        let server = tokio::spawn({
            let shutdown = shutdown.clone();
            async move {
                axum::serve(listener, app)
                    .with_graceful_shutdown(async move { shutdown.cancelled().await })
                    .await
                    .unwrap();
            }
        });

        let http = reqwest::Client::new();
        let unary = http
            .post(&base)
            .header(reqwest::header::AUTHORIZATION, "Bearer wrong-token")
            .json(&build_send_request("hi"))
            .send()
            .await
            .unwrap();
        assert_eq!(unary.status(), reqwest::StatusCode::UNAUTHORIZED);

        let streaming = http
            .post(&base)
            .header(reqwest::header::AUTHORIZATION, "Bearer wrong-token")
            .header(reqwest::header::ACCEPT, "text/event-stream")
            .json(&build_send_streaming_request("hi"))
            .send()
            .await
            .unwrap();
        assert_eq!(streaming.status(), reqwest::StatusCode::UNAUTHORIZED);

        shutdown.cancel();
        let _ = server.await;
    }

    /// Calling `SendStreamingMessage` against a peer whose card does not
    /// advertise `capabilities.streaming` must refuse before ever dialing —
    /// proven with an unreachable URL: a `Transport` error here would mean the
    /// gate let the call through.
    #[tokio::test]
    async fn send_message_streaming_refuses_a_peer_without_the_capability() {
        let client = A2aClient::new("http://127.0.0.1:1/");
        let peer = PeerCard {
            name: "Peer".to_owned(),
            url: "http://127.0.0.1:1/".to_owned(),
            public_key: Vec::new(),
            supports_streaming: false,
        };
        let err = client
            .send_message_streaming(&peer, "hi")
            .await
            .expect_err("must refuse a non-streaming peer");
        assert!(matches!(err, ClientError::PeerDoesNotSupportStreaming));
    }

    #[test]
    fn a_card_endpoint_stays_under_its_configured_origin() {
        assert!(same_origin(
            "https://finance.example/peer",
            "https://finance.example/rpc/v1"
        ));
        assert!(!same_origin(
            "https://finance.example/",
            "https://metadata.google.internal/compute"
        ));
        assert!(!same_origin(
            "https://finance.example/",
            "http://finance.example/rpc"
        ));
        assert!(!same_origin(
            "https://finance.example/",
            "https://finance.example:8443/rpc"
        ));
    }

    #[test]
    fn drain_sse_events_splits_multiple_events_and_keeps_a_trailing_partial() {
        let mut buffer =
            String::from("data: {\"a\":1}\n\ndata: {\"a\":2}\n\ndata: {\"a\":3 (not yet complete)");
        let events = drain_sse_events(&mut buffer);
        assert_eq!(events, vec!["{\"a\":1}".to_owned(), "{\"a\":2}".to_owned()]);
        assert_eq!(
            buffer, "data: {\"a\":3 (not yet complete)",
            "a trailing event with no blank-line terminator yet is retained for the next chunk"
        );
    }

    #[test]
    fn drain_sse_events_ignores_comments_and_non_data_fields() {
        let mut buffer =
            String::from(": keep-alive\r\nevent: message\r\ndata: {\"a\":1}\r\nid: 5\r\n\r\n");
        let events = drain_sse_events(&mut buffer);
        assert_eq!(events, vec!["{\"a\":1}".to_owned()]);
        assert!(buffer.is_empty());
    }

    #[test]
    fn parse_stream_step_classifies_progress_vs_terminal() {
        let submitted = json!({
            "jsonrpc": "2.0", "id": "1",
            "result": { "task": {
                "id": "t1", "contextId": "c1", "status": { "state": "TASK_STATE_SUBMITTED" }
            }}
        });
        assert_eq!(
            parse_stream_step(&serde_json::to_vec(&submitted).unwrap()).unwrap(),
            StreamStep::Progress {
                task_id: Some("t1".to_owned())
            }
        );

        let terminal = json!({
            "jsonrpc": "2.0", "id": "1",
            "result": { "statusUpdate": {
                "taskId": "t1", "contextId": "c1",
                "status": { "state": "TASK_STATE_COMPLETED", "message": {
                    "messageId": "r", "role": "ROLE_AGENT", "parts": [{ "text": "42" }]
                }},
                "final": true
            }}
        });
        assert_eq!(
            parse_stream_step(&serde_json::to_vec(&terminal).unwrap()).unwrap(),
            StreamStep::Terminal(PeerOutcome::Completed {
                text: "42".to_owned()
            })
        );

        // A non-final statusUpdate (`working`) is still in progress.
        let working = json!({
            "jsonrpc": "2.0", "id": "1",
            "result": { "statusUpdate": {
                "taskId": "t1", "contextId": "c1",
                "status": { "state": "TASK_STATE_WORKING" }, "final": false
            }}
        });
        assert_eq!(
            parse_stream_step(&serde_json::to_vec(&working).unwrap()).unwrap(),
            StreamStep::Progress {
                task_id: Some("t1".to_owned())
            }
        );
    }
}