diff --git a/Cargo.lock b/Cargo.lock index 872eed3..5a95598 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,7 +124,7 @@ dependencies = [ [[package]] name = "a2a-slimrpc" -version = "0.2.2" +version = "0.2.3" dependencies = [ "a2a-client-lf", "a2a-lf", @@ -140,6 +140,7 @@ dependencies = [ "futures", "prost", "prost-types", + "serde_json", "tokio", ] diff --git a/a2a-slimrpc/Cargo.toml b/a2a-slimrpc/Cargo.toml index f9dfa7b..9e74f30 100644 --- a/a2a-slimrpc/Cargo.toml +++ b/a2a-slimrpc/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "a2a-slimrpc" -version = "0.2.2" +version = "0.2.3" description = "A2A v1 SLIMRPC protocol binding for client and server" readme = "README.md" edition.workspace = true @@ -25,7 +25,8 @@ slim_rpc = { workspace = true } slim_service = { workspace = true } slim_datapath = { workspace = true } slim_auth = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } [dev-dependencies] -tokio = { workspace = true } slim_config = { workspace = true } diff --git a/a2a-slimrpc/src/client.rs b/a2a-slimrpc/src/client.rs index b877982..9c74a9c 100644 --- a/a2a-slimrpc/src/client.rs +++ b/a2a-slimrpc/src/client.rs @@ -17,15 +17,17 @@ pub type SlimApp = slim_service::app::App< >; use crate::common::{ - A2A_SLIMRPC_SERVICE, METHOD_CANCEL_TASK, METHOD_CREATE_PUSH_CONFIG, METHOD_DELETE_PUSH_CONFIG, - METHOD_GET_EXTENDED_AGENT_CARD, METHOD_GET_PUSH_CONFIG, METHOD_GET_TASK, - METHOD_LIST_PUSH_CONFIGS, METHOD_LIST_TASKS, METHOD_SEND_MESSAGE, - METHOD_SEND_STREAMING_MESSAGE, METHOD_SUBSCRIBE_TO_TASK, decode_proto_response, - encode_proto_message, service_params_to_metadata_opt, + A2A_COLLABORATIVE_CHANNEL_SERVICE, A2A_SLIMRPC_SERVICE, METHOD_CANCEL_TASK, METHOD_COLLABORATE, + METHOD_CREATE_PUSH_CONFIG, METHOD_DELETE_PUSH_CONFIG, METHOD_GET_EXTENDED_AGENT_CARD, + METHOD_GET_PUSH_CONFIG, METHOD_GET_TASK, METHOD_LIST_PUSH_CONFIGS, METHOD_LIST_TASKS, + METHOD_SEND_MESSAGE, METHOD_SEND_STREAMING_MESSAGE, METHOD_SUBSCRIBE_TO_TASK, + SLIM_SRC_METADATA_KEY, decode_proto_response, encode_proto_message, + service_params_to_metadata_opt, }; use crate::errors::rpc_error_to_a2a_error; /// SLIMRPC transport for A2A clients. +#[derive(Clone)] pub struct SlimRpcTransport { channel: slim_rpc::Channel, } @@ -49,6 +51,64 @@ impl SlimRpcTransport { Self { channel } } + /// Build a transport backed by a SLIM **group** channel spanning `members`, + /// for the `Collaborate` many-to-many operation (see + /// [`Self::collaborate`]). Unlike the point-to-point [`Self::new`], this does + /// not correspond to a single `Transport` peer. + pub fn new_group( + app: Arc, + members: Vec>, + ) -> Result { + Self::new_group_with_connection(app, members, None) + } + + pub fn new_group_with_connection( + app: Arc, + members: Vec>, + connection_id: Option, + ) -> Result { + Ok(Self { + channel: slim_rpc::Channel::new_group_with_connection(app, members, connection_id)?, + }) + } + + /// Open a `Collaborate` session on this (group) channel: broadcast every + /// `Message` produced by `outbound` to the group, and yield every `Message` + /// broadcast by other members — each attributed via + /// `metadata["slim-src"]` (the sender's SLIM name), per the SLIMRPC + /// collaborative channel extension spec. Additive: not part of the + /// point-to-point [`Transport`] trait, whose methods assume exactly one + /// response per call. + pub fn collaborate( + &self, + outbound: impl futures::Stream + Send + 'static, + timeout: Option, + ) -> impl futures::Stream> { + let request_stream = + outbound.map(|message| encode_proto_message(&pbconv::to_proto_message(&message))); + let stream = self.channel.multicast_stream_stream::, Vec>( + A2A_COLLABORATIVE_CHANNEL_SERVICE, + METHOD_COLLABORATE, + request_stream, + timeout, + None, + ); + stream.map(|item| { + let item = item.map_err(|error| rpc_error_to_a2a_error(&error))?; + let proto_message = + decode_proto_response::(item.message, "Message")?; + let mut message = pbconv::from_proto_message(&proto_message); + message + .metadata + .get_or_insert_with(std::collections::HashMap::new) + .insert( + SLIM_SRC_METADATA_KEY.to_string(), + serde_json::Value::String(item.context.source.to_string()), + ); + Ok(message) + }) + } + async fn call_unary( &self, params: &ServiceParams, diff --git a/a2a-slimrpc/src/common.rs b/a2a-slimrpc/src/common.rs index 0f80cc6..8513fba 100644 --- a/a2a-slimrpc/src/common.rs +++ b/a2a-slimrpc/src/common.rs @@ -25,6 +25,18 @@ pub const METHOD_LIST_PUSH_CONFIGS: &str = "ListTaskPushNotificationConfigs"; pub const METHOD_DELETE_PUSH_CONFIG: &str = "DeleteTaskPushNotificationConfig"; pub const METHOD_GET_EXTENDED_AGENT_CARD: &str = "GetExtendedAgentCard"; +/// The `Collaborate` operation (many-to-many messaging on a SLIM group channel; +/// see the SLIMRPC collaborative channel extension spec) lives on its own service, +/// separate from the point-to-point `A2AService`. +pub const A2A_COLLABORATIVE_CHANNEL_SERVICE: &str = + "experimental.slimrpc.collaborative_channel.v1.CollaborativeChannelService"; +pub const METHOD_COLLABORATE: &str = "Collaborate"; + +/// `Message.metadata` key the SLIMRPC layer populates with the sender's SLIM name +/// on every inbound `Collaborate` message, per the collaborative channel spec's +/// message-attribution requirement. +pub const SLIM_SRC_METADATA_KEY: &str = "slim-src"; + pub fn encode_proto_message(message: &T) -> Vec where T: Message, diff --git a/a2a-slimrpc/src/lib.rs b/a2a-slimrpc/src/lib.rs index a656aa5..cd2d94a 100644 --- a/a2a-slimrpc/src/lib.rs +++ b/a2a-slimrpc/src/lib.rs @@ -6,4 +6,5 @@ pub mod errors; pub mod server; pub use client::{SlimApp, SlimRpcTransport, SlimRpcTransportFactory, parse_slimrpc_target}; -pub use server::SlimRpcHandler; +pub use common::SLIM_SRC_METADATA_KEY; +pub use server::{SlimRpcHandler, register_collaborate}; diff --git a/a2a-slimrpc/src/server.rs b/a2a-slimrpc/src/server.rs index b2c9ff7..7be16b9 100644 --- a/a2a-slimrpc/src/server.rs +++ b/a2a-slimrpc/src/server.rs @@ -4,17 +4,17 @@ use std::future::Future; use std::pin::Pin; use std::sync::Arc; -use a2a::{A2AError, StreamResponse}; +use a2a::{A2AError, Message, StreamResponse}; use a2a_pb::pbconv; use a2a_pb::proto; use a2a_server::RequestHandler; use futures::StreamExt; use crate::common::{ - A2A_SLIMRPC_SERVICE, METHOD_CANCEL_TASK, METHOD_CREATE_PUSH_CONFIG, METHOD_DELETE_PUSH_CONFIG, - METHOD_GET_EXTENDED_AGENT_CARD, METHOD_GET_PUSH_CONFIG, METHOD_GET_TASK, - METHOD_LIST_PUSH_CONFIGS, METHOD_LIST_TASKS, METHOD_SEND_MESSAGE, - METHOD_SEND_STREAMING_MESSAGE, METHOD_SUBSCRIBE_TO_TASK, ServiceParamsMap, + A2A_COLLABORATIVE_CHANNEL_SERVICE, A2A_SLIMRPC_SERVICE, METHOD_CANCEL_TASK, METHOD_COLLABORATE, + METHOD_CREATE_PUSH_CONFIG, METHOD_DELETE_PUSH_CONFIG, METHOD_GET_EXTENDED_AGENT_CARD, + METHOD_GET_PUSH_CONFIG, METHOD_GET_TASK, METHOD_LIST_PUSH_CONFIGS, METHOD_LIST_TASKS, + METHOD_SEND_MESSAGE, METHOD_SEND_STREAMING_MESSAGE, METHOD_SUBSCRIBE_TO_TASK, ServiceParamsMap, context_to_service_params, decode_proto_request, encode_proto_message, }; use crate::errors::a2a_error_to_rpc_error; @@ -157,6 +157,46 @@ impl SlimRpcHandler { } } +/// Register `server` as a **listen-only** `Collaborate` participant (see the +/// SLIMRPC collaborative channel extension spec): `on_message` is invoked for +/// every inbound `Message` on any `Collaborate` session this app is a member of. +/// This participant sends nothing of its own — its reply stream is empty, which +/// per the spec signals non-participation on the send side without affecting its +/// ability to receive. +/// +/// Independent of [`SlimRpcHandler`]/`RequestHandler` — `Collaborate` is a +/// broadcast channel rather than a request/response operation, so this takes a +/// plain callback instead. +pub fn register_collaborate(server: &slim_rpc::Server, on_message: F) +where + F: Fn(Message) + Send + Sync + 'static, +{ + let on_message = Arc::new(on_message); + server.register_stream_stream::<_, Vec, Vec, _, _>( + A2A_COLLABORATIVE_CHANNEL_SERVICE, + METHOD_COLLABORATE, + move |mut stream: slim_rpc::DecodedStream>, _context: slim_rpc::Context| { + let on_message = on_message.clone(); + async move { + // Drain the broadcast concurrently with returning the (empty) + // reply stream below, so this member's non-participation is + // visible immediately rather than only after the session ends. + tokio::spawn(async move { + while let Some(item) = stream.next().await { + if let Ok(bytes) = item + && let Ok(proto_message) = + decode_proto_request::(bytes, "Message") + { + on_message(pbconv::from_proto_message(&proto_message)); + } + } + }); + Ok(futures::stream::empty::, slim_rpc::RpcError>>()) + } + }, + ); +} + fn register_unary_unary( server: &slim_rpc::Server, handler: Arc, diff --git a/a2a-slimrpc/tests/e2e.rs b/a2a-slimrpc/tests/e2e.rs index 0fc2add..f927014 100644 --- a/a2a-slimrpc/tests/e2e.rs +++ b/a2a-slimrpc/tests/e2e.rs @@ -738,3 +738,106 @@ async fn slimrpc_transport_reports_malformed_payloads() { env.shutdown().await; } + +/// Group `Collaborate`: every member independently broadcasts one intro `Message` +/// to the others over its own group channel, and every other member's listen-only +/// `register_collaborate` handler receives it — proving the SLIMRPC collaborative +/// channel extension's many-to-many delivery end to end (not just moderator-to- +/// members fan-out). +#[tokio::test] +async fn slimrpc_collaborate_broadcasts_to_every_other_member() { + use std::sync::Mutex; + + const MEMBERS: [&str; 3] = ["avatar", "agent-a", "agent-b"]; + + let service = Arc::new(Service::new( + ID::new_with_name(Kind::new("slim").unwrap(), "collab-group").unwrap(), + )); + + struct Member { + name: Arc, + app: Arc, + received: Arc>>, + } + + let mut members = Vec::new(); + for suffix in MEMBERS { + let name = Arc::new(Name::from_strings([ + "org", + "test", + &format!("collab-{suffix}"), + ])); + let (provider, verifier) = auth(&format!("collab-provider-{suffix}")); + let (app, notifications) = service.create_app(&name, provider, verifier).unwrap(); + let app = Arc::new(app); + + let server = Arc::new(Server::new( + app.clone(), + app.app_name().clone(), + notifications, + )); + let received = Arc::new(Mutex::new(Vec::new())); + let received_for_handler = received.clone(); + a2a_slimrpc::register_collaborate(&server, move |message| { + received_for_handler.lock().unwrap().push(message); + }); + tokio::spawn(async move { + let _ = server.serve().await; + }); + + members.push(Member { + name, + app, + received, + }); + } + tokio::time::sleep(Duration::from_millis(200)).await; + + // Every member independently broadcasts its own intro to the other two. + let mut senders = Vec::new(); + for (i, member) in members.iter().enumerate() { + let others: Vec> = members + .iter() + .enumerate() + .filter(|(j, _)| *j != i) + .map(|(_, m)| m.name.clone()) + .collect(); + let transport = SlimRpcTransport::new_group(member.app.clone(), others).unwrap(); + let intro = sample_message( + Role::Agent, + &format!("hi, I am {}", MEMBERS[i]), + "collaborate", + ); + let outbound = stream::once(async move { intro }); + senders.push(tokio::spawn(async move { + let replies = transport.collaborate(outbound, Some(Duration::from_secs(5))); + futures::pin_mut!(replies); + // Everyone else is listen-only, so this drains to nothing; the point + // is proving the broadcast send doesn't error. + while let Some(reply) = replies.next().await { + reply.expect("collaborate reply stream should not error"); + } + })); + } + for sender in senders { + sender.await.expect("collaborate sender task panicked"); + } + + tokio::time::sleep(Duration::from_millis(300)).await; + + for (i, member) in members.iter().enumerate() { + let received = member.received.lock().unwrap(); + for (j, other) in MEMBERS.iter().enumerate() { + if i == j { + continue; + } + let expected = format!("hi, I am {other}"); + assert!( + received.iter().any(|m| m.text() == Some(expected.as_str())), + "{} did not receive {other}'s intro; got {:?}", + MEMBERS[i], + received.iter().map(|m| m.text()).collect::>() + ); + } + } +}