laser_sdk/

a2a.rs

use crate::agent::Laser;
use crate::bridge_hops::hops_metadata;
use crate::error::LaserError;
use crate::provenance::AgentTopic;
use crate::types::ConversationId;
#[cfg(feature = "a2a-http")]
use axum::Router;
#[cfg(feature = "a2a-http")]
use axum::extract::State;
#[cfg(feature = "a2a-http")]
use axum::routing::post;
use laser_wire::agent::{
    self as agdx, AgentEnvelope, AgentId, CapabilityDescriptor, ContentRef, CorrelationId,
    METADATA_BRIDGE_HOPS, OPERATION_CHAT,
};
use laser_wire::content::ContentType;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use serde_json::to_value;
use std::str::FromStr;
#[cfg(feature = "a2a-http")]
use std::sync::Arc;
use strum::{Display, EnumString};

pub use crate::bridge_hops::enter_bridge;
pub use laser_wire::agent::TaskState;

const JSONRPC_VERSION: &str = "2.0";
// JSON-RPC reserved range ends at -32000, and -32000..=-32099 is for application errors.
const APP_ERROR_CODE: i32 = -32000;

/// The A2A JSON-RPC methods the bridge serves. `Display`/`FromStr` (strum) carry
/// the exact wire spelling (v1.0's PascalCase operation names), so the dispatch
/// never matches on bare string literals. v1.0's `ListTasks` is not served: the
/// bridge is stateless over the log, and an unknown method already answers with
/// the bridge's application error code (`APP_ERROR_CODE`, -32000), not JSON-RPC's
/// reserved -32601 method-not-found.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Display, EnumString)]
pub enum A2aMethod {
    #[strum(serialize = "SendMessage")]
    MessageSend,
    #[strum(serialize = "SendStreamingMessage")]
    MessageStream,
    #[strum(serialize = "GetTask")]
    TasksGet,
    #[strum(serialize = "CancelTask")]
    TasksCancel,
}

/// The A2A protocol version this bridge's Agent Card declares.
pub const A2A_PROTOCOL_VERSION: &str = "1.0";

/// The JSON-RPC binding label of a v1.0 `AgentInterface`.
pub const A2A_JSONRPC_BINDING: &str = "JSONRPC";

/// The bridge's A2A Agent Card: the discovery document an A2A client fetches.
/// Field names follow the v1.0 `AgentCard` schema: the endpoint and its
/// protocol version live in `supportedInterfaces` (v1.0 dropped the top-level
/// `protocolVersion` / `url` / `preferredTransport`), and `signatures` carries
/// the optional detached JWS set (`sign` feature).
#[derive(Debug, Clone, Serialize)]
pub struct AgentCard {
    pub name: String,
    pub description: String,
    pub version: String,
    #[serde(rename = "supportedInterfaces")]
    pub supported_interfaces: Vec<AgentInterface>,
    pub capabilities: AgentCardCapabilities,
    #[serde(rename = "defaultInputModes")]
    pub default_input_modes: Vec<String>,
    #[serde(rename = "defaultOutputModes")]
    pub default_output_modes: Vec<String>,
    pub skills: Vec<AgentSkill>,
    /// Detached JWS signatures over the JCS-canonicalized card (RFC 8785 +
    /// RFC 7515), empty when unsigned.
    #[serde(skip_serializing_if = "Vec::is_empty", default)]
    pub signatures: Vec<AgentCardSignature>,
}

/// One v1.0 `AgentInterface`: where the agent serves and over what binding.
#[derive(Debug, Clone, Serialize)]
pub struct AgentInterface {
    pub url: String,
    #[serde(rename = "protocolBinding")]
    pub protocol_binding: String,
    #[serde(rename = "protocolVersion")]
    pub protocol_version: String,
}

/// One v1.0 `AgentCardSignature`: an RFC 7515 JWS over the card, detached (the
/// payload is the JCS-canonicalized card without `signatures`, reconstructed by
/// the verifier).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentCardSignature {
    /// The base64url-encoded protected JWS header (carries `alg`).
    pub protected: String,
    /// The base64url-encoded signature.
    pub signature: String,
}

/// The A2A `AgentCapabilities` flags of an [`AgentCard`].
#[derive(Debug, Clone, Serialize)]
pub struct AgentCardCapabilities {
    /// Whether the agent streams (AGDX chunk channels ride the log, consumed
    /// over Iggy's own transport rather than SSE).
    pub streaming: bool,
    #[serde(rename = "pushNotifications")]
    pub push_notifications: bool,
    #[serde(rename = "stateTransitionHistory")]
    pub state_transition_history: bool,
    /// v1.0: whether an authenticated extended card is served (never, the
    /// well-known card is the whole truth here).
    #[serde(rename = "extendedAgentCard")]
    pub extended_agent_card: bool,
}

