Fix Codex Responses tool schema and pending recovery

crates/aether-ai-formats/src/formats/openai/responses/codex.rs

@@ -375,6 +375,51 @@ fn ensure_codex_chat_reasoning_defaults(
         .or_insert_with(|| json!(CODEX_DEFAULT_REASONING_SUMMARY));
 }
 
+fn codex_tool_type_rejects_top_level_name(tool_type: &str) -> bool {
+    let normalized = tool_type.trim().to_ascii_lowercase();
+    !normalized.is_empty()
+        && normalized != "function"
+        && normalized != "custom"
+        && normalized != "namespace"
+}
+
+fn strip_codex_hosted_tool_names_for_backend(body_object: &mut serde_json::Map<String, Value>) {
+    let Some(tools) = body_object.get_mut("tools").and_then(Value::as_array_mut) else {
+        return;
+    };
+
+    for tool in tools {
+        let Some(tool_object) = tool.as_object_mut() else {
+            continue;
+        };
+        if tool_object
+            .get("type")
+            .and_then(Value::as_str)
+            .is_some_and(codex_tool_type_rejects_top_level_name)
+        {
+            tool_object.remove("name");
+        }
+    }
+}
+
+fn strip_codex_hosted_tool_choice_name_for_backend(
+    body_object: &mut serde_json::Map<String, Value>,
+) {
+    let Some(tool_choice_object) = body_object
+        .get_mut("tool_choice")
+        .and_then(Value::as_object_mut)
+    else {
+        return;
+    };
+    if tool_choice_object
+        .get("type")
+        .and_then(Value::as_str)
+        .is_some_and(codex_tool_type_rejects_top_level_name)
+    {
+        tool_choice_object.remove("name");
+    }
+}
+
 pub fn apply_codex_openai_responses_special_body_edits(
     provider_request_body: &mut Value,
     provider_type: &str,
@@ -420,6 +465,8 @@ pub fn apply_codex_openai_responses_special_body_edits(
     {
         body_object.insert("instructions".to_string(), json!(""));
     }
+    strip_codex_hosted_tool_names_for_backend(body_object);
+    strip_codex_hosted_tool_choice_name_for_backend(body_object);
     if is_openai_image_request(provider_api_format)
         || codex_openai_responses_tool_choice_references_image_generation(body_object)
     {
@@ -620,6 +667,87 @@ mod tests {
         assert_eq!(provider_request_body["parallel_tool_calls"], json!(false));
     }
 
+    #[test]
+    fn codex_responses_body_edits_preserve_function_tools_for_codex_backend() {
+        let mut provider_request_body = json!({
+            "input": [],
+            "model": "gpt-5.4",
+            "tools": [{
+                "type": "function",
+                "name": "lookup_account",
+                "description": "Lookup an account by id.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "account_id": {
+                            "type": "string"
+                        }
+                    },
+                    "required": ["account_id"],
+                    "additionalProperties": false
+                },
+                "strict": true
+            }],
+            "tool_choice": {
+                "type": "function",
+                "name": "lookup_account"
+            }
+        });
+
+        apply_codex_openai_responses_special_body_edits(
+            &mut provider_request_body,
+            "codex",
+            "openai:responses",
+            None,
+            None,
+        );
+
+        assert_eq!(
+            provider_request_body["tools"][0]["name"],
+            json!("lookup_account")
+        );
+        assert_eq!(
+            provider_request_body["tools"][0]["parameters"]["properties"]["account_id"]["type"],
+            json!("string")
+        );
+        assert_eq!(
+            provider_request_body["tool_choice"]["name"],
+            json!("lookup_account")
+        );
+        assert!(provider_request_body["tools"][0].get("function").is_none());
+    }
+
+    #[test]
+    fn codex_responses_body_edits_strip_name_from_hosted_web_search_tool() {
+        let mut provider_request_body = json!({
+            "input": [],
+            "model": "gpt-5.4",
+            "tools": [{
+                "type": "web_search",
+                "name": "web_search"
+            }],
+            "tool_choice": {
+                "type": "web_search",
+                "name": "web_search"
+            }
+        });
+
+        apply_codex_openai_responses_special_body_edits(
+            &mut provider_request_body,
+            "codex",
+            "openai:responses",
+            None,
+            None,
+        );
+
+        assert!(provider_request_body["tools"][0].get("name").is_none());
+        assert!(provider_request_body["tool_choice"].get("name").is_none());
+        assert_eq!(
+            provider_request_body["tool_choice"]["type"],
+            json!("web_search")
+        );
+    }
+
     #[test]
     fn compact_body_edits_strip_include_store_and_stream() {
         let mut provider_request_body = json!({

crates/aether-scheduler-core/src/request_candidate.rs

@@ -488,6 +488,7 @@ pub fn build_local_request_candidate_status_record(
         demoted_by: metadata.demoted_by.clone(),
         routing_trace: metadata.routing_trace.clone(),
     });
+    let extra_data = mark_request_candidate_stream_completed_if_success(status, extra_data);
     let created_at_unix_ms = started_at_unix_ms.or(finished_at_unix_ms);
 
     Some(UpsertRequestCandidateRecord {
@@ -568,7 +569,7 @@ pub fn build_report_request_candidate_status_record(
         error_message,
         latency_ms,
         concurrent_requests: None,
-        extra_data: slot.extra_data,
+        extra_data: mark_request_candidate_stream_completed_if_success(status, slot.extra_data),
         required_capabilities: None,
         created_at_unix_ms: Some(created_at_unix_ms),
         started_at_unix_ms,
@@ -829,6 +830,23 @@ fn build_report_candidate_extra_data(input: ReportCandidateExtraDataInput) -> Op
     (!extra_data.is_empty()).then_some(Value::Object(extra_data))
 }
 
+fn mark_request_candidate_stream_completed_if_success(
+    status: RequestCandidateStatus,
+    extra_data: Option<Value>,
+) -> Option<Value> {
+    if status != RequestCandidateStatus::Success {
+        return extra_data;
+    }
+
+    let mut object = match extra_data {
+        Some(Value::Object(object)) => object,
+        Some(other) => return Some(other),
+        None => Map::new(),
+    };
+    object.insert("stream_completed".to_string(), Value::Bool(true));
+    Some(Value::Object(object))
+}
+
 fn merge_request_candidate_extra_data(
     existing: Option<Value>,
     overlay: Option<Value>,
@@ -1295,6 +1313,53 @@ mod tests {
         assert_eq!(record.finished_at_unix_ms, Some(123));
         assert_eq!(record.created_at_unix_ms, Some(123));
         assert_eq!(record.status, RequestCandidateStatus::Success);
+        assert_eq!(
+            record
+                .extra_data
+                .as_ref()
+                .and_then(|value| value.get("stream_completed")),
+            Some(&json!(true))
+        );
+    }
+
+    #[test]
+    fn local_success_status_marks_stream_completed_for_pending_cleanup_recovery() {
+        let mut plan = sample_plan();
+        plan.candidate_id = Some("cand-1".to_string());
+        let report_context = json!({
+            "request_id": "req-1",
+            "candidate_id": "cand-1",
+            "candidate_index": 0,
+            "retry_index": 0,
+            "user_id": "user-1",
+            "api_key_id": "api-key-1",
+            "client_api_format": "openai:responses",
+            "provider_api_format": "openai:responses",
+        });
+
+        let record =
+            build_local_request_candidate_status_record(LocalRequestCandidateStatusRecordInput {
+                plan: &plan,
+                report_context: Some(&report_context),
+                status_update: SchedulerRequestCandidateStatusUpdate {
+                    status: RequestCandidateStatus::Success,
+                    status_code: Some(200),
+                    error_type: None,
+                    error_message: None,
+                    latency_ms: Some(25),
+                    started_at_unix_ms: Some(1_000),
+                    finished_at_unix_ms: Some(1_025),
+                },
+            })
+            .expect("success status record should build");
+
+        assert_eq!(
+            record
+                .extra_data
+                .as_ref()
+                .and_then(|value| value.get("stream_completed")),
+            Some(&json!(true))
+        );
     }
 
     #[test]