feat(slimrpc): add Collaborate — many-to-many messaging on SLIM group channels - #111
Merged
Conversation
… channels Implements the demo-sufficient subset of the SLIMRPC collaborative channel extension (a2aproject/experimental-cpb-slimrpc#5): rpc Collaborate(stream Message) returns (stream Message); Unlike the existing multicast SendMessage/SendStreamingMessage (one client to many independent agent responses), Collaborate is genuinely many-to-many: any group-channel member may originate messages, and every message is broadcast to every other member, reusing A2A's own Message type in both directions. - SlimRpcTransport::new_group/new_group_with_connection build a transport over a SLIM group channel (slim_rpc::Channel::new_group), instead of the point-to-point new/new_with_connection. - SlimRpcTransport::collaborate(outbound, timeout) drives Channel::multicast_stream_stream with A2A Message protos, and stamps every inbound Message with metadata["slim-src"] = the sender's SLIM name (from MulticastItem.context.source), per the spec's message-attribution requirement. Additive, not part of the point-to-point Transport trait, whose methods assume exactly one response per call. - register_collaborate registers a listen-only Collaborate participant on a slim_rpc::Server: it invokes a callback for every inbound Message and sends nothing of its own (an empty reply stream), matching the spec's non-participation mode. Independent of RequestHandler, since Collaborate is a broadcast channel rather than a request/response operation. Not yet implemented (spec is still draft): Agent Card extension declaration, formal EOS/non-participation signaling beyond the empty reply stream, and server-side per-message sender attribution (Context/DecodedStream don't expose it with the current API; attribution is only available client-side via MulticastItem, which is what collaborate() uses). Test: slimrpc_collaborate_broadcasts_to_every_other_member — 3 members each independently broadcast one intro Message to the other two via collaborate(), and every other member's register_collaborate listener receives it. Bumps a2a-slimrpc to 0.2.3; adds serde_json, promotes tokio to a real dependency (the SLIM stack already requires a tokio runtime). Signed-off-by: Luca Muscariello <muscariello@ieee.org>
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
Consumers that only register a listen-only Collaborate participant (via register_collaborate) need the attribution key name to read Message.metadata["slim-src"] themselves, since server-side attribution isn't populated automatically (see the earlier commit's note on this limitation). Signed-off-by: Luca Muscariello <muscariello@ieee.org>
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements the demo-sufficient subset of the SLIMRPC collaborative channel extension (currently a draft spec, open for review):
Unlike the existing multicast
SendMessage/SendStreamingMessage(one client → many independent agent responses),Collaborateis genuinely many-to-many: any group-channel member may originate messages, and every message is broadcast to every other member. Reuses A2A's ownMessagetype in both directions — no new proto types.Changes
SlimRpcTransport::new_group/new_group_with_connection— build a transport over a SLIM group channel (slim_rpc::Channel::new_group), as opposed to the existing point-to-pointnew/new_with_connection.SlimRpcTransport::collaborate(outbound, timeout)— drivesChannel::multicast_stream_streamwith A2AMessageprotos, and stamps every inboundMessagewithmetadata["slim-src"]= the sender's SLIM name (fromMulticastItem.context.source), per the spec's message-attribution requirement. Additive — not part of the point-to-pointTransporttrait, whose methods assume exactly one response per call.register_collaborate— registers a listen-onlyCollaborateparticipant on aslim_rpc::Server: invokes a callback for every inboundMessage, sends nothing of its own (empty reply stream), matching the spec's non-participation mode. Independent ofRequestHandler, sinceCollaborateis a broadcast channel rather than request/response.Not yet implemented (spec is still draft): Agent Card extension declaration, formal EOS/non-participation signaling beyond the empty reply stream, and server-side per-message sender attribution (
Context/DecodedStreamdon't expose it with the current API — attribution is only available client-side viaMulticastItem, which is whatcollaborate()uses).Bumps
a2a-slimrpcto 0.2.3; addsserde_json; promotestokioto a real dependency (the SLIM stack already requires a tokio runtime transitively).Test
slimrpc_collaborate_broadcasts_to_every_other_member: 3 members each independently broadcast one introMessageto the other two viacollaborate(), and every other member'sregister_collaboratelistener receives it — proving many-to-many delivery end to end. All existing tests + clippy + fmt pass.