/// One A2A `AgentSkill` advertised on the card.
#[derive(Debug, Clone, Serialize)]
pub struct AgentSkill {
    pub id: String,
    pub name: String,
    pub description: String,
    pub tags: Vec<String>,
    /// The skill's accepted input content modes (A2A `inputModes`), derived from
    /// the capability descriptor's input `ContentRef`. Omitted when unspecified.
    #[serde(rename = "inputModes", skip_serializing_if = "Vec::is_empty")]
    pub input_modes: Vec<String>,
    /// The skill's produced output content modes (A2A `outputModes`).
    #[serde(rename = "outputModes", skip_serializing_if = "Vec::is_empty")]
    pub output_modes: Vec<String>,
}

/// Project a capability's [`ContentRef`] to an A2A content mode string: a
/// content-type code to its media type, a registered schema id to an AGDX schema
/// media type. Unknown codes fall back to opaque bytes.
fn content_ref_mode(reference: &ContentRef) -> String {
    match reference {
        ContentRef::ContentType(content_type) => match content_type {
            ContentType::Raw => "application/octet-stream",
            ContentType::Json => "application/json",
            ContentType::Msgpack => "application/msgpack",
            ContentType::Cbor => "application/cbor",
            ContentType::Bson => "application/bson",
            ContentType::Avro => "application/avro",
            ContentType::Protobuf => "application/protobuf",
            ContentType::Arrow => "application/vnd.apache.arrow.stream",
            ContentType::Ref => "application/x-agdx-ref",
            ContentType::Any => "*/*",
            _ => "application/octet-stream",
        }
        .to_owned(),
        ContentRef::SchemaId(id) => format!("application/x-agdx-schema;id={id}"),
    }
}

/// An A2A task's current state. `state` is the ONE wire dictionary
/// (`laser_wire::agent::TaskState`, re-exported below): the bridge maps its
/// codes to A2A's kebab-case names at the JSON boundary, never a second enum.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskStatus {
    #[serde(with = "task_state_json")]
    pub state: TaskState,
}

// The JSON boundary for the task-state dictionary: A2A speaks the kebab-case
// names (the dictionary's `Display`/`FromStr`), while the wire type itself
// rides CBOR as a u8 code. An unknown inbound name is a protocol error.
mod task_state_json {
    use super::TaskState;
    use serde::de::Error;
    use serde::{Deserialize, Deserializer, Serializer};

    pub fn serialize<S: Serializer>(state: &TaskState, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.collect_str(state)
    }

    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<TaskState, D::Error> {
        String::deserialize(deserializer)?
            .parse()
            .map_err(D::Error::custom)
    }
}

/// An output artifact produced by an A2A task.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Artifact {
    pub text: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
/// An A2A task: its id, status, and artifacts.
pub struct Task {
    pub id: String,
    pub status: TaskStatus,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub artifacts: Vec<Artifact>,
}

/// A JSON-RPC request envelope.
#[derive(Debug, Deserialize)]
pub struct JsonRpcRequest {
    #[serde(default)]
    pub id: JsonValue,
    pub method: String,
    #[serde(default)]
    pub params: JsonValue,
}

/// A JSON-RPC response envelope.
#[derive(Debug, Serialize)]
pub struct JsonRpcResponse {
    pub jsonrpc: &'static str,
    pub id: JsonValue,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result: Option<JsonValue>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<JsonRpcError>,
}

/// A JSON-RPC error object.
#[derive(Debug, Serialize)]
pub struct JsonRpcError {
    pub code: i32,
    pub message: String,
}

/// Map an inbound `message/send` onto its on-log `command` envelope: the
/// mapped core is the ids (the fresh task id rides `correlation`), and the
/// tunneled remainder is the ORIGINAL params JSON, carried byte-identical in
/// `body` with `agdx.ct = json`, so a round trip back out returns the foreign
/// payload untouched instead of flattening it into a lossy intermediate.
pub fn command_from_message_send(
    record: agdx::RecordId,
    conversation: agdx::ConversationId,
    source: agdx::AgentId,
    correlation: agdx::CorrelationId,
    params_json: Vec<u8>,
) -> AgentEnvelope {
    AgentEnvelope::command(record, conversation, source, correlation, params_json)
        .with_operation(OPERATION_CHAT)
}

/// The outbound view: the envelope answering a task becomes the A2A task.
/// `task_state` maps through the dictionary when present. Otherwise a
/// `response` reads completed and an `error` reads failed. A non-empty body
/// becomes the text artifact.
pub fn task_from_envelope(task_id: impl Into<String>, envelope: &AgentEnvelope) -> Task {
    let state = envelope.task_state.unwrap_or(match envelope.kind {
        agdx::AgentKind::Error => TaskState::Failed,
        _ => TaskState::Completed,
    });
    let artifacts = if envelope.body.is_empty() {
        Vec::new()
    } else {
        vec![Artifact {
            text: String::from_utf8_lossy(&envelope.body).into_owned(),
        }]
    };
    Task {
        id: task_id.into(),
        status: TaskStatus { state },
        artifacts,
    }
}

