fix(codex): normalize Responses body for ChatGPT backend (strip max_output_tokens, require instructions)

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".

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,13 +184,61 @@ 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);
     apply_headers(response.headers_mut(), rate_limit_headers);
     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<Vec<u8>> {
+    let text = std::str::from_utf8(body).ok()?;
+    let mut completed: Option<serde_json::Value> = 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::<serde_json::Value>(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<String> {
+    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::<Vec<_>>()
+                .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<String> = 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<String> = 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(