diff --git a/changelog.d/20260622_codex_responses_normalization.md b/changelog.d/20260622_codex_responses_normalization.md new file mode 100644 index 0000000..ad52ba7 --- /dev/null +++ b/changelog.d/20260622_codex_responses_normalization.md @@ -0,0 +1,12 @@ +--- +bump: patch +--- + +### Fixed + +- Codex subscription proxy now shapes `/v1/responses` (and projected + `/v1/chat/completions`) request bodies for the ChatGPT Codex backend: the + unsupported `max_output_tokens` parameter is stripped and a default + `instructions` field is injected when the client omits one. Standard OpenAI + Responses clients (e.g. OpenClaw) previously received HTTP 400 + "Unsupported parameter: max_output_tokens" / "Instructions are required". diff --git a/src/subscription_proxy.rs b/src/subscription_proxy.rs index 1b29b43..da6c8ed 100644 --- a/src/subscription_proxy.rs +++ b/src/subscription_proxy.rs @@ -90,10 +90,10 @@ pub async fn forward_subscription_openai( .and_then(serde_json::Value::as_bool) .unwrap_or(false); - // Codex always streams from the ChatGPT backend; reflect that into the body - // so the upstream emits SSE we pass straight through. + // The ChatGPT Codex backend is stricter than the generic Responses API, so + // reshape the body before forwarding (see `normalize_codex_responses_body`). if provider == SubscriptionProvider::Codex { - body["stream"] = serde_json::Value::Bool(true); + normalize_codex_responses_body(&mut body); } let serialized = match serde_json::to_vec(&body) { @@ -148,13 +148,23 @@ pub async fn forward_subscription_openai( // back off intelligently when a subscription upstream throttles us. let rate_limit_headers = rate_limit_headers(upstream_resp.headers()); + let codex = provider == SubscriptionProvider::Codex; if stream_requested || is_event_stream(&content_type) { + // The Codex backend streams SSE but labels it `application/json`; re-label + // so SSE-aware clients treat the body as the stream it is. + let stream_content_type = if codex { + HeaderValue::from_static("text/event-stream") + } else { + content_type + }; let stream = upstream_resp .bytes_stream() .map(|chunk| chunk.map_err(std::io::Error::other)); let mut response = Response::new(Body::from_stream(stream)); *response.status_mut() = status; - response.headers_mut().insert("content-type", content_type); + response + .headers_mut() + .insert("content-type", stream_content_type); apply_headers(response.headers_mut(), rate_limit_headers); return response; } @@ -174,6 +184,24 @@ pub async fn forward_subscription_openai( .metrics .record_bytes(bytes_sent, upstream_body.len() as u64); + // The Codex backend always streams Server-Sent Events even when the client + // asked for a non-streaming (`stream:false`) response, and labels that SSE + // body `application/json`. A non-streaming client (e.g. OpenClaw's gateway) + // then parses the raw event stream as a single JSON object and fails with an + // incomplete result. Collapse the SSE into the final `response.completed` + // payload and return it as a normal JSON Responses object. + if codex && status.is_success() { + if let Some(json) = codex_sse_to_response_json(&upstream_body) { + let mut response = Response::new(Body::from(json)); + *response.status_mut() = status; + response + .headers_mut() + .insert("content-type", HeaderValue::from_static("application/json")); + apply_headers(response.headers_mut(), rate_limit_headers); + return response; + } + } + let mut response = Response::new(Body::from(upstream_body)); *response.status_mut() = status; response.headers_mut().insert("content-type", content_type); @@ -181,6 +209,36 @@ pub async fn forward_subscription_openai( response } +/// Collapse a Codex Responses SSE body into the final Responses JSON object. +/// +/// The ChatGPT Codex backend only streams (`text/event-stream`-style `event:` / +/// `data:` lines). For non-streaming clients we extract the `response` object +/// carried by the terminal `response.completed` event and return it verbatim, so +/// the client receives the single JSON object it expects. Returns `None` if no +/// completed event is present (caller falls back to the raw body). +fn codex_sse_to_response_json(body: &[u8]) -> Option> { + let text = std::str::from_utf8(body).ok()?; + let mut completed: Option = None; + for line in text.lines() { + let Some(payload) = line.strip_prefix("data:") else { + continue; + }; + let payload = payload.trim(); + if payload.is_empty() || payload == "[DONE]" { + continue; + } + let Ok(event) = serde_json::from_str::(payload) else { + continue; + }; + if event.get("type").and_then(serde_json::Value::as_str) == Some("response.completed") { + if let Some(response) = event.get("response") { + completed = Some(response.clone()); + } + } + } + completed.and_then(|value| serde_json::to_vec(&value).ok()) +} + /// Collect rate-limit-related headers (`retry-after`, `x-ratelimit-*`) from an /// upstream response so they can be relayed to the client. fn rate_limit_headers(headers: &HeaderMap) -> Vec<(axum::http::HeaderName, HeaderValue)> { @@ -281,10 +339,164 @@ fn is_event_stream(content_type: &HeaderValue) -> bool { .is_ok_and(|value| value.to_ascii_lowercase().contains("text/event-stream")) } +/// Extract the concatenated text from a Responses `input` message item. +fn input_item_text(item: &serde_json::Value) -> Option { + match item.get("content") { + Some(serde_json::Value::String(s)) => Some(s.clone()), + Some(serde_json::Value::Array(parts)) => { + let text: String = parts + .iter() + .filter_map(|p| p.get("text").and_then(serde_json::Value::as_str)) + .collect::>() + .join(""); + (!text.is_empty()).then_some(text) + } + _ => None, + } +} + +/// Shape a Responses-API request body for the `ChatGPT` Codex backend. +/// +/// The Codex backend is stricter than the generic `OpenAI` Responses API: +/// - it always streams, +/// - **rejects** `max_output_tokens` (HTTP 400 "Unsupported parameter"), +/// - **rejects** `system`/`developer` messages inside `input` +/// (HTTP 400 "System messages are not allowed") — they must be carried in the +/// top-level `instructions` field, +/// - **requires** a non-empty `instructions` field (HTTP 400 "Instructions are +/// required"). +/// +/// Standard Responses clients (e.g. OpenClaw's gateway) send `max_output_tokens` +/// and put the system prompt as a `system` message in `input`, so without this +/// shaping the backend rejects every request. System turns are hoisted into +/// `instructions`; a default is used only if nothing remains. +fn normalize_codex_responses_body(body: &mut serde_json::Value) { + let Some(obj) = body.as_object_mut() else { + return; + }; + // Codex always streams from the ChatGPT backend. + obj.insert("stream".to_string(), serde_json::Value::Bool(true)); + // `max_output_tokens` is not accepted by the Codex backend. + obj.remove("max_output_tokens"); + + // Hoist system/developer turns out of `input` (Codex forbids them there). + let mut hoisted: Vec = Vec::new(); + if let Some(serde_json::Value::Array(items)) = obj.get_mut("input") { + items.retain(|item| { + match item.get("role").and_then(serde_json::Value::as_str) { + Some("system" | "developer") => { + if let Some(text) = input_item_text(item) { + hoisted.push(text); + } + false + } + _ => true, + } + }); + } + + // Merge existing instructions + hoisted system turns; fall back to a default. + let mut parts: Vec = Vec::new(); + if let Some(existing) = obj.get("instructions").and_then(serde_json::Value::as_str) { + if !existing.trim().is_empty() { + parts.push(existing.to_string()); + } + } + parts.extend(hoisted); + let instructions = if parts.is_empty() { + "You are a helpful assistant.".to_string() + } else { + parts.join("\n\n") + }; + obj.insert("instructions".to_string(), serde_json::Value::String(instructions)); +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn codex_normalizes_responses_body_for_chatgpt_backend() { + // OpenClaw-style Responses body: omits `instructions`, sends + // `max_output_tokens`. Both trip the Codex backend without shaping. + let mut body = serde_json::json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "hi"}]}], + "store": false, + "max_output_tokens": 8192, + "reasoning": {"effort": "none"} + }); + normalize_codex_responses_body(&mut body); + assert_eq!(body["stream"], serde_json::Value::Bool(true)); + assert!( + body.get("max_output_tokens").is_none(), + "max_output_tokens must be stripped for Codex" + ); + assert_eq!(body["instructions"], "You are a helpful assistant."); + // Untouched fields are preserved. + assert_eq!(body["store"], serde_json::Value::Bool(false)); + assert_eq!(body["reasoning"]["effort"], "none"); + } + + #[test] + fn codex_hoists_system_messages_into_instructions() { + // OpenClaw gateway shape: system prompt as a `system` message in `input`. + let mut body = serde_json::json!({ + "model": "gpt-5.5", + "input": [ + {"type":"message","role":"system","content":[{"type":"input_text","text":"be terse"}]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]} + ], + "stream": true, + "max_output_tokens": 8192 + }); + normalize_codex_responses_body(&mut body); + // System turn moved out of input (Codex forbids it there)... + let input = body["input"].as_array().unwrap(); + assert_eq!(input.len(), 1); + assert_eq!(input[0]["role"], "user"); + // ...and merged into instructions. + assert_eq!(body["instructions"], "be terse"); + assert!(body.get("max_output_tokens").is_none()); + } + + #[test] + fn codex_preserves_caller_instructions() { + let mut body = serde_json::json!({ + "model": "gpt-5-codex", + "input": [], + "instructions": "be terse", + "max_output_tokens": 100 + }); + normalize_codex_responses_body(&mut body); + assert_eq!(body["instructions"], "be terse"); + assert!(body.get("max_output_tokens").is_none()); + assert_eq!(body["stream"], serde_json::Value::Bool(true)); + } + + #[test] + fn codex_sse_collapses_to_completed_response() { + let sse = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"status\":\"in_progress\"}}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hi\"}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"output\":[{\"type\":\"message\"}]}}\n\n" + ); + let out = codex_sse_to_response_json(sse.as_bytes()).expect("completed payload"); + let value: serde_json::Value = serde_json::from_slice(&out).unwrap(); + assert_eq!(value["id"], "resp_1"); + assert_eq!(value["status"], "completed"); + assert_eq!(value["output"][0]["type"], "message"); + } + + #[test] + fn codex_sse_without_completed_returns_none() { + let sse = "event: response.created\ndata: {\"type\":\"response.created\"}\n\n"; + assert!(codex_sse_to_response_json(sse.as_bytes()).is_none()); + } + #[test] fn codex_url_collapses_v1_responses() { let url = join_subscription_url(