feat(slimrpc): add Collaborate — many-to-many messaging on SLIM group channels

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 }

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<SlimApp>,
+        members: Vec<Arc<ProtoName>>,
+    ) -> Result<Self, slim_rpc::RpcError> {
+        Self::new_group_with_connection(app, members, None)
+    }
+
+    pub fn new_group_with_connection(
+        app: Arc<SlimApp>,
+        members: Vec<Arc<ProtoName>>,
+        connection_id: Option<u64>,
+    ) -> Result<Self, slim_rpc::RpcError> {
+        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<Item = Message> + Send + 'static,
+        timeout: Option<std::time::Duration>,
+    ) -> impl futures::Stream<Item = Result<Message, A2AError>> {
+        let request_stream =
+            outbound.map(|message| encode_proto_message(&pbconv::to_proto_message(&message)));
+        let stream = self.channel.multicast_stream_stream::<Vec<u8>, Vec<u8>>(
+            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::<a2a_pb::proto::Message>(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<Req, Res>(
         &self,
         params: &ServiceParams,

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<T>(message: &T) -> Vec<u8>
 where
     T: Message,

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};

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<H: RequestHandler> SlimRpcHandler<H> {
     }
 }
 
+/// 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<F>(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<u8>, Vec<u8>, _, _>(
+        A2A_COLLABORATIVE_CHANNEL_SERVICE,
+        METHOD_COLLABORATE,
+        move |mut stream: slim_rpc::DecodedStream<Vec<u8>>, _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::<a2a_pb::proto::Message>(bytes, "Message")
+                        {
+                            on_message(pbconv::from_proto_message(&proto_message));
+                        }
+                    }
+                });
+                Ok(futures::stream::empty::<Result<Vec<u8>, slim_rpc::RpcError>>())
+            }
+        },
+    );
+}
+
 fn register_unary_unary<H, ReqProto, ReqNative, ResNative, DecodeReq, EncodeRes>(
     server: &slim_rpc::Server,
     handler: Arc<H>,

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<Name>,
+        app: Arc<SlimApp>,
+        received: Arc<Mutex<Vec<Message>>>,
+    }
+
+    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<Arc<Name>> = 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::<Vec<_>>()
+            );
+        }
+    }
+}