Skip to main content

objectiveai_sdk/http/
client.rs

1//! HTTP client implementation for ObjectiveAI API.
2
3use crate::error;
4use eventsource_stream::Event as MessageEvent;
5use futures::{SinkExt, Stream, StreamExt};
6use reqwest_eventsource::{Event, RequestBuilderExt};
7use std::sync::Arc;
8use tokio_tungstenite::tungstenite;
9
10/// HTTP client for making requests to the ObjectiveAI API.
11///
12/// Handles authentication, request building, and response parsing for both
13/// unary and streaming endpoints.
14///
15/// # Example
16///
17/// ```ignore
18/// let client = HttpClient::new(
19///     reqwest::Client::new(),
20///     None, // Use default address
21///     Some("your-api-key"),
22///     None, // user_agent
23///     None, // x_title
24///     None, // http_referer
25///     None, // x_github_authorization
26///     None, // x_openrouter_authorization
27///     None, // x_mcp_authorization
28///     None, // agent_instance_hierarchy
29///     None, // mcp_session_id
30///     None, // mcp_call_timeout_ms
31/// );
32/// ```
33#[derive(Debug, Clone)]
34pub struct HttpClient {
35    /// The underlying reqwest HTTP client.
36    pub http_client: reqwest::Client,
37    /// Base URL for API requests. Defaults to `https://api.objectiveai.dev`.
38    pub address: String,
39    /// API key for authentication. Sent as `Bearer` token in `Authorization` header.
40    pub authorization: Option<Arc<String>>,
41    /// Value for the `User-Agent` header.
42    pub user_agent: Option<String>,
43    /// Value for the `X-Title` header.
44    pub x_title: Option<String>,
45    /// Value for both `Referer` and `HTTP-Referer` headers.
46    pub http_referer: Option<String>,
47    /// Value for the `X-GITHUB-AUTHORIZATION` header.
48    pub x_github_authorization: Option<Arc<String>>,
49    /// Value for the `X-OPENROUTER-AUTHORIZATION` header.
50    pub x_openrouter_authorization: Option<Arc<String>>,
51    /// Values for the `X-MCP-AUTHORIZATION` header (JSON-encoded).
52    pub x_mcp_authorization:
53        Option<Arc<std::collections::HashMap<String, String>>>,
54    /// Value for the `X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY` header.
55    pub agent_instance_hierarchy: Option<Arc<String>>,
56    /// Value for the `Mcp-Session-Id` header — propagated through to
57    /// `objectiveai-mcp` so server-side tool invocations see a stable
58    /// per-session id. See `objectiveai_sdk::mcp::MCP_SESSION_ID_HEADER`.
59    pub mcp_session_id: Option<Arc<String>>,
60    /// Value for the `X-MCP-CALL-TIMEOUT` header, in integer
61    /// milliseconds: the per-request budget the API applies to each MCP
62    /// CALL its proxy makes on this request's behalf (HTTP and ws://
63    /// upstreams alike; never connects, never laboratory transfers).
64    /// Unset ⇒ no header ⇒ the API applies NO call timeout.
65    pub mcp_call_timeout_ms: Option<u64>,
66}
67
68impl HttpClient {
69    /// Creates a new HTTP client.
70    ///
71    /// # Arguments
72    ///
73    /// * `http_client` - The reqwest client to use for requests
74    /// * `address` - Base URL for API requests (defaults to `https://api.objectiveai.dev`)
75    /// * `authorization` - API key for authentication
76    /// * `user_agent` - Optional User-Agent header value
77    /// * `x_title` - Optional X-Title header value
78    /// * `http_referer` - Optional Referer header value
79    /// * `x_github_authorization` - Optional X-GITHUB-AUTHORIZATION header value
80    /// * `x_openrouter_authorization` - Optional X-OPENROUTER-AUTHORIZATION header value
81    /// * `x_mcp_authorization` - Optional X-MCP-AUTHORIZATION header value (HashMap)
82    /// * `agent_instance_hierarchy` - Optional X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY header value
83    /// * `mcp_session_id` - Optional Mcp-Session-Id header value
84    /// * `mcp_call_timeout_ms` - Optional X-MCP-CALL-TIMEOUT header value
85    ///   (integer ms; no env fallback — config/option-only)
86    pub fn new(
87        http_client: reqwest::Client,
88        address: Option<impl Into<String>>,
89        authorization: Option<impl Into<String>>,
90        user_agent: Option<impl Into<String>>,
91        x_title: Option<impl Into<String>>,
92        http_referer: Option<impl Into<String>>,
93        x_github_authorization: Option<impl Into<String>>,
94        x_openrouter_authorization: Option<impl Into<String>>,
95        x_mcp_authorization: Option<std::collections::HashMap<String, String>>,
96        agent_instance_hierarchy: Option<impl Into<String>>,
97        mcp_session_id: Option<impl Into<String>>,
98        mcp_call_timeout_ms: Option<u64>,
99    ) -> Self {
100        #[cfg(feature = "env")]
101        let env = |name: &str| -> Option<String> { std::env::var(name).ok() };
102
103        Self {
104            http_client,
105            address: match address {
106                Some(base) => base.into(),
107                #[cfg(feature = "env")]
108                None => env("OBJECTIVEAI_ADDRESS").unwrap_or_else(|| {
109                    "https://api.objectiveai.dev".to_string()
110                }),
111                #[cfg(not(feature = "env"))]
112                None => "https://api.objectiveai.dev".to_string(),
113            },
114            authorization: authorization.map(|k| Arc::new(k.into())).or_else(
115                || {
116                    #[cfg(feature = "env")]
117                    {
118                        env("OBJECTIVEAI_AUTHORIZATION").map(Arc::new)
119                    }
120                    #[cfg(not(feature = "env"))]
121                    {
122                        None
123                    }
124                },
125            ),
126            user_agent: user_agent.map(Into::into).or_else(|| {
127                #[cfg(feature = "env")]
128                {
129                    env("USER_AGENT")
130                }
131                #[cfg(not(feature = "env"))]
132                {
133                    None
134                }
135            }),
136            x_title: x_title.map(Into::into).or_else(|| {
137                #[cfg(feature = "env")]
138                {
139                    env("X_TITLE")
140                }
141                #[cfg(not(feature = "env"))]
142                {
143                    None
144                }
145            }),
146            http_referer: http_referer.map(Into::into).or_else(|| {
147                #[cfg(feature = "env")]
148                {
149                    env("HTTP_REFERER")
150                }
151                #[cfg(not(feature = "env"))]
152                {
153                    None
154                }
155            }),
156            x_github_authorization: x_github_authorization
157                .map(|v| Arc::new(v.into()))
158                .or_else(|| {
159                    #[cfg(feature = "env")]
160                    {
161                        env("GITHUB_AUTHORIZATION").map(Arc::new)
162                    }
163                    #[cfg(not(feature = "env"))]
164                    {
165                        None
166                    }
167                }),
168            x_openrouter_authorization: x_openrouter_authorization
169                .map(|v| Arc::new(v.into()))
170                .or_else(|| {
171                    #[cfg(feature = "env")]
172                    {
173                        env("OPENROUTER_AUTHORIZATION").map(Arc::new)
174                    }
175                    #[cfg(not(feature = "env"))]
176                    {
177                        None
178                    }
179                }),
180            x_mcp_authorization: x_mcp_authorization.map(Arc::new).or_else(
181                || {
182                    #[cfg(feature = "env")]
183                    {
184                        env("MCP_AUTHORIZATION")
185                            .and_then(|v| serde_json::from_str(&v).ok())
186                            .map(Arc::new)
187                    }
188                    #[cfg(not(feature = "env"))]
189                    {
190                        None
191                    }
192                },
193            ),
194            agent_instance_hierarchy: agent_instance_hierarchy.map(|v| Arc::new(v.into())).or_else(|| {
195                #[cfg(feature = "env")]
196                {
197                    env("OBJECTIVEAI_AGENT_INSTANCE_HIERARCHY").map(Arc::new)
198                }
199                #[cfg(not(feature = "env"))]
200                {
201                    None
202                }
203            }),
204            mcp_session_id: mcp_session_id.map(|v| Arc::new(v.into())).or_else(
205                || {
206                    #[cfg(feature = "env")]
207                    {
208                        env(crate::mcp::MCP_SESSION_ID_ENV).map(Arc::new)
209                    }
210                    #[cfg(not(feature = "env"))]
211                    {
212                        None
213                    }
214                },
215            ),
216            // Deliberately no env fallback: the caller (e.g. the daemon,
217            // from its `api.mcp_call_timeout_ms` config) supplies it or not.
218            mcp_call_timeout_ms,
219        }
220    }
221
222    /// Builds a request with authentication and custom headers.
223    fn request(
224        &self,
225        method: reqwest::Method,
226        path: &str,
227        body: Option<impl serde::Serialize>,
228    ) -> reqwest::RequestBuilder {
229        let url = format!(
230            "{}/{}",
231            self.address.trim_end_matches('/'),
232            path.trim_start_matches('/')
233        );
234        let mut request = self.http_client.request(method, &url);
235        if let Some(authorization) = &self.authorization {
236            let key = authorization
237                .strip_prefix("Bearer ")
238                .unwrap_or(authorization);
239            request =
240                request.header("authorization", format!("Bearer {}", key));
241        }
242        if let Some(user_agent) = &self.user_agent {
243            request = request.header("user-agent", user_agent);
244        }
245        if let Some(x_title) = &self.x_title {
246            request = request.header("x-title", x_title);
247        }
248        if let Some(http_referer) = &self.http_referer {
249            request = request.header("referer", http_referer);
250            request = request.header("http-referer", http_referer);
251        }
252        if let Some(token) = &self.x_github_authorization {
253            request = request.header("X-GITHUB-AUTHORIZATION", token.as_str());
254        }
255        if let Some(token) = &self.x_openrouter_authorization {
256            request =
257                request.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
258        }
259        if let Some(headers) = &self.x_mcp_authorization {
260            if let Ok(json) = serde_json::to_string(headers.as_ref()) {
261                request = request.header("X-MCP-AUTHORIZATION", json);
262            }
263        }
264        if let Some(id) = &self.agent_instance_hierarchy {
265            request = request.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
266        }
267        if let Some(s) = &self.mcp_session_id {
268            request =
269                request.header(crate::mcp::MCP_SESSION_ID_HEADER, s.as_str());
270        }
271        if let Some(ms) = self.mcp_call_timeout_ms {
272            request = request.header("X-MCP-CALL-TIMEOUT", ms.to_string());
273        }
274        if let Some(body) = body {
275            request = request.json(&body);
276        }
277        request
278    }
279
280    /// Sends a unary (request-response) API call and deserializes the response.
281    ///
282    /// # Type Parameters
283    ///
284    /// * `T` - The expected response type to deserialize into
285    ///
286    /// # Errors
287    ///
288    /// Returns [`super::HttpError`] if the request fails, returns a non-success status,
289    /// or the response cannot be deserialized.
290    pub async fn send_unary<T: serde::de::DeserializeOwned + Send + 'static>(
291        &self,
292        method: reqwest::Method,
293        path: impl AsRef<str>,
294        body: Option<impl serde::Serialize>,
295    ) -> Result<T, super::HttpError> {
296        let response = self
297            .http_client
298            .execute(
299                self.request(method, path.as_ref(), body)
300                    .build()
301                    .map_err(super::HttpError::RequestError)?,
302            )
303            .await
304            .map_err(super::HttpError::HttpError)?;
305        let code = response.status();
306        if code.is_success() {
307            let text =
308                response.text().await.map_err(super::HttpError::HttpError)?;
309            let mut de = serde_json::Deserializer::from_str(&text);
310            match serde_path_to_error::deserialize::<_, T>(&mut de) {
311                Ok(value) => Ok(value),
312                Err(e) => Err(super::HttpError::DeserializationError(e)),
313            }
314        } else {
315            match response.text().await {
316                Ok(text) => Err(super::HttpError::BadStatus {
317                    code,
318                    body: match serde_json::from_str::<serde_json::Value>(&text)
319                    {
320                        Ok(body) => body,
321                        Err(_) => serde_json::Value::String(text),
322                    },
323                }),
324                Err(_) => Err(super::HttpError::BadStatus {
325                    code,
326                    body: serde_json::Value::Null,
327                }),
328            }
329        }
330    }
331
332    /// Sends a unary API call that expects no response body.
333    ///
334    /// Useful for DELETE or other operations that only return a status code.
335    ///
336    /// # Errors
337    ///
338    /// Returns [`super::HttpError`] if the request fails or returns a non-success status.
339    pub async fn send_unary_no_response(
340        &self,
341        method: reqwest::Method,
342        path: impl AsRef<str>,
343        body: Option<impl serde::Serialize>,
344    ) -> Result<(), super::HttpError> {
345        let response = self
346            .http_client
347            .execute(
348                self.request(method, path.as_ref(), body)
349                    .build()
350                    .map_err(super::HttpError::RequestError)?,
351            )
352            .await
353            .map_err(super::HttpError::HttpError)?;
354        let code = response.status();
355        if code.is_success() {
356            Ok(())
357        } else {
358            match response.text().await {
359                Ok(text) => Err(super::HttpError::BadStatus {
360                    code,
361                    body: match serde_json::from_str::<serde_json::Value>(&text)
362                    {
363                        Ok(body) => body,
364                        Err(_) => serde_json::Value::String(text),
365                    },
366                }),
367                Err(_) => Err(super::HttpError::BadStatus {
368                    code,
369                    body: serde_json::Value::Null,
370                }),
371            }
372        }
373    }
374
375    /// Sends a streaming API call using Server-Sent Events (SSE).
376    ///
377    /// Returns a stream of deserialized chunks. The stream automatically handles:
378    /// - SSE `[DONE]` messages (filtered out)
379    /// - Comment lines starting with `:` (filtered out)
380    /// - Empty data lines (filtered out)
381    /// - API errors embedded in stream data
382    ///
383    /// # Type Parameters
384    ///
385    /// * `T` - The expected chunk type to deserialize each SSE message into
386    ///
387    /// # Errors
388    ///
389    /// Returns [`super::HttpError`] if the stream cannot be established.
390    pub async fn send_streaming<
391        T: serde::de::DeserializeOwned + Send + 'static,
392        P: AsRef<str> + Send,
393        B: serde::Serialize + Send,
394    >(
395        &self,
396        method: reqwest::Method,
397        path: P,
398        body: Option<B>,
399    ) -> Result<
400        impl Stream<Item = Result<T, super::HttpError>>
401        + Send
402        + 'static
403        + use<T, P, B>,
404        super::HttpError,
405    > {
406        // Stop the stream at [DONE] to prevent reqwest_eventsource from
407        // auto-reconnecting. Uses take_while on the raw SSE events, then
408        // maps/filters the remaining events into typed chunks.
409        // Stamps X-Transport: sse so the API's transport dispatcher
410        // routes this to the SSE branch (the API default is WS).
411        Ok(
412            self.request(method, path.as_ref(), body)
413                .header("X-Transport", "sse")
414                .eventsource()?
415                .take_while(|result| {
416                    let dominated = matches!(
417                        result,
418                        Ok(Event::Message(MessageEvent { data, .. })) if data == "[DONE]"
419                    );
420                    async move { !dominated }
421                })
422                .then(|result| async {
423                    match result {
424                        Ok(Event::Open) => None,
425                        Ok(Event::Message(MessageEvent { data, .. }))
426                            if data.starts_with(":")
427                                || data.is_empty() =>
428                        {
429                            None
430                        }
431                        Ok(Event::Message(MessageEvent { data, .. })) => {
432                            let mut de =
433                                serde_json::Deserializer::from_str(&data);
434                            Some(
435                                match serde_path_to_error::deserialize::<_, T>(
436                                    &mut de,
437                                ) {
438                                    Ok(value) => Ok(value),
439                                    Err(e) => match serde_json::from_str::<error::ResponseError>(&data) {
440                                        Ok(err) => Err(super::HttpError::ApiError(err)),
441                                        Err(_) => Err(super::HttpError::DeserializationError(e)),
442                                    },
443                                }
444                            )
445                        }
446                        Err(reqwest_eventsource::Error::InvalidStatusCode(
447                            code,
448                            response,
449                        )) => match response.text().await {
450                            Ok(body) => {
451                                Some(Err(super::HttpError::BadStatus {
452                                    code,
453                                    body: match serde_json::from_str::<
454                                        serde_json::Value,
455                                    >(
456                                        &body
457                                    ) {
458                                        Ok(body) => body,
459                                        Err(_) => {
460                                            serde_json::Value::String(body)
461                                        }
462                                    },
463                                }))
464                            }
465                            Err(_) => Some(Err(super::HttpError::BadStatus {
466                                code,
467                                body: serde_json::Value::Null,
468                            })),
469                        },
470                        Err(e) => Some(Err(super::HttpError::StreamError(e))),
471                    }
472                })
473                .filter_map(|x| async { x }),
474        )
475    }
476
477    /// WebSocket variant of [`Self::send_streaming`]. Opens a WS to
478    /// the configured `address`, sends `body` as the first text
479    /// frame, then demultiplexes inbound frames into:
480    ///
481    /// - Chunk frames (yielded on the returned [`Stream`]).
482    /// - [`client_response::Response`](crate::client_objectiveai_mcp::client_response::Response)
483    ///   frames (routed to the [`super::Notifier`]'s pending-id map).
484    /// - [`server_request::Request`](crate::client_objectiveai_mcp::server_request::Request)
485    ///   frames (dispatched to `handler`; the result is written back
486    ///   as a `server_response::Response` echoing the request id).
487    ///
488    /// Both the returned `Stream` and the returned [`super::Notifier`]
489    /// share the underlying WebSocket: dropping both stops the demux
490    /// task and closes the connection cleanly. Dropping only one
491    /// keeps the WS alive — useful when a caller wants to send
492    /// notifies after the chunk stream has finished, or vice-versa.
493    #[cfg(feature = "mcp")]
494    pub async fn send_streaming_ws<Chunk, B, H, P>(
495        &self,
496        method: reqwest::Method,
497        path: P,
498        body: B,
499        handler: H,
500    ) -> Result<
501        (
502            impl Stream<Item = Result<Chunk, super::HttpError>>
503            + Send
504            + Unpin
505            + 'static
506            + use<Chunk, B, H, P>,
507            super::Notifier,
508        ),
509        super::HttpError,
510    >
511    where
512        Chunk: serde::de::DeserializeOwned + Send + 'static,
513        B: serde::Serialize + Send + 'static,
514        H: super::McpHandler,
515        P: AsRef<str>,
516    {
517        use crate::client_objectiveai_mcp::{
518            client_response::Response as ClientResponse,
519            server_request::Request as ServerRequest,
520        };
521        use futures::stream::SplitStream;
522        use tokio::net::TcpStream;
523        use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
524
525        // Translate the configured `address` (http(s)://...) into a
526        // ws(s):// URL. Path is appended directly.
527        let url = format!(
528            "{}/{}",
529            self.address.trim_end_matches('/'),
530            path.as_ref().trim_start_matches('/')
531        );
532        let ws_url = if let Some(rest) = url.strip_prefix("https://") {
533            format!("wss://{rest}")
534        } else if let Some(rest) = url.strip_prefix("http://") {
535            format!("ws://{rest}")
536        } else {
537            url.clone()
538        };
539        let _ = method; // axum's WS route is `any(...)`, method is ignored on the wire.
540
541        // Build the upgrade request manually so we can apply the
542        // same auth + custom headers `request()` does for HTTP.
543        let mut req = tungstenite::handshake::client::Request::builder()
544            .method("GET")
545            .uri(&ws_url)
546            .header(
547                "Host",
548                reqwest::Url::parse(&url)
549                    .ok()
550                    .and_then(|u| u.host_str().map(str::to_owned))
551                    .unwrap_or_default(),
552            )
553            .header("Upgrade", "websocket")
554            .header("Connection", "Upgrade")
555            .header(
556                "Sec-WebSocket-Key",
557                tungstenite::handshake::client::generate_key(),
558            )
559            .header("Sec-WebSocket-Version", "13")
560            .header("X-Transport", "ws");
561        if let Some(authorization) = &self.authorization {
562            let key = authorization
563                .strip_prefix("Bearer ")
564                .unwrap_or(authorization.as_str());
565            req = req.header("authorization", format!("Bearer {}", key));
566        }
567        if let Some(ua) = &self.user_agent {
568            req = req.header("user-agent", ua);
569        }
570        if let Some(x_title) = &self.x_title {
571            req = req.header("x-title", x_title);
572        }
573        if let Some(http_referer) = &self.http_referer {
574            req = req.header("referer", http_referer);
575            req = req.header("http-referer", http_referer);
576        }
577        if let Some(token) = &self.x_github_authorization {
578            req = req.header("X-GITHUB-AUTHORIZATION", token.as_str());
579        }
580        if let Some(token) = &self.x_openrouter_authorization {
581            req = req.header("X-OPENROUTER-AUTHORIZATION", token.as_str());
582        }
583        if let Some(headers) = &self.x_mcp_authorization {
584            if let Ok(json) = serde_json::to_string(headers.as_ref()) {
585                req = req.header("X-MCP-AUTHORIZATION", json);
586            }
587        }
588        if let Some(id) = &self.agent_instance_hierarchy {
589            req = req.header("X-OBJECTIVEAI-AGENT-INSTANCE-HIERARCHY", id.as_str());
590        }
591        if let Some(s) = &self.mcp_session_id {
592            req = req.header(crate::mcp::MCP_SESSION_ID_HEADER, s.as_str());
593        }
594        if let Some(ms) = self.mcp_call_timeout_ms {
595            req = req.header("X-MCP-CALL-TIMEOUT", ms.to_string());
596        }
597        let req = req.body(()).map_err(|e| {
598            super::HttpError::WsConnect(tungstenite::Error::Http(
599                tungstenite::http::Response::builder()
600                    .status(400)
601                    .body(Some(e.to_string().into_bytes()))
602                    .unwrap(),
603            ))
604        })?;
605
606        let (ws_stream, _resp) = tokio_tungstenite::connect_async(req).await?;
607        let (mut sink, rx_stream): (
608            _,
609            SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
610        ) = ws_stream.split();
611
612        // Send the body as the first text frame.
613        let body_frame = serde_json::to_string(&body)
614            .map_err(super::HttpError::NotifySerialize)?;
615        sink.send(tungstenite::Message::Text(body_frame.into()))
616            .await
617            .map_err(super::HttpError::NotifySend)?;
618
619        // Build the per-connection state shared with Notifier + demux.
620        let sink: super::notifier::SharedSink =
621            Arc::new(tokio::sync::Mutex::new(sink));
622        let pending: super::notifier::PendingNotifies =
623            Arc::new(dashmap::DashMap::new());
624
625        // mpsc the demux task pushes chunks (or terminal errors) into.
626        // Use futures::channel::mpsc so the rx side is `impl Stream`
627        // without pulling in tokio-stream.
628        let (chunk_tx, chunk_rx) = futures::channel::mpsc::unbounded::<
629            Result<Chunk, super::HttpError>,
630        >();
631
632        let demux_sink = sink.clone();
633        let demux_pending = pending.clone();
634        let handler = Arc::new(handler);
635        tokio::spawn(async move {
636            let mut rx_stream = rx_stream;
637            let mut chunk_tx = chunk_tx;
638            loop {
639                let msg = match rx_stream.next().await {
640                    Some(m) => m,
641                    None => break,
642                };
643                let text = match msg {
644                    Ok(tungstenite::Message::Text(t)) => {
645                        let s = t.to_string();
646                        s
647                    }
648                    Ok(tungstenite::Message::Binary(_)) => {
649                        continue;
650                    }
651                    Ok(
652                        tungstenite::Message::Ping(_)
653                        | tungstenite::Message::Pong(_),
654                    ) => continue,
655                    Ok(tungstenite::Message::Close(_)) => {
656                        break;
657                    }
658                    Ok(tungstenite::Message::Frame(_)) => continue,
659                    Err(_) => {
660                        break;
661                    }
662                };
663
664                // Classification: try client_response, then
665                // server_request, then chunk. Order matters — chunks
666                // tend to have many fields; the envelopes have a
667                // distinctive `id` + tagged `type`.
668                if let Ok(response) =
669                    serde_json::from_str::<ClientResponse>(&text)
670                {
671                    let id = response.id().to_string();
672                    if let Some((_, tx)) = demux_pending.remove(&id) {
673                        let _ = tx.send(response);
674                    }
675                    continue;
676                }
677                if let Ok(request) =
678                    serde_json::from_str::<ServerRequest>(&text)
679                {
680                    let id = request.id.clone();
681                    let handler = handler.clone();
682                    let demux_sink = demux_sink.clone();
683                    tokio::spawn(async move {
684                        let id = id;
685                        // Handler returns the full response (incl.
686                        // matching id); we just frame + write it.
687                        let response = handler.handle(request).await;
688                        let frame = match serde_json::to_string(&response) {
689                            Ok(s) => s,
690                            Err(_) => {
691                                return;
692                            }
693                        };
694                        let mut guard = demux_sink.lock().await;
695                        let send_result = guard
696                            .send(tungstenite::Message::Text(frame.into()))
697                            .await;
698                    });
699                    continue;
700                }
701
702                // Chunk path.
703                let mut de = serde_json::Deserializer::from_str(&text);
704                match serde_path_to_error::deserialize::<_, Chunk>(&mut de) {
705                    Ok(chunk) => {
706                        if chunk_tx.unbounded_send(Ok(chunk)).is_err() {
707                            break;
708                        }
709                    }
710                    Err(e) => {
711                        // Try to parse as a ResponseError before
712                        // surfacing the deserialization failure.
713                        let err = match serde_json::from_str::<
714                            error::ResponseError,
715                        >(&text)
716                        {
717                            Ok(api_err) => super::HttpError::ApiError(api_err),
718                            Err(_) => super::HttpError::DeserializationError(e),
719                        };
720                        let _ = chunk_tx.unbounded_send(Err(err));
721                        break;
722                    }
723                }
724            }
725            // Make sure any awaiting Notifier futures unblock when
726            // we exit — dropping the dashmap fires every oneshot
727            // Sender's drop, which causes the rx side to error.
728            drop(demux_pending);
729            drop(chunk_tx);
730        });
731
732        let notifier = super::Notifier::new(sink, pending);
733        Ok((chunk_rx, notifier))
734    }
735}