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
87pub fn extract_upstream_message(body: &str) -> String {
88 serde_json::from_str::<serde_json::Value>(body)
89 .ok()
90 .and_then(|v| v["error"]["message"].as_str().map(ToOwned::to_owned))
91 .unwrap_or_else(|| body.chars().take(500).collect())
92}
93
94#[derive(Debug)]
95pub struct OutboundCtx<'a> {
96 pub route: &'a GatewayRoute,
97 pub endpoint: &'a str,
98 pub api_key: &'a str,
99 pub request: &'a CanonicalRequest,
100 pub upstream_model: &'a str,
101 pub model_limits: Option<ModelLimits>,
102 /// Inbound headers cleared for verbatim relay, already stripped of every
103 /// header that identifies the client, user, or session.
104 pub forward_headers: &'a [(String, String)],
105 /// The caller's request body, set only when the caller's wire protocol
106 /// matches the upstream's and the bytes can be relayed untouched.
107 pub raw_body: Option<&'a bytes::Bytes>,
108}
109
110#[expect(
111 missing_debug_implementations,
112 reason = "variants hold streaming bodies that intentionally do not implement Debug"
113)]
114pub enum OutboundOutcome {
115 Buffered(Box<CanonicalResponse>),
116 Streaming(BoxStream<'static, Result<CanonicalEvent, String>>),
117 /// A non-streaming response relayed byte-for-byte, with a canonical parse
118 /// alongside it purely so audit, cost, and safety keep working.
119 RawBuffered {
120 body: bytes::Bytes,
121 content_type: Option<String>,
122 canonical: Box<CanonicalResponse>,
123 },
124 /// A streaming response relayed byte-for-byte. Usage accounting reads a
125 /// copy of the frames as they pass; the bytes the client receives are the
126 /// provider's own.
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// Why: #[async_trait] is required — the upstream registry stores adapters as
144// `Arc<dyn OutboundAdapter>`, so the trait must stay dyn-compatible.
145#[async_trait]
146pub trait OutboundAdapter: Send + Sync {
147 /// Kept separate from [`OutboundAdapter::send`], and sync and pure, so the
148 /// gateway can have governance inspect the same bytes the socket will
149 /// carry before it commits to sending them.
150 fn build_body(&self, ctx: &OutboundCtx<'_>) -> Result<PreparedBody>;
151
152 async fn send(&self, ctx: OutboundCtx<'_>, body: &PreparedBody) -> Result<OutboundOutcome>;
153}
154
155#[derive(Debug, Clone, Copy)]
156pub struct OutboundAdapterRegistration {
157 pub tag: &'static str,
158 pub factory: fn() -> Arc<dyn OutboundAdapter>,
159}
160
161inventory::collect!(OutboundAdapterRegistration);