/// Bridges the synchronous A2A JSON-RPC edge onto durable agent topics: a request
/// becomes a message on `request_topic` keyed by a fresh task (conversation) id,
/// and a task lookup replays `reply_topic` for that conversation. The log is the
/// source of truth, so tasks survive a bridge restart and stay replayable.
pub struct A2aBridge {
    laser: Laser,
    source: AgentId,
    request_topic: AgentTopic<'static>,
    reply_topic: AgentTopic<'static>,
    capabilities: Vec<CapabilityDescriptor>,
    // The `bridge_hops` path stamped on every record the bridge publishes,
    // ending in its own id.
    hops: Vec<String>,
    /// When set, control records the bridge emits (the cancel) are signed, so a
    /// verifying consumer can authorize them against the enrolled key. Closes the
    /// unauthenticated-cancel hole. Requires the `sign` feature.
    #[cfg(feature = "sign")]
    signing_key: Option<crate::sign::SigningKey>,
    /// When set, every task id is bound to the caller that submitted it.
    task_binding: Option<TaskBinding>,
}

/// The key and lifetime task ids are bound with.
#[derive(Clone)]
struct TaskBinding {
    key: [u8; 32],
    ttl: std::time::Duration,
}

impl TaskBinding {
    /// The keyed digest a bound task id carries: the caller, the stream
    /// generation, the conversation and the expiry, so an id works only for
    /// its caller, on the stream incarnation it was made on, until it expires.
    fn mac(
        &self,
        caller: &str,
        stream_generation: u64,
        task: ConversationId,
        expires: u64,
    ) -> blake3::Hash {
        let mut hasher = blake3::Hasher::new_keyed(&self.key);
        hasher.update(b"laserdata a2a task v1\0");
        hasher.update(&(caller.len() as u64).to_le_bytes());
        hasher.update(caller.as_bytes());
        hasher.update(&stream_generation.to_le_bytes());
        hasher.update(&task.as_u128().to_le_bytes());
        hasher.update(&expires.to_le_bytes());
        hasher.finalize()
    }
}

/// The authenticated caller of an A2A HTTP request. A host that binds tasks
/// to callers inserts it into the request extensions from its own
/// authentication, and the router passes it to the bridge.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct A2aCallerPrincipal(pub String);

/// The bridge acting for one authenticated caller. Build it with
/// [`A2aBridge::caller`].
pub struct A2aCaller<'a> {
    bridge: &'a A2aBridge,
    principal: String,
}

impl A2aCaller<'_> {
    /// [`A2aBridge::submit`] for this caller.
    pub async fn submit(&self, params_json: Vec<u8>) -> Result<Task, LaserError> {
        self.bridge
            .submit_addressed(None, None, params_json, Some(&self.principal))
            .await
    }

    /// [`A2aBridge::submit_to`] for this caller.
    pub async fn submit_to(
        &self,
        target: impl Into<AgentId>,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        self.bridge
            .submit_addressed(
                Some(target.into()),
                None,
                params_json,
                Some(&self.principal),
            )
            .await
    }

    /// [`A2aBridge::submit_in`] for this caller.
    pub async fn submit_in(
        &self,
        parent: ConversationId,
        root: ConversationId,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        self.bridge
            .submit_addressed(
                None,
                Some((parent, root)),
                params_json,
                Some(&self.principal),
            )
            .await
    }

    /// [`A2aBridge::submit_in_to`] for this caller.
    pub async fn submit_in_to(
        &self,
        target: impl Into<AgentId>,
        parent: ConversationId,
        root: ConversationId,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        self.bridge
            .submit_addressed(
                Some(target.into()),
                Some((parent, root)),
                params_json,
                Some(&self.principal),
            )
            .await
    }

    /// [`A2aBridge::task`] for this caller. Another caller's task id answers
    /// like an unknown one.
    pub async fn task(&self, id: &str) -> Result<Task, LaserError> {
        self.bridge.task_for(id, Some(&self.principal)).await
    }

    /// [`A2aBridge::cancel`] for this caller. Another caller's task id is
    /// refused before anything is published.
    pub async fn cancel(&self, id: &str) -> Result<Task, LaserError> {
        self.bridge.cancel_for(id, Some(&self.principal)).await
    }

    /// [`A2aBridge::handle_rpc`] for this caller.
    pub async fn handle_rpc(&self, request: JsonValue) -> JsonRpcResponse {
        self.bridge
            .handle_rpc_for(request, Some(&self.principal))
            .await
    }
}

impl A2aBridge {
    /// A bridge mapping A2A JSON-RPC methods onto agent topics over `laser`,
    /// publishing as `source` (the bridge's agent id). Every topic lives on the
    /// stream of `laser`. Pass `laser.with_default_stream(stream)` to run the bridge on a
    /// non-default stream (the unit of multi-stream topologies and per-stream
    /// Iggy RBAC). `request_topic`/`reply_topic` are any [`AgentTopic`], including
    /// `AgentTopic::Custom` for an arbitrary name.
    pub fn new(
        laser: Laser,
        source: impl Into<AgentId>,
        request_topic: AgentTopic<'static>,
        reply_topic: AgentTopic<'static>,
    ) -> Self {
        let source = source.into();
        let hops = vec![source.as_str().to_owned()];
        Self {
            laser,
            source,
            request_topic,
            reply_topic,
            capabilities: Vec::new(),
            hops,
            #[cfg(feature = "sign")]
            signing_key: None,
            task_binding: None,
        }
    }

