lean_ctx/proxy/
chatgpt.rs

1use axum::{
2    body::Body,
3    extract::State,
4    http::{HeaderName, Request, StatusCode},
5    response::Response,
6};
7
8use super::{ProxyState, forward, openai_responses};
9
10/// Codex subscription model turns hit ChatGPT's Responses-compatible rail:
11/// `/backend-api/codex/responses`. Forward through the same compressor/metering
12/// path as OpenAI Responses, but target `https://chatgpt.com`.
13pub async fn codex_responses_handler(
14    State(state): State<ProxyState>,
15    mut req: Request<Body>,
16) -> Result<Response, StatusCode> {
17    // Drop the "responses-lite" marker before forwarding. Codex requests the
18    // reduced lite transport on its HTTP path (`supports_websockets = false`),
19    // but chatgpt.com rejects newer subscription models there
20    // ("This model is not supported when using X-OpenAI-Internal-Codex-Responses-Lite",
21    // seen with gpt-5.5). Stripping it makes chatgpt.com serve the full Responses
22    // rail every model supports; Codex parses the full stream identically
23    // (verified single- + multi-turn `previous_response_id` continuation). #623
24    req.headers_mut().remove(CODEX_RESPONSES_LITE_HEADER);
25    let upstream = state.chatgpt_upstream();
26    forward::forward_request(
27        State(state),
28        req,
29        &upstream,
30        "/backend-api/codex/responses",
31        openai_responses::compress_request_body,
32        "ChatGPT",
33        &[],
34    )
35    .await
36}
37
38/// Codex's HTTP fallback marks the reduced "responses-lite" transport with this
39/// header. chatgpt.com gates newer subscription models behind the full rail, so
40/// the Codex ChatGPT handler strips it (see [`codex_responses_handler`]).
41const CODEX_RESPONSES_LITE_HEADER: &str = "x-openai-internal-codex-responses-lite";
42
43/// ChatGPT's Codex rail rejects WS-only continuation fields such as
44/// `previous_response_id`; ask Codex to retry through the HTTP/SSE path.
45pub async fn codex_responses_ws_handler(
46    State(_state): State<ProxyState>,
47    _headers: axum::http::HeaderMap,
48    _ws: axum::extract::ws::WebSocketUpgrade,
49) -> Response {
50    chatgpt_responses_ws_fallback_response()
51}
52
53fn chatgpt_responses_ws_fallback_response() -> Response {
54    Response::builder()
55        .status(StatusCode::UPGRADE_REQUIRED)
56        .header("content-type", "application/json")
57        .body(Body::from(
58            r#"{"error":{"type":"unsupported_transport","message":"ChatGPT codex responses use HTTP/SSE; retry without WebSocket."}}"#,
59        ))
60        .expect("static response is valid")
61}
62
63/// ChatGPT backend calls outside the model rail are not model JSON and must not be
64/// compressed or cost-metered. They are credential-preserving passthroughs.
65///
66/// A WebSocket upgrade (Codex Desktop remote-control pairing, #597) is tunnelled
67/// verbatim to chatgpt.com by [`super::chatgpt_ws`]; plain HTTP/SSE falls through
68/// to the reqwest forward below.
69pub async fn backend_api_handler(
70    State(state): State<ProxyState>,
71    req: Request<Body>,
72) -> Result<Response, StatusCode> {
73    if super::chatgpt_ws::is_websocket_upgrade(req.headers()) {
74        return Ok(super::chatgpt_ws::passthrough(state, req).await);
75    }
76
77    let (parts, body) = req.into_parts();
78    let body_bytes = axum::body::to_bytes(body, forward::max_body_bytes())
79        .await
80        .map_err(|_| StatusCode::PAYLOAD_TOO_LARGE)?;
81    let upstream = state.chatgpt_upstream();
82    let path = parts
83        .uri
84        .path_and_query()
85        .map_or("/backend-api", axum::http::uri::PathAndQuery::as_str);
86    let url = format!("{upstream}{path}");
87
88    let mut upstream_req = state.client.request(parts.method.clone(), &url);
89    for (key, value) in &parts.headers {
90        if is_backend_passthrough_request_header(key) {
91            upstream_req = upstream_req.header(key.clone(), value.clone());
92        }
93    }
94
95    let response = upstream_req
96        .body(body_bytes.to_vec())
97        .send()
98        .await
99        .map_err(|e| {
100            tracing::error!("lean-ctx proxy: ChatGPT backend upstream error: {e}");
101            StatusCode::BAD_GATEWAY
102        })?;
103
104    let status = StatusCode::from_u16(response.status().as_u16()).unwrap_or(StatusCode::OK);
105    let headers = response.headers().clone();
106    let is_stream = headers
107        .get("content-type")
108        .and_then(|v| v.to_str().ok())
109        .is_some_and(|ct| ct.contains("text/event-stream"));
110
111    let mut out = Response::builder().status(status);
112    for (key, value) in &headers {
113        if is_backend_passthrough_response_header(key) {
114            out = out.header(key, value);
115        }
116    }
117
118    if is_stream {
119        return out
120            .body(Body::from_stream(super::sse_keepalive::keepalive_stream(
121                Box::pin(response.bytes_stream()),
122            )))
123            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR);
124    }
125
126    let bytes = response
127        .bytes()
128        .await
129        .map_err(|_| StatusCode::BAD_GATEWAY)?;
130
131    out.body(Body::from(bytes))
132        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
133}
134
135fn is_backend_passthrough_request_header(name: &HeaderName) -> bool {
136    let lower = name.as_str().to_ascii_lowercase();
137    !matches!(
138        lower.as_str(),
139        "host"
140            | "connection"
141            | "content-length"
142            | "transfer-encoding"
143            | "upgrade"
144            | "keep-alive"
145            | "proxy-authenticate"
146            | "proxy-authorization"
147            | "te"
148            | "trailer"
149            | "accept-encoding"
150    )
151}
152
153fn is_backend_passthrough_response_header(name: &HeaderName) -> bool {
154    let lower = name.as_str().to_ascii_lowercase();
155    !matches!(
156        lower.as_str(),
157        "connection"
158            | "content-length"
159            | "transfer-encoding"
160            | "upgrade"
161            | "keep-alive"
162            | "proxy-authenticate"
163            | "proxy-authorization"
164            | "te"
165            | "trailer"
166    )
167}
168
169#[cfg(test)]
170mod tests {
171    use std::sync::Arc;
172    use std::time::Duration;
173
174    use tokio::io::{AsyncReadExt, AsyncWriteExt};
175
176    use super::*;
177    use crate::core::config::Upstreams;
178
179    fn proxy_state(chatgpt_upstream: String) -> ProxyState {
180        let (_tx, rx) = tokio::sync::watch::channel(Arc::new(Upstreams {
181            anthropic: "https://api.anthropic.com".into(),
182            openai: "https://api.openai.com".into(),
183            chatgpt: chatgpt_upstream,
184            gemini: "https://generativelanguage.googleapis.com".into(),
185            providers: Vec::new(),
186        }));
187        ProxyState {
188            client: reqwest::Client::new(),
189            port: 0,
190            stats: Arc::new(crate::proxy::ProxyStats::default()),
191            break_even: Arc::new(crate::proxy::break_even::BreakEvenCalculator::new(1500)),
192            introspect: Arc::new(crate::proxy::introspect::IntrospectState::default()),
193            ocla_cache: None,
194            upstreams: rx,
195            chatgpt_cookies: crate::proxy::chatgpt_cookies::shared_chatgpt_cloudflare_cookie_store(
196            ),
197            mcp_servers: Arc::new(Vec::new()),
198            web_app_tracker: Arc::new(std::sync::Mutex::new(
199                crate::proxy::web_app::conversation_tracker::ConversationTracker::default(),
200            )),
201        }
202    }
203
204    #[test]
205    fn codex_responses_ws_requests_trigger_http_fallback() {
206        let response = chatgpt_responses_ws_fallback_response();
207        assert_eq!(response.status(), StatusCode::UPGRADE_REQUIRED);
208        assert_eq!(
209            response
210                .headers()
211                .get(axum::http::header::CONTENT_TYPE)
212                .unwrap(),
213            "application/json"
214        );
215    }
216
217    async fn spawn_streaming_upstream() -> (String, tokio::sync::oneshot::Receiver<String>) {
218        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
219        let addr = listener.local_addr().unwrap();
220        let (tx, rx) = tokio::sync::oneshot::channel();
221        tokio::spawn(async move {
222            let (mut socket, _) = listener.accept().await.unwrap();
223            let mut buf = Vec::new();
224            loop {
225                let mut chunk = [0_u8; 1024];
226                let n = socket.read(&mut chunk).await.unwrap();
227                if n == 0 {
228                    break;
229                }
230                buf.extend_from_slice(&chunk[..n]);
231                if buf.windows(4).any(|w| w == b"\r\n\r\n") {
232                    break;
233                }
234            }
235            let _ = tx.send(String::from_utf8_lossy(&buf).into_owned());
236            socket
237                .write_all(
238                    b"HTTP/1.1 200 OK\r\n\
239                      content-type: text/event-stream\r\n\
240                      mcp-session-id: server-session\r\n\
241                      cache-control: no-cache\r\n\
242                      x-custom-backend-state: passthrough\r\n\
243                      \r\n\
244                      event: message\n\
245                      data: {\"jsonrpc\":\"2.0\"}\n\n",
246                )
247                .await
248                .unwrap();
249            tokio::time::sleep(Duration::from_secs(2)).await;
250        });
251        (format!("http://{addr}"), rx)
252    }
253
254    #[tokio::test]
255    async fn backend_api_streams_mcp_sse_and_preserves_session_headers() {
256        let (upstream, seen_request) = spawn_streaming_upstream().await;
257        let state = proxy_state(upstream);
258        let req = Request::builder()
259            .method("POST")
260            .uri("/backend-api/ps/mcp?transport=streamable")
261            .header("Authorization", "Bearer codex-token")
262            .header("Mcp-Session-Id", "client-session")
263            .header("Last-Event-ID", "event-7")
264            .header("X-OpenAI-Product-Sku", "codex")
265            .header("X-OpenAI-Internal-Codex-Residency", "us")
266            .header("Originator", "codex_cli_rs")
267            .header("Accept", "application/json, text/event-stream")
268            .body(Body::empty())
269            .unwrap();
270
271        let response = tokio::time::timeout(
272            Duration::from_millis(500),
273            backend_api_handler(State(state), req),
274        )
275        .await
276        .expect("SSE passthrough must return after upstream headers")
277        .expect("backend request should succeed");
278
279        assert_eq!(response.status(), StatusCode::OK);
280        assert_eq!(
281            response.headers().get("mcp-session-id").unwrap(),
282            "server-session"
283        );
284        assert_eq!(
285            response.headers().get("x-custom-backend-state").unwrap(),
286            "passthrough"
287        );
288
289        let request = seen_request.await.unwrap().to_ascii_lowercase();
290        assert!(request.contains("post /backend-api/ps/mcp?transport=streamable http/1.1"));
291        assert!(request.contains("authorization: bearer codex-token"));
292        assert!(request.contains("mcp-session-id: client-session"));
293        assert!(request.contains("last-event-id: event-7"));
294        assert!(request.contains("x-openai-product-sku: codex"));
295        assert!(request.contains("x-openai-internal-codex-residency: us"));
296        assert!(request.contains("originator: codex_cli_rs"));
297    }
298}
lean_ctx/proxy/chatgpt.rs

lean_ctx/proxy/
chatgpt.rs
