systemprompt_api/services/gateway/protocol/outbound/
mod.rs

1//! Outbound protocol adapters: canonical model to upstream provider.
2//!
3//! The [`OutboundAdapter`] trait sends a [`CanonicalRequest`] to an upstream
4//! provider and yields an [`OutboundOutcome`] — a buffered response or a stream
5//! of canonical events. Adapters register themselves via
6//! [`OutboundAdapterRegistration`] (collected by `inventory`) so the upstream
7//! registry can resolve one by provider tag. Implementations cover Anthropic,
8//! `OpenAI` Chat Completions, and `OpenAI` Responses.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13pub mod anthropic;
14pub mod gemini;
15pub mod openai_chat;
16pub mod openai_responses;
17
18use std::sync::Arc;
19
20use anyhow::Result;
21use async_trait::async_trait;
22use futures_util::stream::BoxStream;
23use systemprompt_models::profile::GatewayRoute;
24use systemprompt_models::services::ai::ModelLimits;
25use thiserror::Error;
26
27use super::canonical::CanonicalRequest;
28use super::canonical_response::{CanonicalEvent, CanonicalResponse};
29
30/// Upstream provider failure, carried inside the `anyhow::Error` an adapter
31/// returns so the route layer can recover the real HTTP status by downcast
32/// instead of flattening every failure to 502.
33#[derive(Debug, Error)]
34pub enum UpstreamError {
35    #[error("{provider} returned {status}: {message}")]
36    Status {
37        provider: String,
38        status: u16,
39        message: String,
40        /// The provider's error response verbatim.
41        ///
42        /// Claude Code recovers from several upstream rejections by matching on
43        /// the error's own wording and then disabling the rejected capability
44        /// for the rest of the conversation. Re-wrapping the error in the
45        /// gateway's envelope defeats that even when the status code survives,
46        /// so the original bytes are carried here and relayed unchanged.
47        body: bytes::Bytes,
48        /// `retry-after` as sent by the provider, when present.
49        retry_after: Option<String>,
50        /// The provider's own request id, for correlating with their support.
51        request_id: Option<String>,
52    },
53    #[error("{provider} request failed: {source}")]
54    Transport {
55        provider: String,
56        #[source]
57        source: reqwest::Error,
58    },
59}
60
61impl UpstreamError {
62    /// Builds a [`UpstreamError::Status`] from a non-success upstream response,
63    /// preserving the body and the headers a client needs to retry correctly.
64    pub async fn from_response(provider: &str, response: reqwest::Response) -> Self {
65        let status = response.status().as_u16();
66        let header = |name: &str| {
67            response
68                .headers()
69                .get(name)
70                .and_then(|v| v.to_str().ok())
71                .map(ToOwned::to_owned)
72        };
73        let retry_after = header("retry-after");
74        let request_id = header("request-id").or_else(|| header("x-request-id"));
75        let body = response.bytes().await.unwrap_or_default();
76        Self::Status {
77            provider: provider.to_owned(),
78            status,
79            message: extract_upstream_message(&String::from_utf8_lossy(&body)),
80            body,
81            retry_after,
82            request_id,
83        }
84    }
85}
86
87// Why: one process-wide client — a client per request would open a fresh
88// connection pool and TLS handshake on every gateway call.
89pub(in crate::services::gateway) fn http_client() -> &'static reqwest::Client {
90    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
91    CLIENT.get_or_init(reqwest::Client::new)
92}
93
94pub(in crate::services::gateway) async fn send_checked(
95    provider: &str,
96    req: reqwest::RequestBuilder,
97) -> Result<reqwest::Response> {
98    let response = req.send().await.map_err(|e| {
99        anyhow::Error::new(UpstreamError::Transport {
100            provider: provider.to_owned(),
101            source: e,
102        })
103    })?;
104    if !response.status().is_success() {
105        return Err(anyhow::Error::new(
106            UpstreamError::from_response(provider, response).await,
107        ));
108    }
109    Ok(response)
110}
111
112pub fn extract_upstream_message(body: &str) -> String {
113    serde_json::from_str::<serde_json::Value>(body)
114        .ok()
115        .and_then(|v| v["error"]["message"].as_str().map(ToOwned::to_owned))
116        .unwrap_or_else(|| body.chars().take(500).collect())
117}
118
119#[derive(Debug)]
120pub struct OutboundCtx<'a> {
121    pub route: &'a GatewayRoute,
122    pub endpoint: &'a str,
123    pub api_key: &'a str,
124    pub request: &'a CanonicalRequest,
125    pub upstream_model: &'a str,
126    pub model_limits: Option<ModelLimits>,
127    /// Inbound headers cleared for verbatim relay, already stripped of every
128    /// header that identifies the client, user, or session.
129    pub forward_headers: &'a [(String, String)],
130    /// The caller's request body, set only when the caller's wire protocol
131    /// matches the upstream's and the bytes can be relayed untouched.
132    pub raw_body: Option<&'a bytes::Bytes>,
133}
134
135#[expect(
136    missing_debug_implementations,
137    reason = "variants hold streaming bodies that intentionally do not implement Debug"
138)]
139pub enum OutboundOutcome {
140    Buffered(Box<CanonicalResponse>),
141    Streaming(BoxStream<'static, Result<CanonicalEvent, String>>),
142    /// A non-streaming response relayed byte-for-byte, with a canonical parse
143    /// alongside it purely so audit, cost, and safety keep working.
144    RawBuffered {
145        body: bytes::Bytes,
146        content_type: Option<String>,
147        canonical: Box<CanonicalResponse>,
148    },
149    /// A streaming response relayed byte-for-byte. Usage accounting reads a
150    /// copy of the frames as they pass; the bytes the client receives are the
151    /// provider's own.
152    RawStreaming {
153        content_type: Option<String>,
154        stream: BoxStream<'static, Result<bytes::Bytes, String>>,
155    },
156}
157
158/// The exact bytes an adapter will put on the wire.
159///
160/// `raw_lane` records that the bytes started as the caller's own; they are
161/// still normalised in place, so they are not byte-identical to what arrived.
162#[derive(Debug, Clone)]
163pub struct PreparedBody {
164    pub bytes: bytes::Bytes,
165    pub raw_lane: bool,
166}
167
168// Why: #[async_trait] is required — the upstream registry stores adapters as
169// `Arc<dyn OutboundAdapter>`, so the trait must stay dyn-compatible.
170#[async_trait]
171pub trait OutboundAdapter: Send + Sync {
172    /// Kept separate from [`OutboundAdapter::send`], and sync and pure, so the
173    /// gateway can have governance inspect the same bytes the socket will
174    /// carry before it commits to sending them.
175    fn build_body(&self, ctx: &OutboundCtx<'_>) -> Result<PreparedBody>;
176
177    async fn send(&self, ctx: OutboundCtx<'_>, body: &PreparedBody) -> Result<OutboundOutcome>;
178}
179
180#[derive(Debug, Clone, Copy)]
181pub struct OutboundAdapterRegistration {
182    pub tag: &'static str,
183    pub factory: fn() -> Arc<dyn OutboundAdapter>,
184}
185
186inventory::collect!(OutboundAdapterRegistration);
systemprompt_api/services/gateway/protocol/outbound/mod.rs

systemprompt_api/services/gateway/protocol/outbound/
mod.rs
