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;
17pub mod retry;
18
19use std::sync::Arc;
20
21use anyhow::Result;
22use async_trait::async_trait;
23use futures_util::stream::BoxStream;
24use systemprompt_models::services::GatewayRoute;
25use systemprompt_models::services::ai::ModelLimits;
26use thiserror::Error;
27
28use super::canonical::CanonicalRequest;
29use super::canonical_response::{CanonicalEvent, CanonicalResponse};
30
31/// Upstream provider failure, carried inside the `anyhow::Error` an adapter
32/// returns so the route layer can recover the real HTTP status by downcast
33/// instead of flattening every failure to 502.
34#[derive(Debug, Error)]
35pub enum UpstreamError {
36    #[error("{provider} returned {status}: {message}")]
37    Status {
38        provider: String,
39        status: u16,
40        message: String,
41        body: bytes::Bytes,
42        retry_after: Option<String>,
43        request_id: Option<String>,
44    },
45    #[error("{provider} request failed: {source}")]
46    Transport {
47        provider: String,
48        #[source]
49        source: reqwest::Error,
50    },
51}
52
53impl UpstreamError {
54    pub async fn from_response(provider: &str, response: reqwest::Response) -> Self {
55        let status = response.status().as_u16();
56        let header = |name: &str| {
57            response
58                .headers()
59                .get(name)
60                .and_then(|v| v.to_str().ok())
61                .map(ToOwned::to_owned)
62        };
63        let retry_after = header("retry-after");
64        let request_id = header("request-id").or_else(|| header("x-request-id"));
65        let body = response
66            .bytes()
67            .await
68            .unwrap_or_else(|e| bytes::Bytes::from(format!("<unreadable body: {e}>")));
69        Self::Status {
70            provider: provider.to_owned(),
71            status,
72            message: extract_upstream_message(&String::from_utf8_lossy(&body)),
73            body,
74            retry_after,
75            request_id,
76        }
77    }
78}
79
80pub(in crate::services::gateway) fn http_client() -> &'static reqwest::Client {
81    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
82    CLIENT.get_or_init(reqwest::Client::new)
83}
84
85pub(in crate::services::gateway) async fn send_checked(
86    provider: &str,
87    req: reqwest::RequestBuilder,
88) -> Result<reqwest::Response> {
89    let policy = retry::current_policy();
90    let (response, _retries) = retry::send_with_retry(provider, req, &policy).await?;
91    Ok(response)
92}
93
94pub fn extract_upstream_message(body: &str) -> String {
95    serde_json::from_str::<serde_json::Value>(body)
96        .ok()
97        .and_then(|v| v["error"]["message"].as_str().map(ToOwned::to_owned))
98        .unwrap_or_else(|| body.chars().take(500).collect())
99}
100
101#[derive(Debug)]
102pub struct OutboundCtx<'a> {
103    pub route: &'a GatewayRoute,
104    pub endpoint: &'a str,
105    pub api_key: &'a str,
106    pub api_key_is_bearer: bool,
107    pub request: &'a CanonicalRequest,
108    pub upstream_model: &'a str,
109    pub model_limits: Option<ModelLimits>,
110    pub forward_headers: &'a [(String, String)],
111    pub raw_body: Option<&'a bytes::Bytes>,
112}
113
114#[expect(
115    missing_debug_implementations,
116    reason = "variants hold streaming bodies that intentionally do not implement Debug"
117)]
118pub enum OutboundOutcome {
119    Buffered(Box<CanonicalResponse>),
120    Streaming(BoxStream<'static, Result<CanonicalEvent, String>>),
121    RawBuffered {
122        body: bytes::Bytes,
123        content_type: Option<String>,
124        canonical: Box<CanonicalResponse>,
125    },
126    RawStreaming {
127        content_type: Option<String>,
128        stream: BoxStream<'static, Result<bytes::Bytes, String>>,
129    },
130}
131
132/// The exact bytes an adapter will put on the wire.
133///
134/// `raw_lane` records that the bytes started as the caller's own; they are
135/// still normalised in place, so they are not byte-identical to what arrived.
136#[derive(Debug, Clone)]
137pub struct PreparedBody {
138    pub bytes: bytes::Bytes,
139    pub raw_lane: bool,
140}
141
142/// Adapters are looked up by wire name as `Arc<dyn OutboundAdapter>`;
143/// `#[async_trait]` keeps the trait object-safe.
144#[async_trait]
145pub trait OutboundAdapter: Send + Sync {
146    fn build_body(&self, ctx: &OutboundCtx<'_>) -> Result<PreparedBody>;
147
148    async fn send(&self, ctx: OutboundCtx<'_>, body: &PreparedBody) -> Result<OutboundOutcome>;
149}
150
151#[derive(Debug, Clone, Copy)]
152pub struct OutboundAdapterRegistration {
153    pub tag: &'static str,
154    pub factory: fn() -> Arc<dyn OutboundAdapter>,
155}
156
157inventory::collect!(OutboundAdapterRegistration);
158
159const DEFECTIVE_BODY_STATUS: u16 = 502;
160
161pub(in crate::services::gateway) fn reject_defective_body(
162    provider: &str,
163    wire: &str,
164    defect: &systemprompt_models::wire::defect::BodyDefect,
165    body: &bytes::Bytes,
166) -> anyhow::Error {
167    let excerpt: String = String::from_utf8_lossy(body).chars().take(512).collect();
168    tracing::warn!(
169        provider = %provider,
170        wire = %wire,
171        defect = %defect,
172        body = %excerpt,
173        "upstream returned a success status with a body carrying no turn"
174    );
175    anyhow::Error::new(UpstreamError::Status {
176        provider: provider.to_owned(),
177        status: DEFECTIVE_BODY_STATUS,
178        message: format!("{defect}: {excerpt}"),
179        body: body.clone(),
180        retry_after: None,
181        request_id: None,
182    })
183}
184
185pub(in crate::services::gateway) fn reject_unparsable_body(
186    provider: &str,
187    wire: &str,
188    error: &systemprompt_models::wire::error::WireParseError,
189    body: &bytes::Bytes,
190) -> anyhow::Error {
191    let excerpt: String = String::from_utf8_lossy(body).chars().take(512).collect();
192    tracing::error!(
193        provider = %provider,
194        wire = %wire,
195        error = %error,
196        body = %excerpt,
197        "upstream returned a success status with a body that does not parse"
198    );
199    anyhow::Error::new(UpstreamError::Status {
200        provider: provider.to_owned(),
201        status: DEFECTIVE_BODY_STATUS,
202        message: format!("{error}: {excerpt}"),
203        body: body.clone(),
204        retry_after: None,
205        request_id: None,
206    })
207}
systemprompt_api/services/gateway/protocol/outbound/mod.rs

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