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 automatic_prompt_caching: bool,
111    pub forward_headers: &'a [(String, String)],
112    pub raw_body: Option<&'a bytes::Bytes>,
113}
114
115#[expect(
116    missing_debug_implementations,
117    reason = "variants hold streaming bodies that intentionally do not implement Debug"
118)]
119pub enum OutboundOutcome {
120    Buffered(Box<CanonicalResponse>),
121    Streaming(BoxStream<'static, Result<CanonicalEvent, String>>),
122    RawBuffered {
123        body: bytes::Bytes,
124        content_type: Option<String>,
125        canonical: Box<CanonicalResponse>,
126    },
127    RawStreaming {
128        content_type: Option<String>,
129        stream: BoxStream<'static, Result<bytes::Bytes, String>>,
130    },
131}
132
133/// The exact bytes an adapter will put on the wire.
134///
135/// `raw_lane` records that the bytes started as the caller's own; they are
136/// still normalised in place, so they are not byte-identical to what arrived.
137#[derive(Debug, Clone)]
138pub struct PreparedBody {
139    pub bytes: bytes::Bytes,
140    pub raw_lane: bool,
141}
142
143/// Adapters are looked up by wire name as `Arc<dyn OutboundAdapter>`;
144/// `#[async_trait]` keeps the trait object-safe.
145#[async_trait]
146pub trait OutboundAdapter: Send + Sync {
147    fn build_body(&self, ctx: &OutboundCtx<'_>) -> Result<PreparedBody>;
148
149    async fn send(&self, ctx: OutboundCtx<'_>, body: &PreparedBody) -> Result<OutboundOutcome>;
150}
151
152#[derive(Debug, Clone, Copy)]
153pub struct OutboundAdapterRegistration {
154    pub tag: &'static str,
155    pub factory: fn() -> Arc<dyn OutboundAdapter>,
156}
157
158inventory::collect!(OutboundAdapterRegistration);
159
160const DEFECTIVE_BODY_STATUS: u16 = 502;
161
162pub(in crate::services::gateway) fn reject_defective_body(
163    provider: &str,
164    wire: &str,
165    defect: &systemprompt_models::wire::defect::BodyDefect,
166    body: &bytes::Bytes,
167) -> anyhow::Error {
168    let excerpt: String = String::from_utf8_lossy(body).chars().take(512).collect();
169    tracing::warn!(
170        provider = %provider,
171        wire = %wire,
172        defect = %defect,
173        body = %excerpt,
174        "upstream returned a success status with a body carrying no turn"
175    );
176    anyhow::Error::new(UpstreamError::Status {
177        provider: provider.to_owned(),
178        status: DEFECTIVE_BODY_STATUS,
179        message: format!("{defect}: {excerpt}"),
180        body: body.clone(),
181        retry_after: None,
182        request_id: None,
183    })
184}
185
186pub(in crate::services::gateway) fn reject_unparsable_body(
187    provider: &str,
188    wire: &str,
189    error: &systemprompt_models::wire::error::WireParseError,
190    body: &bytes::Bytes,
191) -> anyhow::Error {
192    let excerpt: String = String::from_utf8_lossy(body).chars().take(512).collect();
193    tracing::error!(
194        provider = %provider,
195        wire = %wire,
196        error = %error,
197        body = %excerpt,
198        "upstream returned a success status with a body that does not parse"
199    );
200    anyhow::Error::new(UpstreamError::Status {
201        provider: provider.to_owned(),
202        status: DEFECTIVE_BODY_STATUS,
203        message: format!("{error}: {excerpt}"),
204        body: body.clone(),
205        retry_after: None,
206        request_id: None,
207    })
208}
systemprompt_api/services/gateway/protocol/outbound/mod.rs

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