    /// Bind every task id to the caller that submitted it, keyed with `key`
    /// and valid for `ttl`. A bound id names the caller, the stream
    /// incarnation, the conversation and an expiry under a keyed digest, so
    /// another caller cannot read or cancel the task. Once set, task
    /// operations need a caller: use [`caller`](Self::caller), and the HTTP
    /// router reads [`A2aCallerPrincipal`] from the request extensions.
    #[must_use]
    pub fn with_task_binding(mut self, key: [u8; 32], ttl: std::time::Duration) -> Self {
        self.task_binding = Some(TaskBinding { key, ttl });
        self
    }

    /// The bridge acting for one authenticated caller.
    pub fn caller(&self, principal: impl Into<String>) -> A2aCaller<'_> {
        A2aCaller {
            bridge: self,
            principal: principal.into(),
        }
    }

    /// The creation time of the bridge's stream, which tells a recreated
    /// stream apart from the one a task id was made on.
    async fn stream_generation(&self) -> Result<u64, LaserError> {
        let stream = iggy::prelude::Identifier::named(self.laser.stream_required()?)?;
        let details =
            iggy::prelude::StreamClient::get_stream(self.laser.client().as_ref(), &stream)
                .await?
                .ok_or_else(|| {
                    LaserError::HandlerConfig("the bridge stream does not exist".to_owned())
                })?;
        Ok(details.created_at.as_micros())
    }

    /// The id a new task is known by.
    async fn task_id(
        &self,
        task: ConversationId,
        caller: Option<&str>,
    ) -> Result<String, LaserError> {
        let Some(binding) = &self.task_binding else {
            return Ok(task.to_string());
        };
        let caller = caller.ok_or_else(unbound_caller)?;
        let generation = self.stream_generation().await?;
        let expires =
            now_micros().saturating_add(u64::try_from(binding.ttl.as_micros()).unwrap_or(u64::MAX));
        let mac = binding.mac(caller, generation, task, expires);
        Ok(format!("{task}.{expires}.{}", mac.to_hex()))
    }

    /// The conversation a task id names, refused like a malformed id when it
    /// is not this caller's, expired, or from another stream incarnation.
    async fn open_task(
        &self,
        id: &str,
        caller: Option<&str>,
    ) -> Result<ConversationId, LaserError> {
        let invalid = || LaserError::HandlerConfig(format!("invalid task id `{id}`"));
        let Some(binding) = &self.task_binding else {
            return id.parse::<ConversationId>().map_err(|_| invalid());
        };
        let caller = caller.ok_or_else(unbound_caller)?;
        let mut parts = id.split('.');
        let (Some(task), Some(expires), Some(mac), None) =
            (parts.next(), parts.next(), parts.next(), parts.next())
        else {
            return Err(invalid());
        };
        let task = task.parse::<ConversationId>().map_err(|_| invalid())?;
        let expires = expires.parse::<u64>().map_err(|_| invalid())?;
        let mac = blake3::Hash::from_hex(mac).map_err(|_| invalid())?;
        if expires < now_micros() {
            return Err(invalid());
        }
        let generation = self.stream_generation().await?;
        // `blake3::Hash` compares in constant time.
        if binding.mac(caller, generation, task, expires) != mac {
            return Err(invalid());
        }
        Ok(task)
    }

    /// Continue the loop-guard path of work that already crossed other
    /// bridges: `previous` is its `bridge_hops`, and this bridge's id is
    /// appended to it. Every record the bridge publishes carries the path.
    /// Refuses a path that already holds this bridge, the loop the guard
    /// exists to stop.
    pub fn with_bridge_hops(mut self, previous: &[String]) -> Result<Self, LaserError> {
        self.hops = enter_bridge(self.source.as_str(), previous)?;
        Ok(self)
    }

    /// Sign the control records the bridge emits (the cancel) with `key`, so a
    /// verifying consumer authorizes them against the enrolled key registry. This
    /// is what corrects the otherwise-unauthenticated cancel. Requires the `sign`
    /// feature.
    #[cfg(feature = "sign")]
    #[must_use]
    pub fn with_signing_key(mut self, key: crate::sign::SigningKey) -> Self {
        self.signing_key = Some(key);
        self
    }

    /// Advertise the agent's structured capabilities on the A2A card. Each
    /// [`CapabilityDescriptor`] is projected to an A2A `AgentSkill` by
    /// [`card`](Self::card), so a remote A2A client discovers the agent's skills
    /// and their I/O shapes. Consumes and returns self for chaining.
    #[must_use]
    pub fn with_capabilities(mut self, capabilities: Vec<CapabilityDescriptor>) -> Self {
        self.capabilities = capabilities;
        self
    }

    /// `message/send`: publish the params as a typed AGDX `command` on a fresh
    /// task conversation, tunneling the foreign JSON byte-identical in the body
    /// (`agdx.ct = json`), and returns Submitted. The task id is the conversation, and
    /// the A2A task identity rides `correlation` (derived from it, so the lookup
    /// stays stateless). The command is addressed to every agent (`agdx.to = *`),
    /// so on a shared session topic every listening agent receives it. Use
    /// [`submit_to`](Self::submit_to) to address one agent.
    pub async fn submit(&self, params_json: Vec<u8>) -> Result<Task, LaserError> {
        self.submit_addressed(None, None, params_json, None).await
    }

    /// [`submit`](Self::submit) addressed to `target`, so only that agent
    /// handles the task on a shared session topic.
    pub async fn submit_to(
        &self,
        target: impl Into<AgentId>,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        let target = target.into();
        self.submit_addressed(Some(target), None, params_json, None)
            .await
    }

    /// `message/send` as a child session of `parent`, in the tree rooted at
    /// `root`: the task's submitted start lands on `agent.sessions` with the
    /// ancestry before the command, and the command carries it too. The
    /// agent that handles the task ends the child. The command is addressed to
    /// every agent, like [`submit`](Self::submit).
    pub async fn submit_in(
        &self,
        parent: ConversationId,
        root: ConversationId,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        self.submit_addressed(None, Some((parent, root)), params_json, None)
            .await
    }

    /// [`submit_in`](Self::submit_in) addressed to `target`.
    pub async fn submit_in_to(
        &self,
        target: impl Into<AgentId>,
        parent: ConversationId,
        root: ConversationId,
        params_json: Vec<u8>,
    ) -> Result<Task, LaserError> {
        let target = target.into();
        self.submit_addressed(Some(target), Some((parent, root)), params_json, None)
            .await
    }

    async fn submit_addressed(
        &self,
        target: Option<AgentId>,
        ancestry: Option<(ConversationId, ConversationId)>,
        params_json: Vec<u8>,
        caller: Option<&str>,
    ) -> Result<Task, LaserError> {
        let task = ConversationId::new();
        // Minted before anything is published, so an unbound caller on a
        // binding bridge publishes nothing.
        let id = self.task_id(task, caller).await?;
        if let Some((parent, root)) = ancestry {
            crate::agent::contract::start_child(
                &self.laser,
                &self.source,
                &self.source,
                task.into(),
                parent.into(),
                root.into(),
            )
            .await?;
        }
        let producer =
            self.laser
                .agdx(self.request_topic.clone(), self.source.clone(), task.into());
        let mut command = producer
            .command(correlation_of(task), params_json)
            .with_operation(OPERATION_CHAT)
            .with_metadata(METADATA_BRIDGE_HOPS, hops_metadata(&self.hops))
            .content_type(ContentType::Json);
        if let Some((parent, root)) = ancestry {
            command = command.with_ancestry(Some(parent.into()), Some(root.into()));
        }
        if let Some(target) = target {
            command = command.with_target(target);
        }
        command.send().await?;
        Ok(Task {
            id,
            status: TaskStatus {
                state: TaskState::Submitted,
            },
            artifacts: Vec::new(),
        })
    }

    /// `tasks/get`: replay the reply topic for the task and map the answering
    /// AGDX envelope (the `response`/`error` carrying this task's `correlation`)
    /// to the A2A task. Still Working until one lands, and Unknown when the
    /// bounded scan could not reach the end of the reply topic.
    pub async fn task(&self, id: &str) -> Result<Task, LaserError> {
        self.task_for(id, None).await
    }

    async fn task_for(&self, id: &str, caller: Option<&str>) -> Result<Task, LaserError> {
        let conversation = self.open_task(id, caller).await?;
        let correlation = correlation_of(conversation);
        // A point lookup over the reply topic via the forward reply reader (no
        // full re-scan + sort of the conversation each call).
        let answer = self
            .laser
            .find_agdx_reply(self.reply_topic.clone(), correlation)
            .await?;
        Ok(match answer {
            crate::agent::ReplyLookup::Found(envelope) => task_from_envelope(id, &envelope),
            crate::agent::ReplyLookup::NotFound => Task {
                id: id.to_owned(),
                status: TaskStatus {
                    state: TaskState::Working,
                },
                artifacts: Vec::new(),
            },
            // The bounded scan did not reach the end of the reply topic, so
            // the task may already be answered. Unknown asks the caller to
            // look again rather than claiming the task still works.
            crate::agent::ReplyLookup::Incomplete => Task {
                id: id.to_owned(),
                status: TaskStatus {
                    state: TaskState::Unknown,
                },
                artifacts: Vec::new(),
            },
        })
    }

    /// `tasks/cancel`: publish an AGDX `error` terminal (code `Cancelled`,
    /// `task_state = Canceled`) correlated to the task, so a later `tasks/get`
    /// also reports Canceled, and returns the canceled task.
    pub async fn cancel(&self, id: &str) -> Result<Task, LaserError> {
        self.cancel_for(id, None).await
    }

    async fn cancel_for(&self, id: &str, caller: Option<&str>) -> Result<Task, LaserError> {
        // Checked before the cancel is published, so a foreign id has no effect.
        let conversation = self.open_task(id, caller).await?;
        let error = agdx::AgentErrorBody {
            code: agdx::AgentErrorCode::Cancelled,
            message: Some("canceled by the A2A client".to_owned()),
            retryable: false,
            detail: None,
        };
        let producer = self.laser.agdx(
            self.reply_topic.clone(),
            self.source.clone(),
            conversation.into(),
        );
        let send = producer
            .fail(correlation_of(conversation), &error)?
            .with_task_state(TaskState::Canceled)
            .with_metadata(METADATA_BRIDGE_HOPS, hops_metadata(&self.hops));
        // Sign the cancel when a key is configured, so a verifying consumer can
        // authorize it. Without a key it rides unsigned, and a topic that
        // mandates verification rejects it.
        #[cfg(feature = "sign")]
        let send = match &self.signing_key {
            Some(key) => send.signed_by(key),
            None => send,
        };
        send.send().await?;
        Ok(Task {
            id: id.to_owned(),
            status: TaskStatus {
                state: TaskState::Canceled,
            },
            artifacts: Vec::new(),
        })
    }

    /// The bridge's A2A Agent Card, for discovery. Streaming is advertised
    /// because AGDX chunk channels replay off the log. Each advertised
    /// [`CapabilityDescriptor`] (set via [`with_capabilities`](Self::with_capabilities))
    /// projects to an A2A `AgentSkill`: the skill id is its id and name, and the
    /// I/O `ContentRef`s become the skill's input and output modes.
    pub fn card(&self) -> AgentCard {
        let skills = self
            .capabilities
            .iter()
            .map(|capability| AgentSkill {
                id: capability.skill_id.clone(),
                name: capability.skill_id.clone(),
                description: String::new(),
                tags: Vec::new(),
                input_modes: capability.input.iter().map(content_ref_mode).collect(),
                output_modes: capability.output.iter().map(content_ref_mode).collect(),
            })
            .collect();
        AgentCard {
            name: self.source.as_str().to_owned(),
            description: "LaserData AGDX bridge over the durable log".to_owned(),
            version: env!("CARGO_PKG_VERSION").to_owned(),
            supported_interfaces: vec![AgentInterface {
                url: "/".to_owned(),
                protocol_binding: A2A_JSONRPC_BINDING.to_owned(),
                protocol_version: A2A_PROTOCOL_VERSION.to_owned(),
            }],
            capabilities: AgentCardCapabilities {
                streaming: true,
                push_notifications: false,
                state_transition_history: false,
                extended_agent_card: false,
            },
            default_input_modes: vec!["text/plain".to_owned()],
            default_output_modes: vec!["text/plain".to_owned()],
            skills,
            signatures: Vec::new(),
        }
    }

    /// The card with a detached JWS signature attached (`sign` feature): the
    /// JCS-canonicalized card (RFC 8785, minus `signatures`) signed EdDSA with
    /// the same enrolled Ed25519 key the envelope scheme uses, so one principal
    /// key covers both surfaces. Verify with
    /// [`verify_card`](crate::sign::verify_card).
    #[cfg(feature = "sign")]
    pub fn signed_card(&self, key: &crate::sign::SigningKey) -> Result<AgentCard, LaserError> {
        let mut card = self.card();
        let value = serde_json::to_value(&card)
            .map_err(|error| LaserError::Codec(format!("card to JSON: {error}")))?;
        let signature = crate::sign::sign_card_value(key, &value)?;
        card.signatures.push(signature);
        Ok(card)
    }

    pub async fn handle_rpc(&self, request: JsonValue) -> JsonRpcResponse {
        self.handle_rpc_for(request, None).await
    }

    async fn handle_rpc_for(&self, request: JsonValue, caller: Option<&str>) -> JsonRpcResponse {
        let id = request.get("id").cloned().unwrap_or(JsonValue::Null);
        match self.dispatch_rpc(&request, caller).await {
            Ok(result) => JsonRpcResponse {
                jsonrpc: JSONRPC_VERSION,
                id,
                result: Some(result),
                error: None,
            },
            Err(error) => {
                tracing::warn!(error = %error, "A2A request failed");
                JsonRpcResponse {
                    jsonrpc: JSONRPC_VERSION,
                    id,
                    result: None,
                    error: Some(JsonRpcError {
                        code: APP_ERROR_CODE,
                        message: crate::error::public_error_message(&error).to_owned(),
                    }),
                }
            }
        }
    }

    async fn dispatch_rpc(
        &self,
        request: &JsonValue,
        caller: Option<&str>,
    ) -> Result<JsonValue, LaserError> {
        // Serde also reads a struct from an array, so the object check comes first.
        if !request.is_object() {
            return Err(LaserError::HandlerConfig(
                "A2A JSON-RPC request must be an object".to_owned(),
            ));
        }
        let JsonRpcRequest { method, params, .. } = JsonRpcRequest::deserialize(request)
            .map_err(|_| LaserError::HandlerConfig("A2A method must be a string".to_owned()))?;
        let params = if params.is_null() {
            JsonValue::Object(Default::default())
        } else {
            params
        };
        let task = match A2aMethod::from_str(&method) {
            Ok(A2aMethod::MessageSend | A2aMethod::MessageStream) => {
                let body = serde_json::to_vec(&params)
                    .map_err(|error| LaserError::Codec(error.to_string()))?;
                self.submit_addressed(None, None, body, caller).await?
            }
            Ok(operation @ (A2aMethod::TasksGet | A2aMethod::TasksCancel)) => {
                let id = params
                    .as_object()
                    .and_then(|params| params.get("id"))
                    .and_then(JsonValue::as_str)
                    .ok_or_else(|| {
                        LaserError::HandlerConfig("A2A task id must be a string".to_owned())
                    })?;
                if operation == A2aMethod::TasksGet {
                    self.task_for(id, caller).await?
                } else {
                    self.cancel_for(id, caller).await?
                }
            }
            Err(_) => {
                return Err(LaserError::HandlerConfig(format!(
                    "unknown A2A method `{method}`"
                )));
            }
        };
        to_value(task).map_err(|error| LaserError::Codec(error.to_string()))
    }

    /// An axum router: the JSON-RPC endpoint at `/` plus the Agent Card at the
    /// A2A well-known discovery path. Requires the `a2a-http` feature. The bridge
    /// adapter (`submit` / `task` / `cancel` / `card`) is usable without it.
    #[cfg(feature = "a2a-http")]
    pub fn router(self: Arc<Self>) -> Router {
        Router::new()
            .route("/", post(handle_rpc))
            .route(
                "/.well-known/agent-card.json",
                axum::routing::get(handle_card),
            )
            .with_state(self)
    }
}

