Fix agent jobs finalization race and reduce status polling churn

codex-rs/app-server/src/codex_message_processor.rs

@@ -6694,6 +6694,13 @@ impl CodexMessageProcessor {
             None => None,
         };
 
+        if let Some(chatgpt_user_id) = self
+            .auth_manager
+            .auth_cached()
+            .and_then(|auth| auth.get_chatgpt_user_id())
+        {
+            tracing::info!(target: "feedback_tags", chatgpt_user_id);
+        }
         let snapshot = self.feedback.snapshot(conversation_id);
         let thread_id = snapshot.thread_id.clone();
         let sqlite_feedback_logs = if include_logs {

codex-rs/core/src/auth.rs

@@ -266,6 +266,12 @@ impl CodexAuth {
         self.get_current_token_data().and_then(|t| t.id_token.email)
     }
 
+    /// Returns `None` if `is_chatgpt_auth()` is false.
+    pub fn get_chatgpt_user_id(&self) -> Option<String> {
+        self.get_current_token_data()
+            .and_then(|t| t.id_token.chatgpt_user_id)
+    }
+
     /// Account-facing plan classification derived from the current token.
     /// Returns a high-level `AccountPlanType` (e.g., Free/Plus/Pro/Team/…)
     /// mapped from the ID token's internal plan value. Prefer this when you
@@ -1466,6 +1472,7 @@ mod tests {
             .unwrap();
         assert_eq!(None, auth.api_key());
         assert_eq!(AuthMode::Chatgpt, auth.auth_mode());
+        assert_eq!(auth.get_chatgpt_user_id().as_deref(), Some("user-12345"));
 
         let auth_dot_json = auth
             .get_current_auth_json()

codex-rs/core/src/client_common.rs

@@ -160,6 +160,7 @@ pub(crate) mod tools {
     use codex_protocol::config_types::WebSearchUserLocationType;
     use serde::Deserialize;
     use serde::Serialize;
+    use serde_json::Value;
 
     /// When serialized as JSON, this produces a valid "Tool" in the OpenAI
     /// Responses API.
@@ -268,6 +269,8 @@ pub(crate) mod tools {
         /// `properties` must be present in `required`.
         pub(crate) strict: bool,
         pub(crate) parameters: JsonSchema,
+        #[serde(skip)]
+        pub(crate) output_schema: Option<Value>,
     }
 }
 

codex-rs/core/src/codex_tests.rs

@@ -58,6 +58,7 @@ use codex_app_server_protocol::AppInfo;
 use codex_otel::TelemetryAuthMode;
 use codex_protocol::models::BaseInstructions;
 use codex_protocol::models::ContentItem;
+use codex_protocol::models::McpToolOutput;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::openai_models::ModelsResponse;
@@ -1607,7 +1608,7 @@ fn prefers_structured_content_when_present() {
         meta: None,
     };
 
-    let got = FunctionCallOutputPayload::from(&ctr);
+    let got = McpToolOutput::from(&ctr).into_function_call_output_payload();
     let expected = FunctionCallOutputPayload {
         body: FunctionCallOutputBody::Text(
             serde_json::to_string(&json!({
@@ -1689,7 +1690,7 @@ fn falls_back_to_content_when_structured_is_null() {
         meta: None,
     };
 
-    let got = FunctionCallOutputPayload::from(&ctr);
+    let got = McpToolOutput::from(&ctr).into_function_call_output_payload();
     let expected = FunctionCallOutputPayload {
         body: FunctionCallOutputBody::Text(
             serde_json::to_string(&vec![text_block("hello"), text_block("world")]).unwrap(),
@@ -1709,7 +1710,7 @@ fn success_flag_reflects_is_error_true() {
         meta: None,
     };
 
-    let got = FunctionCallOutputPayload::from(&ctr);
+    let got = McpToolOutput::from(&ctr).into_function_call_output_payload();
     let expected = FunctionCallOutputPayload {
         body: FunctionCallOutputBody::Text(
             serde_json::to_string(&json!({ "message": "bad" })).unwrap(),
@@ -1729,7 +1730,7 @@ fn success_flag_true_with_no_error_and_content_used() {
         meta: None,
     };
 
-    let got = FunctionCallOutputPayload::from(&ctr);
+    let got = McpToolOutput::from(&ctr).into_function_call_output_payload();
     let expected = FunctionCallOutputPayload {
         body: FunctionCallOutputBody::Text(
             serde_json::to_string(&vec![text_block("alpha")]).unwrap(),

codex-rs/core/src/mcp_tool_call.rs

@@ -29,9 +29,7 @@ use crate::protocol::McpToolCallBeginEvent;
 use crate::protocol::McpToolCallEndEvent;
 use crate::state_db;
 use codex_protocol::mcp::CallToolResult;
-use codex_protocol::models::FunctionCallOutputBody;
-use codex_protocol::models::FunctionCallOutputPayload;
-use codex_protocol::models::ResponseInputItem;
+use codex_protocol::models::McpToolOutput;
 use codex_protocol::openai_models::InputModality;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::ReviewDecision;
@@ -58,7 +56,7 @@ pub(crate) async fn handle_mcp_tool_call(
     server: String,
     tool_name: String,
     arguments: String,
-) -> ResponseInputItem {
+) -> McpToolOutput {
     // Parse the `arguments` as JSON. An empty string is OK, but invalid JSON
     // is not.
     let arguments_value = if arguments.trim().is_empty() {
@@ -68,13 +66,7 @@ pub(crate) async fn handle_mcp_tool_call(
             Ok(value) => Some(value),
             Err(e) => {
                 error!("failed to parse tool call arguments: {e}");
-                return ResponseInputItem::FunctionCallOutput {
-                    call_id: call_id.clone(),
-                    output: FunctionCallOutputPayload {
-                        body: FunctionCallOutputBody::Text(format!("err: {e}")),
-                        success: Some(false),
-                    },
-                };
+                return McpToolOutput::from_error_text(format!("err: {e}"));
             }
         }
     };
@@ -118,7 +110,7 @@ pub(crate) async fn handle_mcp_tool_call(
         turn_context
             .session_telemetry
             .counter("codex.mcp.call", 1, &[("status", status)]);
-        return ResponseInputItem::McpToolCallOutput { call_id, result };
+        return McpToolOutput::from_result(result);
     }
 
     if let Some(decision) = maybe_request_mcp_tool_approval(
@@ -212,7 +204,7 @@ pub(crate) async fn handle_mcp_tool_call(
             .session_telemetry
             .counter("codex.mcp.call", 1, &[("status", status)]);
 
-        return ResponseInputItem::McpToolCallOutput { call_id, result };
+        return McpToolOutput::from_result(result);
     }
 
     let tool_call_begin_event = EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
@@ -258,7 +250,7 @@ pub(crate) async fn handle_mcp_tool_call(
         .session_telemetry
         .counter("codex.mcp.call", 1, &[("status", status)]);
 
-    ResponseInputItem::McpToolCallOutput { call_id, result }
+    McpToolOutput::from_result(result)
 }
 
 async fn maybe_mark_thread_memory_mode_polluted(sess: &Session, turn_context: &TurnContext) {

codex-rs/core/src/shell.rs

@@ -90,22 +90,62 @@ impl Eq for Shell {}
 
 #[cfg(unix)]
 fn get_user_shell_path() -> Option<PathBuf> {
-    use libc::getpwuid;
-    use libc::getuid;
+    let uid = unsafe { libc::getuid() };
     use std::ffi::CStr;
+    use std::mem::MaybeUninit;
+    use std::ptr;
+
+    let mut passwd = MaybeUninit::<libc::passwd>::uninit();
+
+    // We cannot use getpwuid here: it returns pointers into libc-managed
+    // storage, which is not safe to read concurrently on all targets (the musl
+    // static build used by the CLI can segfault when parallel callers race on
+    // that buffer). getpwuid_r keeps the passwd data in caller-owned memory.
+    let suggested_buffer_len = unsafe { libc::sysconf(libc::_SC_GETPW_R_SIZE_MAX) };
+    let buffer_len = usize::try_from(suggested_buffer_len)
+        .ok()
+        .filter(|len| *len > 0)
+        .unwrap_or(1024);
+    let mut buffer = vec![0; buffer_len];
+
+    loop {
+        let mut result = ptr::null_mut();
+        let status = unsafe {
+            libc::getpwuid_r(
+                uid,
+                passwd.as_mut_ptr(),
+                buffer.as_mut_ptr().cast(),
+                buffer.len(),
+                &mut result,
+            )
+        };
 
-    unsafe {
-        let uid = getuid();
-        let pw = getpwuid(uid);
+        if status == 0 {
+            if result.is_null() {
+                return None;
+            }
 
-        if !pw.is_null() {
-            let shell_path = CStr::from_ptr((*pw).pw_shell)
+            let passwd = unsafe { passwd.assume_init_ref() };
+            if passwd.pw_shell.is_null() {
+                return None;
+            }
+
+            let shell_path = unsafe { CStr::from_ptr(passwd.pw_shell) }
                 .to_string_lossy()
                 .into_owned();
-            Some(PathBuf::from(shell_path))
-        } else {
-            None
+            return Some(PathBuf::from(shell_path));
+        }
+
+        if status != libc::ERANGE {
+            return None;
+        }
+
+        // Retry with a larger buffer until libc can materialize the passwd entry.
+        let new_len = buffer.len().checked_mul(2)?;
+        if new_len > 1024 * 1024 {
+            return None;
         }
+        buffer.resize(new_len, 0);
     }
 }
 
@@ -500,7 +540,7 @@ mod tests {
     }
 
     #[test]
-    fn finds_poweshell() {
+    fn finds_powershell() {
         if !cfg!(windows) {
             return;
         }

codex-rs/core/src/stream_events_utils.rs

@@ -359,14 +359,8 @@ pub(crate) fn response_input_to_response_item(input: &ResponseInputItem) -> Opti
                 output: output.clone(),
             })
         }
-        ResponseInputItem::McpToolCallOutput { call_id, result } => {
-            let output = match result {
-                Ok(call_tool_result) => FunctionCallOutputPayload::from(call_tool_result),
-                Err(err) => FunctionCallOutputPayload {
-                    body: FunctionCallOutputBody::Text(err.clone()),
-                    success: Some(false),
-                },
-            };
+        ResponseInputItem::McpToolCallOutput { call_id, output } => {
+            let output = output.as_function_call_output_payload();
             Some(ResponseItem::FunctionCallOutput {
                 call_id: call_id.clone(),
                 output,