protocol_droid/adapters/primary/transport/legacy_sse/frame.rs
1//! The two frame kinds a legacy stream carries, and the port that fills them.
2//!
3//! # Why the ring holds a type rather than a `Value`
4//!
5//! Because one of the two frames is not JSON. The `endpoint` event's data is
6//! a bare URI, and `Value::String("/messages?sessionId=x")` would put quotes
7//! in it — `data: "/messages?sessionId=x"` — which is not a URI and is not
8//! what the specification says to send. A client doing the specified thing
9//! with the event hands that to a URL parser and fails.
10//!
11//! So the payload is an enum and the *variant* picks the name, and it is
12//! impossible to send a `message` event carrying a URI or an `endpoint` event
13//! carrying JSON. That is the reason this is an enum rather than a
14//! `MessageEndpoint` plus a `Value` in a struct: the struct would let a
15//! caller pair the wrong one with the wrong name, and the mismatch would be
16//! found by a peer in production.
17//!
18//! # Why the `endpoint` frame rides the ring
19//!
20//! Because the specification requires it to be the **first** event, and
21//! ordering enforced by a struct field is a convention while ordering enforced
22//! by the ring's insertion order is a guarantee. The router pushes the
23//! `endpoint` frame before it opens the stream and before the session can
24//! receive anything, so a `message` frame physically cannot precede it. A
25//! router that prepended the frame to the response body instead would have
26//! two things to keep in order — the frame and the stream — and one place to
27//! get it wrong.
28//!
29//! The cost is that the frame is a ring entry, so it is replayed by a client
30//! that reconnects. That is harmless and in fact useful: a reconnecting
31//! client needs the endpoint again. The legacy transport has no `Last-Event-ID`
32//! and no resume, so the ring is used here for ordering, not for replay.
33
34use serde_json::Value;
35
36use crate::adapters::primary::transport::http::{
37 EventTail, FramePayload, Refusal, SseError, SseEvent,
38};
39use crate::domain::registry::AuthContext;
40use crate::domain::session::SessionId;
41use crate::domain::{EventId, McpSessionId, MessageEndpoint, names};
42
43/// One frame on a legacy stream.
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub enum LegacyFrame {
46 /// The `endpoint` event: where the client POSTs its messages.
47 ///
48 /// Exactly one of these per session, and the first frame on the stream.
49 Endpoint(MessageEndpoint),
50 /// A JSON-RPC message, as a `message` event.
51 Message(Value),
52}
53
54impl LegacyFrame {
55 /// The endpoint, if this is one.
56 #[must_use]
57 pub const fn endpoint(&self) -> Option<&MessageEndpoint> {
58 match self {
59 Self::Endpoint(endpoint) => Some(endpoint),
60 Self::Message(_) => None,
61 }
62 }
63
64 /// The message, if this is one.
65 #[must_use]
66 pub const fn message(&self) -> Option<&Value> {
67 match self {
68 Self::Endpoint(_) => None,
69 Self::Message(value) => Some(value),
70 }
71 }
72}
73
74impl FramePayload for LegacyFrame {
75 /// # Errors
76 ///
77 /// [`SseError::Unserialisable`] for a [`Message`](crate::domain::a2a::message::Message) whose value has no JSON
78 /// form — a non-string map key. A [`Endpoint`](Self::Endpoint) cannot
79 /// fail: the URI is already parsed, and `MessageEndpoint` refused every
80 /// value that could terminate a line.
81 fn to_frame(&self, id: EventId) -> Result<SseEvent, SseError> {
82 match self {
83 Self::Endpoint(endpoint) => Ok(SseEvent::named_with_id(
84 names::ENDPOINT,
85 id,
86 endpoint.as_str().to_owned(),
87 )),
88 Self::Message(value) => SseEvent::json(id, value),
89 }
90 }
91}
92
93/// A legacy session's server→client tail.
94pub type LegacyTail = EventTail<LegacyFrame>;
95
96/// One legacy session: the stream the server writes to, and whatever the
97/// dispatcher keeps for it.
98///
99/// The store holds these, so a session that expires or is deleted takes its
100/// dispatcher state with it. A dispatcher that kept its own table keyed by
101/// session id would have no way to learn the session had gone.
102#[derive(Debug)]
103pub struct LegacySession<S> {
104 tail: LegacyTail,
105 state: S,
106}
107
108impl<S> LegacySession<S> {
109 /// A session writing to `tail`, with `state` kept for the dispatcher.
110 #[must_use]
111 pub const fn new(tail: LegacyTail, state: S) -> Self {
112 Self { tail, state }
113 }
114
115 /// The stream the server writes this session's frames to.
116 #[must_use]
117 pub const fn tail(&self) -> &LegacyTail {
118 &self.tail
119 }
120
121 /// What the dispatcher kept when the session was opened.
122 #[must_use]
123 pub const fn state(&self) -> &S {
124 &self.state
125 }
126}
127
128/// What the protocol layer has to be able to do for the legacy `POST`.
129///
130/// A port, and it belongs here because the router is its consumer. One method,
131/// because the router asks one question: *what should this client frame cause
132/// the server to send back?*
133///
134/// It is not [`RequestHandler`](crate::adapters::primary::transport::http::RequestHandler),
135/// and the difference is not cosmetic. That trait returns an HTTP `Response`,
136/// which is right when a request's answer travels in the response. Here it
137/// does not: the answer goes onto the SSE stream, and the `POST`'s own response
138/// carries only a status. Handing the router a `Response` it has to take apart
139/// to find the body, and push that body somewhere else, is a shape that invites
140/// the mistake of forwarding the `POST`'s response body directly to the client
141/// — which is exactly the bug the 2025-03-26 client fallback has to avoid.
142///
143/// Returns a `Vec` because one frame can produce a response *and* notifications
144/// the server chose to send, and a single-output signature would force the
145/// caller to pick one.
146///
147/// # Examples
148///
149/// ```
150/// use protocol_droid::adapters::primary::transport::http::Refusal;
151/// use protocol_droid::adapters::primary::transport::legacy_sse::{LegacyDispatcher, LegacyTail};
152/// use protocol_droid::domain::McpSessionId;
153/// use protocol_droid::domain::session::SessionId;
154/// use protocol_droid::domain::registry::AuthContext;
155///
156/// /// Answers every request with the frame it was sent, which is all this test needs.
157/// struct Echo;
158///
159/// impl LegacyDispatcher for Echo {
160/// type Session = ();
161///
162/// fn open(&self, _id: SessionId, _tail: &LegacyTail) {}
163///
164/// async fn dispatch(
165/// &self,
166/// _auth: &AuthContext,
167/// _id: &McpSessionId,
168/// _session: &(),
169/// frame: &serde_json::Value,
170/// ) -> Result<Vec<serde_json::Value>, Refusal> {
171/// Ok(vec![frame.clone()])
172/// }
173/// }
174/// # let _ = Echo;
175/// ```
176pub trait LegacyDispatcher: Send + Sync + 'static {
177 /// What the dispatcher keeps for one session.
178 ///
179 /// Dropped with the session, so a subscription or a negotiated state held
180 /// here ends when the session does.
181 type Session: Send + Sync + 'static;
182
183 /// Called once when a stream opens, before the `endpoint` event is sent.
184 ///
185 /// `tail` is the stream the server's own frames go to: notifications, and
186 /// requests the server makes of the client. Keep it if you will write to it.
187 fn open(&self, id: SessionId, tail: &LegacyTail) -> Self::Session;
188
189 /// Handles one client frame, returning every frame to send back.
190 ///
191 /// `auth` is the principal the router resolved for **this** request, after
192 /// the `Origin` check and the endpoint's [`AuthPolicy`] have both passed.
193 /// It is the only place a legacy-SSE deployment gets to see who is asking:
194 /// the router authenticates and this port is where that result goes, so an
195 /// implementation that ignores it is serving every caller as anonymous.
196 ///
197 /// Per request rather than per session, for the reason it is on
198 /// Streamable HTTP too — a bearer token can be refreshed or withdrawn
199 /// between two `POST`s to one session id.
200 ///
201 /// A dispatcher may also write its answer to the session's tail itself and
202 /// return nothing. The `POST` is answered `202` when this returns, so a
203 /// handler that has to wait on the client (an elicitation, a sampling
204 /// request) can answer on the stream once it has the client's reply instead
205 /// of holding the `POST` open.
206 ///
207 /// # Errors
208 ///
209 /// [`Refusal::Handler`] if the frame cannot be handled. The frames are
210 /// published to the session's tail in order, and a refusal here publishes
211 /// nothing — a client whose request failed gets an error on the stream
212 /// rather than a half-written batch, because a partial batch is
213 /// indistinguishable from a complete one on a stream that carries no
214 /// transaction.
215 ///
216 /// [`AuthPolicy`]: crate::adapters::primary::transport::http::AuthPolicy
217 fn dispatch(
218 &self,
219 auth: &AuthContext,
220 id: &McpSessionId,
221 session: &Self::Session,
222 frame: &Value,
223 ) -> impl core::future::Future<Output = Result<Vec<Value>, Refusal>> + Send;
224}
225
226#[cfg(test)]
227mod tests;