#[cfg(feature = "a2a-http")]
async fn handle_card(State(bridge): State<Arc<A2aBridge>>) -> axum::Json<AgentCard> {
    axum::Json(bridge.card())
}

// The A2A task identity rides AGDX `correlation`, and deriving it from the task's
// conversation keeps the bridge stateless: `tasks/get` reconstructs it from
// the id alone with no mapping table.
fn correlation_of(conversation: ConversationId) -> CorrelationId {
    CorrelationId::from_u128(conversation.as_u128())
}

#[cfg(feature = "a2a-http")]
async fn handle_rpc(
    State(bridge): State<Arc<A2aBridge>>,
    principal: Option<axum::Extension<A2aCallerPrincipal>>,
    axum::Json(request): axum::Json<JsonValue>,
) -> axum::Json<JsonRpcResponse> {
    let caller = principal.as_ref().map(|principal| principal.0.0.as_str());
    axum::Json(bridge.handle_rpc_for(request, caller).await)
}

fn now_micros() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |elapsed| {
            u64::try_from(elapsed.as_micros()).unwrap_or(u64::MAX)
        })
}

fn unbound_caller() -> LaserError {
    LaserError::HandlerConfig(
        "this bridge binds tasks to callers, so the request needs an authenticated caller"
            .to_owned(),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use laser_wire::framing::{decode_named, encode_named};
    use serde_json::{from_str, to_string};

    #[test]
    fn given_a_task_binding_when_any_bound_field_changes_then_the_digest_should_change() {
        let binding = TaskBinding {
            key: [7; 32],
            ttl: std::time::Duration::from_secs(60),
        };
        let task = ConversationId::derive("task-42");
        let base = binding.mac("alice", 10, task, 1_000);
        assert_eq!(base, binding.mac("alice", 10, task, 1_000));
        assert_ne!(base, binding.mac("bob", 10, task, 1_000), "another caller");
        assert_ne!(
            base,
            binding.mac("alice", 11, task, 1_000),
            "a recreated stream"
        );
        assert_ne!(
            base,
            binding.mac("alice", 10, ConversationId::derive("task-43"), 1_000),
            "another task"
        );
        assert_ne!(
            base,
            binding.mac("alice", 10, task, 2_000),
            "another expiry"
        );
        let other_key = TaskBinding {
            key: [8; 32],
            ..binding.clone()
        };
        assert_ne!(base, other_key.mac("alice", 10, task, 1_000), "another key");
        // A caller name cannot be shifted into the following fields.
        assert_ne!(
            binding.mac("ab", 10, task, 1),
            binding.mac("a", 10, task, 1)
        );
        // The shared vector the Python and TypeScript SDKs check as well.
        assert_eq!(
            base.to_hex().as_str(),
            "d1ea2040a6c1451ff74ad5eb597715ce67fa2d144aa0a2b369d9e56ebf0fee1b"
        );
    }

    #[test]
    fn given_task_states_when_serialized_then_should_use_the_kebab_case_wire_names() {
        let status = TaskStatus {
            state: TaskState::InputRequired,
        };
        assert_eq!(
            to_string(&status).expect("status serializes"),
            r#"{"state":"input-required"}"#
        );
        let back: TaskStatus = from_str(r#"{"state":"auth-required"}"#).expect("status parses");
        assert_eq!(back.state, TaskState::AuthRequired);
        // An unknown inbound name is a protocol error, never a guess.
        assert!(from_str::<TaskStatus>(r#"{"state":"nope"}"#).is_err());
    }

    #[cfg(feature = "a2a-http")]
    #[test]
    fn given_a2a_methods_when_round_tripped_then_should_match_the_wire_names() {
        assert_eq!(A2aMethod::MessageSend.to_string(), "SendMessage");
        assert_eq!(
            "GetTask".parse::<A2aMethod>().expect("the method parses"),
            A2aMethod::TasksGet
        );
        // The 0.3.0 path-style spelling is gone: v1.0 clients speak PascalCase.
        assert!("tasks/get".parse::<A2aMethod>().is_err());
        assert!("nope/now".parse::<A2aMethod>().is_err());
    }

    #[test]
    fn given_a_message_send_when_mapped_then_the_foreign_payload_should_tunnel_byte_identical() {
        let params = br#"{"message":{"role":"user","parts":[{"kind":"text","text":"hi"}]},"metadata":{"trace":"abc"}}"#;
        let envelope = command_from_message_send(
            agdx::RecordId::from_u128(1),
            agdx::ConversationId::from_u128(2),
            "source-agent".parse().expect("valid agent id"),
            agdx::CorrelationId::from_u128(4),
            params.to_vec(),
        );
        agdx::validate(&envelope).expect("the mapped command validates");
        let payload = encode_named(&envelope).expect("encodes");
        let back: AgentEnvelope = decode_named(&payload).expect("decodes");
        assert_eq!(
            back.body,
            params.to_vec(),
            "the tunneled remainder must round-trip byte-identical"
        );
        assert_eq!(back.correlation, envelope.correlation);

        // The way back out: the answering envelope renders as the A2A task.
        let reply = AgentEnvelope::response(
            agdx::RecordId::from_u128(5),
            agdx::ConversationId::from_u128(2),
            "responder-agent".parse().expect("valid agent id"),
            agdx::CorrelationId::from_u128(4),
            b"plan ready".to_vec(),
        )
        .with_task_state(TaskState::Completed);
        let task = task_from_envelope("t-1", &reply);
        assert_eq!(
            to_string(&task).expect("task serializes"),
            r#"{"id":"t-1","status":{"state":"completed"},"artifacts":[{"text":"plan ready"}]}"#
        );

        // A failure terminal reads failed without a task_state.
        let failure = AgentEnvelope::error(
            agdx::RecordId::from_u128(7),
            agdx::ConversationId::from_u128(2),
            "responder-agent".parse().expect("valid agent id"),
            agdx::CorrelationId::from_u128(4),
            b"boom".to_vec(),
        );
        assert_eq!(
            task_from_envelope("t-1", &failure).status.state,
            TaskState::Failed
        );
    }

    fn bridge() -> A2aBridge {
        let laser = Laser::from_client(iggy::prelude::IggyClient::default());
        A2aBridge::new(
            laser,
            "a2a-edge".parse::<AgentId>().expect("valid agent id"),
            AgentTopic::Sessions,
            AgentTopic::Sessions,
        )
    }

    #[tokio::test]
    async fn given_malformed_rpc_requests_when_dispatched_then_should_echo_ids_and_return_public_errors()
     {
        for request in [
            serde_json::json!(null),
            serde_json::json!([]),
            serde_json::json!([7, "GetTask", {"id": "invalid"}]),
            serde_json::json!({"id": 7}),
            serde_json::json!({"id": 7, "method": "unknown"}),
            serde_json::json!({"id": 7, "method": "GetTask", "params": {"id": 12}}),
            serde_json::json!({"id": 7, "method": "CancelTask", "params": []}),
            serde_json::json!({"id": 7, "method": "GetTask", "params": {"id": "invalid"}}),
        ] {
            let expected = request.get("id").cloned().unwrap_or(JsonValue::Null);
            let response = bridge().handle_rpc(request).await;
            assert_eq!(response.id, expected);
            assert_eq!(response.jsonrpc, "2.0");
            assert!(response.result.is_none());
            let error = response.error.expect("malformed request returns an error");
            assert_eq!(error.code, -32000);
            assert_eq!(error.message, "invalid request");
        }
    }

    #[test]
    fn given_a_new_bridge_when_built_then_should_stamp_its_own_id_as_the_hop_path() {
        assert_eq!(bridge().hops, ["a2a-edge"]);
    }

    #[test]
    fn given_an_upstream_hop_path_when_continued_then_should_append_this_bridge() {
        let bridge = bridge()
            .with_bridge_hops(&["mcp-edge".to_owned()])
            .expect("no loop");
        assert_eq!(bridge.hops, ["mcp-edge", "a2a-edge"]);
        assert!(matches!(
            bridge.with_bridge_hops(&["a2a-edge".to_owned()]),
            Err(LaserError::Invalid(_))
        ));
    }

    #[test]
    fn given_a_jsonrpc_message_send_when_parsed_then_should_expose_the_text_part() {
        let request: JsonRpcRequest = from_str(
            r#"{"jsonrpc":"2.0","id":1,"method":"SendMessage",
                "params":{"message":{"role":"user","parts":[{"kind":"text","text":"hi"}]}}}"#,
        )
        .expect("the request parses");
        assert_eq!(request.method, "SendMessage");
        assert_eq!(
            request
                .params
                .pointer("/message/parts/0/text")
                .and_then(JsonValue::as_str),
            Some("hi")
        );
    }
}