Fix OpenAI Codex contract, SSE parsing, and default xhigh reasoning

src/auth/openai_oauth.rs

@@ -14,7 +14,7 @@ pub const OPENAI_OAUTH_CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann";
 pub const OPENAI_OAUTH_AUTHORIZE_URL: &str = "https://auth.openai.com/oauth/authorize";
 pub const OPENAI_OAUTH_TOKEN_URL: &str = "https://auth.openai.com/oauth/token";
 pub const OPENAI_OAUTH_DEVICE_CODE_URL: &str = "https://auth.openai.com/oauth/device/code";
-pub const OPENAI_OAUTH_REDIRECT_URI: &str = "http://127.0.0.1:1455/auth/callback";
+pub const OPENAI_OAUTH_REDIRECT_URI: &str = "http://localhost:1455/auth/callback";
 
 #[derive(Debug, Clone)]
 pub struct PkceState {

src/providers/openai_codex.rs

@@ -1,12 +1,15 @@
 use crate::auth::AuthService;
+use crate::auth::openai_oauth::extract_account_id_from_jwt;
 use crate::providers::traits::Provider;
 use crate::providers::ProviderRuntimeOptions;
 use async_trait::async_trait;
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use std::path::PathBuf;
 
 const CODEX_RESPONSES_URL: &str = "https://chatgpt.com/backend-api/codex/responses";
+const DEFAULT_CODEX_INSTRUCTIONS: &str = "You are ZeroClaw, a concise and helpful coding assistant.";
 
 pub struct OpenAiCodexProvider {
     auth: AuthService,
@@ -18,15 +21,38 @@ pub struct OpenAiCodexProvider {
 struct ResponsesRequest {
     model: String,
     input: Vec<ResponsesInput>,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    instructions: String,
-    instructions: Option<String>,
+    store: bool,
     stream: bool,
+    text: ResponsesTextOptions,
+    reasoning: ResponsesReasoningOptions,
+    include: Vec<String>,
+    tool_choice: String,
+    parallel_tool_calls: bool,
 }
 
 #[derive(Debug, Serialize)]
 struct ResponsesInput {
     role: String,
-    content: String,
+    content: Vec<ResponsesInputContent>,
+}
+
+#[derive(Debug, Serialize)]
+struct ResponsesInputContent {
+    #[serde(rename = "type")]
+    kind: String,
+    text: String,
+}
+
+#[derive(Debug, Serialize)]
+struct ResponsesTextOptions {
+    verbosity: String,
+}
+
+#[derive(Debug, Serialize)]
+struct ResponsesReasoningOptions {
+    effort: String,
+    summary: String,
 }
 
 #[derive(Debug, Deserialize)]
@@ -88,6 +114,51 @@ fn first_nonempty(text: Option<&str>) -> Option<String> {
     })
 }
 
+fn resolve_instructions(system_prompt: Option<&str>) -> String {
+    first_nonempty(system_prompt).unwrap_or_else(|| DEFAULT_CODEX_INSTRUCTIONS.to_string())
+}
+
+fn normalize_model_id(model: &str) -> &str {
+    model.rsplit('/').next().unwrap_or(model)
+}
+
+fn clamp_reasoning_effort(model: &str, effort: &str) -> String {
+    let id = normalize_model_id(model);
+    if (id.starts_with("gpt-5.2") || id.starts_with("gpt-5.3")) && effort == "minimal" {
+        return "low".to_string();
+    }
+    if id == "gpt-5.1" && effort == "xhigh" {
+        return "high".to_string();
+    }
+    if id == "gpt-5.1-codex-mini" {
+        return if effort == "high" || effort == "xhigh" {
+            "high".to_string()
+        } else {
+            "medium".to_string()
+        };
+    }
+    effort.to_string()
+}
+
+fn resolve_reasoning_effort(model: &str) -> String {
+    let raw = std::env::var("ZEROCLAW_CODEX_REASONING_EFFORT")
+        .ok()
+        .and_then(|value| first_nonempty(Some(&value)))
+        .unwrap_or_else(|| "xhigh".to_string())
+        .to_ascii_lowercase();
+    clamp_reasoning_effort(model, &raw)
+}
+
+fn nonempty_preserve(text: Option<&str>) -> Option<String> {
+    text.and_then(|value| {
+        if value.is_empty() {
+            None
+        } else {
+            Some(value.to_string())
+        }
+    })
+}
+
 fn extract_responses_text(response: &ResponsesResponse) -> Option<String> {
     if let Some(text) = first_nonempty(response.output_text.as_deref()) {
         return Some(text);
@@ -114,6 +185,155 @@ fn extract_responses_text(response: &ResponsesResponse) -> Option<String> {
     None
 }
 
+fn extract_stream_event_text(event: &Value, saw_delta: bool) -> Option<String> {
+    let event_type = event.get("type").and_then(Value::as_str);
+    match event_type {
+        Some("response.output_text.delta") => {
+            nonempty_preserve(event.get("delta").and_then(Value::as_str))
+        }
+        Some("response.output_text.done") if !saw_delta => {
+            nonempty_preserve(event.get("text").and_then(Value::as_str))
+        }
+        Some("response.completed") | Some("response.done") => event
+            .get("response")
+            .and_then(|value| serde_json::from_value::<ResponsesResponse>(value.clone()).ok())
+            .and_then(|response| extract_responses_text(&response)),
+        _ => None,
+    }
+}
+
+fn parse_sse_text(body: &str) -> anyhow::Result<Option<String>> {
+    let mut saw_delta = false;
+    let mut delta_accumulator = String::new();
+    let mut fallback_text = None;
+    let mut buffer = body.to_string();
+
+    let mut process_event = |event: Value| -> anyhow::Result<()> {
+        if let Some(message) = extract_stream_error_message(&event) {
+            return Err(anyhow::anyhow!("OpenAI Codex stream error: {message}"));
+        }
+        if let Some(text) = extract_stream_event_text(&event, saw_delta) {
+            let event_type = event.get("type").and_then(Value::as_str);
+            if event_type == Some("response.output_text.delta") {
+                saw_delta = true;
+                delta_accumulator.push_str(&text);
+            } else if fallback_text.is_none() {
+                fallback_text = Some(text);
+            }
+        }
+        Ok(())
+    };
+
+    let mut process_chunk = |chunk: &str| -> anyhow::Result<()> {
+        let data_lines: Vec<String> = chunk
+            .lines()
+            .filter_map(|line| line.strip_prefix("data:"))
+            .map(|line| line.trim().to_string())
+            .collect();
+        if data_lines.is_empty() {
+            return Ok(());
+        }
+
+        let joined = data_lines.join("\n");
+        let trimmed = joined.trim();
+        if trimmed.is_empty() || trimmed == "[DONE]" {
+            return Ok(());
+        }
+
+        if let Ok(event) = serde_json::from_str::<Value>(trimmed) {
+            return process_event(event);
+        }
+
+        for line in data_lines {
+            let line = line.trim();
+            if line.is_empty() || line == "[DONE]" {
+                continue;
+            }
+            if let Ok(event) = serde_json::from_str::<Value>(line) {
+                process_event(event)?;
+            }
+        }
+
+        Ok(())
+    };
+
+    loop {
+        let Some(idx) = buffer.find("\n\n") else {
+            break;
+        };
+
+        let chunk = buffer[..idx].to_string();
+        buffer = buffer[idx + 2..].to_string();
+        process_chunk(&chunk)?;
+    }
+
+    if !buffer.trim().is_empty() {
+        process_chunk(&buffer)?;
+    }
+
+    if saw_delta {
+        return Ok(nonempty_preserve(Some(&delta_accumulator)));
+    }
+
+    Ok(fallback_text)
+}
+
+fn extract_stream_error_message(event: &Value) -> Option<String> {
+    let event_type = event.get("type").and_then(Value::as_str);
+
+    if event_type == Some("error") {
+        return first_nonempty(
+            event
+                .get("message")
+                .and_then(Value::as_str)
+                .or_else(|| event.get("code").and_then(Value::as_str))
+                .or_else(|| {
+                    event
+                        .get("error")
+                        .and_then(|error| error.get("message"))
+                        .and_then(Value::as_str)
+                }),
+        );
+    }
+
+    if event_type == Some("response.failed") {
+        return first_nonempty(
+            event
+                .get("response")
+                .and_then(|response| response.get("error"))
+                .and_then(|error| error.get("message"))
+                .and_then(Value::as_str),
+        );
+    }
+
+    None
+}
+
+async fn decode_responses_body(response: reqwest::Response) -> anyhow::Result<String> {
+    let body = response.text().await?;
+
+    if let Some(text) = parse_sse_text(&body)? {
+        return Ok(text);
+    }
+
+    let body_trimmed = body.trim_start();
+    let looks_like_sse = body_trimmed.starts_with("event:") || body_trimmed.starts_with("data:");
+    if looks_like_sse {
+        return Err(anyhow::anyhow!(
+            "No response from OpenAI Codex stream payload: {}",
+            super::sanitize_api_error(&body)
+        ));
+    }
+
+    let parsed: ResponsesResponse = serde_json::from_str(&body).map_err(|err| {
+        anyhow::anyhow!(
+            "OpenAI Codex JSON parse failed: {err}. Payload: {}",
+            super::sanitize_api_error(&body)
+        )
+    })?;
+    extract_responses_text(&parsed).ok_or_else(|| anyhow::anyhow!("No response from OpenAI Codex"))
+}
+
 #[async_trait]
 impl Provider for OpenAiCodexProvider {
     async fn chat_with_system(
@@ -123,6 +343,9 @@ impl Provider for OpenAiCodexProvider {
         model: &str,
         _temperature: f64,
     ) -> anyhow::Result<String> {
+        let profile = self
+            .auth
+            .get_profile("openai-codex", self.auth_profile_override.as_deref())?;
         let access_token = self
             .auth
             .get_valid_openai_access_token(self.auth_profile_override.as_deref())
@@ -132,21 +355,47 @@ impl Provider for OpenAiCodexProvider {
                     "OpenAI Codex auth profile not found. Run `zeroclaw auth login --provider openai-codex`."
                 )
             })?;
+        let account_id = profile
+            .and_then(|profile| profile.account_id)
+            .or_else(|| extract_account_id_from_jwt(&access_token))
+            .ok_or_else(|| {
+                anyhow::anyhow!(
+                    "OpenAI Codex account id not found in auth profile/token. Run `zeroclaw auth login --provider openai-codex` again."
+                )
+            })?;
 
         let request = ResponsesRequest {
             model: model.to_string(),
             input: vec![ResponsesInput {
                 role: "user".to_string(),
-                content: message.to_string(),
+                content: vec![ResponsesInputContent {
+                    kind: "input_text".to_string(),
+                    text: message.to_string(),
                 }],
-            instructions: system_prompt.map(str::to_string),
+            }],
-            stream: false,
+            instructions: resolve_instructions(system_prompt),
+            store: false,
+            stream: true,
+            text: ResponsesTextOptions {
+                verbosity: "medium".to_string(),
+            },
+            reasoning: ResponsesReasoningOptions {
+                effort: resolve_reasoning_effort(model),
+                summary: "auto".to_string(),
+            },
+            include: vec!["reasoning.encrypted_content".to_string()],
+            tool_choice: "auto".to_string(),
+            parallel_tool_calls: true,
         };
 
         let response = self
             .client
             .post(CODEX_RESPONSES_URL)
             .header("Authorization", format!("Bearer {access_token}"))
+            .header("chatgpt-account-id", account_id)
+            .header("OpenAI-Beta", "responses=experimental")
+            .header("originator", "pi")
+            .header("accept", "text/event-stream")
             .header("Content-Type", "application/json")
             .json(&request)
             .send()
@@ -156,10 +405,7 @@ impl Provider for OpenAiCodexProvider {
             return Err(super::api_error("OpenAI Codex", response).await);
         }
 
-        let parsed: ResponsesResponse = response.json().await?;
+        decode_responses_body(response).await
-
-        extract_responses_text(&parsed)
-            .ok_or_else(|| anyhow::anyhow!("No response from OpenAI Codex"))
     }
 }
 
@@ -195,4 +441,77 @@ mod tests {
         let path = default_zeroclaw_dir();
         assert!(!path.as_os_str().is_empty());
     }
+
+    #[test]
+    fn resolve_instructions_uses_default_when_missing() {
+        assert_eq!(
+            resolve_instructions(None),
+            DEFAULT_CODEX_INSTRUCTIONS.to_string()
+        );
+    }
+
+    #[test]
+    fn resolve_instructions_uses_default_when_blank() {
+        assert_eq!(
+            resolve_instructions(Some("   ")),
+            DEFAULT_CODEX_INSTRUCTIONS.to_string()
+        );
+    }
+
+    #[test]
+    fn resolve_instructions_uses_system_prompt_when_present() {
+        assert_eq!(
+            resolve_instructions(Some("Be strict")),
+            "Be strict".to_string()
+        );
+    }
+
+    #[test]
+    fn clamp_reasoning_effort_adjusts_known_models() {
+        assert_eq!(
+            clamp_reasoning_effort("gpt-5.3-codex", "minimal"),
+            "low".to_string()
+        );
+        assert_eq!(
+            clamp_reasoning_effort("gpt-5.1", "xhigh"),
+            "high".to_string()
+        );
+        assert_eq!(
+            clamp_reasoning_effort("gpt-5.1-codex-mini", "low"),
+            "medium".to_string()
+        );
+        assert_eq!(
+            clamp_reasoning_effort("gpt-5.1-codex-mini", "xhigh"),
+            "high".to_string()
+        );
+        assert_eq!(
+            clamp_reasoning_effort("gpt-5.3-codex", "xhigh"),
+            "xhigh".to_string()
+        );
+    }
+
+    #[test]
+    fn parse_sse_text_reads_output_text_delta() {
+        let payload = r#"data: {"type":"response.created","response":{"id":"resp_123"}}
+
+data: {"type":"response.output_text.delta","delta":"Hello"}
+data: {"type":"response.output_text.delta","delta":" world"}
+data: {"type":"response.completed","response":{"output_text":"Hello world"}}
+data: [DONE]
+"#;
+
+        assert_eq!(
+            parse_sse_text(payload).unwrap().as_deref(),
+            Some("Hello world")
+        );
+    }
+
+    #[test]
+    fn parse_sse_text_falls_back_to_completed_response() {
+        let payload = r#"data: {"type":"response.completed","response":{"output_text":"Done"}}
+data: [DONE]
+"#;
+
+        assert_eq!(parse_sse_text(payload).unwrap().as_deref(), Some("Done"));
+    }
 }